airbyte-ci: ClickPipelineContext handles dagger logging

airbyte-ci/connectors/pipelines/README.md

@@ -502,6 +502,7 @@ This command runs the Python tests for a airbyte-ci poetry package.
 
 | Version | PR                                                         | Description                                                                                               |
 | ------- | ---------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
+| 2.10.9  | [#33419](https://github.com/airbytehq/airbyte/pull/33419)  | Make ClickPipelineContext handle dagger logging.                                                                |
 | 2.10.8  | [#33249](https://github.com/airbytehq/airbyte/pull/33249)  | Exclude git ignored files from formatting.                                                                |
 | 2.10.7  | [#33248](https://github.com/airbytehq/airbyte/pull/33248)  | Fix bug which broke airbyte-ci connectors tests when optional DockerHub credentials env vars are not set. |
 | 2.10.6  | [#33170](https://github.com/airbytehq/airbyte/pull/33170)  | Remove Dagger logs from console output of `format`.                                                       |

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/format/format_command.py

@@ -121,16 +121,8 @@ async def invoke(self, ctx: click.Context, click_pipeline_context: ClickPipeline
         Returns:
             Any: The result of running the command
         """
-        dagger_logs_file_descriptor, dagger_logs_temp_file_path = tempfile.mkstemp(
-            dir="/tmp", prefix=f"format_{self.formatter.value}_dagger_logs_", suffix=".log"
-        )
-        # Create a FileIO object from the file descriptor
-        dagger_logs = io.FileIO(dagger_logs_file_descriptor, "w+")
-        self.logger.info(f"Running {self.formatter.value} formatter. Logging dagger output to {dagger_logs_temp_file_path}")
 
-        dagger_client = await click_pipeline_context.get_dagger_client(
-            pipeline_name=f"Format {self.formatter.value}", log_output=dagger_logs
-        )
+        dagger_client = await click_pipeline_context.get_dagger_client(pipeline_name=f"Format {self.formatter.value}")
         dir_to_format = self.get_dir_to_format(dagger_client)
 
         container = self.get_format_container_fn(dagger_client, dir_to_format)

airbyte-ci/connectors/pipelines/pipelines/models/contexts/click_pipeline_context.py

@@ -2,13 +2,16 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import io
 import sys
-from typing import Any, Callable, Optional, TextIO
+import tempfile
+from typing import Any, Callable, Optional, TextIO, Tuple
 
 import anyio
 import dagger
 from asyncclick import Context, get_current_context
 from dagger.api.gen import Client, Container
+from pipelines import main_logger
 from pipelines.cli.click_decorators import LazyPassDecorator
 from pydantic import BaseModel, Field, PrivateAttr
 
@@ -76,15 +79,18 @@ def __init__(self, **data: dict[str, Any]):
 
     _dagger_client_lock: anyio.Lock = PrivateAttr(default_factory=anyio.Lock)
 
-    async def get_dagger_client(self, pipeline_name: Optional[str] = None, log_output: Optional[TextIO] = sys.stdout) -> Client:
+    async def get_dagger_client(self, pipeline_name: Optional[str] = None) -> Client:
         """
         Get (or initialize) the Dagger Client instance.
         """
         if not self._dagger_client:
             async with self._dagger_client_lock:
                 if not self._dagger_client:
+                    if self.params.get("show_dagger_logs", False):
+                        log_output = sys.stdout
+                    else:
+                        log_output, self._click_context().obj["dagger_logs_path"] = self._create_dagger_client_log_file()
                     connection = dagger.Connection(dagger.Config(log_output=log_output))
-
                     """
                     Sets up the '_dagger_client' attribute, intended for single-threaded use within connectors.
 
@@ -97,6 +103,14 @@ async def get_dagger_client(self, pipeline_name: Optional[str] = None, log_outpu
         assert self._dagger_client, "Error initializing Dagger client"
         return self._dagger_client.pipeline(pipeline_name) if pipeline_name else self._dagger_client
 
+    def _create_dagger_client_log_file(self) -> Tuple[io.FileIO, str]:
+        """
+        Create the dagger client log file.
+        """
+        dagger_logs_file_descriptor, dagger_logs_temp_file_path = tempfile.mkstemp(dir="/tmp", prefix=f"dagger_client_", suffix=".log")
+        main_logger.info(f"Dagger client logs stored in {dagger_logs_temp_file_path}")
+        return io.FileIO(dagger_logs_file_descriptor, "w+"), dagger_logs_temp_file_path
+
 
 # Create @pass_pipeline_context decorator for use in click commands
 pass_pipeline_context: LazyPassDecorator = LazyPassDecorator(ClickPipelineContext)

airbyte-ci/connectors/pipelines/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pipelines"
-version = "2.10.8"
+version = "2.10.9"
 description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
 authors = ["Airbyte <contact@airbyte.io>"]
 

airbyte-ci/connectors/pipelines/tests/test_models/test_click_pipeline_context.py

@@ -15,6 +15,8 @@ def cli():
         pass
 
     ctx = click.Context(cli)
+    ctx.obj = {"foo": "bar"}
+    ctx.params = {"baz": "qux"}
 
     async with ctx.scope():
         click_pipeline_context = ClickPipelineContext()