Dataflow job logs router

docs/README.md

@@ -32,6 +32,8 @@ external to the above-listed teams to contribute code to this repository.
   compute & storage backends.
 - [Bakeries: job status monitoring](#bakeries-job-status-monitoring) - This is how the FastAPI app
   knows when compute jobs have concluded. Terraform for this infra is run on each Heroku deploy.
+- [Bakeries: querying job logs](#bakeries-querying-job-logs) - Query bakery job logs for recipe runs.
+  Especially useful for diagnosing failed jobs.
 - [Security](#security) - Some notes on security.
 - [Testing](#testing) - Automated testing in both local envs and containerized contexts.
 - [GitHub App: manual API calls](#github-app-manual-api-calls) - How to manually call the GitHub API _as a GitHub App_.
@@ -639,6 +641,18 @@ This project layout is based directly on https://blog.gruntwork.io/how-to-manage
 
 See also [Resource locations](#resource-locations) for a table referencing these terraform envs.
 
+# Bakeries: querying job logs
+
+To query logs for recipe runs, see the API docs at the `/docs#/logs` route of the deployment
+(https://api.pangeo-forge.org/docs#/logs for the production deployment).
+These routes are protected, due to the risk of leaking secrets in the logs. The
+`PANGEO_FORGE_API_KEY` is available to admins via in the deployments secrets config:
+
+```console
+$ sops -d -i secrets/config.${PANGEO_FORGE_DEPLOYMENT}.yaml
+# cat secrets/config.${PANGEO_FORGE_DEPLOYMENT}.yaml
+```
+
 # Security
 
 ## Database API

pangeo_forge_orchestrator/api.py

@@ -8,6 +8,7 @@
 from .http import http_session
 from .metadata import app_metadata
 from .routers.github_app import github_app_router
+from .routers.logs import logs_router
 from .routers.model_router import router as model_router
 from .routers.repr import repr_router
 from .routers.stats import stats_router
@@ -41,6 +42,7 @@ def on_shutdown():
     http_session.stop()
 
 
+app.include_router(logs_router)
 app.include_router(model_router)
 app.include_router(stats_router)
 app.include_router(github_app_router)

pangeo_forge_orchestrator/routers/github_app.py

@@ -620,6 +620,16 @@ async def run(
         try:
             out = subprocess.check_output(cmd)
             logger.debug(f"Command output is {out.decode('utf-8')}")
+            out = subprocess.check_output(cmd)
+            for line in out.splitlines():
+                p = json.loads(line)
+                if p["status"] == "submitted":
+                    existing_message = json.loads(recipe_run.message) if recipe_run.message else {}
+                    recipe_run.message = json.dumps(
+                        existing_message | dict(job_name=p["job_name"], job_id=p["job_id"])
+                    )
+                    db_session.add(recipe_run)
+                    db_session.commit()
         except subprocess.CalledProcessError as e:
             for line in e.output.splitlines():
                 p = json.loads(line)

pangeo_forge_orchestrator/routers/logs.py

@@ -0,0 +1,109 @@
+import json
+import subprocess
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import PlainTextResponse
+from sqlmodel import Session, SQLModel, select
+from starlette import status
+
+from ..dependencies import check_authentication_header, get_session as get_database_session
+from ..models import MODELS
+
+logs_router = APIRouter()
+
+DEFAULT_SEVERITY = Query(
+    "ERROR",
+    description=(
+        "A valid gcloud logging severity as defined in "
+        "https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity"
+    ),
+)
+DEFAULT_LIMIT = Query(1, description="Max number of log entries to return.")
+
+
+def job_id_from_recipe_run(recipe_run: SQLModel) -> str:
+    try:
+        job_id = json.loads(recipe_run.message)["job_id"]
+    except (KeyError, json.JSONDecodeError) as e:
+        detail = (
+            f"Message field of {recipe_run = } missing 'job_id'."
+            if type(e) == KeyError
+            else f"Message field of {recipe_run = } not JSON decodable."
+        )
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)
+
+    return job_id
+
+
+def get_logs(
+    job_id: str,
+    severity: str,
+    limit: int,
+):
+    log_name_prefix = "projects/pangeo-forge-4967/logs/dataflow.googleapis.com"
+    query = (
+        'resource.type="dataflow_step" '
+        f'AND resource.labels.job_id="{job_id}" '
+        f'AND logName=("{log_name_prefix}%2Fjob-message" OR "{log_name_prefix}%2Flauncher") '
+        f'AND severity="{severity}"'
+    )
+    cmd = "gcloud logging read".split() + [query]
+
+    if limit:
+        cmd += [f"--limit={limit}"]
+
+    logs = subprocess.check_output(cmd)
+    return logs
+
+
+@logs_router.get(
+    "/recipe_runs/{id}/logs",
+    summary="Get job logs for a recipe_run, specified by database id.",
+    tags=["recipe_run", "logs", "admin"],
+    response_class=PlainTextResponse,
+    dependencies=[Depends(check_authentication_header)],
+)
+async def logs_from_recipe_run_id(
+    id: int,
+    *,
+    db_session: Session = Depends(get_database_session),
+    severity: str = DEFAULT_SEVERITY,
+    limit: int = DEFAULT_LIMIT,
+):
+    recipe_run = db_session.exec(
+        select(MODELS["recipe_run"].table).where(MODELS["recipe_run"].table.id == id)
+    ).one()
+    job_id = job_id_from_recipe_run(recipe_run)
+    logs = get_logs(job_id, severity, limit)
+    return logs
+
+
+@logs_router.get(
+    "/feedstocks/{feedstock_spec:path}/{commit}/{recipe_id}/logs",
+    summary="Get job logs for a recipe run, specified by feedstock_spec, commit, and recipe_id.",
+    tags=["feedstock", "logs", "admin"],
+    response_class=PlainTextResponse,
+    dependencies=[Depends(check_authentication_header)],
+)
+async def logs_from_feedstock_spec_commit_and_recipe_id(
+    feedstock_spec: str,
+    commit: str,
+    recipe_id: str,
+    *,
+    db_session: Session = Depends(get_database_session),
+    severity: str = DEFAULT_SEVERITY,
+    limit: int = DEFAULT_LIMIT,
+):
+    feedstock = db_session.exec(
+        select(MODELS["feedstock"].table).where(MODELS["feedstock"].table.spec == feedstock_spec)
+    ).one()
+    statement = (
+        select(MODELS["recipe_run"].table)
+        .where(MODELS["recipe_run"].table.recipe_id == recipe_id)
+        .where(MODELS["recipe_run"].table.head_sha == commit)
+        .where(MODELS["recipe_run"].table.feedstock_id == feedstock.id)
+    )
+    recipe_run = db_session.exec(statement).one()
+    job_id = job_id_from_recipe_run(recipe_run)
+    logs = get_logs(job_id, severity, limit)
+    return logs

tests/github_app/mock_pangeo_forge_runner.py

@@ -44,7 +44,15 @@ def mock_subprocess_check_output(cmd: List[str]):
                 '{"message": "Expansion complete", "status": "completed", "meta": {"title": "Global Precipitation Climatology Project", "description": "Global Precipitation Climatology Project (GPCP) Daily Version 1.3 gridded, merged ty satellite/gauge precipitation Climate data Record (CDR) from 1996 to present.\\n", "pangeo_forge_version": "0.9.0", "pangeo_notebook_version": "2022.06.02", "recipes": [{"id": "gpcp", "object": "recipe:recipe"}], "provenance": {"providers": [{"name": "NOAA NCEI", "description": "National Oceanographic & Atmospheric Administration National Centers for Environmental Information", "roles": ["host", "licensor"], "url": "https://www.ncei.noaa.gov/products/global-precipitation-climatology-project"}, {"name": "University of Maryland", "description": "University of Maryland College Park Earth System Science Interdisciplinary Center (ESSIC) and Cooperative Institute for Climate and Satellites (CICS).\\n", "roles": ["producer"], "url": "http://gpcp.umd.edu/"}], "license": "No constraints on data access or use."}, "maintainers": [{"name": "Ryan Abernathey", "orcid": "0000-0001-5999-4917", "github": "rabernat"}], "bakery": {"id": "pangeo-ldeo-nsf-earthcube"}}}\n'
             )
         elif cmd[1] == "bake":
-            return '{"message": ""}'.encode("utf-8")
+            loglines = [
+                {
+                    "message": "Submitted job ABC for recipe run EFG",
+                    "job_id": "ABC",
+                    "job_name": "a132456",
+                    "status": "submitted",
+                },
+            ]
+            return "\n".join([json.dumps(line) for line in loglines]).encode("utf-8")
         else:
             raise NotImplementedError(f"Command {cmd} not implemented in tests.")
     else:

tests/github_app/test_helpers_run.py

@@ -2,7 +2,9 @@
 This module tests the `run` function, which is a special case.
 """
 
+import json
 import subprocess
+from typing import List
 
 import pytest
 import pytest_asyncio
@@ -15,10 +17,7 @@
 
 from ..conftest import clear_database
 from .fixtures import _MockGitHubBackend, get_mock_github_session
-from .mock_pangeo_forge_runner import (
-    mock_subprocess_check_output,
-    mock_subprocess_check_output_raises_called_process_error,
-)
+from .mock_pangeo_forge_runner import mock_subprocess_check_output_raises_called_process_error
 
 
 @pytest_asyncio.fixture(
@@ -107,11 +106,38 @@ async def run_fixture(
 
 
 @pytest.mark.asyncio
-async def test_run(mocker, run_fixture):
+@pytest.mark.parametrize(
+    "job_name, job_id",
+    [
+        (
+            "a6170692e70616e67656f2d666f7267652e6f7267251162",
+            "2022-09-28_13_30_25-9666562633575851936",
+        )
+    ],
+)
+async def test_run(mocker, run_fixture, job_name, job_id, async_app_client):
     run_kws = run_fixture
+
+    def mock_subprocess_check_output(cmd: List[str]):
+        loglines = [
+            {
+                "message": f"Submitted job {job_id} for recipe casm-soil-moisture",
+                "recipe": "casm-soil-moisture",
+                "job_name": job_name,
+                "job_id": job_id,
+                "status": "submitted",
+            },
+        ]
+        return "\n".join([json.dumps(line) for line in loglines]).encode("utf-8")
+
     mocker.patch.object(subprocess, "check_output", mock_subprocess_check_output)
     await run(**run_kws)
 
+    recipe_run = await async_app_client.get("recipe_runs/1")
+    recipe_run_message = recipe_run.json()["message"]
+    assert json.loads(recipe_run_message)["job_name"] == job_name
+    assert json.loads(recipe_run_message)["job_id"] == job_id
+
 
 @pytest.mark.xfail(reason="https://github.com/pangeo-forge/pangeo-forge-orchestrator/issues/132")
 @pytest.mark.asyncio