import json import traceback import yaml from kubernetes import client, config from flask import jsonify import dmPython # Load in-cluster config config.load_incluster_config() v1 = client.CoreV1Api() namespace = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace").read() IP = "36.138.114.105" def list_pods(): ret = v1.list_namespaced_pod(namespace, watch=False) return ({ "list": [{ "name": item.metadata.name, "ip": item.status.pod_ip } for item in ret.items] }) def list_services(): ret = v1.list_namespaced_service(namespace) return { "list": [{ "ip": item.spec.cluster_ip, "name": item.metadata.name, "ports": [port.to_dict() for port in item.spec.ports] } for item in ret.items] } def create_pod(type, name): # 读取 pod yaml 文件 with open(f"assets/type{type}-pod.yaml") as f: pod_manifest = yaml.safe_load(f) pod_manifest["metadata"]["name"] = name pod_manifest["metadata"]["labels"]["app"] = name # 创建 Pod resp = v1.create_namespaced_pod( body=pod_manifest, namespace=namespace ) # print(f"pod '{resp.metadata.name}' 创建成功") # 创建 Service with open(f"assets/type{type}-service.yaml") as f: service_pod = yaml.safe_load(f) service_pod["metadata"]["name"] = name + "-service" service_pod["spec"]["selector"]["app"] = name resp = v1.create_namespaced_service( body=service_pod, namespace=namespace ) # print(f"服务 '{resp.metadata.name}' 创建成功") if type == 1: # 创建成功,再次查询,拿到端口 service = v1.read_namespaced_service(name=name + "-service", namespace=namespace) # print(service) # print("访问地址:" + IP + ":" + str(service.spec.ports[0].node_port)) else: service = v1.read_namespaced_service(name=name + "-service", namespace=namespace) # print(service) # print("访问地址:" + IP + ":" + str(service.spec.ports[0].node_port)) return "cjcg" # create_pod(0,'20240101-zt-9-0') def delete_pod(name): v1.delete_namespaced_pod(name=name, namespace=namespace) print('deleting pod ' + name) v1.delete_namespaced_service(name=name + "-service", namespace=namespace) return "删除成功" def check_dm(name): try: service = v1.read_namespaced_service(name=name + "-service", namespace=namespace) print(service.spec.cluster_ip) db = dmPython.connect(user='SYSDBA', password='SYSDBA', host=service.spec.cluster_ip) cursor = db.cursor() records = cursor.execute("SELECT * FROM V$VERSION").fetchall() for record in records: print(record) return "OK" except: traceback.print_exc() return "NO"