python (k8s) 1. config
1. remote_cluster.py# -*- coding: utf8 -*-"""和远程 kube 集群通信认证需要创建 barer tokenhttps://kubernetes.io/docs/tasks/access-application-cluster/access-cluster/"""from kubernetes import client, con...
·
1. remote_cluster.py
# -*- coding: utf8 -*-
"""
和远程 kube 集群通信
认证需要创建 barer token
https://kubernetes.io/docs/tasks/access-application-cluster/access-cluster/
"""
from kubernetes import client, config
def main():
aToken = "<token>"
# 创建一个配置对象
aConfiguration = client.Configuration()
# 对k8s集群指定一个终端
aConfiguration.host = "http://xxx.xxx.xxx.xxx:443"
# 为了简单,本例不使用ssl认证
aConfiguration.verify_ssl = False
"""
如果有需要:
configuration.verify_ssl = True
configuration.ssl_ca_cert = 'certificate' # 证书路径
"""
aConfiguration.api_key = {"authorization": "Bearer" + aToken}
# 使用配置创建 api client
aApiClient = client.ApiClient(aConfiguration)
# 调用
v1 = client.CoreV1Api(aApiClient)
print("Listing pods with their Ips:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
print("%s\t%s\t%s" % (
i.status.pod_ip,
i.metadata.namespace,
i.metadata.name
))
if __name__ == '__main__':
main()
2. cluster_config.py
# -*- coding: utf8 -*-
"""
从集群外部导入k8s的配置
"""
from kubernetes import client, config
def main():
config.load_kube_config()
v1 = client.CoreV1Api()
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
print("%s\t%s\t%s" %
(i.status.pod_ip,
i.metadata.namespace,
i.metadata.name))
if __name__ == '__main__':
main()
3. pod_config.py
# -*- coding: utf8 -*-
from kubernetes import client, config
from kubernetes.client import configuration
from pick import pick
def main():
contexts, active_context = config.list_kube_config_contexts()
if not contexts:
print("Cannot find any context in kube-config file.")
return
contexts = [context['name'] for context in contexts]
active_index = contexts.index(active_context['name'])
option, _ = pick(contexts, title='Pick the context to load', default_index=active_index)
config.load_kube_config(context=option)
print("Active host is %s" % configuration.Configuration().host)
v1 = client.CoreV1Api()
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for item in ret.items:
print("%s\t%s\t%s" %
(item.status.pod_ip,
item.metadata.namespace,
item.metadata.name))
if __name__ == '__main__':
main()
4. multiple_cluster.py
# -*- coding: utf8 -*-
from kubernetes import client, config
from kubernetes.client import configuration
from pick import pick
def main():
contexts, active_context = config.list_kube_config_contexts()
if not contexts:
print("Cannot find any context in kube-config file.")
return
contexts = [context['name'] for context in contexts]
active_index = contexts.index(active_context['name'])
cluster1, first_index = pick(contexts, title='Pick the first context', default_index=active_index)
cluster2, _ = pick(contexts, title='Pick the second context', default_index=first_index)
client1 = client.CoreV1Api(
api_client=config.new_client_from_config(context=cluster1)
)
client2 = client.CoreV1Api(
api_client=config.new_client_from_config(context=cluster2)
)
print('\nList of pods on %s:' % cluster1)
for i in client1.list_pod_for_all_namespaces().items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print("\n\nList of pods on %s:" % cluster2)
for i in client2.list_pod_for_all_namespaces().items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
if __name__ == '__main__':
main()
5. api_discovery.py
# -*- coding: utf8 -*-
# kubectl api-versions
from kubernetes import client, config
def main():
# Configs 可以从配置类中直接设置,如果没有参数提供,将从默认地址加载
config.load_kube_config()
print('支持的Apis (*标号的是更好的的版本')
print('%-40s %s' % ('core', ','.join(client.CoreApi().get_api_versions().versions))) # .versions
for api in client.ApisApi().get_api_versions().groups: # groups
versions = []
for v in api.versions: # .versions
name = ''
if v.version == api.preferred_version.version and len(api.versons) > 1: # .preferred_version.version
name += '*'
name += v.version
versions.append(name)
print('%-40s %s' % (api.name, ','.join(versions)))
if __name__ == '__main__':
main()
更多推荐
已为社区贡献12条内容
所有评论(0)