本頁面包含基於各種編程語言使用 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