gcpdiag.queries.dataflow

39class Job(models.Resource):
40  """Represents Dataflow job.
41
42  resource_data is of the form similar to:
43  {'id': 'my_job_id',
44  'projectId': 'my_project_id',
45  'name': 'pubsubtogcs-20240328-122953',
46  'environment': {},
47  'currentState': 'JOB_STATE_FAILED',
48  'currentStateTime': '2024-03-28T12:34:27.383249Z',
49  'createTime': '2024-03-28T12:29:55.284524Z',
50  'location': 'europe-west2',
51  'startTime': '2024-03-28T12:29:55.284524Z'}
52  """
53
54  _resource_data: dict
55  project_id: str
56
57  def __init__(self, project_id: str, resource_data: dict):
58    super().__init__(project_id)
59    self._resource_data = resource_data
60
61  @property
62  def full_path(self) -> str:
63    return self._resource_data.get('name', '')
64
65  @property
66  def id(self) -> str:
67    return self._resource_data['id']
68
69  @property
70  def state(self) -> str:
71    return self._resource_data['currentState']
72
73  @property
74  def job_type(self) -> str:
75    return self._resource_data['type']
76
77  @property
78  def location(self) -> str:
79    return self._resource_data['location']
80
81  @property
82  def sdk_support_status(self) -> str:
83    return self._resource_data['jobMetadata']['sdkVersion']['sdkSupportStatus']
84
85  @property
86  def sdk_language(self) -> str:
87    return self._resource_data['jobMetadata']['sdkVersion']['versionDisplayName']
88
89  @property
90  def minutes_in_current_state(self) -> int:
91    timestamp = datetime.strptime(self._resource_data['currentStateTime'], '%Y-%m-%dT%H:%M:%S.%fZ')
92    delta = datetime.now() - timestamp
93    return int(delta.total_seconds() // 60)

 96def get_region_dataflow_jobs(api, context: models.Context, region: str) -> List[Job]:
 97  response = apis_utils.list_all(
 98    request=api.projects().locations().jobs().list(projectId=context.project_id, location=region),
 99    next_function=api.projects().locations().jobs().list_next,
100    response_keyword='jobs',
101  )
102  jobs = []
103  for job in response:
104    location = job.get('location', '')
105    labels = job.get('labels', {})
106    name = job.get('name', '')
107
108    # add job id as one of labels for filtering
109    labels['id'] = job.get('id', '')
110
111    # we could get the specific job but correctly matching the location will take too
112    # much effort. Hence get all the jobs and filter afterwards
113    # https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs/list#query-parameters
114    if not context.match_project_resource(location=location, labels=labels, resource=name):
115      continue
116    jobs.append(Job(context.project_id, job))
117  return jobs

120@caching.cached_api_call
121def get_all_dataflow_jobs(context: models.Context) -> List[Job]:
122  api = apis.get_api('dataflow', 'v1b3', context.project_id)
123
124  if not apis.is_enabled(context.project_id, 'dataflow'):
125    return []
126
127  result: List[Job] = []
128  executor = get_executor(context)
129  for jobs in executor.map(lambda r: get_region_dataflow_jobs(api, context, r), DATAFLOW_REGIONS):
130    result += jobs
131
132  print(f'\n\nFound {len(result)} Dataflow jobs\n')
133
134  # print one Dataflow job id when it is found
135  if context.labels and result and 'id' in context.labels:
136    print(f'{result[0].full_path} - {result[0].id}\n')
137
138  return result

141@caching.cached_api_call
142def get_job(project_id: str, job: str, region: str) -> Union[Job, None]:
143  """Fetch a specific Dataflow job."""
144  api = apis.get_api('dataflow', 'v1b3', project_id)
145
146  if not apis.is_enabled(project_id, 'dataflow'):
147    return None
148
149  query = api.projects().locations().jobs().get(projectId=project_id, location=region, jobId=job)
150  try:
151    resp = query.execute(num_retries=config.API_RETRIES)
152    return Job(project_id, resp)
153  except googleapiclient.errors.HttpError as err:
154    raise utils.GcpApiError(err) from err

157@caching.cached_api_call
158def get_all_dataflow_jobs_for_project(
159  project_id: str,
160  filter_str: Optional[str] = None,
161) -> Union[List[Job], None]:
162  """Fetch all Dataflow jobs for a project."""
163  api = apis.get_api('dataflow', 'v1b3', project_id)
164
165  if not apis.is_enabled(project_id, 'dataflow'):
166    return None
167
168  jobs: List[Job] = []
169
170  request = api.projects().jobs().aggregated(projectId=project_id, filter=filter_str)
171  logging.debug('listing dataflow jobs of project %s', project_id)
172
173  while request:  # Continue as long as there are pages
174    response = request.execute(num_retries=config.API_RETRIES)
175    if 'jobs' in response:
176      jobs.extend([Job(project_id, job) for job in response['jobs']])
177    request = (
178      api.projects().jobs().aggregated_next(previous_request=request, previous_response=response)
179    )
180  return jobs

183@caching.cached_api_call
184def logs_excluded(project_id: str) -> Union[bool, None]:
185  """Check if Dataflow Logs are excluded."""
186
187  if not apis.is_enabled(project_id, 'dataflow'):
188    return None
189
190  exclusions = logs.exclusions(project_id)
191  if exclusions is None:
192    return None
193  else:
194    for log_exclusion in exclusions:
195      if 'resource.type="dataflow_step"' in log_exclusion.filter and log_exclusion.disabled:
196        return True
197  return False