gcpdiag.queries.gcf

29class CloudFunction(models.Resource):
30  """Represents a GCF instance."""
31  _resource_data: dict
32
33  def __init__(self, project_id, resource_data):
34    super().__init__(project_id=project_id)
35    self._resource_data = resource_data
36    self._metadata_dict = None
37
38  @property
39  def name(self) -> str:
40    m = re.search(r'/functions/([^/]+)$', self._resource_data['name'])
41    if not m:
42      raise RuntimeError('can\'t determine name of cloudfunction %s' %
43                         (self._resource_data['name']))
44    return m.group(1)
45
46  @property
47  def description(self) -> str:
48    return self._resource_data['description']
49
50  @property
51  def full_path(self) -> str:
52    return self._resource_data['name']
53
54  @property
55  def short_path(self) -> str:
56    path = self.project_id + '/' + self.name
57    return path
58
59  @property
60  def runtime(self) -> str:
61    return self._resource_data['runtime']
62
63  @property
64  def memory(self) -> str:
65    return self._resource_data['availableMemoryMb']

 68@caching.cached_api_call
 69def get_cloudfunctions(context: models.Context) -> Mapping[str, CloudFunction]:
 70  """Get a list of CloudFunctions matching the given context, indexed by CloudFunction name."""
 71  cloudfunctions: Dict[str, CloudFunction] = {}
 72  if not apis.is_enabled(context.project_id, 'cloudfunctions'):
 73    return cloudfunctions
 74  gcf_api = apis.get_api('cloudfunctions', 'v1', context.project_id)
 75  logging.info('fetching list of GCF functions in project %s',
 76               context.project_id)
 77  query = gcf_api.projects().locations().functions().list(
 78      parent=f'projects/{context.project_id}/locations/-')
 79  try:
 80    resp = query.execute(num_retries=config.API_RETRIES)
 81    if 'functions' not in resp:
 82      return cloudfunctions
 83    for f in resp['functions']:
 84      # verify that we have some minimal data that we expect
 85      if 'name' not in f or 'runtime' not in f:
 86        raise RuntimeError(
 87            'missing data in projects.locations.functions.list response')
 88      # projects/*/locations/*/functions/*
 89      result = re.match(
 90          r'projects/[^/]+/(?:locations)/([^/]+)/functions/([^/]+)', f['name'])
 91      if not result:
 92        logging.error('invalid cloud functions data: %s', f['name'])
 93        continue
 94
 95      location = result.group(1)
 96      labels = f.get('labels', {})
 97      name = f.get('name', '')
 98      if not context.match_project_resource(
 99          location=location, labels=labels, resource=name):
100        continue
101
102      cloudfunctions[f['name']] = CloudFunction(project_id=context.project_id,
103                                                resource_data=f)
104  except googleapiclient.errors.HttpError as err:
105    raise utils.GcpApiError(err) from err
106  return cloudfunctions