用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/microwind/ai-skills --skill kubernetes命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | Kubernetes验证器 |
| description | 当验证Kubernetes配置时，分析YAML清单语法，检查资源定义规范，验证部署配置。验证集群资源，设计安全策略，和最佳实践。 |
| license | MIT |
Kubernetes配置验证是确保部署成功的关键步骤。无效的Kubernetes清单会静默部署然后神秘失败。在部署前进行全面验证可以避免生产环境问题。
核心原则: 好的配置验证应该覆盖语法、语义和最佳实践，确保部署的稳定性和安全性。坏的配置会导致部署失败、资源浪费和安全风险。
始终:
触发短语:
问题:
Pod没有设置CPU和内存限制
错误示例:
apiVersion: v1
kind: Pod
spec:
containers:
- name: app
image: nginx
# 没有resources字段
后果:
- Pod可能耗尽所有可用资源
- 影响其他Pod正常运行
- 集群变得不稳定
解决方案:
1. 设置resources.requests确保调度
2. 设置resources.limits防止资源耗尽
3. 根据应用特点合理配置
4. 监控资源使用情况并调整
问题:
Deployment没有配置存活性和就绪性探针
错误示例:
apiVersion: apps/v1
kind: Deployment
spec:
template:
spec:
containers:
- name: app
image: nginx
# 没有livenessProbe和readinessProbe
后果:
- 故障Pod仍被标记为健康
- 流量被发送到故障Pod
- 用户体验差
解决方案:
1. 配置livenessProbe检测容器健康
2. 配置readinessProbe检测服务就绪
3. 设置合适的检查间隔和阈值
4. 配置startupProbe处理启动慢的应用
问题:
镜像不存在或仓库地址错误
错误示例:
spec:
containers:
- name: app
image: nonexistent/app:latest # 镜像不存在
后果:
- Pod无法启动
- ImagePullBackOff错误
- 服务不可用
解决方案:
1. 验证镜像在仓库中存在
2. 使用正确的仓库地址和命名空间
3. 避免使用latest标签
4. 配置正确的镜像拉取策略
问题:
容器以特权模式运行或使用root用户
错误示例:
spec:
containers:
- name: app
securityContext:
privileged: true # 特权模式
runAsUser: 0 # root用户
后果:
- 容器获得主机权限
- 安全风险增加
- 可能影响主机稳定性
解决方案:
1. 避免使用特权模式
2. 使用非root用户运行
3. 配置只读根文件系统
4. 限制Linux capabilities
import yaml
import json
import re
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from pathlib import Path
@dataclass
class ValidationIssue:
"""验证问题"""
severity: str # critical, high, medium, low
type: str
resource: str
field: str
message: str
suggestion: str
line_number: Optional[int] = None
@dataclass
class ValidationResult:
"""验证结果"""
is_valid: bool
issues: List[ValidationIssue]
warnings: List[ValidationIssue]
summary: Dict[str, Any]
class KubernetesConfigValidator:
def __init__(self):
self.issues: List[ValidationIssue] = []
self.warnings: List[ValidationIssue] = []
# Kubernetes资源必填字段定义
self.required_fields = {
: {
: ,
: ,
: ,
: ,
:
},
: {
: ,
: ,
: ,
: ,
: ,
:
},
: {
: ,
: ,
: ,
: ,
: ,
:
}
}
.best_practice_rules = {
: ,
: ,
: ,
: ,
:
}
() -> ValidationResult:
:
(file_path, , encoding=) f:
content = f.read()
.validate_yaml_content(content, file_path)
FileNotFoundError:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
._generate_result()
Exception e:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
._generate_result()
() -> ValidationResult:
:
.issues = []
.warnings = []
documents = yaml.safe_load_all(content)
doc documents:
doc :
.validate_document(doc, file_name)
._generate_result()
yaml.YAMLError e:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
._generate_result()
() -> :
(doc, ):
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
kind = doc.get(, )
api_version = doc.get(, )
kind:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
.validate_required_fields(doc, kind)
kind == :
.validate_pod(doc)
kind == :
.validate_deployment(doc)
kind == :
.validate_service(doc)
.validate_best_practices(doc, kind)
() -> :
required = .required_fields.get(kind, {})
field, is_required required.items():
is_required:
value = .get_nested_field(doc, field)
value :
.issues.append(ValidationIssue(
severity=,
=,
resource=kind,
field=field,
message=,
suggestion=
))
() -> :
keys = field_path.split()
value = doc
key keys:
(value, ) key value:
value = value[key]
:
value
() -> :
spec = doc.get(, {})
containers = spec.get(, [])
containers:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
i, container (containers):
container_name = container.get(, )
container.get():
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
image = container.get(, )
image:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
image:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
resources = container.get(, {})
resources:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
:
requests = resources.get(, {})
requests:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
limits = resources.get(, {})
limits:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
liveness_probe = container.get()
readiness_probe = container.get()
liveness_probe readiness_probe:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
security_context = container.get(, {})
security_context.get(, ):
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
security_context.get() == :
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
() -> :
spec = doc.get(, {})
template = spec.get(, {})
pod_spec = template.get(, {})
replicas = spec.get(, )
replicas == :
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
selector = spec.get(, {})
match_labels = selector.get(, {})
template_labels = template.get(, {}).get(, {})
match_labels:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
key, value match_labels.items():
template_labels.get(key) != value:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
pod_doc = {
: ,
: pod_spec
}
.validate_pod(pod_doc)
() -> :
spec = doc.get(, {})
selector = spec.get(, {})
ports = spec.get(, [])
ports:
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
service_type = spec.get(, )
service_type != selector:
.warnings.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
i, port (ports):
port.get():
.issues.append(ValidationIssue(
severity=,
=,
resource=,
field=,
message=,
suggestion=
))
target_port = port.get(, port.get())
(target_port, ) target_port.isdigit():
() -> :
metadata = doc.get(, {})
labels = metadata.get(, {})
labels:
.warnings.append(ValidationIssue(
severity=,
=,
resource=kind,
field=,
message=,
suggestion=
))
namespace = metadata.get(, )
namespace == kind != :
.warnings.append(ValidationIssue(
severity=,
=,
resource=kind,
field=,
message=,
suggestion=
))
() -> ValidationResult:
critical_issues = [i i .issues i.severity == ]
high_issues = [i i .issues i.severity == ]
is_valid = (critical_issues) == (high_issues) ==
summary = {
: (.issues),
: (.warnings),
: (critical_issues),
: (high_issues),
: ([i i .issues i.severity == ]),
: ([i i .issues i.severity == ])
}
ValidationResult(
is_valid=is_valid,
issues=.issues,
warnings=.warnings,
summary=summary
)
:
():
.kubeconfig_path = kubeconfig_path
.load_kubernetes_config()
.v1 = kubernetes.client.CoreV1Api()
.apps_v1 = kubernetes.client.AppsV1Api()
():
:
.kubeconfig_path:
kubernetes.config.load_kube_config(config_file=.kubeconfig_path)
:
kubernetes.config.load_incluster_config()
Exception e:
:
kubernetes.config.load_kube_config()
Exception e2:
Exception()
() -> [, ]:
validation_results = {
: namespace,
: .validate_pods(namespace),
: .validate_deployments(namespace),
: .validate_services(namespace),
: {}
}
total_issues =
resource_type, result validation_results.items():
(result, ) result:
total_issues += (result[])
validation_results[] = {
: total_issues,
: total_issues ==
}
validation_results
() -> [, ]:
:
pods = .v1.list_namespaced_pod(namespace)
pod_results = []
pod pods.items:
issues = []
pod.status.phase == :
issues.append({
: ,
: ,
:
})
pod.status.phase == :
issues.append({
: ,
: ,
:
})
restart_count =
pod.status.container_statuses:
container_status pod.status.container_statuses:
restart_count += container_status.restart_count
restart_count > :
issues.append({
: ,
: ,
:
})
pod_results.append({
: pod.metadata.name,
: pod.status.phase,
: restart_count,
: issues
})
{
: (pods.items),
: pod_results,
: [issue pod pod_results issue pod[]]
}
Exception e:
{: }
() -> [, ]:
:
deployments = .apps_v1.list_namespaced_deployment(namespace)
deployment_results = []
deployment deployments.items:
issues = []
replicas = deployment.spec.replicas
ready_replicas = deployment.status.ready_replicas
ready_replicas < replicas:
issues.append({
: ,
: ,
:
})
replicas == :
issues.append({
: ,
: ,
:
})
strategy = deployment.spec.strategy
strategy strategy. == :
strategy strategy.rolling_update:
max_unavailable = strategy.rolling_update.max_unavailable
max_unavailable:
issues.append({
: ,
: ,
:
})
deployment_results.append({
: deployment.metadata.name,
: replicas,
: ready_replicas,
: issues
})
{
: (deployments.items),
: deployment_results,
: [issue deployment deployment_results issue deployment[]]
}
Exception e:
{: }
() -> [, ]:
:
services = .v1.list_namespaced_service(namespace)
service_results = []
service services.items:
issues = []
service_type = service.spec.
service_type == :
service.status.load_balancer service.status.load_balancer.ingress:
issues.append({
: ,
: ,
:
})
service.spec.selector:
:
endpoints = .v1.read_namespaced_endpoints(
name=service.metadata.name,
namespace=namespace
)
endpoints.subsets:
issues.append({
: ,
: ,
:
})
:
issues.append({
: ,
: ,
:
})
ports = service.spec.ports []
ports:
issues.append({
: ,
: ,
:
})
service_results.append({
: service.metadata.name,
: service_type,
: service.spec.cluster_ip,
: (ports),
: issues
})
{
: (services.items),
: service_results,
: [issue service service_results issue service[]]
}
Exception e:
{: }
():
validator = KubernetesConfigValidator()
result = validator.validate_yaml_file()
()
()
()
()
issue result.issues:
()
warning result.warnings:
()
:
cluster_validator = ClusterResourceValidator()
cluster_result = cluster_validator.validate_cluster_resources()
()
()
()
Exception e:
()
__name__ == :
main()
import yaml
import json
from typing import List, Dict, Any, Optional
from pathlib import Path
class KubernetesConfigFixer:
def __init__(self):
self.fixes_applied = []
def fix_yaml_file(self, file_path: str, issues: List[ValidationIssue]) -> Dict[str, Any]:
"""修复YAML文件中的问题"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
fixed_content, fixes = self.fix_yaml_content(content, issues)
# 写入修复后的内容
with open(file_path, 'w', encoding='utf-8') as f:
f.write(fixed_content)
return {
'file_path': file_path,
'fixes_applied': fixes,
'success': True
}
except Exception as e:
return {
'file_path': file_path,
: ,
:
}
() -> [, []]:
fixes = []
fixed_content = content
:
documents = (yaml.safe_load_all(content))
fixed_documents = []
doc documents:
doc :
fixed_doc, doc_fixes = .fix_document(doc, issues)
fixed_documents.append(fixed_doc)
fixes.extend(doc_fixes)
fixed_content = yaml.dump_all(fixed_documents, default_flow_style=, allow_unicode=)
Exception e:
fixes.append()
fixed_content, fixes
() -> [[, ], []]:
fixes = []
fixed_doc = doc.copy()
critical_issues = [i i issues i.severity == ]
high_issues = [i i issues i.severity == ]
medium_issues = [i i issues i.severity == ]
all_issues = critical_issues + high_issues + medium_issues
issue all_issues:
fix = .fix_issue(fixed_doc, issue)
fix:
fixed_doc = fix
fixes.append()
fixed_doc, fixes
() -> [[, ]]:
issue. == :
.fix_missing_field(doc, issue)
issue. == :
.apply_best_practice(doc, issue)
issue. == :
.fix_security_issue(doc, issue)
() -> [, ]:
fixed_doc = doc.copy()
issue.field == :
fixed_doc[] =
issue.field == :
fixed_doc[] =
issue.field == :
fixed_doc[] = {: }
issue.field == :
fixed_doc[] = {}
issue.field.startswith():
.set_nested_field(fixed_doc, issue.field, .get_default_value(issue.field))
fixed_doc
() -> [, ]:
fixed_doc = doc.copy()
issue.message:
.add_default_resources(fixed_doc)
issue.message:
.add_default_health_checks(fixed_doc)
issue.message:
.set_replicas(fixed_doc, )
fixed_doc
() -> [, ]:
fixed_doc = doc.copy()
issue.message:
.remove_privileged_mode(fixed_doc)
issue.message:
.set_non_root_user(fixed_doc)
fixed_doc
() -> :
keys = field_path.split()
current = doc
key keys[:-]:
key current:
current[key] = {}
current = current[key]
current[keys[-]] = value
() -> :
defaults = {
: ,
: {: },
: {: },
: [],
: []
}
defaults.get(field_path, {})
() -> :
kind = doc.get(, )
kind [, ]:
containers = .get_containers(doc)
container containers:
container:
container[] = {
: {
: ,
:
},
: {
: ,
:
}
}
() -> :
containers = .get_containers(doc)
container containers:
container:
container[] = {
: {
: ,
:
},
: ,
:
}
container:
container[] = {
: {
: ,
:
},
: ,
:
}
() -> :
doc.get() == :
doc:
doc[] = {}
doc[][] = replicas
() -> :
containers = .get_containers(doc)
container containers:
container:
container[].pop(, )
() -> :
containers = .get_containers(doc)
container containers:
container:
container[] = {}
container[][] =
container[][] =
() -> [[, ]]:
kind = doc.get(, )
containers = []
kind == :
containers = doc.get(, {}).get(, [])
kind == :
containers = doc.get(, {}).get(, {}).get(, {}).get(, [])
containers
():
validator = KubernetesConfigValidator()
fixer = KubernetesConfigFixer()
result = validator.validate_yaml_file()
result.is_valid:
()
fix_result = fixer.fix_yaml_file(, result.issues)
fix_result[]:
()
fix fix_result[]:
()
:
()
:
()
__name__ == :
main()