kubernetes: surface log-unavailable reason and observability link in placeholder

When log acquisition fails, replace the empty return value with a human-
readable message that includes the pod name, namespace, and — when
TANGLE_LOG_SEARCH_URL_TEMPLATE is set — a direct link to the pod's logs in
the configured observability platform.

The URL template supports two placeholders substituted at runtime:
  {pod_name}   — Kubernetes pod name
  {start_time} — relative start derived from started_at (e.g. "now-125m",
                 adding 5 min of padding); falls back to "now-1440m" (24 h)
                 if the start time is not available in memory.

Both started_at values (LaunchedKubernetesContainer from pod container state,
LaunchedKubernetesJob from job status) are in-memory reads — no additional
database queries are required to compute the time range.

The placeholder is stored in GCS via upload_log and returned verbatim by the
log-read API, so it surfaces wherever logs are displayed without any frontend
or schema changes.

Author: morgan-wowk

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -69,6 +69,43 @@
 # Environment variables for multi-node execution.
 _MULTI_NODE_NODE_INDEX_ENV_VAR_NAME = "_TANGLE_MULTI_NODE_NODE_INDEX"
 
+# Optional URL template for linking to pod logs in an external observability
+# platform when log acquisition fails. Set via TANGLE_LOG_SEARCH_URL_TEMPLATE.
+# Two placeholders are substituted at runtime:
+#   {pod_name}   — the Kubernetes pod name
+#   {start_time} — relative start time, e.g. "now-125m" (elapsed + 5 min pad),
+#                  or "now-1440m" (24 h) when the pod start time is unavailable.
+# Example for Observe: see oasis-backend deployment config.
+_LOG_SEARCH_URL_TEMPLATE: str | None = os.environ.get("TANGLE_LOG_SEARCH_URL_TEMPLATE")
+
+
+def _format_log_unavailable_message(
+    pod_name: str,
+    namespace: str,
+    started_at: "datetime.datetime | None",
+) -> str:
+    """Return a placeholder log string when the Kubernetes API cannot be read."""
+    msg = (
+        f"[Log unavailable: Kubernetes API returned a malformed response. "
+        f"Pod: {pod_name}, Namespace: {namespace}."
+    )
+    if _LOG_SEARCH_URL_TEMPLATE:
+        if started_at is not None:
+            elapsed_seconds = (
+                datetime.datetime.now(tz=datetime.timezone.utc) - started_at
+            ).total_seconds()
+            # Add 5-minute padding so the window opens before the first log line.
+            elapsed_minutes = max(1, int(elapsed_seconds / 60) + 5)
+            start_time = f"now-{elapsed_minutes}m"
+        else:
+            start_time = "now-1440m"  # 24 h fallback when start time is unknown
+        url = _LOG_SEARCH_URL_TEMPLATE.replace("{pod_name}", pod_name).replace(
+            "{start_time}", start_time
+        )
+        msg += f" Search: {url}"
+    msg += "]\n"
+    return msg
+
 
 _T = typing.TypeVar("_T")
 
@@ -944,7 +981,9 @@ def get_log(self) -> str:
                 self._pod_name,
                 exc_info=True,
             )
-            return ""
+            return _format_log_unavailable_message(
+                self._pod_name, self._namespace, self.started_at
+            )
 
     def upload_log(self):
         launcher = self._get_launcher()
@@ -1528,7 +1567,9 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None:
                 pod_name,
                 exc_info=True,
             )
-            return None
+            return _format_log_unavailable_message(
+                pod_name, self._namespace, self.started_at
+            )
 
     def _get_all_logs(self) -> dict[str, str]:
         logs = {}