القائمة
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام 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()