gcpdiag.queries.billing

32class BillingAccount(models.Resource):
33  """Represents a Cloud Billing Account.
34
35  See also the API documentation:
36  https://cloud.google.com/billing/docs/reference/rest/v1/billingAccounts
37  """
38
39  @property
40  def full_path(self) -> str:
41    return self._resource_data['name']
42
43  @property
44  def name(self) -> str:
45    return self._resource_data['name']
46
47  @property
48  def display_name(self) -> str:
49    return self._resource_data['displayName']
50
51  def is_open(self) -> bool:
52    return self._resource_data['open']
53
54  def is_master(self) -> bool:
55    return len(self._resource_data['masterBillingAccount']) > 0
56
57  def list_projects(self, context) -> list:
58    return get_all_projects_in_billing_account(context, self.name)
59
60  def __init__(self, project_id, resource_data):
61    super().__init__(project_id=project_id)
62    self._resource_data = resource_data

65class ProjectBillingInfo(models.Resource):
66  """Represents a Billing Information about a Project.
67
68  See also the API documentation:
69  https://cloud.google.com/billing/docs/reference/rest/v1/ProjectBillingInfo
70  """
71
72  @property
73  def full_path(self) -> str:
74    return self._resource_data['name']
75
76  @property
77  def name(self) -> str:
78    return self._resource_data['name']
79
80  @property
81  def project_id(self) -> str:
82    return self._resource_data['projectId']
83
84  @property
85  def billing_account_name(self) -> str:
86    return self._resource_data['billingAccountName']
87
88  def is_billing_enabled(self) -> bool:
89    return self._resource_data['billingEnabled']
90
91  def __init__(self, project_id, resource_data):
92    super().__init__(project_id=project_id)
93    self._resource_data = resource_data

 96class CostInsights(models.Resource):
 97  """Represents a Costs Insights object"""
 98
 99  @property
100  def full_path(self) -> str:
101    return self._resource_data['name']
102
103  @property
104  def description(self) -> str:
105    return self._resource_data['description']
106
107  @property
108  def anomaly_details(self) -> dict:
109    return self._resource_data['content']['anomalyDetails']
110
111  @property
112  def forecasted_units(self) -> str:
113    return self.anomaly_details['forecastedCostData']['cost']['units']
114
115  @property
116  def forecasted_currency(self) -> str:
117    return self.anomaly_details['forecastedCostData']['cost']['currencyCode']
118
119  @property
120  def actual_units(self) -> str:
121    return self.anomaly_details['actualCostData']['cost']['units']
122
123  @property
124  def actual_currency(self) -> str:
125    return self.anomaly_details['actualCostData']['cost']['currencyCode']
126
127  @property
128  def start_time(self) -> str:
129    return self.anomaly_details['costSlice']['startTime']
130
131  @property
132  def end_time(self) -> str:
133    return self.anomaly_details['costSlice']['endTime']
134
135  @property
136  def anomaly_type(self) -> str:
137    return 'Below' if self._resource_data['insightSubtype'] == 'COST_BELOW_FORECASTED' else 'Above'
138
139  def is_anomaly(self) -> bool:
140    if 'description' in self._resource_data.keys():
141      return 'This is a cost anomaly' in self.description
142    return False
143
144  def build_anomaly_description(self):
145    return (
146      self.description
147      + '\nCost '
148      + self.anomaly_type
149      + ' forecast, Forecasted: '
150      + self.forecasted_units
151      + ' '
152      + self.forecasted_currency
153      + ', Actual: '
154      + self.actual_units
155      + ' '
156      + self.actual_currency
157      + '\nAnomaly Period From: '
158      + self.start_time
159      + ', To: '
160      + self.end_time
161    )
162
163  def __init__(self, project_id, resource_data):
164    super().__init__(project_id=project_id)
165    self._resource_data = resource_data

168@caching.cached_api_call
169def get_billing_info(project_id) -> ProjectBillingInfo:
170  """Get Billing Information for a project, caching the result."""
171  project_api = apis.get_api('cloudbilling', 'v1', project_id)
172  project_id = 'projects/' + project_id if 'projects/' not in project_id else project_id
173  query = project_api.projects().getBillingInfo(name=project_id)
174  logging.debug('fetching Billing Information for project %s', project_id)
175  try:
176    resource_data = query.execute(num_retries=config.API_RETRIES)
177  except googleapiclient.errors.HttpError as err:
178    raise GcpApiError(err) from err
179  return ProjectBillingInfo(project_id, resource_data)

182@caching.cached_api_call
183def get_billing_account(project_id: str) -> Optional[BillingAccount]:
184  """Get a Billing Account object by its project name, caching the result."""
185  if not apis.is_enabled(project_id, 'cloudbilling'):
186    return None
187  billing_info = get_billing_info(project_id)
188  if not billing_info.is_billing_enabled():
189    return None
190
191  billing_account_api = apis.get_api('cloudbilling', 'v1', project_id)
192  query = billing_account_api.billingAccounts().get(name=billing_info.billing_account_name)
193  logging.debug('fetching Billing Account for project %s', project_id)
194  try:
195    resource_data = query.execute(num_retries=config.API_RETRIES)
196  except googleapiclient.errors.HttpError as error:
197    e = utils.GcpApiError(error)
198    if ('The caller does not have permission' in e.message) or ('PERMISSION_DENIED' in e.reason):
199      # billing rules cannot be tested without permissions on billing account
200      return None
201    else:
202      raise GcpApiError(error) from error
203  return BillingAccount(project_id, resource_data)

206@caching.cached_api_call
207def get_all_billing_accounts(project_id: str) -> Optional[List[BillingAccount]]:
208  """Get all Billing Accounts that current user has permission to view"""
209  accounts = []
210  if not apis.is_enabled(project_id, 'cloudbilling'):
211    return None
212  api = apis.get_api('cloudbilling', API_VERSION, project_id)
213
214  try:
215    for account in apis_utils.list_all(
216      request=api.billingAccounts().list(),
217      next_function=api.billingAccounts().list_next,
218      response_keyword='billingAccounts',
219    ):
220      accounts.append(BillingAccount(project_id, account))
221  except utils.GcpApiError as e:
222    if ('The caller does not have permission' in e.message) or ('PERMISSION_DENIED' in e.reason):
223      # billing rules cannot be tested without permissions on billing account
224      return None
225    else:
226      raise e
227  return accounts

230@caching.cached_api_call
231def get_all_projects_in_billing_account(
232  context: models.Context, billing_account_name: str
233) -> List[ProjectBillingInfo]:
234  """Get all projects associated with the Billing Account that current user has
235  permission to view"""
236  projects = []
237  api = apis.get_api('cloudbilling', API_VERSION, context.project_id)
238
239  for p in apis_utils.list_all(
240    request=api.billingAccounts()
241    .projects()
242    .list(
243      name=billing_account_name,
244    ),
245    next_function=api.billingAccounts().projects().list_next,
246    response_keyword='projectBillingInfo',
247  ):
248    try:
249      crm_api = apis.get_api('cloudresourcemanager', 'v3', p['projectId'])
250      p_name = 'projects/' + p['projectId'] if 'projects/' not in p['projectId'] else p['projectId']
251      request = crm_api.projects().get(name=p_name)
252      response = request.execute(num_retries=config.API_RETRIES)
253      projects.append(ProjectBillingInfo(response['projectId'], p))
254    except (utils.GcpApiError, googleapiclient.errors.HttpError) as error:
255      if isinstance(error, googleapiclient.errors.HttpError):
256        error = utils.GcpApiError(error)
257      if error.reason in ['IAM_PERMISSION_DENIED', 'USER_PROJECT_DENIED', 'SERVICE_DISABLED']:
258        # skip projects that user does not have permissions on
259        continue
260      else:
261        print(
262          f'[ERROR]: An Http Error occurred whiles accessing projects.get \n\n{error}',
263          file=sys.stderr,
264        )
265      raise error from error
266  return projects

269@caching.cached_api_call
270def get_cost_insights_for_a_project(project_id: str):
271  """Get cost insights for the project"""
272  billing_account = get_billing_account(project_id)
273
274  # If Billing Account is closed or is a reseller account then Cost Insights
275  # are not available
276  if (not billing_account.is_open()) or billing_account.is_master():
277    return None
278
279  api = apis.get_api('recommender', 'v1', project_id)
280
281  insight_name = billing_account.name + '/locations/global/insightTypes/google.billing.CostInsight'
282  insights = []
283  for insight in apis_utils.list_all(
284    request=api.billingAccounts().locations().insightTypes().insights().list(parent=insight_name),
285    next_function=api.billingAccounts().locations().insightTypes().insights().list_next,
286    response_keyword='insights',
287  ):
288    insights.append(CostInsights(project_id, insight))
289  return insights