本页面包含基于各种编程语言使用 Kubernetes API 的客户端库概述。
在使用 Kubernetes REST API 编写应用程序时， 您并不需要自己实现 API 调用和 “请求/响应” 类型。 您可以根据自己的编程语言需要选择使用合适的客户端库。
客户端库通常为您处理诸如身份验证之类的常见任务。 如果 API 客户端在 Kubernetes 集群中运行，大多数客户端库可以发现并使用 Kubernetes 服务帐户进行身份验证， 或者能够理解 kubeconfig 文件 格式来读取凭据和 API 服务器地址。
官方支持的 Kubernetes 客户端库
以下客户端库由 Kubernetes SIG API Machinery 正式维护。
| 语言 | 客户端库 | 样例程序 |
|---|---|---|
| Go | github.com/kubernetes/client-go/ | 浏览 |
| Python | github.com/kubernetes-client/python/ | 浏览 |
| Java | github.com/kubernetes-client/java | 浏览 |
| dotnet | github.com/kubernetes-client/csharp | 浏览 |
| JavaScript | github.com/kubernetes-client/javascript | 浏览 |
| Haskell | github.com/kubernetes-client/haskell | 浏览 |
社区维护的客户端库
以下 Kubernetes API 客户端库是由社区，而非 Kubernetes 团队支持、维护的。
def get_k8s_services(context=None):
values = {}
services = []
try:
config.load_kube_config()
v1 = client.CoreV1Api()
ret = v1.list_namespaced_service(namespace='testns')
for i in ret.items:
port = []
for info in i.spec.ports:
port.append(int(info.port))
print(info)
services.append(i.metadata.name)
values[i.metadata.name] = port[0]
except Exception as e:
print(e)
Python kubernetes.client.CoreV1Api() Examples
The following are 30 code examples for showing how to use kubernetes.client.CoreV1Api(). These examples are extracted from open source projects. You can vote up the ones you like or vote down the ones you don't like, and go to the original project or source file by following the links above each example.
You may want to check out the right sidebar which shows the related API usage.
You may also want to check out all available functions/classes of the module kubernetes.client , or try the search function 
.
Example 1
| Project: python Author: kubernetes-client File: pod_config_list.py License: Apache License 2.0 | 7 votes |   |
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) # Configs can be set in Configuration class directly or using helper # utility 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))
Example 2
| Project: mars Author: mars-project File: client.py License: Apache License 2.0 | 6 votes | |
def stop(self, wait=False, timeout=0): from kubernetes.client import CoreV1Api api = CoreV1Api(self._api_client) api.delete_namespace(self._namespace) if wait: start_time = time.time() while True: try: api.read_namespace(self._namespace) except K8SApiException as ex: if ex.status != 404: # pragma: no cover raise break else: time.sleep(1) if timeout and time.time() - start_time > timeout: # pragma: no cover raise TimeoutError
Example 3
| Project: mars Author: mars-project File: core.py License: Apache License 2.0 | 6 votes | |
def __init__(self, k8s_config=None, k8s_namespace=None, label_selector=None): from kubernetes import config, client from gevent.threadpool import ThreadPool if k8s_config is not None: self._k8s_config = k8s_config elif os.environ.get('KUBE_API_ADDRESS'): self._k8s_config = client.Configuration() self._k8s_config.host = os.environ['KUBE_API_ADDRESS'] else: self._k8s_config = config.load_incluster_config() self._k8s_namespace = k8s_namespace or os.environ.get('MARS_K8S_POD_NAMESPACE') or 'default' self._label_selector = label_selector self._client = client.CoreV1Api(client.ApiClient(self._k8s_config)) self._pool = ThreadPool(1) self._pod_to_ep = None
Example 4
| Project: chaostoolkit-kubernetes Author: chaostoolkit File: actions.py License: Apache License 2.0 | 6 votes | |
def delete_pods(name: str, ns: str = "default", label_selector: str = "name in ({name})", secrets: Secrets = None): """ Delete pods by `name` in the namespace `ns`. The pods are deleted without a graceful period to trigger an abrupt termination. The selected resources are matched by the given `label_selector`. """ label_selector = label_selector.format(name=name) api = create_k8s_api_client(secrets) v1 = client.CoreV1Api(api) if label_selector: ret = v1.list_namespaced_pod(ns, label_selector=label_selector) else: ret = v1.list_namespaced_pod(ns) logger.debug("Found {d} pods named '{n}'".format( d=len(ret.items), n=name)) body = client.V1DeleteOptions() for p in ret.items: v1.delete_namespaced_pod(p.metadata.name, ns, body=body)
Example 5
| Project: chaostoolkit-kubernetes Author: chaostoolkit File: actions.py License: Apache License 2.0 | 6 votes | |
def create_service_endpoint(spec_path: str, ns: str = "default", secrets: Secrets = None): """ Create a service endpoint described by the service config, which must be the path to the JSON or YAML representation of the service. """ api = create_k8s_api_client(secrets) with open(spec_path) as f: p, ext = os.path.splitext(spec_path) if ext == '.json': service = json.loads(f.read()) elif ext in ['.yml', '.yaml']: service = yaml.safe_load(f.read()) else: raise ActivityFailed( "cannot process {path}".format(path=spec_path)) v1 = client.CoreV1Api(api) v1.create_namespaced_service(ns, body=service)
Example 6
| Project: chaostoolkit-kubernetes Author: chaostoolkit File: probes.py License: Apache License 2.0 | 6 votes | |
def service_is_initialized(name: str, ns: str = "default", label_selector: str = "name in ({name})", secrets: Secrets = None): """ Lookup a service endpoint by its name and raises :exc:`FailedProbe` when the service was not found or not initialized. """ label_selector = label_selector.format(name=name) api = create_k8s_api_client(secrets) v1 = client.CoreV1Api(api) if label_selector: ret = v1.list_namespaced_service(ns, label_selector=label_selector) else: ret = v1.list_namespaced_service(ns) logger.debug("Found {d} service(s) named '{n}' ins ns '{s}'".format( d=len(ret.items), n=name, s=ns)) if not ret.items: raise ActivityFailed( "service '{name}' is not initialized".format(name=name)) return True
Example 7
| Project: eks-rolling-update Author: hellofresh File: k8s.py License: Apache License 2.0 | 6 votes | |
def get_k8s_nodes(exclude_node_label_key=app_config["EXCLUDE_NODE_LABEL_KEY"]): """ Returns a list of kubernetes nodes """ try: config.load_incluster_config() except config.ConfigException: try: config.load_kube_config() except config.ConfigException: raise Exception("Could not configure kubernetes python client") k8s_api = client.CoreV1Api() logger.info("Getting k8s nodes...") response = k8s_api.list_node() if exclude_node_label_key is not None: nodes = [] for node in response.items: if exclude_node_label_key not in node.metadata.labels: nodes.append(node) response.items = nodes logger.info("Current k8s node count is {}".format(len(response.items))) return response.items
Example 8
| Project: eks-rolling-update Author: hellofresh File: k8s.py License: Apache License 2.0 | 6 votes | |
def delete_node(node_name): """ Deletes a kubernetes node from the cluster """ try: config.load_incluster_config() except config.ConfigException: try: config.load_kube_config() except config.ConfigException: raise Exception("Could not configure kubernetes python client") configuration = client.Configuration() # create an instance of the API class k8s_api = client.CoreV1Api(client.ApiClient(configuration)) logger.info("Deleting k8s node {}...".format(node_name)) try: if not app_config['DRY_RUN']: k8s_api.delete_node(node_name) else: k8s_api.delete_node(node_name, dry_run="true") logger.info("Node deleted") except ApiException as e: logger.info("Exception when calling CoreV1Api->delete_node: {}".format(e))
Example 9
| Project: airflow Author: apache File: pod_launcher.py License: Apache License 2.0 | 6 votes | |
def __init__(self, kube_client: client.CoreV1Api = None, in_cluster: bool = True, cluster_context: Optional[str] = None, extract_xcom: bool = False): """ Creates the launcher. :param kube_client: kubernetes client :param in_cluster: whether we are in cluster :param cluster_context: context of the cluster :param extract_xcom: whether we should extract xcom """ super().__init__() self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context) self._watch = watch.Watch() self.extract_xcom = extract_xcom
Example 10
| Project: airflow Author: apache File: kubernetes_executor.py License: Apache License 2.0 | 6 votes | |
def run(self) -> None: """Performs watching""" kube_client: client.CoreV1Api = get_kube_client() if not self.worker_uuid: raise AirflowException(NOT_STARTED_MESSAGE) while True: try: self.resource_version = self._run(kube_client, self.resource_version, self.worker_uuid, self.kube_config) except ReadTimeoutError: self.log.warning("There was a timeout error accessing the Kube API. " "Retrying request.", exc_info=True) time.sleep(1) except Exception: self.log.exception('Unknown error in KubernetesJobWatcher. Failing') raise else: self.log.warning('Watch died gracefully, starting back up with: ' 'last resource_version: %s', self.resource_version)
Example 11
| Project: airflow Author: apache File: kubernetes_executor.py License: Apache License 2.0 | 6 votes | |
def start(self) -> None: """Starts the executor""" self.log.info('Start Kubernetes executor') self.worker_uuid = KubeWorkerIdentifier.get_or_create_current_kube_worker_uuid() if not self.worker_uuid: raise AirflowException("Could not get worker uuid") self.log.debug('Start with worker_uuid: %s', self.worker_uuid) # always need to reset resource version since we don't know # when we last started, note for behavior below # https://github.com/kubernetes-client/python/blob/master/kubernetes/docs # /CoreV1Api.md#list_namespaced_pod KubeResourceVersion.reset_resource_version() self.kube_client = get_kube_client() self.kube_scheduler = AirflowKubernetesScheduler( self.kube_config, self.task_queue, self.result_queue, self.kube_client, self.worker_uuid ) self._inject_secrets() self.clear_not_launched_queued_tasks()
Example 12
| Project: python Author: kubernetes-client File: node_labels.py License: Apache License 2.0 | 6 votes | |
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) pprint(api_response)
Example 13
| Project: python Author: kubernetes-client File: ingress_create.py License: Apache License 2.0 | 6 votes | |
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 )] ) ) # Creation of the Deployment in specified namespace # (Can replace "default" with a namespace you may have created) core_v1_api.create_namespaced_service(namespace="default", body=body)
Example 14
| Project: python Author: kubernetes-client File: pick_kube_config_context.py License: Apache License 2.0 | 6 votes | |
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) # Configs can be set in Configuration class directly or using helper # utility 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))
Example 15
| Project: python Author: kubernetes-client File: test_utils.py License: Apache License 2.0 | 6 votes | |
def test_create_deployment_non_default_namespace_from_yaml(self): """ Should be able to create a namespace "dep", and then create a deployment in the just-created namespace. """ k8s_client = client.ApiClient(configuration=self.config) utils.create_from_yaml( k8s_client, self.path_prefix + "dep-namespace.yaml") utils.create_from_yaml( k8s_client, self.path_prefix + "dep-deployment.yaml") core_api = client.CoreV1Api(k8s_client) ext_api = client.AppsV1Api(k8s_client) nmsp = core_api.read_namespace(name="dep") self.assertIsNotNone(nmsp) dep = ext_api.read_namespaced_deployment(name="nginx-deployment", namespace="dep") self.assertIsNotNone(dep) ext_api.delete_namespaced_deployment( name="nginx-deployment", namespace="dep", body={}) core_api.delete_namespace(name="dep", body={})
Example 16
| Project: python Author: kubernetes-client File: test_utils.py License: Apache License 2.0 | 6 votes | |
def test_create_general_list_from_yaml(self): """ Should be able to create a service and a deployment from a kind: List yaml file """ k8s_client = client.api_client.ApiClient(configuration=self.config) utils.create_from_yaml( k8s_client, self.path_prefix + "list.yaml") core_api = client.CoreV1Api(k8s_client) ext_api = client.AppsV1Api(k8s_client) svc = core_api.read_namespaced_service(name="list-service-test", namespace="default") self.assertIsNotNone(svc) dep = ext_api.read_namespaced_deployment(name="list-deployment-test", namespace="default") self.assertIsNotNone(dep) core_api.delete_namespaced_service(name="list-service-test", namespace="default", body={}) ext_api.delete_namespaced_deployment(name="list-deployment-test", namespace="default", body={})
Example 17
| Project: python Author: kubernetes-client File: test_utils.py License: Apache License 2.0 | 6 votes | |
def test_create_from_multi_resource_yaml(self): """ Should be able to create a service and a replication controller from a multi-resource yaml file """ k8s_client = client.api_client.ApiClient(configuration=self.config) utils.create_from_yaml( k8s_client, self.path_prefix + "multi-resource.yaml") core_api = client.CoreV1Api(k8s_client) svc = core_api.read_namespaced_service(name="mock", namespace="default") self.assertIsNotNone(svc) ctr = core_api.read_namespaced_replication_controller( name="mock", namespace="default") self.assertIsNotNone(ctr) core_api.delete_namespaced_replication_controller( name="mock", namespace="default", body={}) core_api.delete_namespaced_service(name="mock", namespace="default", body={})
Example 18
| Project: python Author: kubernetes-client File: test_utils.py License: Apache License 2.0 | 6 votes | |
def test_create_from_list_in_multi_resource_yaml(self): """ Should be able to create the items in the PodList and a deployment specified in the multi-resource file """ k8s_client = client.api_client.ApiClient(configuration=self.config) utils.create_from_yaml( k8s_client, self.path_prefix + "multi-resource-with-list.yaml") core_api = client.CoreV1Api(k8s_client) app_api = client.AppsV1Api(k8s_client) pod_0 = core_api.read_namespaced_pod( name="mock-pod-0", namespace="default") self.assertIsNotNone(pod_0) pod_1 = core_api.read_namespaced_pod( name="mock-pod-1", namespace="default") self.assertIsNotNone(pod_1) dep = app_api.read_namespaced_deployment( name="mock", namespace="default") self.assertIsNotNone(dep) core_api.delete_namespaced_pod( name="mock-pod-0", namespace="default", body={}) core_api.delete_namespaced_pod( name="mock-pod-1", namespace="default", body={}) app_api.delete_namespaced_deployment( name="mock", namespace="default", body={})
Example 19
| Project: wrapanapi Author: RedHatQE File: rhopenshift.py License: MIT License | 6 votes | |
def _connect(self): url = '{proto}://{host}:{port}'.format(proto=self.protocol, host=self.hostname, port=self.port) token = 'Bearer {token}'.format(token=self.token) config = ociclient.Configuration() config.host = url config.verify_ssl = self.verify_ssl config.debug = self.debug config.api_key['authorization'] = token self.ociclient = ociclient self.kclient = kubeclient self.oapi_client = ociclient.ApiClient(config=config) self.kapi_client = kubeclient.ApiClient(config=config) self.o_api = ociclient.OapiApi(api_client=self.oapi_client) self.k_api = kubeclient.CoreV1Api(api_client=self.kapi_client) self.security_api = self.ociclient.SecurityOpenshiftIoV1Api(api_client=self.oapi_client) self.batch_api = self.kclient.BatchV1Api(api_client=self.kapi_client) # for job api
Example 20
| Project: paasta Author: Yelp File: kubernetes_tools.py License: Apache License 2.0 | 6 votes | |
def __init__(self) -> None: kube_config.load_kube_config( config_file=os.environ.get("KUBECONFIG", KUBE_CONFIG_PATH), context=os.environ.get("KUBECONTEXT"), ) models.V1beta1PodDisruptionBudgetStatus.disrupted_pods = property( fget=lambda *args, **kwargs: models.V1beta1PodDisruptionBudgetStatus.disrupted_pods( *args, **kwargs ), fset=_set_disrupted_pods, ) self.deployments = kube_client.AppsV1Api() self.core = kube_client.CoreV1Api() self.policy = kube_client.PolicyV1beta1Api() self.apiextensions = kube_client.ApiextensionsV1beta1Api() self.custom = kube_client.CustomObjectsApi() self.autoscaling = kube_client.AutoscalingV2beta1Api() self.request = kube_client.ApiClient().request
Example 21
| Project: sparrow Author: wylok File: ajax_api.py License: GNU General Public License v3.0 | 6 votes | |
def get_k8s_services(context=None): values = defaultdict() services = [] try: config.load_kube_config(config_file,context) v1 = client.CoreV1Api() ret = v1.list_namespaced_service(namespace='default') for i in ret.items: port = [] for info in i.spec.ports: port.append(int(info.port)) services.append(i.metadata.name) values[i.metadata.name] = port[0] except Exception as e: logging.error(e) finally: return jsonify({'values': values,'services':services})
Example 22
| Project: mars Author: mars-project File: client.py License: Apache License 2.0 | 5 votes | |
def _create_kube_service(self): """ :type kube_api: kubernetes.client.CoreV1Api """ if self._service_type != 'NodePort': # pragma: no cover raise NotImplementedError('Service type %s not supported' % self._service_type) service_config = ServiceConfig( 'marsservice', service_type='NodePort', port=self._web_service_port, selector=dict(name=MarsWebsConfig.rc_name), ) self._core_api.create_namespaced_service(self._namespace, service_config.build()) time.sleep(1) svc_data = self._core_api.read_namespaced_service('marsservice', self._namespace).to_dict() port = svc_data['spec']['ports'][0]['node_port'] web_pods = self._core_api.list_namespaced_pod( self._namespace, label_selector='name=' + MarsWebsConfig.rc_name).to_dict() host_ip = random.choice(web_pods['items'])['status']['host_ip'] # docker desktop use a VM to hold docker processes, hence # we need to use API address instead desktop_nodes = self._core_api.list_node(field_selector='metadata.name=docker-desktop').to_dict() if desktop_nodes['items']: # pragma: no cover addresses = set( addr_info['address'] for addr_info in desktop_nodes['items'][0].get('status', {}).get('addresses', ()) ) if host_ip in addresses: host_ip = urlparse(self._core_api.api_client.configuration.host).netloc.split(':', 1