fix: Launchers - Kubernetes - Fix getting logs when logs arte not valid UTF-8 (#277)

The Kubernetes client's default read_namespaced_pod_log path does a strict
.decode('utf8') over the full log payload before checking HTTP status.
When a pod with high-volume tqdm progress bars (block glyphs █▉▊▋▌▍▎▏,
3-byte UTF-8) runs with num_proc>1, concurrent writes to the same fd can
split a multi-byte glyph across a chunk boundary, leaving an orphaned
continuation byte.  The strict decode throws UnicodeDecodeError, which
bubbles through the log-upload retry wrapper and marks an otherwise-healthy
training run as SYSTEM_ERROR.

Fix: pass _preload_content=False to get the raw urllib3 response and decode
manually with errors="replace".  This is applied to both the single-pod
(LaunchedKubernetesContainer.get_log) and multi-pod
(LaunchedKubernetesJob._get_log_by_pod_key) log-read paths.
A warning is logged whenever replacement characters are injected, so the
next occurrence is observable in Observe without requiring a separate
debug build.

The existing "Bad Request" catch for PodInitializing is unaffected:
the kubernetes client's status check runs outside the _preload_content
block and still raises ApiException with the correct reason phrase.

## User experience: before and after

### Orchestrator log-upload path (run lifecycle)

**Before** — the UnicodeDecodeError bubbles out of the retry wrapper. The run is marked `SYSTEM_ERROR`, no logs are uploaded, and all downstream tasks (e.g. Upload HF, Upload Training Summary) are skipped. The user sees a failed run with no log output and no indication that their training code was healthy.

**After** — the log is decoded successfully and uploaded. The run continues to completion. One or two progress-bar characters are substituted with `?` (U+FFFD) at the point of corruption, but the rest of the log is intact and readable.

### API log-read path (viewing logs for a running execution)

**Before** — the request throws before returning a response. The user gets a 500 error in the UI when trying to view logs mid-run.

**After** — the full log is returned. The substituted character appears inline exactly where the torn byte was, typically mid-progress-bar where it is visually unnoticeable.

---

### Example log output

**Before** (UnicodeDecodeError thrown at byte 5,115,152 — nothing returned):
```
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe2 in position 5115152: unexpected end of data
```

**After** (log returned; `?` marks the single substituted byte at the corruption point):
```
2024-06-17T20:29:11Z Map (num_proc=64):  77%|██████████████████████████████████████████████████████████████████████████████████████▌         | 137412/178432 [00:34<00:10, 3991.03 examples/s]
2024-06-17T20:29:12Z Map (num_proc=64):  79%|███████████████████████████████████████████████████████████████?██████  | 140876/178432 [00:35<00:09, 3987.11 examples/s]
2024-06-17T20:29:13Z Map (num_proc=64):  81%|█████████████████████████████████████████████████████████████████████▏  | 144501/178432 [00:36<00:08, 3994.77 examples/s]
2024-06-17T20:29:55Z ***** Running training *****
2024-06-17T20:29:55Z   Num examples = 144,501
2024-06-17T20:29:55Z   Num Epochs = 3
```

The `?` on line 2 is where one torn block glyph was replaced. All structured log lines above and below it — training config, loss values, eval metrics — are fully intact.

Fixes: #281

Author: morgan-wowk

api_server_main.py

@@ -1,3 +1,4 @@
+import json
 import os
 import traceback
 
@@ -72,9 +73,18 @@ def get_user_details(request: fastapi.Request):
     )
 
 
+_slack_workflow_urls_raw = os.environ.get("TANGLE_SLACK_WORKFLOW_URLS")
+_slack_workflow_urls: dict[str, str] | None = None
+if _slack_workflow_urls_raw:
+    try:
+        _slack_workflow_urls = json.loads(_slack_workflow_urls_raw)
+    except json.JSONDecodeError:
+        pass
+
 api_router.setup_routes(
     app=app,
     db_engine=db_engine,
     user_details_getter=get_user_details,
     default_component_library_owner_username=ADMIN_USER_NAME,
+    slack_workflow_urls=_slack_workflow_urls,
 )

cloud_pipelines_backend/api_router.py

@@ -15,6 +15,7 @@
 from . import component_library_api_server as components_api
 from . import database_ops
 from . import errors
+from . import integration_api
 from .instrumentation import contextual_logging
 
 if typing.TYPE_CHECKING:
@@ -40,6 +41,7 @@ def setup_routes(
     container_launcher_for_log_streaming: "launcher_interfaces.ContainerTaskLauncher[launcher_interfaces.LaunchedContainer] | None" = None,
     default_component_library_owner_username: str = "admin",
     do_skip_backfill: bool = False,
+    slack_workflow_urls: dict[str, str] | None = None,
 ):
     def get_session():
         with orm.Session(autocommit=False, autoflush=False, bind=db_engine) as session:
@@ -76,6 +78,7 @@ async def lifespan(app: fastapi.FastAPI):
         user_details_getter=user_details_getter,
         get_launcher=get_launcher,
         lifespan=lifespan,
+        slack_workflow_urls=slack_workflow_urls,
     )
 
 
@@ -93,6 +96,7 @@ def _setup_routes_internal(
         | typing.Callable[..., abc.Iterator[typing.Any]]
         | None
     ) = None,
+    slack_workflow_urls: dict[str, str] | None = None,
 ):
     # We request `app: fastapi.FastAPI` instead of just returning the router
     # because we want to add exception handler which is only supported for `FastAPI`.
@@ -615,6 +619,14 @@ async def get_sql_engine_connection_pool_status(
     # # Needs to be called after all routes have been added to the router
     # app.include_router(router)
 
+    if slack_workflow_urls:
+        integration_api.register_routes(
+            router=router,
+            service=integration_api.IntegrationApiService(
+                workflow_urls=slack_workflow_urls
+            ),
+        )
+
     app.include_router(router=router)
 
 

cloud_pipelines_backend/integration_api.py

@@ -0,0 +1,105 @@
+"""
+Integration API — proxy layer for external workflow triggers.
+
+Exposes ``POST /api/integration/slack_workflow_trigger`` which accepts a
+logical workflow identifier and an arbitrary payload, resolves the target
+webhook URL server-side, and forwards the payload.  Keeping URLs on the
+server means the frontend config contains no sensitive values and the
+mapping can be updated without a frontend rebuild.
+
+OSS deployments can populate the URL mapping from environment variables
+(e.g. ``TANGLE_SLACK_WORKFLOW_URLS``) to wire up their own integrations.
+Shopify's oasis-backend supplies the mapping via an EJSON secret.
+
+The endpoint is only registered when at least one workflow URL is configured;
+when the mapping is empty the route does not exist and returns 404 for all
+callers.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+import typing
+import urllib.error
+import urllib.request
+
+import fastapi
+
+
+@dataclasses.dataclass
+class SlackWorkflowTriggerRequest:
+    workflow_internal_id: str
+    webhook_payload: dict[str, typing.Any]
+
+
+class IntegrationApiService:
+    """Routes workflow trigger requests to their registered webhook URLs.
+
+    Args:
+        workflow_urls: Mapping of logical workflow id to the full webhook URL.
+            When empty the service is constructed but no routes are registered.
+    """
+
+    def __init__(self, workflow_urls: dict[str, str]) -> None:
+        self._workflow_urls = workflow_urls
+
+    def trigger_workflow(
+        self,
+        *,
+        workflow_internal_id: str,
+        webhook_payload: dict[str, typing.Any],
+    ) -> None:
+        """Forward *webhook_payload* to the URL registered for *workflow_internal_id*.
+
+        Raises:
+            fastapi.HTTPException 404: workflow id not in the configured mapping.
+            fastapi.HTTPException 502: upstream webhook returned an error status.
+        """
+        url = self._workflow_urls.get(workflow_internal_id)
+        if not url:
+            raise fastapi.HTTPException(
+                status_code=404,
+                detail=f"No workflow registered for id '{workflow_internal_id}'.",
+            )
+
+        body = json.dumps(webhook_payload).encode("utf-8")
+        req = urllib.request.Request(
+            url,
+            data=body,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        try:
+            urllib.request.urlopen(req, timeout=10)
+        except urllib.error.HTTPError as exc:
+            raise fastapi.HTTPException(
+                status_code=502,
+                detail=f"Workflow webhook returned HTTP {exc.code}.",
+            ) from exc
+        except OSError as exc:
+            raise fastapi.HTTPException(
+                status_code=502,
+                detail=f"Workflow webhook unreachable: {exc}",
+            ) from exc
+
+
+def register_routes(
+    router: fastapi.APIRouter,
+    service: IntegrationApiService,
+) -> None:
+    """Attach integration routes to *router*.
+
+    Called from ``api_router._setup_routes_internal`` only when the service
+    has at least one configured workflow URL.
+    """
+
+    @router.post("/api/integration/slack_workflow_trigger", tags=["integration"])
+    def trigger_slack_workflow(
+        body: SlackWorkflowTriggerRequest,
+    ) -> dict[str, str]:
+        service.trigger_workflow(
+            workflow_internal_id=body.workflow_internal_id,
+            webhook_payload=body.webhook_payload,
+        )
+        return {"status": "ok"}

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -922,7 +922,8 @@ def get_refreshed(self) -> "LaunchedKubernetesContainer":
     def get_log(self) -> str:
         launcher = self._get_launcher()
         core_api_client = k8s_client_lib.CoreV1Api(api_client=launcher._api_client)
-        return core_api_client.read_namespaced_pod_log(
+        # _preload_content=False bypasses the kubernetes client's strict .decode('utf8'); see _get_log_by_pod_key.
+        response = core_api_client.read_namespaced_pod_log(
             name=self._pod_name,
             namespace=self._namespace,
             container=_MAIN_CONTAINER_NAME,
@@ -931,7 +932,17 @@ def get_log(self) -> str:
             # HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"PodLogOptions \"task-pod-xxxxx\" is invalid: stream: Forbidden: may not be specified","reason":"Invalid","details":{"name":"task-pod-xxxxx","kind":"PodLogOptions","causes":[{"reason":"FieldValueForbidden","message":"Forbidden: may not be specified","field":"stream"}]},"code":422}
             # stream="All",
             _request_timeout=launcher._request_timeout,
+            _preload_content=False,
         )
+        try:
+            log = response.data.decode("utf-8", errors="replace")
+            if "\N{REPLACEMENT CHARACTER}" in log:
+                _logger.warning(
+                    f"Pod log for {self._pod_name} contained invalid UTF-8 bytes; substituted replacement characters."
+                )
+            return log
+        finally:
+            response.release_conn()
 
     def upload_log(self):
         launcher = self._get_launcher()
@@ -1490,14 +1501,26 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None:
         launcher = self._get_launcher()
         core_api_client = k8s_client_lib.CoreV1Api(api_client=launcher._api_client)
         try:
-            log = core_api_client.read_namespaced_pod_log(
+            # _preload_content=False bypasses the kubernetes client's strict .decode('utf8'),
+            # which would raise UnicodeDecodeError on torn multi-byte chars in pod logs (e.g.
+            # tqdm block glyphs split across concurrent-writer chunk boundaries).
+            response = core_api_client.read_namespaced_pod_log(
                 name=pod_name,
                 namespace=self._namespace,
                 container=_MAIN_CONTAINER_NAME,
                 timestamps=True,
                 _request_timeout=launcher._request_timeout,
+                _preload_content=False,
             )
-            return log
+            try:
+                log = response.data.decode("utf-8", errors="replace")
+                if "\N{REPLACEMENT CHARACTER}" in log:
+                    _logger.warning(
+                        f"Pod log for {pod_name} contained invalid UTF-8 bytes; substituted replacement characters."
+                    )
+                return log
+            finally:
+                response.release_conn()
         except kubernetes.client.exceptions.ApiException as ex:
             if ex.reason == "Bad Request":
                 # Kubernetes client raises kubernetes.client.exceptions.ApiException when Pod is still in PodInitializing phase