Sources Fixes for GCP onboarding and some sentry errors (#2620)

koku/providers/gcp/provider.py

@@ -4,6 +4,7 @@
 import google.auth
 from google.cloud import bigquery
 from google.cloud.exceptions import GoogleCloudError
+from google.cloud.exceptions import NotFound
 from googleapiclient import discovery
 from googleapiclient.errors import HttpError
 from rest_framework import serializers
@@ -50,6 +51,24 @@ def update_source_data_source(self, credentials, data_source):
         except Sources.DoesNotExist:
             LOG.info("Source not found, unable to update data source.")
 
+    def _detect_billing_export_table(self, data_source, credentials):
+        """Verify that dataset and billing export table exists."""
+        proj_table = f"{credentials.get('project_id')}.{data_source.get('dataset')}"
+        try:
+            bigquery_table_id = self.get_table_id(proj_table)
+            if bigquery_table_id:
+                data_source["table_id"] = bigquery_table_id
+                self.update_source_data_source(credentials, data_source)
+            else:
+                raise SkipStatusPush("Table ID not ready.")
+        except NotFound as e:
+            key = "billing_source.dataset"
+            LOG.info(error_obj(key, e.message))
+            message = (
+                f"Unable to find dataset: {data_source.get('dataset')} in project: {credentials.get('project_id')}"
+            )
+            raise serializers.ValidationError(error_obj(key, message))
+
     def cost_usage_source_is_reachable(self, credentials, data_source):
         """
         Verify that the GCP bucket exists and is reachable.
@@ -60,7 +79,7 @@ def cost_usage_source_is_reachable(self, credentials, data_source):
 
         """
         try:
-            project = credentials.get("project_id")
+            project = credentials.get("project_id", "")
             gcp_credentials, _ = google.auth.default()
             # https://github.com/googleapis/google-api-python-client/issues/299
             service = discovery.build("cloudresourcemanager", "v1", credentials=gcp_credentials, cache_discovery=False)
@@ -82,18 +101,14 @@ def cost_usage_source_is_reachable(self, credentials, data_source):
             raise serializers.ValidationError(error_obj(key, e.message))
         except HttpError as err:
             reason = err._get_reason()
+            if reason == "Not Found":
+                reason = "Project ID not found"
             key = "authentication.project_id"
             LOG.info(error_obj(key, reason))
             raise serializers.ValidationError(error_obj(key, reason))
 
         if not data_source.get("table_id"):
-            proj_table = f"{credentials.get('project_id')}.{data_source.get('dataset')}"
-            bigquery_table_id = self.get_table_id(proj_table)
-            if bigquery_table_id:
-                data_source["table_id"] = bigquery_table_id
-                self.update_source_data_source(credentials, data_source)
-            else:
-                raise SkipStatusPush("Table ID not ready.")
+            self._detect_billing_export_table(data_source, credentials)
 
         return True
 

koku/providers/test/gcp/tests_provider.py

@@ -5,13 +5,15 @@
 from django.test import TestCase
 from faker import Faker
 from google.cloud.exceptions import GoogleCloudError
+from google.cloud.exceptions import NotFound
 from rest_framework.serializers import ValidationError
 
 from api.models import Provider
 from providers.gcp.provider import GCPProvider
 from providers.gcp.provider import REQUIRED_IAM_PERMISSIONS
 from providers.provider_errors import SkipStatusPush
 
+
 FAKE = Faker()
 
 
@@ -134,3 +136,20 @@ def test_cost_usage_source_is_reachable_no_table_id_notready(self, mock_auth, mo
         provider = GCPProvider()
         with self.assertRaises(SkipStatusPush):
             provider.cost_usage_source_is_reachable(credentials_param, billing_source_param)
+
+    @patch("providers.gcp.provider.bigquery")
+    @patch("providers.gcp.provider.discovery")
+    @patch("providers.gcp.provider.google.auth.default")
+    def test_cost_usage_source_is_reachable_dataset_not_found(self, mock_auth, mock_discovery, mock_bigquery):
+        """Test that cost_usage_source_is_reachable throws appropriate error when dataset is not found."""
+        mock_bigquery.Client.side_effect = NotFound(message="Not Found")
+        gcp_creds = MagicMock()
+        mock_auth.return_value = (gcp_creds, MagicMock())
+        mock_discovery.build.return_value.projects.return_value.testIamPermissions.return_value.execute.return_value.get.return_value = (  # noqa: E501
+            REQUIRED_IAM_PERMISSIONS
+        )
+        billing_source_param = {"dataset": FAKE.word()}
+        credentials_param = {"project_id": FAKE.word()}
+        provider = GCPProvider()
+        with self.assertRaises(ValidationError):
+            provider.cost_usage_source_is_reachable(credentials_param, billing_source_param)

koku/sources/kafka_listener.py

@@ -376,12 +376,10 @@ def cost_mgmt_msg_filter(msg_data):
 
         if msg_data.get("resource_type") == "Application":
             source_id = sources_network.get_source_id_from_applications_id(msg_data.get("resource_id"))
-        msg_data["source_id"] = source_id
-        if not sources_network.get_application_type_is_cost_management(source_id):
-            LOG.info(f"Resource id {msg_data.get('resource_id')} not associated with cost-management.")
-            return None
-    else:
-        source_id = msg_data.get("source_id")
+            msg_data["source_id"] = source_id
+            if not sources_network.get_application_type_is_cost_management(source_id):
+                LOG.info(f"Resource id {msg_data.get('resource_id')} not associated with cost-management.")
+                return None
 
     return msg_data
 

koku/sources/storage.py

@@ -436,7 +436,7 @@ def save_status(source_id, status):
         status (dict) - source status json
 
     """
-    source = get_source(source_id, f"Source ID {source_id} does not exist.", LOG.error)
+    source = get_source(source_id, f"Source ID {source_id} does not exist.", LOG.warning)
     if source and source.status != status:
         source.status = status
         source.save()