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'] == \
138                         'COST_BELOW_FORECASTED' else 'Above'
139
140  def is_anomaly(self) -> bool:
141    if 'description' in self._resource_data.keys():
142      return 'This is a cost anomaly' in self.description
143    return False
144
145  def build_anomaly_description(self):
146    return self.description + '\nCost ' + self.anomaly_type + \
147           ' forecast, Forecasted: ' + self.forecasted_units + \
148           ' ' + self.forecasted_currency + ', Actual: ' + \
149           self.actual_units + ' ' + self.actual_currency + \
150           '\nAnomaly Period From: ' + self.start_time + ', To: ' + self.end_time
151
152  def __init__(self, project_id, resource_data):
153    super().__init__(project_id=project_id)
154    self._resource_data = resource_data

157@caching.cached_api_call
158def get_billing_info(project_id) -> ProjectBillingInfo:
159  """Get Billing Information for a project, caching the result."""
160  project_api = apis.get_api('cloudbilling', 'v1', project_id)
161  project_id = 'projects/' + project_id if 'projects/' not in project_id else project_id
162  query = project_api.projects().getBillingInfo(name=project_id)
163  logging.debug('fetching Billing Information for project %s', project_id)
164  try:
165    resource_data = query.execute(num_retries=config.API_RETRIES)
166  except googleapiclient.errors.HttpError as err:
167    raise GcpApiError(err) from err
168  return ProjectBillingInfo(project_id, resource_data)

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

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

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

257@caching.cached_api_call
258def get_cost_insights_for_a_project(project_id: str):
259  """Get cost insights for the project"""
260  billing_account = get_billing_account(project_id)
261
262  # If Billing Account is closed or is a reseller account then Cost Insights
263  # are not available
264  if (not billing_account.is_open()) or billing_account.is_master():
265    return None
266
267  api = apis.get_api('recommender', 'v1', project_id)
268
269  insight_name = billing_account.name + '/locations/global/insightTypes/google.billing.CostInsight'
270  insights = []
271  for insight in apis_utils.list_all(
272      request=api.billingAccounts().locations().insightTypes().insights().list(
273          parent=insight_name),
274      next_function=api.billingAccounts().locations().insightTypes().insights(
275      ).list_next,
276      response_keyword='insights'):
277    insights.append(CostInsights(project_id, insight))
278  return insights