gcpdiag.queries.managedkafka

26class Cluster(models.Resource):
27  """Represents a Managed Service for Apache Kafka Cluster."""
28
29  def __init__(self, project_id: str, resource_data: dict):
30    super().__init__(project_id=project_id)
31    self._resource_data = resource_data
32
33  @property
34  def name(self) -> str:
35    return self._resource_data.get('name', '').split('/')[-1]
36
37  @property
38  def full_path(self) -> str:
39    return self._resource_data.get('name', '')
40
41  @property
42  def short_path(self) -> str:
43    return '/'.join(self.full_path.split('/')[-4:])
44
45  @property
46  def state(self) -> str:
47    return self._resource_data.get('state', 'STATE_UNSPECIFIED')
48
49  @property
50  def location(self) -> str:
51    return self.full_path.split('/')[3] if len(self.full_path.split('/')) > 3 else ''
52
53  @property
54  def broker_details(self) -> List[dict]:
55    """Returns details of each broker in the cluster (only populated in FULL view)."""
56    return self._resource_data.get('brokerDetails', [])

59class Topic(models.Resource):
60  """Represents a Managed Service for Apache Kafka Topic."""
61
62  def __init__(self, project_id: str, resource_data: dict):
63    super().__init__(project_id=project_id)
64    self._resource_data = resource_data
65
66  @property
67  def name(self) -> str:
68    return self._resource_data.get('name', '').split('/')[-1]
69
70  @property
71  def full_path(self) -> str:
72    return self._resource_data.get('name', '')
73
74  @property
75  def short_path(self) -> str:
76    # projects/{project_id}/locations/{location_id}/clusters/{cluster_id}/topics/{topic_id}
77    return '/'.join(self.full_path.split('/')[-6:])
78
79  @property
80  def partition_count(self) -> int:
81    return int(self._resource_data.get('partitionCount', 0))
82
83  @property
84  def replication_factor(self) -> int:
85    return int(self._resource_data.get('replicationFactor', 0))
86
87  @property
88  def configs(self) -> Dict[str, str]:
89    return self._resource_data.get('configs', {})
90
91  @property
92  def is_internal(self) -> bool:
93    """Returns True if the topic is a system/internal topic."""
94    return self.name.startswith('__')

 97class ConsumerGroup(models.Resource):
 98  """Represents a Managed Service for Apache Kafka Consumer Group."""
 99
100  def __init__(self, project_id: str, resource_data: dict):
101    super().__init__(project_id=project_id)
102    self._resource_data = resource_data
103
104  @property
105  def name(self) -> str:
106    return self._resource_data.get('name', '').split('/')[-1]
107
108  @property
109  def full_path(self) -> str:
110    return self._resource_data.get('name', '')
111
112  @property
113  def short_path(self) -> str:
114    # projects/{project_id}/locations/{location_id}/clusters/{cluster_id}/consumerGroups/{group_id}
115    return '/'.join(self.full_path.split('/')[-6:])
116
117  @property
118  def state(self) -> str:
119    return self._resource_data.get('state', 'STATE_UNSPECIFIED')

122class Acl(models.Resource):
123  """Represents a Managed Service for Apache Kafka ACL."""
124
125  def __init__(self, project_id: str, resource_data: dict):
126    super().__init__(project_id=project_id)
127    self._resource_data = resource_data
128
129  @property
130  def name(self) -> str:
131    full = self._resource_data.get('name', '')
132    return full.split('/acls/')[-1] if '/acls/' in full else full.split('/')[-1]
133
134  @property
135  def full_path(self) -> str:
136    return self._resource_data.get('name', '')
137
138  @property
139  def short_path(self) -> str:
140    # projects/{project_id}/locations/{location_id}/clusters/{cluster_id}/acls/{acl_id}
141    return '/'.join(self.full_path.split('/')[-6:])
142
143  @property
144  def acl_entries(self) -> List[dict]:
145    return self._resource_data.get('aclEntries', [])

184@caching.cached_api_call
185def get_clusters(context: models.Context) -> Dict[str, Cluster]:
186  """Get a list of Kafka Clusters from the given GCP project."""
187  clusters: Dict[str, Cluster] = {}
188  if not apis.is_enabled(context.project_id, 'managedkafka'):
189    return clusters
190  kafka_api = apis.get_api('managedkafka', 'v1', context.project_id)
191  locations_to_scan = _get_locations_to_scan(context, kafka_api)
192  for loc_id in locations_to_scan:
193    try:
194      parent_path = f'projects/{context.project_id}/locations/{loc_id}'
195      request = kafka_api.projects().locations().clusters().list(parent=parent_path)
196      for c in apis_utils.list_all(
197        request=request,
198        next_function=kafka_api.projects().locations().clusters().list_next,
199        response_keyword='clusters',
200      ):
201        if not context.match_project_resource(resource=c.get('name', '')):
202          continue
203        cluster = Cluster(project_id=context.project_id, resource_data=c)
204        clusters[cluster.full_path] = cluster
205    except utils.GcpApiError as err:
206      logging.warning('Could not list Kafka clusters for location %s: %s', loc_id, err)
207      continue
208  return clusters

211@caching.cached_api_call
212def get_cluster(project_id: str, location: str, cluster_name: str) -> Optional[Cluster]:
213  """Retrieve a single Managed Kafka cluster by name and location with full details."""
214  if not apis.is_enabled(project_id, 'managedkafka'):
215    return None
216  kafka_api = apis.get_api('managedkafka', 'v1', project_id)
217  logging.debug('Fetching Managed Kafka cluster: %s', cluster_name)
218  request = (
219    kafka_api.projects()
220    .locations()
221    .clusters()
222    .get(
223      name=f'projects/{project_id}/locations/{location}/clusters/{cluster_name}',
224      view='CLUSTER_VIEW_FULL',
225    )
226  )
227  try:
228    resp = request.execute(num_retries=config.API_RETRIES)
229    return Cluster(project_id=project_id, resource_data=resp)
230  except googleapiclient.errors.HttpError as err:
231    if err.resp.status == 404:
232      return None
233    raise utils.GcpApiError(err) from err

236@caching.cached_api_call
237def get_topics(
238  context: models.Context, cluster_name: str, location: Optional[str] = None
239) -> Dict[str, Topic]:
240  """Get Kafka Topics for a specific cluster."""
241  topics: Dict[str, Topic] = {}
242  if not apis.is_enabled(context.project_id, 'managedkafka'):
243    return topics
244
245  kafka_api = apis.get_api('managedkafka', 'v1', context.project_id)
246  parent_path = _resolve_cluster_path(context, cluster_name, location)
247
248  try:
249    request = kafka_api.projects().locations().clusters().topics().list(parent=parent_path)
250    for t in apis_utils.list_all(
251      request=request,
252      next_function=kafka_api.projects().locations().clusters().topics().list_next,
253      response_keyword='topics',
254    ):
255      if not context.match_project_resource(resource=t.get('name', '')):
256        continue
257      topic = Topic(project_id=context.project_id, resource_data=t)
258      topics[topic.full_path] = topic
259  except utils.GcpApiError as err:
260    logging.warning('Could not list Kafka topics for cluster %s: %s', cluster_name, err)
261  return topics

264@caching.cached_api_call
265def get_topic(
266  project_id: str, location: str, cluster_name: str, topic_name: str
267) -> Optional[Topic]:
268  """Retrieve a single Managed Kafka topic by name."""
269  if not apis.is_enabled(project_id, 'managedkafka'):
270    return None
271  kafka_api = apis.get_api('managedkafka', 'v1', project_id)
272  request = (
273    kafka_api.projects()
274    .locations()
275    .clusters()
276    .topics()
277    .get(
278      name=f'projects/{project_id}/locations/{location}/clusters/{cluster_name}/topics/{topic_name}'
279    )
280  )
281  try:
282    resp = request.execute(num_retries=config.API_RETRIES)
283    return Topic(project_id=project_id, resource_data=resp)
284  except googleapiclient.errors.HttpError as err:
285    if err.resp.status == 404:
286      return None
287    raise utils.GcpApiError(err) from err

290@caching.cached_api_call
291def get_consumer_groups(
292  context: models.Context, cluster_name: str, location: Optional[str] = None
293) -> Dict[str, ConsumerGroup]:
294  """Get Kafka Consumer Groups for a specific cluster."""
295  groups: Dict[str, ConsumerGroup] = {}
296  if not apis.is_enabled(context.project_id, 'managedkafka'):
297    return groups
298
299  kafka_api = apis.get_api('managedkafka', 'v1', context.project_id)
300  parent_path = _resolve_cluster_path(context, cluster_name, location)
301
302  try:
303    request = kafka_api.projects().locations().clusters().consumerGroups().list(parent=parent_path)
304    for g in apis_utils.list_all(
305      request=request,
306      next_function=kafka_api.projects().locations().clusters().consumerGroups().list_next,
307      response_keyword='consumerGroups',
308    ):
309      if not context.match_project_resource(resource=g.get('name', '')):
310        continue
311      group = ConsumerGroup(project_id=context.project_id, resource_data=g)
312      groups[group.full_path] = group
313  except utils.GcpApiError as err:
314    logging.warning('Could not list Kafka consumer groups for cluster %s: %s', cluster_name, err)
315  return groups

318@caching.cached_api_call
319def get_consumer_group(
320  project_id: str, location: str, cluster_name: str, group_name: str
321) -> Optional[ConsumerGroup]:
322  """Retrieve a single Managed Kafka consumer group by name."""
323  if not apis.is_enabled(project_id, 'managedkafka'):
324    return None
325  kafka_api = apis.get_api('managedkafka', 'v1', project_id)
326  request = (
327    kafka_api.projects()
328    .locations()
329    .clusters()
330    .consumerGroups()
331    .get(
332      name=f'projects/{project_id}/locations/{location}/clusters/{cluster_name}/consumerGroups/{group_name}'
333    )
334  )
335  try:
336    resp = request.execute(num_retries=config.API_RETRIES)
337    return ConsumerGroup(project_id=project_id, resource_data=resp)
338  except googleapiclient.errors.HttpError as err:
339    if err.resp.status == 404:
340      return None
341    raise utils.GcpApiError(err) from err

344@caching.cached_api_call
345def get_acls(
346  context: models.Context, cluster_name: str, location: Optional[str] = None
347) -> Dict[str, Acl]:
348  """Get Kafka ACLs for a specific cluster."""
349  acls: Dict[str, Acl] = {}
350  if not apis.is_enabled(context.project_id, 'managedkafka'):
351    return acls
352
353  kafka_api = apis.get_api('managedkafka', 'v1', context.project_id)
354  parent_path = _resolve_cluster_path(context, cluster_name, location)
355
356  try:
357    request = kafka_api.projects().locations().clusters().acls().list(parent=parent_path)
358    for a in apis_utils.list_all(
359      request=request,
360      next_function=kafka_api.projects().locations().clusters().acls().list_next,
361      response_keyword='acls',
362    ):
363      if not context.match_project_resource(resource=a.get('name', '')):
364        continue
365      acl = Acl(project_id=context.project_id, resource_data=a)
366      acls[acl.full_path] = acl
367  except utils.GcpApiError as err:
368    logging.warning('Could not list Kafka ACLs for cluster %s: %s', cluster_name, err)
369  return acls

372@caching.cached_api_call
373def get_acl(project_id: str, location: str, cluster_name: str, acl_name: str) -> Optional[Acl]:
374  """Retrieve a single Managed Kafka ACL by name."""
375  if not apis.is_enabled(project_id, 'managedkafka'):
376    return None
377  kafka_api = apis.get_api('managedkafka', 'v1', project_id)
378  request = (
379    kafka_api.projects()
380    .locations()
381    .clusters()
382    .acls()
383    .get(name=f'projects/{project_id}/locations/{location}/clusters/{cluster_name}/acls/{acl_name}')
384  )
385  try:
386    resp = request.execute(num_retries=config.API_RETRIES)
387    return Acl(project_id=project_id, resource_data=resp)
388  except googleapiclient.errors.HttpError as err:
389    if err.resp.status == 404:
390      return None
391    raise utils.GcpApiError(err) from err