connectors-insights: high-level error handling (airbytehq#40253)

.github/workflows/connectors_insights.yml

@@ -48,4 +48,4 @@ jobs:
         run: echo "GOOGLE_APPLICATION_CREDENTIALS=$HOME/gcp-sa-key.json" >> $GITHUB_ENV
       - name: Run connectors insights
         run: |
-          poetry -C airbyte-ci/connectors/connectors_insights run connectors-insights generate --gcs-uri=gs://prod-airbyte-cloud-connector-metadata-service/connector_insights --connector-directory airbyte-integrations/connectors/ --concurrency 10 ${{ inputs.rewrite == true && '--rewrite' || ''}}
+          poetry -C airbyte-ci/connectors/connectors_insights run connectors-insights generate --gcs-uri=gs://prod-airbyte-cloud-connector-metadata-service/connector_insights --connector-directory airbyte-integrations/connectors/ --concurrency 10 ${{ inputs.rewrite == 'true' && '--rewrite' || ''}}

airbyte-ci/connectors/connectors_insights/README.md

@@ -56,6 +56,9 @@ This CLI is currently running nightly in GitHub Actions. The workflow can be fou
 
 ## Changelog
 
+### 0.2.1
+- Implement a high-level error handling to not fail the entire process if a connector fails to generate insights.
+
 ### 0.2.0
 - Detect deprecated class and module use in connectors.
 - Fix missing CDK version for connectors not declaring a CDK name in their metadata.

airbyte-ci/connectors/connectors_insights/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "connectors-insights"
-version = "0.2.0"
+version = "0.2.1"
 description = ""
 authors = ["Airbyte <[email protected]>"]
 readme = "README.md"

airbyte-ci/connectors/connectors_insights/src/connectors_insights/cli.py

@@ -106,13 +106,19 @@ async def generate(
         logger.info(f"Generating insights for {len(connectors)} connectors.")
 
     semaphore = Semaphore(concurrency)
+    soon_results = []
     async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as dagger_client:
         async with asyncer.create_task_group() as connector_task_group:
             for connector in connectors:
-                connector_task_group.soonify(generate_insights_for_connector)(
-                    dagger_client.pipeline(connector.technical_name),
-                    connector,
-                    semaphore,
-                    rewrite,
-                    result_backends=result_backends,
+                soon_results.append(
+                    connector_task_group.soonify(generate_insights_for_connector)(
+                        dagger_client.pipeline(connector.technical_name),
+                        connector,
+                        semaphore,
+                        rewrite,
+                        result_backends=result_backends,
+                    )
                 )
+    failing_connector_names = [soon_result.value[1].technical_name for soon_result in soon_results if not soon_result.value[0]]
+    if failing_connector_names:
+        raise click.ClickException("Failed to generate insights for the following connectors: " + ", ".join(failing_connector_names))

airbyte-ci/connectors/connectors_insights/src/connectors_insights/insights.py

@@ -16,7 +16,7 @@
 from connectors_insights.sbom import get_json_sbom
 
 if TYPE_CHECKING:
-    from typing import Dict, List
+    from typing import Dict, List, Tuple
 
     import dagger
     from anyio import Semaphore
@@ -203,7 +203,7 @@ async def generate_insights_for_connector(
     semaphore: Semaphore,
     rewrite: bool = False,
     result_backends: List[ResultBackend] | None = None,
-) -> ConnectorInsights | None:
+) -> Tuple[bool, Connector]:
     """Aggregate insights for a connector and write them to the result backends.
 
     Args:
@@ -213,7 +213,7 @@ async def generate_insights_for_connector(
         rewrite (bool): whether to rewrite the insights if they already exist.
         result_backend (List[ResultBackend] | None): the result backends to write the insights to.
     Returns:
-        ConnectorInsights: the insights for the connector.
+        Tuple[bool, Connector]: a tuple of whether the insights were generated and the connector.
     """
     logger = logging.getLogger(__name__)
     insights_file = FileToPersist("insights.json")
@@ -223,17 +223,21 @@ async def generate_insights_for_connector(
     async with semaphore:
         if should_skip_generation(result_backends, connector, files_to_persist, rewrite):
             logger.info(f"Skipping insights generation for {connector.technical_name} because it is already generated.")
-            return None
+            return True, connector
 
         logger.info(f"Generating insights for {connector.technical_name}")
         result_backends = result_backends or []
-        pylint_output = await get_pylint_output(dagger_client, connector)
-        raw_sbom = await fetch_sbom(dagger_client, connector)
-        if raw_sbom:
-            sbom_file.set_file_content(raw_sbom)
-
-        insights = generate_insights(connector, raw_sbom, pylint_output)
-        insights_file.set_file_content(insights.json())
-        persist_files(connector, files_to_persist, result_backends, rewrite, logger)
-        logger.info(f"Finished generating insights for {connector.technical_name}")
-        return insights
+        try:
+            pylint_output = await get_pylint_output(dagger_client, connector)
+            raw_sbom = await fetch_sbom(dagger_client, connector)
+            if raw_sbom:
+                sbom_file.set_file_content(raw_sbom)
+
+            insights = generate_insights(connector, raw_sbom, pylint_output)
+            insights_file.set_file_content(insights.json())
+            persist_files(connector, files_to_persist, result_backends, rewrite, logger)
+            logger.info(f"Finished generating insights for {connector.technical_name}")
+            return True, connector
+        except Exception as e:
+            logger.error(f"Failed to generate insights for {connector.technical_name}: {e}")
+            return False, connector