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 billing_account_name(self) -> str:
82    return self._resource_data['billingAccountName']
83
84  def is_billing_enabled(self) -> bool:
85    return self._resource_data['billingEnabled']
86
87  def __init__(self, project_id, resource_data):
88    super().__init__(project_id=project_id)
89    self._resource_data = resource_data

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

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

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

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

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

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