[golang] Add apim weblog variant for azure/apim-callout

docs/understand/scenarios/README.md

@@ -70,26 +70,34 @@ Please note that it requires the docker daemon to support IPv6. It should be ok
 
 A user has seen his network function altered after running it on a linux laptop (to be investigated). If it happen, `docker network prune` may solve the issue.
 
-### Go proxies (Envoy and HAProxy) scenario
+### Go proxies (Envoy, HAProxy, and APIM) scenario
 
 ```mermaid
 flowchart LR
 %% Nodes
     A("Test runner")
     B("Proxy (Envoy or HAProxy)")
     C("Go security processor")
-    D("HTTP app")
-    E("Proxy")
-    F("Agent")
-    G("Backend")
+    D("http-app<br/>:8080")
+    E("apim-gateway<br/>(shim, :80 -> host 127.0.0.1:7777)")
+    F("apim-callout<br/>(POST :8080, health :8081)<br/>library under test")
+    G("Proxy")
+    H("Agent")
+    I("Backend")
 
 %% Edge connections between nodes
     A --> B --> D
     B --> C --> B
-    C --> E --> F --> G
+    A --> E
+    E --> D
+    E --> F
+    C --> G
+    F --> G --> H --> I
     %% D -- Mermaid js --> I --> J
 ```
 
+For a proxy variant with `build_mode: none`, there is nothing to build: run `./utils/scripts/load-binary.sh golang` to write the image pointers, then `./run.sh <SCENARIO> --weblog <variant>`.
+
 ## Scenario lifecycle
 
 System tests spawn several services before starting. Here is the lifecycle:

tests/external_processing/test_apim_callout.py

@@ -0,0 +1,182 @@
+import re
+from collections import defaultdict
+from pathlib import Path
+
+from utils import context, features, interfaces, irrelevant, weblog
+from utils._weblog import HttpResponse
+from utils.dd_types import DataDogLibrarySpan
+
+
+CALLOUT_LOG_PATTERN = re.compile(
+    r'apim-gateway callout phase=(?P<phase><[^>]+>) request-id="(?P<request_id>[^"]+)" path="(?P<path>[^"]+)" outcome=ok'
+)
+# Deferred (default) body mode: the bodies are not inlined, so the <RequestHeaders> callout answers
+# with `allowed-body-size` and the gateway makes a separate <RequestBody> call. These three phases
+# are guaranteed, in this order.
+DEFERRED_PHASES = ("<RequestHeaders>", "<RequestBody>", "<ResponseHeaders>")
+# Inline body mode: both bodies ride along on the header calls, which is exactly what suppresses
+# `allowed-body-size`, so no body phase is ever requested on either side.
+INLINE_PHASES = ("<RequestHeaders>", "<ResponseHeaders>")
+# A fourth <ResponseBody> phase is possible but NOT guaranteed, so it is tolerated and never
+# required. The gateway only makes it when the <ResponseHeaders> callout answers with
+# `allowed-body-size`, and the callout only asks for the response body when the upstream returned
+# one it can parse. That is a property of the upstream, not of the gateway: the stock `http-app`
+# (jasonrm/dummy-server) answers every request with `text/plain` containing the status code, so
+# this phase never fires in CI. Do not turn it back into a required phase after validating against
+# a substituted upstream -- a JSON-returning stand-in makes it fire and hides this distinction.
+UPSTREAM_DEPENDENT_PHASE = "<ResponseBody>"
+APIM_SPAN = ("web", "server", "apim-callout")
+DEFAULT_PROBE_PATH = "/?probe=default-deferred-body"
+INLINE_PROBE_PATH = "/?probe=inline-two-call"
+TRACE_DEFAULT_PROBE_PATH = "/?probe=trace-default"
+TRACE_INLINE_PROBE_PATH = "/?probe=trace-inline"
+STATE_CLOSURE_PROBE_PATH = "/?probe=inline-state-closure"
+
+
+def _container_stderr(container_name: str) -> str:
+    log_path = Path(context.scenario.host_log_folder) / "docker" / container_name / "stderr.log"
+    return log_path.read_text(encoding="utf-8")
+
+
+def _callout_phases_by_request_id_and_path() -> dict[tuple[str, str], list[str]]:
+    phases_by_request_id_and_path: defaultdict[tuple[str, str], list[str]] = defaultdict(list)
+    for line in _container_stderr("apim-gateway").splitlines():
+        if match := CALLOUT_LOG_PATTERN.fullmatch(line):
+            key = match.group("request_id"), match.group("path")
+            phases_by_request_id_and_path[key].append(match.group("phase"))
+
+    return dict(phases_by_request_id_and_path)
+
+
+def _probe_phase_group(probe_path: str) -> tuple[str, list[str]]:
+    """Return the (request-id, phases) of the single callout group that belongs to `probe_path`.
+
+    Every request reaching the gateway -- including its own bodiless healthcheck -- appends to the
+    same stderr log, so phases are correlated by (request-id, path) and never counted globally.
+    Each probe path is requested exactly once, so exactly one group must match. Zero groups means
+    the gateway log format drifted away from CALLOUT_LOG_PATTERN, which would otherwise make every
+    phase assertion below pass vacuously.
+    """
+    matching_groups = {
+        request_id: phases
+        for (request_id, path), phases in _callout_phases_by_request_id_and_path().items()
+        if path == probe_path
+    }
+    assert len(matching_groups) == 1, (
+        f"expected exactly one callout request-id group for probe {probe_path}, "
+        f"got {len(matching_groups)}: {sorted(matching_groups)}"
+    )
+    return next(iter(matching_groups.items()))
+
+
+def _assert_deferred_probe_phases(probe_path: str) -> None:
+    """Assert `probe_path` was served in deferred body mode: a separate <RequestBody> callout.
+
+    A single trailing <ResponseBody> is accepted because it depends on the upstream returning a
+    parseable response body (see UPSTREAM_DEPENDENT_PHASE); everything before it is required.
+    """
+    request_id, phases = _probe_phase_group(probe_path)
+    required_phases = phases[:-1] if phases[-1:] == [UPSTREAM_DEPENDENT_PHASE] else phases
+    assert required_phases == list(DEFERRED_PHASES), (
+        f"deferred probe {probe_path} (request-id {request_id}) hit callout phases {phases}, "
+        f"expected {list(DEFERRED_PHASES)} optionally followed by {UPSTREAM_DEPENDENT_PHASE}"
+    )
+
+
+def _assert_inline_probe_phases(probe_path: str) -> str:
+    """Assert `probe_path` was served in inline body mode: header phases only, no body phase.
+
+    Returns the request-id, so a caller can scope further assertions to this exchange.
+    """
+    request_id, phases = _probe_phase_group(probe_path)
+    assert phases == list(INLINE_PHASES), (
+        f"inline probe {probe_path} (request-id {request_id}) hit callout phases {phases}, "
+        f"expected {list(INLINE_PHASES)}"
+    )
+    return request_id
+
+
+def _span_structure(span: DataDogLibrarySpan) -> tuple[str, str, str]:
+    return span["type"], span["meta"]["span.kind"], span["meta"]["component"]
+
+
+def _assert_apim_span(span: DataDogLibrarySpan) -> None:
+    assert _span_structure(span) == APIM_SPAN
+
+
+def _trace_structure(request: HttpResponse) -> list[tuple[str, str, str]]:
+    interfaces.library.assert_trace_exists(request=request)
+    assert _span_structure(interfaces.library.get_root_span(request=request)) == APIM_SPAN
+    interfaces.library.validate_all_spans(request=request, validator=_assert_apim_span)
+    return sorted(_span_structure(span) for _, _, span in interfaces.library.get_spans(request=request))
+
+
+@irrelevant(context.weblog_variant != "apim")
+@features.go_proxies
+class Test_ApimCallout:
+    def setup_default_body_mode_defers_request_body_callout(self) -> None:
+        self.r = weblog.post(DEFAULT_PROBE_PATH, json={"body": "default"})
+
+    def test_default_body_mode_defers_request_body_callout(self) -> None:
+        """Without the inline header, the request body is fetched by its own <RequestBody> callout."""
+        assert self.r.status_code == 200, f"deferred probe returned {self.r.status_code}, expected 200"
+        _assert_deferred_probe_phases(DEFAULT_PROBE_PATH)
+
+    def setup_inline_body_mode_uses_two_callouts(self) -> None:
+        self.r = weblog.post(INLINE_PROBE_PATH, json={"body": "inline"}, headers={"X-Datadog-Apim-Body-Mode": "inline"})
+
+    def test_inline_body_mode_uses_two_callouts(self) -> None:
+        """Inlining the body on the header calls removes the body phases, leaving exactly two calls."""
+        assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200"
+        _assert_inline_probe_phases(INLINE_PROBE_PATH)
+
+    def setup_inline_body_mode_preserves_trace_structure(self) -> None:
+        self.default_response = weblog.post(TRACE_DEFAULT_PROBE_PATH, json={"body": "default trace"})
+        self.inline_response = weblog.post(
+            TRACE_INLINE_PROBE_PATH,
+            json={"body": "inline trace"},
+            headers={"X-Datadog-Apim-Body-Mode": "inline"},
+        )
+
+    def test_inline_body_mode_preserves_trace_structure(self) -> None:
+        assert self.default_response.status_code == 200, (
+            f"deferred probe returned {self.default_response.status_code}, expected 200"
+        )
+        assert self.inline_response.status_code == 200, (
+            f"inline probe returned {self.inline_response.status_code}, expected 200"
+        )
+        _assert_deferred_probe_phases(TRACE_DEFAULT_PROBE_PATH)
+        _assert_inline_probe_phases(TRACE_INLINE_PROBE_PATH)
+        default_spans = _trace_structure(self.default_response)
+        inline_spans = _trace_structure(self.inline_response)
+        assert default_spans == inline_spans, (
+            f"inline body delivery changed the trace: deferred spans {default_spans}, inline spans {inline_spans}"
+        )
+
+    def setup_inline_body_mode_closes_request_state(self) -> None:
+        self.r = weblog.post(
+            STATE_CLOSURE_PROBE_PATH,
+            json={"body": "inline state"},
+            headers={"X-Datadog-Apim-Body-Mode": "inline"},
+        )
+
+    def test_inline_body_mode_closes_request_state(self) -> None:
+        assert self.r.status_code == 200, f"inline probe returned {self.r.status_code}, expected 200"
+        request_id = _assert_inline_probe_phases(STATE_CLOSURE_PROBE_PATH)
+        # The callout evicts cached request state after a 30s TTL and logs one warning naming the
+        # request-id. Inline mode deletes the state on the response-headers call, so that warning
+        # must never appear for this exchange. No wait is needed here: every setup_ runs before any
+        # test_, and the containers are stopped and their logs collected in between, so far more
+        # than the TTL has elapsed by the time this assertion reads the log.
+        #
+        # Scoped to this request-id on purpose. Matching the bare warning text would also fail on
+        # an unrelated request that legitimately orphaned state, and attribute it to inline mode.
+        orphaned = [
+            line
+            for line in _container_stderr("apim-callout").splitlines()
+            if "closing orphaned span" in line and request_id in line
+        ]
+        assert not orphaned, (
+            f"apim-callout orphaned the cached state for inline request-id {request_id}, "
+            f"so inline mode did not close it: {orphaned}"
+        )

tests/external_processing/test_apm.py

@@ -1,7 +1,7 @@
 from utils import weblog, interfaces, features, context, irrelevant
 
 
-@irrelevant(context.weblog_variant not in ("haproxy", "envoy"))
+@irrelevant(context.weblog_variant not in ("haproxy", "envoy", "apim"))
 @features.go_proxies
 class Test_GoProxies_Tracing:
     def setup_correct_span_structure(self):

tests/test_semantic_conventions.py

@@ -36,6 +36,7 @@
     "django-py3.13": "django",
     "python3.12": "django",
     "gin": "gin-gonic/gin",
+    "apim": "apim-callout",
     "haproxy": "haproxy-spoa",
     "gqlgen": "99designs/gqlgen",
     "graph-gophers": "graph-gophers/graphql-go",

tests/test_the_test/test_ci_orchestrator.py

@@ -286,7 +286,7 @@ def _is_uds_weblog(weblog: str) -> bool:
             return False
 
     # Go proxies
-    if weblog.name in ("envoy", "haproxy"):
+    if weblog.name in ("envoy", "haproxy", "apim"):
         if scenario.name not in ("DEFAULT", "APPSEC_BLOCKING"):
             return False
 

tests/test_the_test/test_get_image_list.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import subprocess
+import sys
+from typing import TYPE_CHECKING
+
+import pytest
+import yaml
+
+from utils import scenarios
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+SCRIPT = Path("utils/scripts/get-image-list.py")
+SCENARIOS = "APPSEC_BLOCKING,DEFAULT"
+
+# .github/actions/pull_images redirects the script's stdout into compose.yaml, then feeds that file
+# to `docker compose`. Container objects are built during that call, so anything a container
+# constructor writes on stdout becomes part of the compose document.
+#
+# The go proxy weblogs are the ones exposed to this: each of them resolves its processor image from
+# an optional pointer file in binaries/, a branch that has historically been tempting to report on
+# stdout.
+GO_PROXY_POINTER_FILES = {
+    "apim": Path("binaries/golang-apim-callout-image"),
+    "envoy": Path("binaries/golang-service-extensions-callout-image"),
+    "haproxy": Path("binaries/golang-haproxy-spoa-image"),
+}
+
+
+@contextmanager
+def _pointer_file(path: Path, *, present: bool) -> Iterator[None]:
+    """Force the presence or the absence of a binaries/ image pointer file.
+
+    A pointer file already sitting there is a local build artifact: move it aside and put it back,
+    so both states can be tested whatever the checkout looks like.
+    """
+
+    backup = path.parent / f"{path.name}.test_the_test_backup"
+    existed = path.is_file()
+
+    if existed:
+        path.rename(backup)
+
+    try:
+        if present:
+            path.write_text("ghcr.io/datadog/system-tests/fake-processor:test-the-test\n", encoding="utf-8")
+
+        yield
+    finally:
+        path.unlink(missing_ok=True)
+
+        if existed:
+            backup.rename(path)
+
+
+def _run_get_image_list(weblog: str) -> str:
+    result = subprocess.run(
+        [sys.executable, str(SCRIPT), SCENARIOS, "-l=golang", f"-w={weblog}"],
+        check=False,
+        capture_output=True,
+        text=True,
+        env={**os.environ, "PYTHONPATH": "."},
+    )
+
+    assert result.returncode == 0, result.stderr
+
+    return result.stdout
+
+
+@scenarios.test_the_test
+class Test_GetImageList:
+    @pytest.mark.parametrize("weblog", sorted(GO_PROXY_POINTER_FILES))
+    @pytest.mark.parametrize("pointer_present", [False, True], ids=["pointer_absent", "pointer_present"])
+    def test_stdout_is_only_a_compose_document(self, weblog: str, pointer_present: bool):  # noqa: FBT001
+        """get-image-list.py stdout is a compose file, it must not carry anything else.
+
+        A container constructor calling logger.stdout() would inject its message as an extra
+        top-level key, and `docker compose` would then reject the generated compose.yaml.
+        """
+
+        with _pointer_file(GO_PROXY_POINTER_FILES[weblog], present=pointer_present):
+            stdout = _run_get_image_list(weblog)
+
+        document = yaml.safe_load(stdout)
+
+        assert isinstance(document, dict), f"stdout for weblog {weblog} is not a YAML mapping:\n{stdout}"
+        assert sorted(document) == ["services"], (
+            f"stdout for weblog {weblog} must only contain the services key. "
+            f"A container constructor is writing on stdout instead of logging:\n{stdout}"
+        )
+        assert isinstance(document["services"], dict), f"services must be a mapping:\n{stdout}"