gcpdiag.queries.gcf

29class CloudFunction(models.Resource):
30  """Represents a GCF instance."""
31
32  _resource_data: dict
33
34  def __init__(self, project_id, resource_data):
35    super().__init__(project_id=project_id)
36    self._resource_data = resource_data
37    self._metadata_dict = None
38
39  @property
40  def name(self) -> str:
41    m = re.search(r'/functions/([^/]+)$', self._resource_data['name'])
42    if not m:
43      raise RuntimeError("can't determine name of cloudfunction %s" % (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.debug('fetching list of GCF functions in project %s', context.project_id)
 76  query = (
 77    gcf_api.projects()
 78    .locations()
 79    .functions()
 80    .list(parent=f'projects/{context.project_id}/locations/-')
 81  )
 82  try:
 83    resp = query.execute(num_retries=config.API_RETRIES)
 84    if 'functions' not in resp:
 85      return cloudfunctions
 86    for f in resp['functions']:
 87      # verify that we have some minimal data that we expect
 88      if 'name' not in f or 'runtime' not in f:
 89        raise RuntimeError('missing data in projects.locations.functions.list response')
 90      # projects/*/locations/*/functions/*
 91      result = re.match(r'projects/[^/]+/(?:locations)/([^/]+)/functions/([^/]+)', f['name'])
 92      if not result:
 93        logging.error('invalid cloud functions data: %s', f['name'])
 94        continue
 95
 96      location = result.group(1)
 97      labels = f.get('labels', {})
 98      name = f.get('name', '')
 99      if not context.match_project_resource(location=location, labels=labels, resource=name):
100        continue
101
102      cloudfunctions[f['name']] = CloudFunction(project_id=context.project_id, resource_data=f)
103  except googleapiclient.errors.HttpError as err:
104    raise utils.GcpApiError(err) from err
105  return cloudfunctions