python (k8s) 2. object
1. custom_object.py# -*- coding: utf8 -*-"""使用自定义资源创建一个自定义对象Custom Resource Definition (CRD)本案例中是一个CronTabhttps://kubernetes.io/docs/tasks/access-kubernetes-api/custom-resources/custom-resour...
·
1. custom_object.py
# -*- coding: utf8 -*-
"""
使用自定义资源创建一个自定义对象
Custom Resource Definition (CRD)
本案例中是一个CronTab
https://kubernetes.io/docs/tasks/access-kubernetes-api/custom-resources/custom-resource-definitions/
"""
"""
编组清单manifest: (.yaml文件)
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
name: crontabs.stable.example.com
spec:
group: stable.example
versions:
- name: v1
served: true
storage: true
scope: Namespaced
names:
plural: crontabs
singular: crontab
kind: CronTab
shortNames:
- ct
"""
from pprint import pprint
from kubernetes import client, config
def main():
config.load_kube_config()
api = client.CustomObjectsApi()
# 资源定义成字典形式
my_resource = {
'apiVersion': 'stable.example.com/v1',
'kind': 'CronTab',
'metadata': {'name': 'my-new-cron-object'},
'cronSpec': '* * * * */5',
'image': 'my-awesome-cron-image',
}
# 创建资源
api.create_namespaced_custom_object(
group='stable.example.com',
version='v1',
namespace='default',
plural='crontabs',
body=my_resource,
)
print("Resource created")
# 获取资源, 输出数据
resource = api.get_namespaced_custom_object(
group='stable.example.com',
version='v1',
name='my-new-cron-object',
namespace='default',
plural='crontabs',
)
print('Resource details:')
pprint(resource)
# 删除资源
api.delete_namespaced_custom_object(
group='stable.example.com',
version='v1',
name='my-new-cron-object',
namespace='default',
plural='crontabs',
body=client.V1DeleteOptions(),
)
print('Resource deleted')
if __name__ == '__main__':
main()
2. deployment_crud.py
# -*- coding: utf8 -*-
"""
使用 AppsV1Api 增 改 删 deployment
"""
from kubernetes import client, config
DEPLOYMENT_NAME = 'nginx-deployment'
def create_deployment_object():
# 配置一个 Pod 模板容器
container = client.V1Container(
name='nginx',
image='nginx:1.15.4',
ports=[client.V1ContainerPort(container_port=80)]
)
# 创建 并 配置 spec部分
template = client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(labels={"app": "nginx"}),
spec=client.V1PodSpec(containers=[container])
)
# 创建 deployment 的详细定义
spec = client.V1DeploymentSpec(
replicas=3,
template=template,
selector={'matchLabels': {'app': 'nginx'}}
)
# 实例化 deployment 对象
deployment = client.V1Deployment(
api_version='apps/v1',
kind='Deployment',
metadata=client.V1ObjectMeta(name=DEPLOYMENT_NAME),
spec=spec
)
return deployment
def create_deployment(api_instance, deployment):
# 创建 部署
api_response = api_instance.create_namespaced_deployment(
body=deployment,
namespace='default'
)
print("Deployment created. status = '%s'" % str(api_response.status))
def update_deployment(api_instance, deployment):
# 更新 container image
deployment.spec.template.spec.containers[0].image = 'nginx:1.16.0'
# 更新部署
api_response = api_instance.patch_namespaced_deployment(
name=DEPLOYMENT_NAME,
namespace='default',
body=deployment
)
print("Deployment updated. status='%s'" % str(api_response.status))
def delete_deployment(api_instance):
# 删除部署
api_response = api_instance.delete_namespaced_deployment(
name=DEPLOYMENT_NAME,
namespace='default',
body=client.V1DeleteOptions(
propagation_policy='Foreground',
grace_period_seconds=5 # 宽限期 秒数
)
)
print("Deployment deleted. status='%s'" % str(api_response.status))
def main():
config.load_kube_config()
apps_v1 = client.AppsV1Api()
# 使用 client-python api 创建 deployment 部署,同nginx-deployment.yaml
deployment = create_deployment_object()
create_deployment(apps_v1, deployment)
update_deployment(apps_v1, deployment)
delete_deployment(apps_v1)
3. node_labels.py
# -*- coding: utf8 -*-
from pprint import pprint
from kubernetes import client, config
def main():
config.load_kube_config()
api_instance = client.CoreV1Api()
body = {
'metadata': {
'labels': {
'foo': 'bar',
'baz': None
}
}
}
api_response = api_instance.patch_node('minikube', body=body)
pprint(api_response)
if __name__ == '__main__':
main()
4. job_crud.py
# -*- coding: utf8 -*-
from kubernetes import client, config
from os import path
import yaml
JOB_NAME = 'pi'
def create_job_object():
# Configureate Pod template container
container = client.V1Container(
name="pi",
image="perl",
command=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"])
# Create and configurate a spec section
template = client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(labels={"app": "pi"}),
spec=client.V1PodSpec(restart_policy="Never", containers=[container]))
# Create the specification of deployment
spec = client.V1JobSpec(
template=template,
backoff_limit=4)
# Instantiate the job object
job = client.V1Job(
api_version="batch/v1",
kind="Job",
metadata=client.V1ObjectMeta(name=JOB_NAME),
spec=spec)
return job
def create_job(api_instance, job):
api_response = api_instance.create_namespaced_job(
body=job,
namespace="default")
print("Job created. status='%s'" % str(api_response.status))
def update_job(api_instance, job):
job.spec.template.spec.containers[0].image = "perl"
api_response = api_instance.patch_namespaced_job(
name=JOB_NAME,
namespace="default",
body=job)
print("Job updated. status='%s'" % str(api_response.status))
def delete_job(api_instance):
api_response = api_instance.delete_namespaced_job(
name=JOB_NAME,
namespace="default",
body=client.V1DeleteOptions(
propagation_policy='Foreground',
grace_period_seconds=5))
print("Job deleted. status='%s'" % str(api_response.status))
def main():
config.load_kube_config()
batch_v1 = client.BatchV1Api()
jog = create_job_object()
create_job(batch_v1, jog)
update_job(batch_v1, jog)
delete_job(batch_v1)
if __name__ == '__main__':
main()
5. ingress_create.py
# -*- coding: utf8 -*-
"""
ingress 允许外部网络进入集群
"""
from kubernetes import client, config
def create_deployment(apps_v1_api):
# create_deployment()
container = client.V1Container(
name="deployment",
image="gcr.io/google-appengine/fluentd-logger",
image_pull_policy="Never",
ports=[client.V1ContainerPort(container_port=5678)],
)
# Template
template = client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(labels={"app": "deployment"}),
spec=client.V1PodSpec(containers=[container]))
# Spec
spec = client.V1DeploymentSpec(
replicas=1,
template=template)
# Deployment
deployment = client.V1Deployment(
api_version="apps/v1",
kind="Deployment",
metadata=client.V1ObjectMeta(name="deployment"),
spec=spec)
apps_v1_api.create_namespaced_deployment(namespace='default', body=deployment)
def create_service():
core_v1_api = client.CoreV1Api()
body = client.V1Service(
api_version='v1',
kind='Service',
metadata=client.V1ObjectMeta(
name='service-example'
),
spec=client.V1ServiceSpec(
selector={'app': 'deployment'},
ports=[client.V1ServicePort(
port=5678,
target_port=5678
)]
)
)
core_v1_api.create_namespaced_service(namespace='default', body=body)
def create_ingress(networking_v1_beta1_api):
# create_ingress
body = client.V1beta1Ingress(
api_version='networking.k8s.io/v1beta1',
kind='Ingress',
metadata=client.V1ObjectMeta(name='ingress-example', annotations={
'nginx.ingress.kubernetes.io/rewrite-target': '/'
}),
spec=client.V1beta1IngressSpec(
rules=[client.V1beta1IngressRule(
host='example.com',
http=client.V1beta1HTTPIngressRuleValue(
paths=[client.V1beta1HTTPIngressPath(
path='/',
backend=client.V1beta1IngressBackend(
service_port=5678,
service_name='service-example'
)
)]
)
)]
)
)
networking_v1_beta1_api.create_namespaced_ingress(
namespace='default',
body=body
)
def main():
config.load_kube_config()
apps_v1_api = client.AppsV1Api()
networking_v1_beta1_api = client.ExtensionsV1beta1Api() # 未找到 NetworkingV1beta1Api()
create_deployment(apps_v1_api)
create_service()
create_ingress(networking_v1_beta1_api)
pass
if __name__ == '__main__':
main()
更多推荐
已为社区贡献12条内容
所有评论(0)