n2vc/kubectl.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import logging
  16
  17 from kubernetes import client, config
  18 from kubernetes.client.rest import ApiException
  19
  20
  21 CORE_CLIENT = "core_v1"
  22 STORAGE_CLIENT = "storage_v1"
  23 RBAC_CLIENT = "rbac_v1"
  24
  25
  26 class Kubectl:
  27     def __init__(self, config_file=None):
  28         config.load_kube_config(config_file=config_file)
  29         self._clients = {
  30             "core_v1": client.CoreV1Api(),
  31             "storage_v1": client.StorageV1Api(),
  32             "rbac_v1": client.RbacAuthorizationV1Api(),
  33         }
  34         self._configuration = config.kube_config.Configuration()
  35         self.logger = logging.getLogger("Kubectl")
  36
  37     @property
  38     def configuration(self):
  39         return self._configuration
  40
  41     @property
  42     def clients(self):
  43         return self._clients
  44
  45     def get_services(self, field_selector=None, label_selector=None):
  46         kwargs = {}
  47         if field_selector:
  48             kwargs["field_selector"] = field_selector
  49         if label_selector:
  50             kwargs["label_selector"] = label_selector
  51         try:
  52             result = self.clients[CORE_CLIENT].list_service_for_all_namespaces(**kwargs)
  53             return [
  54                 {
  55                     "name": i.metadata.name,
  56                     "cluster_ip": i.spec.cluster_ip,
  57                     "type": i.spec.type,
  58                     "ports": [
  59                         {
  60                             "name": p.name,
  61                             "node_port": p.node_port,
  62                             "port": p.port,
  63                             "protocol": p.protocol,
  64                             "target_port": p.target_port,
  65                         }
  66                         for p in i.spec.ports
  67                     ]
  68                     if i.spec.ports
  69                     else [],
  70                     "external_ip": [i.ip for i in i.status.load_balancer.ingress]
  71                     if i.status.load_balancer.ingress
  72                     else None,
  73                 }
  74                 for i in result.items
  75             ]
  76         except ApiException as e:
  77             self.logger.error("Error calling get services: {}".format(e))
  78             raise e
  79
  80     def get_default_storage_class(self) -> str:
  81         """
  82         Default storage class
  83
  84         :return:    Returns the default storage class name, if exists.
  85                     If not, it returns the first storage class.
  86                     If there are not storage classes, returns None
  87         """
  88         storage_classes = self.clients[STORAGE_CLIENT].list_storage_class()
  89         selected_sc = None
  90         default_sc_annotations = {
  91             "storageclass.kubernetes.io/is-default-class": "true",
  92             # Older clusters still use the beta annotation.
  93             "storageclass.beta.kubernetes.io/is-default-class": "true",
  94         }
  95         for sc in storage_classes.items:
  96             if not selected_sc:
  97                 # Select the first storage class in case there is no a default-class
  98                 selected_sc = sc.metadata.name
  99             annotations = sc.metadata.annotations
 100             if any(
 101                 k in annotations and annotations[k] == v
 102                 for k, v in default_sc_annotations.items()
 103             ):
 104                 # Default storage
 105                 selected_sc = sc.metadata.name
 106                 break
 107         return selected_sc