gcpdiag.queries.kubectl

31def get_config_path():
32  return config.get_cache_dir() + '/gcpdiag-config'

 35class KubectlExecutor:
 36  """Represents a kubectl executor."""
 37
 38  lock: threading.Lock
 39
 40  def __init__(self, cluster: gke.Cluster):
 41    self.cluster = cluster
 42    self.lock = threading.Lock()
 43
 44  def make_kube_config(self) -> bool:
 45    """Add a new kubernetes context for kubectl plugin CLIs."""
 46
 47    cfg: dict = {}
 48    if not os.path.isfile(get_config_path()):
 49      cfg['apiVersion'] = 'v1'
 50      cfg['users'] = [
 51        {
 52          'name': 'gcpdiag',
 53          'user': {
 54            'exec': {
 55              'apiVersion': 'client.authentication.k8s.io/v1beta1',
 56              'command': 'gke-gcloud-auth-plugin',
 57              'installHint': 'x',
 58              'provideClusterInfo': True,
 59            },
 60          },
 61        }
 62      ]
 63      cfg['clusters'] = []
 64      cfg['contexts'] = []
 65    else:
 66      with open(get_config_path(), encoding='UTF-8') as f:
 67        cfg = yaml.safe_load(f)
 68
 69    if self.cluster.endpoint is None:
 70      logging.warning(
 71        'No kubernetes API server endpoint found for cluster %s', self.cluster.short_path
 72      )
 73      return False
 74
 75    kubecontext = 'gcpdiag-ctx-' + self.cluster.name
 76
 77    cfg['clusters'].append(
 78      {
 79        'cluster': {
 80          'certificate-authority-data': self.cluster.cluster_ca_certificate,
 81          'server': 'https://' + self.cluster.endpoint,
 82        },
 83        'name': self.cluster.short_path,
 84      }
 85    )
 86    cfg['contexts'].append(
 87      {
 88        'context': {
 89          'cluster': self.cluster.short_path,
 90          'user': 'gcpdiag',
 91        },
 92        'name': kubecontext,
 93      }
 94    )
 95
 96    self.kubecontext = kubecontext
 97
 98    config_text = yaml.dump(cfg, default_flow_style=False)
 99    with open(get_config_path(), 'w', encoding='UTF-8') as config_file:
100      config_file.write(config_text)
101      config_file.close()
102
103    return True
104
105  def kubectl_execute(self, command_list: list[str]):
106    """Execute a kubectl command.
107
108    Will take a list of strings which contains all the command and parameters to be executed
109    and return the stdout and stderr of the execution.
110    """
111    res = subprocess.run(command_list, check=False, capture_output=True, text=True)
112    return res.stdout, res.stderr

115def verify_auth(executor: KubectlExecutor) -> bool:
116  """Verify the authentication for kubernetes by running kubeclt cluster-info.
117
118  Will raise a warning and return False if authentication failed.
119  """
120  _, stderr = executor.kubectl_execute(
121    [
122      'kubectl',
123      'cluster-info',
124      '--kubeconfig',
125      get_config_path(),
126      '--context',
127      executor.kubecontext,
128    ]
129  )
130  if stderr:
131    logging.warning(
132      'Failed to authenticate kubectl for cluster %s: %s',
133      executor.cluster.short_path,
134      stderr.strip('\n'),
135    )
136    return False
137  return True

140def check_gke_ingress(executor: KubectlExecutor):
141  return executor.kubectl_execute(
142    [
143      'kubectl',
144      'check-gke-ingress',
145      '--kubeconfig',
146      get_config_path(),
147      '--context',
148      executor.kubecontext,
149    ]
150  )

153@functools.lru_cache()
154def get_kubectl_executor(c: gke.Cluster):
155  """Create a kubectl_executor for a GKE cluster."""
156  executor = KubectlExecutor(cluster=c)
157  with executor.lock:
158    if not executor.make_kube_config():
159      return None
160  try:
161    if not verify_auth(executor):
162      logging.warning('Authentication failed for cluster %s', c.short_path)
163      return None
164  except FileNotFoundError as err:
165    logging.warning('Can not inspect Kubernetes resources: %s: %s', type(err).__name__, err)
166    return None
167  return executor

170def clean_up():
171  """Delete the kubeconfig file generated for gcpdiag."""
172  try:
173    os.remove(get_config_path())
174  except OSError as err:
175    logging.debug(
176      'Error cleaning up kubeconfig file used by gcpdiag: %s: %s', type(err).__name__, err
177    )

180def error_message(rule_name, kind, namespace, name, message) -> str:
181  return f'Check rule {rule_name} on {kind} {namespace}/{name} failed: {message}\n'