python通过api管理k8s

连接k8s集群

from kubernetes  import  client, config
# kubernetes生成时/root目录下.kube目录下的认证配置文件
config.kube_config.load_kube_config(config_file="/root/.kube/config")
# config.load_kube_config(kube_conf)
api_instance = client.AppsV1Api()

查看deployment内容

def read_deployment(name, namespace, update_image):
    body = api_instance.read_namespaced_deployment(name, namespace)
    return body

修改deployment镜像版本

def replace_deployment(name, namespace, update_image):
    body = api_instance.read_namespaced_deployment(name, namespace)
    body.spec.template.spec.containers[0].image = update_image
    try:
        api_response = api_instance.replace_namespaced_deployment(name, namespace, body)
        print("\033[0;32m{} 中deployment : {} {}更新完成\033[0m".format(namespace, name, update_image))
    except ApiException as e:
        print("Exception when calling AppsV1Api->replace_namespaced_deployment: %s\n" % e)    
    return body

注意：

patch_namespaced_deployment ：局部更新YAML
replace_namespaced_deployment：替换整个YAML

创建，删除，service，deployment

1. 连接k8s集群

模块：Connect.py

from kubernetes  import  client, config
from kubernetes.client.rest import ApiException
import yaml
# kubernetes生成时/root目录下.kube目录下的认证配置文件
config.kube_config.load_kube_config(config_file="/root/.kube/config")
# config.load_kube_config(kube_conf)

#管理deployment时使用
api_instance = client.AppsV1Api()

#管理service时使用
CoreV1Api = client.CoreV1Api()

模板文件：service.yaml

apiVersion: v1
kind: Service
metadata:
  annotations:
    field.cattle.io/targetWorkloadIds: '["deployment:default:nginx"]'
  name: projectname
  namespace: NS
  labels:
    app: appname
    project: NS

spec:
  type: NodePort
  selector:
    app: projectname
  ports:
  - protocol: TCP
    port: 8080
    targetPort: 8080

创建service

from module.Connect import CoreV1Api
import yaml
import time
import os

# if __name__ != '__main__':
#     os.chdir('module')


def create_service(namespace, name, container_port, type='ClusterIP'):
    with open('service.yaml', mode='r') as f:
        body = yaml.safe_load(f)
    body['metadata']['name'] = name
    body['metadata']['namespace'] = namespace
    body['metadata']['labels']['app'] = name
    body['metadata']['annotations']['field.cattle.io/targetWorkloadIds'] = '["deployment:{namespace}:{name}"]'.format(namespace=namespace, name=name)
    if type == 'ClusterIP':
        body['metadata']['spec'] = dict(type=type,
                                        selector={'app': name},
                                        ports=[
                                            {'protocol': 'TCP', 'port': container_port, 'targetPort': container_port}])
    elif type == 'NodePort':
        body['metadata']['spec'] = dict(type=type,
                                        selector={'app': name},
                                        ports=[
                                            {'protocol': 'TCP', 'port': container_port, 'targetPort': container_port}])
    else:
        print("""输入错误！！请输入:
        'NodePort' 或者 'ClusterIP'
        默认：'ClusterIP'
        """)
        pass
    CoreV1Api.create_namespaced_service(
        namespace=namespace,
        body=body,
    )

if __name__ == '__main__':
    namespace = 'default'
    name = 'nginx2'
    port = 80
    create_service(namespace=namespace, name=name, container_port=port)

模板文件：deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  annotations:
    deployment.kubernetes.io/revision: "22"
  labels:
  name: tomcat8-test3
  namespace: default
spec:
  progressDeadlineSeconds: 600
  replicas: 4
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      workload.user.cattle.io/workloadselector: deployment-default-tomcat8-test
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
    type: RollingUpdate
  template:
    metadata:
      annotations:
      creationTimestamp: null
      labels:
        workload.user.cattle.io/workloadselector: deployment-default-tomcat8-test
    spec:
      containers:
      - image: test.com/tomcat/centos6.5-tomcat8:prod
        imagePullPolicy: IfNotPresent
        name: tomcat8-test
        resources: {}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities: {}
          privileged: false
          readOnlyRootFilesystem: false
          runAsNonRoot: false
        stdin: true
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        tty: true

      dnsConfig: {}
      dnsPolicy: ClusterFirst
      hostAliases:
      - hostnames:
        - dccas.finupgroup.com
        ip: 192.168.16.247

      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30
      volumes:

创建deployment

from module.Connect import api_instance
import yaml
import time
import os
#os.chdir('module')

def create_deployment(namespace,name,update_image):
    deployment_info = api_instance.read_namespaced_deployment(namespace='default',name='nginx')
    print(deployment_info.spec.template.spec.containers[0].image)
    with open('deployment.yaml', mode='r') as f:
        body = yaml.safe_load(f)

    body['metadata']['name'] = name
    body['metadata']['namespace'] = namespace
    api_instance.create_namespaced_deployment(namespace='default', body=body)
    # api_instance.delete_namespaced_deployment(namespace=namespace, name=name)
    time.sleep(3)

    body = api_instance.read_namespaced_deployment(namespace=namespace, name=name)
    body.spec.template.spec.containers[0].image = update_image
    body.spec.template.spec.containers[0].name = name
    api_instance.replace_namespaced_deployment(namespace=namespace, name=name, body=body)
   
 
if __name__ == '__main__':
    namespace = 'default'
    name = 'nginx2'
    port = 80
    create_service(namespace=namespace, name=name, container_port=port)

删除deployment、service

from Connect import CoreV1Api, api_instance
namespace = 'default'
name = 'nginx2'

# 删除deployment

api_instance.delete_namespaced_deployment(namespace=namespace,name=name)
CoreV1Api.delete_namespaced_service(namespace=namespace,name=name)