K8S 容器资源 Python 运维开发#

1
import json, time, yaml
2
from kubernetes.client import CoreV1Api
3
from kubernetes.config import load_kube_config
4

5
load_kube_config()
6
api = CoreV1Api()
7

8
# 删除同名 Pod（忽略不存在的情况）
9
try:
10
    api.delete_namespaced_pod(name='nginx-pod', namespace='default')
11
    print("删除同名 Pod")
12
except Exception:
13
    pass
14

15
time.sleep(5)
16

17
# 从 YAML 读取配置并创建 Pod
18
with open("nginx-pod.yaml", encoding='utf-8') as f:
19
    data = yaml.safe_load(f)
20

21
create = api.create_namespaced_pod(body=data, namespace='default')
22
print(f"创建成功: {create.metadata.name}")
23

24
# 查询 Pod
25
get = api.read_namespaced_pod(name='nginx-pod', namespace='default')
26
print(f"状态: {get.status.phase}")
27

28
# 保存结果为 JSON
29
with open("pod_sdk_dev.json", "w") as f:
30
    json.dump(get.to_dict(), f, default=str, indent=4)

1
import json, yaml, time
2
from kubernetes.client import AppsV1Api
3
from kubernetes.config import load_kube_config
4

5
load_kube_config()
6
api = AppsV1Api()
7

8
# 删除同名 Deployment
9
try:
10
    api.delete_namespaced_deployment(name='nginx-deployment', namespace='default')
11
    print('同名删除')
12
except Exception:
13
    pass
14

15
time.sleep(5)
16

17
# 从 YAML 创建 Deployment
18
with open('nginx-deployment.yaml', encoding='utf-8') as f:
19
    data = yaml.safe_load(f)
20

21
create = api.create_namespaced_deployment(body=data, namespace='default')
22
print(f"创建成功: {create.metadata.name}")
23

24
# 查询 Deployment
25
get = api.read_namespaced_deployment(name="nginx-deployment", namespace="default")
26
print(f"副本数: {get.spec.replicas}, 就绪: {get.status.ready_replicas}")
27

28
# 保存为 JSON
29
with open("deployment_sdk_dev.json", "w") as f:
30
    json.dump(get.to_dict(), f, default=str, indent=4)

1
import time, yaml
2
from kubernetes import client, config
3

4
config.load_kube_config()
5
api = client.BatchV1Api()
6

7
# 删除同名 Job
8
try:
9
    api.delete_namespaced_job(name='pi', namespace='default')
10
    print("删除同名 Job")
11
except Exception:
12
    pass
13

14
time.sleep(10)
15

16
# 从 YAML 创建 Job
17
with open('job.yaml', encoding='utf-8') as f:
18
    data = yaml.safe_load(f)
19

20
create = api.create_namespaced_job(body=data, namespace='default')
21
print(f"创建成功: {create.metadata.name}")
22

23
# 查询 Job
24
get = api.read_namespaced_job(name='pi', namespace='default')
25
print(f"状态: {get.status.conditions}")

1
import requests, yaml, json
2

3
token = 'bearer <your-token>'
4
url = 'https://<master-ip>:6443/api/v1/namespaces/default/services'
5

6
header = {
7
    "Content-Type": "application/json",
8
    "Authorization": token
9
}
10
# Patch 操作需要不同的 Content-Type
11
header_patch = {
12
    "Content-Type": "application/strategic-merge-patch+json",
13
    "Authorization": token
14
}
15

16
# 查询并删除同名 Service
17
resp = requests.get(f"{url}/nginx-svc", headers=header, verify=False)
18
if resp.status_code == 200:
19
    requests.delete(f"{url}/nginx-svc", headers=header, verify=False)
20
    print("同名删除完成")
21

22
# 创建 Service
23
with open('service.yaml', encoding='utf-8') as f:
24
    data = json.dumps(yaml.safe_load(f))
25
create = requests.post(url, headers=header, data=data, verify=False).text
26
print(f"创建信息: {create}")
27

28
# 查询并保存
29
get = requests.get(f"{url}/nginx-svc", headers=header, verify=False)
30
with open('service_api_dev.json', 'w') as f:
31
    json.dump(get.json(), f, indent=4)
32

33
# 更新 Service（Patch）
34
with open('service_update.yaml', encoding='utf-8') as f:
35
    update_data = json.dumps(yaml.safe_load(f))
36
update = requests.patch(f"{url}/nginx-svc", headers=header_patch, data=update_data, verify=False)
37
print(f"更新状态: {update.status_code}")

1
import random
2
from kubernetes import client, config
3

4
namespace = 'default'
5
config.load_kube_config()
6
api = client.CoreV1Api()
7

8

9
def get_ready_nodes():
10
    """获取所有 Ready 状态的节点"""
11
    nodelist = api.list_node()
12
    ready_nodes = []
13
    for node in nodelist.items:
14
        for condition in node.status.conditions:
15
            if condition.type == 'Ready' and condition.status == 'True':
16
                ready_nodes.append(node.metadata.name)
17
                break
18
    return ready_nodes
19

20

21
def get_pending_pods():
22
    """获取所有 Pending 状态且未调度的 Pod"""
23
    podlist = api.list_namespaced_pod(namespace)
24
    pending_pods = []
25
    for pod in podlist.items:
26
        if pod.status.phase == 'Pending' and pod.spec.node_name is None:
27
            pending_pods.append(pod.metadata.name)
28
    return pending_pods
29

30

31
def bind_pod_to_node(nodename, podname):
32
    """将 Pod 绑定到指定节点"""
33
    target = client.V1ObjectReference(kind='node', api_version='v1', name=nodename)
34
    meta = client.V1ObjectMeta(name=podname)
35
    body = client.V1Binding(target=target, metadata=meta)
36
    try:
37
        api.create_namespaced_binding(namespace, body)
38
        return True
39
    except client.exceptions.ApiException as e:
40
        print(f"绑定失败: {e.reason}")
41
        return False
42

43

44
def schedule_pods(nodes, pods):
45
    """随机选择节点调度所有 Pending Pod"""
46
    if not nodes or not pods:
47
        print("无可用节点或无待调度 Pod")
48
        return
49
    nodename = random.choice(nodes)
50
    print(f"选择节点: {nodename}")
51
    for podname in pods:
52
        scheduler = api.read_namespaced_pod(podname, namespace).spec.scheduler_name
53
        print(f"调度 Pod: {podname} (scheduler: {scheduler})")
54
        if bind_pod_to_node(nodename, podname):
55
            print(f"  -> 绑定成功")
56

57

58
if __name__ == "__main__":
59
    nodes = get_ready_nodes()
60
    pods = get_pending_pods()
61
    schedule_pods(nodes, pods)

1
import json, requests, yaml
2
from http.server import HTTPServer, BaseHTTPRequestHandler
3

4

5
class PodHandler(BaseHTTPRequestHandler):
6
    api = "https://<master-ip>:6443/api/v1/namespaces/default/pods"
7
    auth = {
8
        'verify': False,
9
        'headers': {
10
            "Content-Type": "application/json",
11
            "Authorization": "Bearer <token>"
12
        }
13
    }
14

15
    def _parse_name(self):
16
        return self.path.split('/')[-1]
17

18
    def do_GET(self):
19
        """GET /pod/<name> - 查询 Pod"""
20
        if self.path.startswith('/pod/'):
21
            res = requests.get(f"{self.api}/{self._parse_name()}", **self.auth)
22
            self.send_response(200)
23
            self.end_headers()
24
            self.wfile.write(json.dumps(res.json()).encode())
25

26
    def do_POST(self):
27
        """POST /pod/<yaml-file> - 从 YAML 文件创建 Pod"""
28
        if self.path.startswith('/pod/'):
29
            with open(f"/root/{self._parse_name()}") as f:
30
                body = json.dumps(yaml.safe_load(f))
31
            res = requests.post(self.api, data=body, **self.auth)
32
            self.send_response(201)
33
            self.end_headers()
34
            self.wfile.write(json.dumps(res.json()).encode())
35

36
    def do_DELETE(self):
37
        """DELETE /pod/<name> - 删除 Pod"""
38
        if self.path.startswith('/pod/'):
39
            requests.delete(f"{self.api}/{self._parse_name()}", **self.auth)
40
            self.send_response(204)
41
            self.end_headers()
42

43

44
if __name__ == "__main__":
45
    HTTPServer(('localhost', 8889), PodHandler).serve_forever()

1
import json, requests, yaml, threading
2
from http.server import HTTPServer, BaseHTTPRequestHandler
3

4

5
class ServiceHandler(BaseHTTPRequestHandler):
6
    api = "https://<master-ip>:6443/api/v1/namespaces/default/services"
7
    auth = {
8
        'headers': {
9
            "Content-Type": "application/json",
10
            "Authorization": "Bearer <token>"
11
        },
12
        'verify': False
13
    }
14

15
    def _parse_name(self):
16
        return self.path.split('/')[-1]
17

18
    def do_GET(self):
19
        """GET /services/<name> - 查询 Service"""
20
        if self.path.startswith('/services/'):
21
            res = requests.get(f"{self.api}/{self._parse_name()}", **self.auth)
22
            self.send_response(200)
23
            self.end_headers()
24
            self.wfile.write(json.dumps(res.json()).encode())
25

26
    def do_POST(self):
27
        """POST /services/<yaml-file> - 从 YAML 文件创建 Service"""
28
        if self.path.startswith('/services/'):
29
            with open(f"/root/{self._parse_name()}") as f:
30
                body = json.dumps(yaml.safe_load(f))
31
            res = requests.post(self.api, data=body, **self.auth)
32
            self.send_response(201)
33
            self.end_headers()
34
            self.wfile.write(json.dumps(res.json()).encode())
35

36
    def do_DELETE(self):
37
        """DELETE /services/<name> - 删除 Service"""
38
        if self.path.startswith('/services/'):
39
            requests.delete(f"{self.api}/{self._parse_name()}", **self.auth)
40
            self.send_response(204)
41
            self.end_headers()
42

43

44
if __name__ == "__main__":
45
    threading.Thread(
46
        target=lambda: HTTPServer(('localhost', 8888), ServiceHandler).serve_forever()
47
    ).start()