Menu
The review-first prompt is selected by default. You can switch to a direct command or download a local copy.
Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/microwind/ai-skills --skill kubernetesThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Showing SKILL.md
Based on SOC occupation classification
| name | Kubernetes编排 |
| description | 当实施Kubernetes编排时，分析集群架构设计，优化容器编排性能，解决K8s相关问题。验证资源配置，设计服务部署策略，和最佳实践。 |
| license | MIT |
Kubernetes是容器编排的事实标准。不当的K8s配置会导致资源浪费、性能问题和运维困难。在设计Kubernetes架构前需要仔细分析应用需求。
核心原则: 好的Kubernetes编排应该提升应用可用性和可扩展性，同时保证资源利用率。坏的编排会增加运维复杂性，甚至影响服务稳定性。
始终:
触发短语:
问题:
Pod资源配置不合理导致性能问题
错误示例:
- 没有设置资源请求和限制
- CPU和内存配置过高或过低
- 忽略资源使用监控
- 不合理的QoS类别
解决方案:
1. 设置合理的requests和limits
2. 实施资源监控和告警
3. 优化Pod调度策略
4. 配置合适的QoS类别
问题:
Kubernetes网络配置导致通信故障
错误示例:
- Service无法访问Pod
- Ingress配置错误
- 网络策略阻止通信
- DNS解析失败
解决方案:
1. 检查Service和Endpoint配置
2. 验证Ingress规则设置
3. 调整NetworkPolicy策略
4. 排查CoreDNS问题
问题:
持久化存储挂载失败或数据丢失
错误示例:
- PV和PVC配置不匹配
- 存储类选择错误
- 权限配置问题
- 数据备份策略缺失
解决方案:
1. 正确配置PV和PVC
2. 选择合适的StorageClass
3. 设置正确的访问权限
4. 实施定期备份策略
import kubernetes
import yaml
import json
import time
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class ClusterMetrics:
"""集群指标"""
node_count: int
pod_count: int
service_count: int
deployment_count: int
namespace_count: int
cpu_capacity: float
memory_capacity: float
cpu_usage: float
memory_usage: float
@dataclass
class PodMetrics:
"""Pod指标"""
name: str
namespace: str
status: str
node_name: str
cpu_requests: float
cpu_limits: float
memory_requests: float
memory_limits: float
restart_count: int
age_seconds: int
@dataclass
class ClusterIssue:
"""集群问题"""
severity: str # critical, high, medium, low
type:
resource:
namespace:
message:
suggestion:
:
():
.kubeconfig_path = kubeconfig_path
.load_kubernetes_config()
.v1 = kubernetes.client.CoreV1Api()
.apps_v1 = kubernetes.client.AppsV1Api()
.networking_v1 = kubernetes.client.NetworkingV1Api()
.issues: [ClusterIssue] = []
():
:
.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()
() -> [, ]:
:
cluster_metrics = .get_cluster_metrics()
pod_analysis = .analyze_pods()
node_analysis = .analyze_nodes()
resource_analysis = .analyze_resource_usage()
network_analysis = .analyze_network()
report = {
: cluster_metrics,
: pod_analysis,
: node_analysis,
: resource_analysis,
: network_analysis,
: .issues,
: .generate_recommendations(),
: .calculate_health_score(cluster_metrics)
}
report
Exception e:
{: }
() -> ClusterMetrics:
:
nodes = .v1.list_node()
node_count = (nodes.items)
pods = .v1.list_pod_for_all_namespaces()
pod_count = (pods.items)
services = .v1.list_service_for_all_namespaces()
service_count = (services.items)
deployments = .apps_v1.list_deployment_for_all_namespaces()
deployment_count = (deployments.items)
namespaces = .v1.list_namespace()
namespace_count = (namespaces.items)
cpu_capacity =
memory_capacity =
node nodes.items:
cpu_capacity += .parse_cpu(node.status.capacity.get(, ))
memory_capacity += .parse_memory(node.status.capacity.get(, ))
ClusterMetrics(
node_count=node_count,
pod_count=pod_count,
service_count=service_count,
deployment_count=deployment_count,
namespace_count=namespace_count,
cpu_capacity=cpu_capacity,
memory_capacity=memory_capacity,
cpu_usage=,
memory_usage=
)
Exception e:
Exception()
() -> [, ]:
:
pods = .v1.list_pod_for_all_namespaces()
pod_metrics = []
status_counts = defaultdict()
namespace_counts = defaultdict()
pod pods.items:
metrics = .get_pod_metrics(pod)
pod_metrics.append(metrics)
status_counts[pod.status.phase] +=
namespace_counts[pod.metadata.namespace] +=
.check_pod_issues(pod, metrics)
{
: (pods.items),
: (status_counts),
: (namespace_counts),
: pod_metrics,
: [issue issue .issues issue..startswith()]
}
Exception e:
{: }
() -> PodMetrics:
:
cpu_requests =
cpu_limits =
memory_requests =
memory_limits =
pod.spec.containers:
container pod.spec.containers:
container.resources:
container.resources.requests:
cpu_requests += .parse_cpu(container.resources.requests.get(, ))
memory_requests += .parse_memory(container.resources.requests.get(, ))
container.resources.limits:
cpu_limits += .parse_cpu(container.resources.limits.get(, ))
memory_limits += .parse_memory(container.resources.limits.get(, ))
restart_count =
pod.status.container_statuses:
container_status pod.status.container_statuses:
restart_count += container_status.restart_count
creation_time = pod.metadata.creation_timestamp
age_seconds = ((time.time() - creation_time.timestamp()))
PodMetrics(
name=pod.metadata.name,
namespace=pod.metadata.namespace,
status=pod.status.phase,
node_name=pod.spec.node_name ,
cpu_requests=cpu_requests,
cpu_limits=cpu_limits,
memory_requests=memory_requests,
memory_limits=memory_limits,
restart_count=restart_count,
age_seconds=age_seconds
)
Exception e:
Exception()
() -> :
pod.status.phase == :
.issues.append(ClusterIssue(
severity=,
=,
resource=pod.metadata.name,
namespace=pod.metadata.namespace,
message=,
suggestion=
))
pod.status.phase == :
.issues.append(ClusterIssue(
severity=,
=,
resource=pod.metadata.name,
namespace=pod.metadata.namespace,
message=,
suggestion=
))
metrics.restart_count > :
.issues.append(ClusterIssue(
severity=,
=,
resource=pod.metadata.name,
namespace=pod.metadata.namespace,
message=,
suggestion=
))
metrics.cpu_requests == metrics.memory_requests == :
.issues.append(ClusterIssue(
severity=,
=,
resource=pod.metadata.name,
namespace=pod.metadata.namespace,
message=,
suggestion=
))
metrics.cpu_limits == metrics.memory_limits == :
.issues.append(ClusterIssue(
severity=,
=,
resource=pod.metadata.name,
namespace=pod.metadata.namespace,
message=,
suggestion=
))
() -> [, ]:
:
nodes = .v1.list_node()
node_analysis = []
node nodes.items:
analysis = {
: node.metadata.name,
: .get_node_status(node),
: .get_node_roles(node),
: node.status.capacity,
: node.status.allocatable,
: node.status.conditions,
: []
}
condition node.status.conditions []:
condition. == condition.status != :
analysis[].append({
: ,
: ,
:
})
node_analysis.append(analysis)
{
: (nodes.items),
: node_analysis,
: ([n n node_analysis n[] == ])
}
Exception e:
{: }
() -> :
condition node.status.conditions []:
condition. == :
condition.status ==
() -> []:
roles = []
labels = node.metadata.labels {}
label labels:
label.startswith():
role = label.replace(, )
role:
roles.append(role)
roles []
() -> [, ]:
:
{
: ,
:
}
Exception e:
{: }
() -> [, ]:
:
services = .v1.list_service_for_all_namespaces()
service_analysis = []
service services.items:
analysis = {
: service.metadata.name,
: service.metadata.namespace,
: service.spec.,
: service.spec.cluster_ip,
: service.spec.external_i_ps [],
: service.spec.ports [],
: service.spec.selector {},
: []
}
service.spec. == service.spec.cluster_ip:
analysis[].append({
: ,
: ,
:
})
service.spec.selector service.spec.selector:
analysis[].append({
: ,
: ,
:
})
service_analysis.append(analysis)
:
ingresses = .networking_v1.list_ingress_for_all_namespaces()
ingress_analysis = []
ingress ingresses.items:
analysis = {
: ingress.metadata.name,
: ingress.metadata.namespace,
: ingress.spec.rules [],
: ingress.spec.tls [],
: ingress.spec.backend,
: []
}
ingress_analysis.append(analysis)
:
ingress_analysis = []
{
: (services.items),
: service_analysis,
: (ingress_analysis) ingress_analysis ,
: ingress_analysis
}
Exception e:
{: }
() -> [[, ]]:
recommendations = []
issue_counts = defaultdict()
issue .issues:
issue_counts[issue.] +=
issue_counts[] > :
recommendations.append({
: ,
: ,
: ,
:
})
issue_counts[] > :
recommendations.append({
: ,
: ,
: ,
:
})
issue_counts[] > :
recommendations.append({
: ,
: ,
: ,
:
})
recommendations
() -> :
score =
issue .issues:
issue.severity == :
score -=
issue.severity == :
score -=
issue.severity == :
score -=
issue.severity == :
score -=
metrics.node_count == :
score =
(, (score))
() -> :
cpu_str:
cpu_str = cpu_str.lower()
cpu_str.endswith():
(cpu_str[:-]) /
:
(cpu_str)
() -> :
memory_str:
memory_str = memory_str.upper()
memory_str.endswith():
(memory_str[:-]) *
memory_str.endswith():
(memory_str[:-]) * *
memory_str.endswith():
(memory_str[:-]) * * *
memory_str.endswith():
(memory_str[:-]) *
memory_str.endswith():
(memory_str[:-]) * *
memory_str.endswith():
(memory_str[:-]) * * *
:
(memory_str)
:
():
.kubeconfig_path = kubeconfig_path
.load_kubernetes_config()
.apps_v1 = kubernetes.client.AppsV1Api()
.v1 = kubernetes.client.CoreV1Api()
():
:
.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()
() -> [, ]:
:
deployment = .apps_v1.read_namespaced_deployment(
name=deployment_name,
namespace=namespace
)
current_analysis = .analyze_deployment_config(deployment)
optimization_plan = .generate_deployment_optimizations(deployment)
optimized_deployment = .generate_optimized_deployment(deployment, optimization_plan)
{
: namespace,
: deployment_name,
: current_analysis,
: optimization_plan,
: optimized_deployment,
: .estimate_deployment_improvements(deployment, optimized_deployment)
}
Exception e:
{: }
() -> [, ]:
analysis = {
: deployment.spec.replicas,
: deployment.spec.strategy. deployment.spec.strategy ,
: {
: []
},
: []
}
deployment.spec.template.spec.containers:
container deployment.spec.template.spec.containers:
container_analysis = {
: container.name,
: container.image,
: {
: container.resources.requests container.resources {},
: container.resources.limits container.resources {}
},
: container.ports [],
: (container.env []),
: (container.volume_mounts []),
: container.liveness_probe ,
: container.readiness_probe ,
: container.startup_probe
}
container.resources:
analysis[].append({
: container.name,
: ,
: ,
:
})
container.resources.requests:
analysis[].append({
: container.name,
: ,
: ,
:
})
container.liveness_probe:
analysis[].append({
: container.name,
: ,
: ,
:
})
container.readiness_probe:
analysis[].append({
: container.name,
: ,
: ,
:
})
analysis[][].append(container_analysis)
analysis
() -> [[, ]]:
optimizations = []
deployment.spec.replicas == :
optimizations.append({
: ,
: ,
: ,
: ,
:
})
deployment.spec.strategy deployment.spec.strategy. == :
strategy = deployment.spec.strategy
strategy strategy.rolling_update:
max_unavailable = strategy.rolling_update.max_unavailable
max_surge = strategy.rolling_update.max_surge
max_unavailable max_unavailable == :
optimizations.append({
: ,
: ,
: ,
:
})
deployment.spec.template.spec.containers:
container deployment.spec.template.spec.containers:
container.image:
optimizations.append({
: ,
: ,
: ,
:
})
optimizations
() -> [, ]:
optimized_deployment = original_deployment
optimization optimizations:
optimization[] == optimization.get():
optimized_deployment.spec.replicas = optimization[]
optimization[] == :
optimized_deployment.spec.strategy:
optimized_deployment.spec.strategy = kubernetes.client.V1DeploymentStrategy(
=,
rolling_update=kubernetes.client.V1RollingUpdateDeployment(
max_unavailable=,
max_surge=
)
)
optimized_deployment
() -> [, ]:
improvements = {
: ,
: ,
: ,
:
}
original.spec.replicas == optimized.spec.replicas > :
improvements[] +=
optimized.spec.template.spec.containers:
container optimized.spec.template.spec.containers:
container.resources container.resources.requests:
improvements[] +=
optimized.spec.template.spec.containers:
container optimized.spec.template.spec.containers:
container.image:
improvements[] +=
improvements[] = (
improvements[] +
improvements[] +
improvements[]
) //
improvements
():
analyzer = KubernetesClusterAnalyzer()
cluster_report = analyzer.analyze_cluster()
()
()
()
()
optimizer = KubernetesDeploymentOptimizer()
deployment_optimization = optimizer.optimize_deployment(, )
()
opt deployment_optimization[]:
()
__name__ == :
main()
import kubernetes
import yaml
import json
from typing import List, Dict, Any, Optional
from pathlib import Path
class KubernetesResourceManager:
def __init__(self, kubeconfig_path: Optional[str] = None):
self.kubeconfig_path = kubeconfig_path
self.load_kubernetes_config()
self.v1 = kubernetes.client.CoreV1Api()
self.apps_v1 = kubernetes.client.AppsV1Api()
self.networking_v1 = kubernetes.client.NetworkingV1Api()
def load_kubernetes_config(self):
"""加载Kubernetes配置"""
try:
if self.kubeconfig_path:
kubernetes.config.load_kube_config(config_file=self.kubeconfig_path)
else:
kubernetes.config.load_incluster_config()
except Exception as e:
try:
kubernetes.config.load_kube_config()
except Exception as e2:
raise Exception(f'无法加载Kubernetes配置: {e2}')
def create_resource_from_yaml(self, yaml_file: , namespace: = ) -> [, ]:
:
(yaml_file, , encoding=) f:
yaml_content = yaml.safe_load_all(f)
created_resources = []
resource_dict yaml_content:
resource_dict:
resource_kind = resource_dict.get()
resource_name = resource_dict.get(, {}).get(, )
resource_kind == :
deployment = .create_deployment_from_dict(resource_dict, namespace)
created_resources.append({
: ,
: resource_name,
: namespace,
:
})
resource_kind == :
service = .create_service_from_dict(resource_dict, namespace)
created_resources.append({
: ,
: resource_name,
: namespace,
:
})
resource_kind == :
configmap = .create_configmap_from_dict(resource_dict, namespace)
created_resources.append({
: ,
: resource_name,
: namespace,
:
})
resource_kind == :
secret = .create_secret_from_dict(resource_dict, namespace)
created_resources.append({
: ,
: resource_name,
: namespace,
:
})
{
: created_resources,
: (created_resources),
: namespace
}
Exception e:
{: }
():
deployment = kubernetes.client.V1Deployment(
api_version=,
kind=,
metadata=kubernetes.client.V1ObjectMeta(
name=deployment_dict[][],
namespace=namespace,
labels=deployment_dict[].get(, {})
),
spec=kubernetes.client.V1DeploymentSpec(
replicas=deployment_dict[].get(, ),
selector=kubernetes.client.V1LabelSelector(
match_labels=deployment_dict[][][]
),
template=.create_pod_template_from_dict(deployment_dict[][])
)
)
.apps_v1.create_namespaced_deployment(
namespace=namespace,
body=deployment
)
() -> kubernetes.client.V1PodTemplateSpec:
containers = []
container_dict template_dict[][]:
container = kubernetes.client.V1Container(
name=container_dict[],
image=container_dict[],
ports=[
kubernetes.client.V1ContainerPort(
container_port=port[],
protocol=port.get(, )
) port container_dict.get(, [])
],
env=[
kubernetes.client.V1EnvVar(
name=env[],
value=env.get(),
value_from=env.get()
) env container_dict.get(, [])
],
resources=.create_resource_requirements_from_dict(
container_dict.get(, {})
) container_dict ,
liveness_probe=.create_probe_from_dict(
container_dict.get(, {})
) container_dict ,
readiness_probe=.create_probe_from_dict(
container_dict.get(, {})
) container_dict
)
containers.append(container)
kubernetes.client.V1PodTemplateSpec(
metadata=kubernetes.client.V1ObjectMeta(
labels=template_dict[].get(, {}),
annotations=template_dict[].get(, {})
),
spec=kubernetes.client.V1PodSpec(
containers=containers,
restart_policy=template_dict[].get(, )
)
)
() -> kubernetes.client.V1ResourceRequirements:
kubernetes.client.V1ResourceRequirements(
requests=resources_dict.get(, {}),
limits=resources_dict.get(, {})
)
() -> kubernetes.client.V1Probe:
kubernetes.client.V1Probe(
http_get=kubernetes.client.V1HTTPGetAction(
path=probe_dict.get(, {}).get(, ),
port=probe_dict.get(, {}).get(, )
) probe_dict ,
tcp_socket=kubernetes.client.V1TCPSocketAction(
port=probe_dict.get(, {}).get(, )
) probe_dict ,
initial_delay_seconds=probe_dict.get(, ),
period_seconds=probe_dict.get(, ),
timeout_seconds=probe_dict.get(, ),
failure_threshold=probe_dict.get(, )
)
():
service = kubernetes.client.V1Service(
api_version=,
kind=,
metadata=kubernetes.client.V1ObjectMeta(
name=service_dict[][],
namespace=namespace,
labels=service_dict[].get(, {})
),
spec=kubernetes.client.V1ServiceSpec(
=service_dict[].get(, ),
selector=service_dict[].get(, {}),
ports=[
kubernetes.client.V1ServicePort(
port=port[],
target_port=port.get(, port[]),
protocol=port.get(, )
) port service_dict[][]
]
)
)
.v1.create_namespaced_service(
namespace=namespace,
body=service
)
():
configmap = kubernetes.client.V1ConfigMap(
api_version=,
kind=,
metadata=kubernetes.client.V1ObjectMeta(
name=configmap_dict[][],
namespace=namespace,
labels=configmap_dict[].get(, {})
),
data=configmap_dict.get(, {}),
binary_data=configmap_dict.get(, {})
)
.v1.create_namespaced_config_map(
namespace=namespace,
body=configmap
)
():
base64
data = {}
secret_dict:
key, value secret_dict[].items():
(value, ):
data[key] = base64.b64encode(value.encode()).decode()
:
data[key] = value
secret = kubernetes.client.V1Secret(
api_version=,
kind=,
metadata=kubernetes.client.V1ObjectMeta(
name=secret_dict[][],
namespace=namespace,
labels=secret_dict[].get(, {})
),
=secret_dict.get(, ),
data=data,
string_data=secret_dict.get(, {})
)
.v1.create_namespaced_secret(
namespace=namespace,
body=secret
)
() -> [, ]:
:
deployment = .apps_v1.read_namespaced_deployment(
name=deployment_name,
namespace=namespace
)
deployment.spec.replicas = replicas
updated_deployment = .apps_v1.patch_namespaced_deployment(
name=deployment_name,
namespace=namespace,
body=deployment
)
{
: deployment_name,
: namespace,
: deployment.spec.replicas,
: replicas,
:
}
Exception e:
{: }
() -> [, ]:
:
deployment = .apps_v1.read_namespaced_deployment(
name=deployment_name,
namespace=namespace
)
deployment.spec.template.metadata :
deployment.spec.template.metadata = kubernetes.client.V1ObjectMeta()
deployment.spec.template.metadata.annotations = {
: time.strftime()
}
updated_deployment = .apps_v1.patch_namespaced_deployment(
name=deployment_name,
namespace=namespace,
body=deployment
)
{
: deployment_name,
: namespace,
: ,
: time.strftime()
}
Exception e:
{: }
():
resource_manager = KubernetesResourceManager()
result = resource_manager.create_resource_from_yaml()
()
scale_result = resource_manager.scale_deployment(, , )
()
restart_result = resource_manager.restart_deployment(, )
()
__name__ == :
main()