SEP-1941: Add a diagnostics-delivery target to the connectivity check (#1441)

Adds a `delivery` target to `POST /api/sep/admin/connectivity-check/`,
so an
operator can test the configured diagnostics-delivery receiver without
sending a
bundle. It fans out alongside the existing four targets and is isolated
the same
way — a delivery failure never fails the response or the other targets.

**The probe request is declared on the delivery plan, not derived from
its send
steps.** Replaying the send steps is not possible three ways over: a
resolution
step may be a mutating `POST` (`sidecar/settings.yaml:111` declares
one), its
values may cite the `source_ref` / `case_ref` / `manifest` send inputs
that only
exist during a real send, and the upload step's values may cite an
earlier step's
outputs. So `DeliveryPlan` gains an optional `probe` field, sibling to
`upload`:

- `ProbeStep` carries `path`, `headers` and `query` only — no `name` (so
it joins
no step namespace and cannot collide with a configured step), no
`method` (GET
  by construction), no `body`, no `outputs`.
- Its value maps are typed with `ProbeValue`, a discriminated union
narrowed to
`literal` and `secret`. The other three sources are refused by pydantic
at
  parse time with `union_tag_invalid`, rather than by a validator —
`_check_value` (`app/sep/bundle_upload/plan.py`) branches only on
`SecretValue`
and `StepOutputValue`, so cross-reference validation alone would let
`input`
and `manifest_key` through. The two rules that validator does own — a
secret
  must be declared, and no secret in `query` — are reused unchanged.
- `ProbeStep.path` is validated as a relative reference.
`URL(origin).join()`
resolves a `//host/p` or `https://host/p` path *instead of* the endpoint
rather
than under it, which would send the probe's credentials to a host the
plan
  never named. This is a misconfiguration guard, not an attacker guard:
`DIAGNOSTICS_DELIVERY` is settable through the settings file and
environment
variables only, never the settings API, so anyone who can write a probe
path
  can already repoint `endpoint`.

`DeliveryPlanExecutor.probe()` issues that one GET through
`get_delivery_executor`, reusing the existing per-call, unpooled
transport and
`redact_headers(_secret_valued_keys(...))`. It runs no resolution step
and no
upload, and records no step trail.

**It reads no response body either.** `RemoteAPI.request()` parses every
body
but a `204` before it checks the status, so a receiver whose health
route
answers `200 text/plain` reaches the caller as a `2xx` `HTTPException`
and
would have been reported as an upstream error. `upload()` already
documents
and tolerates that case inline; this promotes its condition to
`is_non_json_success()` in `app/core/requests/remote_api.py` and applies
it in
both callers. A non-JSON *error* status still fails (a `401 text/html`
still
reports `auth_failed`), and an unfollowed redirect still fails — that
branch
raises before the body is parsed, so it carries no non-JSON stamp.

`_rebased_plan` gains a `probe` branch. It enumerates step kinds by
hand, so
without it a plan endpoint carrying a path prefix or query would have
probed the
wrong URL — and only a prefixed-endpoint test can detect that, since the
unrebased path is already correct under a root endpoint.
`_rebase_query`'s first
parameter widens from `dict` to `Mapping` so the probe's narrower value
map goes
through the same helper; no behaviour changes.

`DeliveryPlanResolution` gains a `DeliveryUnavailableCode`, so the route
tells an
unconfigured deployment from one whose stored inputs have drifted
without
matching on the resolver's prose; that prose still travels verbatim as
the result
`detail`. The dataclass invariant now requires the code and reason to be
set
together. `unavailable()` takes the code as a required argument, and a
new
`unconfigured()` classmethod names the outcome its three call sites
share. Both
existing consumers (`app/sep/apps/atw/deps.py`,
`app/sep/apps/atw/send.py`) read
only `.unavailable_reason` and are unchanged.

`ConnectivityStatusEnum` widens by three members — `not_configured`,
`inputs_drifted`, `probe_undeclared` — each with a `_DEFAULT_DETAILS`
entry,
which is required rather than optional: `build_connectivity_result`
indexes that
dict unguarded. `classify_connectivity_error` is unchanged; the three
are only
ever passed explicitly.

The delivery probe takes its own `EXTERNAL_PROBE_TIMEOUT_SECONDS = 15`,
declared
beside `PROBE_TIMEOUT_SECONDS`, which stays 5s for the four
intra-cluster
targets. Because targets are gathered concurrently, this caps the
worst-case
whole-response wait at 15s rather than lengthening it per target.

The Settings → Test connection panel requests the new target and labels
all three
new statuses. `ALL_TARGETS` and `STATUS_CHIP` are hardcoded in
`TestConnectionButton.tsx`, so a backend-only change would have shipped
a target
no UI ever asks for; a test now asserts the requested target list. The
settings
e2e spec gains the fifth row and two of the three new statuses — its
comment
claiming every status is rendered end to end was true for four targets
and is
now stated accurately, with `inputs_drifted` covered by the component
test
instead.

**Behaviour change on an existing target:** an unconfigured PMM reports
`not_configured` instead of `unreachable`. `reachable` stays `false` and
the
`detail` text is unchanged, so only a consumer branching on the status
value is
affected. It ships with its own changelog fragment.

## Known limitations

- **The side-car's baked probe block is not included.**
`sidecar/settings.yaml`
points at production ServiceNow, and this branch has no production
credential
to confirm the connector key may read the probe target. The path is
confirmed
  working against the development instance (`perconadev.service-now.com`
returns 200 with a real key, 401 with a bogus one or none), but
production
authenticates before resolving a table, so a credential-free check there
establishes nothing about the read ACL. Shipping an unverified block
would let
the side-car report `auth_failed` against a receiver that would accept a
real
send — worse than no probe, because it tells an operator to fix a
credential
that is fine. The side-car answers `probe_undeclared` until a follow-up
supplies a verified path; the ticket's In Scope explicitly provides for
this.
  The owner to ask is the ServiceNow connector owner named on
  https://perconadev.atlassian.net/browse/PMM-15390
  <!-- followup -->
- **`ResolutionStep.path` and `UploadStep.path` carry the same
off-origin hazard
the new `ProbeStep.path` validator closes.** They are not fixed here:
adding
the validator to them would change validation for every already-deployed
plan,
and a plan that fails validation resolves to "delivery unavailable",
turning
  delivery off on upgrade.
  <!-- followup -->
- **A connect-phase failure reports `unreachable`, not `timeout`.**
`BaseRemoteAPI.__aenter__` sets a client timeout of `total=300,
connect=5,
sock_connect=5, sock_read=120` in app/core/requests/remote_api.py:357,
so a
receiver slow to *connect* fails at aiohttp's 5s connector bound before
the
probe's own 15s bound is reached. This is deliberate and left alone: the
probe's timeout profile then matches a real send's, which is what a
"test this
connection" button should report. Widening `RemoteAPI` would change
every
  caller.
- **A misspelled `probe:` key is silently ignored.** `DeliveryPlan`
declares no
`model_config`, so pydantic's default `extra='ignore'` applies. This is
the
same property that makes a settings file carrying `probe:` safe to
deploy
against an older image — it is ignored, not rejected — so it is kept
rather
  than closed.

Author: yyyyyyyan

app/core/requests/connectivity.py

@@ -31,6 +31,13 @@
 #: timeouts.
 PROBE_TIMEOUT_SECONDS = 5
 
+#: Upper bound for a probe that leaves the cluster. Named for the property that
+#: justifies the wider bound -- a round trip over the public internet -- rather
+#: than for any one caller, so this module stays free of feature knowledge.
+#: Probes run concurrently, so this caps the whole fan-out rather than adding to
+#: it.
+EXTERNAL_PROBE_TIMEOUT_SECONDS = 15
+
 #: HTTP statuses that mean "the server answered, but rejected our credentials".
 _AUTH_FAILURE_STATUSES = frozenset(
     {status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN}
@@ -46,18 +53,27 @@ class ConnectivityStatusEnum(StrEnum):
     UNREACHABLE = "unreachable"
     SSL_ERROR = "ssl_error"
     TIMEOUT = "timeout"
+    NOT_CONFIGURED = "not_configured"
+    INPUTS_DRIFTED = "inputs_drifted"
+    PROBE_UNDECLARED = "probe_undeclared"
 
 
 #: Human-readable default ``detail`` per outcome. Deliberately fixed strings so
 #: the probe never echoes the configured API key or any credential embedded in
-#: an endpoint URL.
+#: an endpoint URL. Every member needs an entry: ``build_connectivity_result``
+#: indexes this unguarded whenever a caller passes no explicit ``detail``.
 _DEFAULT_DETAILS: dict[ConnectivityStatusEnum, str] = {
     ConnectivityStatusEnum.REACHABLE: "Reachable.",
     ConnectivityStatusEnum.AUTH_FAILED: "Authentication failed.",
     ConnectivityStatusEnum.ERROR: "Endpoint returned an error response.",
     ConnectivityStatusEnum.UNREACHABLE: "Connection failed.",
     ConnectivityStatusEnum.SSL_ERROR: "SSL verification failed.",
     ConnectivityStatusEnum.TIMEOUT: "Connection timed out.",
+    ConnectivityStatusEnum.NOT_CONFIGURED: "Not configured.",
+    ConnectivityStatusEnum.INPUTS_DRIFTED: (
+        "Stored inputs no longer match the configured plan."
+    ),
+    ConnectivityStatusEnum.PROBE_UNDECLARED: "No connectivity probe is declared.",
 }
 
 

app/core/requests/remote_api.py

@@ -20,6 +20,7 @@
     "BaseRemoteAPI",
     "RemoteAPI",
     "exception_for_status",
+    "is_non_json_success",
 ]
 
 import asyncio
@@ -159,6 +160,24 @@ def exception_for_status(
     return exc_class(detail, headers=headers)
 
 
+def is_non_json_success(exc: HTTPException) -> bool:
+    """Return whether ``exc`` reports a successful answer whose body was not JSON.
+
+    :meth:`RemoteAPI.request` parses every body but a ``204`` before it checks
+    the status, so a receiver answering ``200 text/plain`` (an acknowledgement
+    string, an HTML health page, an empty non-``204`` body) surfaces as a
+    ``2xx`` :class:`fastapi.HTTPException` rather than as the success it is.
+    Callers that do not need the parsed body use this to tell that case from a
+    real upstream error.
+
+    :param exc: The exception :meth:`RemoteAPI.request` raised.
+    :return: ``True`` when the status is below 400 and the body was not JSON.
+    """
+    return exc.status_code < status.HTTP_400_BAD_REQUEST and bool(
+        (exc.headers or {}).get(UPSTREAM_NON_JSON_HEADER)
+    )
+
+
 def _sanitize_request_kwargs(
     kwargs: dict[str, Any],
     *,
@@ -1104,8 +1123,6 @@ async def upload(
                 "POST", path, data=payload, headers=headers, **kwargs
             )
         except HTTPException as exc:
-            if exc.status_code < status.HTTP_400_BAD_REQUEST and (
-                exc.headers or {}
-            ).get(UPSTREAM_NON_JSON_HEADER):
+            if is_non_json_success(exc):
                 return None
             raise

app/sep/api/routes/connectivity_check.py

@@ -16,12 +16,12 @@
 """Define the admin-only ``/api/sep/admin/connectivity-check`` endpoint.
 
 Expose a single generic ``POST`` that probes the caller-specified external /
-inter-service endpoints (PMM, Inventory, Tasks, Nomad) on demand and reports
-normalized per-endpoint connectivity status, so an admin can confirm an
-endpoint or credential change is reachable and valid before relying on it. The
-request must name which services to probe (``targets``, required, no default),
-so the settings flow can validate only the endpoint being edited while a full
-sweep still names all four.
+inter-service endpoints (PMM, Inventory, Tasks, Nomad, and the diagnostics
+delivery receiver) on demand and reports normalized per-endpoint connectivity
+status, so an admin can confirm an endpoint or credential change is reachable
+and valid before relying on it. The request must name which services to probe
+(``targets``, required, no default), so the settings flow can validate only the
+endpoint being edited while a full sweep still names them all.
 
 The probes are driven by the overridable ``RemoteAPI.check_connectivity``
 capability and fan out concurrently. A failure for one endpoint is captured and
@@ -46,9 +46,15 @@
     classify_connectivity_error,
     ConnectivityResult,
     ConnectivityStatusEnum,
+    EXTERNAL_PROBE_TIMEOUT_SECONDS,
     PROBE_TIMEOUT_SECONDS,
 )
 from app.core.requests.remote_api import UPSTREAM_NON_JSON_HEADER
+from app.sep.bundle_upload.factory import get_delivery_executor
+from app.sep.bundle_upload.resolver import (
+    DeliveryUnavailableCode,
+    resolve_delivery_plan,
+)
 from app.sep.deps import InventoryAPI, PMMAPIDep, TaskAPI
 
 router = APIRouter()
@@ -61,6 +67,7 @@ class ServiceEnum(StrEnum):
     INVENTORY = "inventory"
     TASKS = "tasks"
     NOMAD = "nomad"
+    DELIVERY = "delivery"
 
 
 class ConnectivityCheckRequest(BaseModel):
@@ -92,7 +99,7 @@ async def _probe_pmm(pmm_api: PMMAPIDep) -> ConnectivityResult:
     if pmm_api is None:
         return build_connectivity_result(
             ServiceEnum.PMM,
-            ConnectivityStatusEnum.UNREACHABLE,
+            ConnectivityStatusEnum.NOT_CONFIGURED,
             detail="PMM is not configured.",
         )
     return await pmm_api.check_connectivity(ServiceEnum.PMM)
@@ -150,6 +157,59 @@ async def _probe_tasks_and_nomad(
     )
 
 
+#: The connectivity status each unavailability code reports as. Delivery being
+#: unconfigured and its stored inputs having drifted are separate outcomes
+#: because only the second is fixed by re-supplying the inputs.
+_DELIVERY_UNAVAILABLE_STATUS: dict[DeliveryUnavailableCode, ConnectivityStatusEnum] = {
+    DeliveryUnavailableCode.UNCONFIGURED: ConnectivityStatusEnum.NOT_CONFIGURED,
+    DeliveryUnavailableCode.DRIFTED_INPUTS: ConnectivityStatusEnum.INPUTS_DRIFTED,
+}
+
+
+async def _probe_delivery() -> ConnectivityResult:
+    """Report whether the configured diagnostics-delivery receiver answers.
+
+    Issues the delivery plan's own declared probe request rather than replaying
+    its send steps: a resolution step may mutate state, and its values may cite
+    inputs that exist only during a real send. A plan that declares no probe is
+    reported as such instead of being guessed at.
+
+    The resolver's prose reason is passed through verbatim as the detail, while
+    the status carries the same distinction machine-readably.
+
+    Branches on the resolution's code rather than its plan, so the unavailable
+    outcomes map through one table instead of being re-derived from the prose.
+    That branch narrows nothing for a type checker, so the ``plan is None`` arm
+    below is what narrows the optional in place of an assertion; the
+    resolution's own invariant is what makes that arm unreachable.
+
+    :return: The delivery connectivity result.
+    """
+    resolution = resolve_delivery_plan()
+    if (code := resolution.code) is not None:
+        return build_connectivity_result(
+            ServiceEnum.DELIVERY,
+            _DELIVERY_UNAVAILABLE_STATUS[code],
+            detail=resolution.unavailable_reason,
+        )
+    plan = resolution.plan
+    if plan is None or plan.probe is None:
+        return build_connectivity_result(
+            ServiceEnum.DELIVERY, ConnectivityStatusEnum.PROBE_UNDECLARED
+        )
+    try:
+        async with asyncio.timeout(EXTERNAL_PROBE_TIMEOUT_SECONDS):
+            async with get_delivery_executor(plan) as executor:
+                await executor.probe()
+    except Exception as exc:  # noqa: BLE001 -- classified, never re-raised
+        return build_connectivity_result(
+            ServiceEnum.DELIVERY, classify_connectivity_error(exc)
+        )
+    return build_connectivity_result(
+        ServiceEnum.DELIVERY, ConnectivityStatusEnum.REACHABLE
+    )
+
+
 @router.post("/")
 async def check_connectivity(
     body: ConnectivityCheckRequest,
@@ -188,6 +248,8 @@ async def check_connectivity(
         )
     if want_tasks or want_nomad:
         probes["_tasks_nomad"] = _probe_tasks_and_nomad(tasks_api)
+    if ServiceEnum.DELIVERY in targets:
+        probes[ServiceEnum.DELIVERY] = _probe_delivery()
 
     completed = dict(zip(probes, await asyncio.gather(*probes.values()), strict=False))
 

app/sep/bundle_upload/factory.py

@@ -24,7 +24,7 @@
 
 __all__ = ["get_delivery_executor", "split_endpoint"]
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
 from urllib.parse import parse_qsl, urlparse
 
@@ -69,10 +69,14 @@ def _rebase_path(prefix: str, path: str) -> str:
 
 
 def _rebase_query(
-    query: dict[str, PlanValue], endpoint_query: dict[str, str]
+    query: Mapping[str, PlanValue], endpoint_query: dict[str, str]
 ) -> dict[str, PlanValue]:
     """Merge the endpoint's query pairs into a step's own, without displacing them.
 
+    Accepts any value map the plan admits, so a step whose values are drawn
+    from a narrower set than :data:`~app.sep.bundle_upload.plan.PlanValue`,
+    as the probe's are, rebases through the same helper.
+
     :param query: The step's configured query map.
     :param endpoint_query: The query pairs carried by the configured endpoint.
     :return: The merged query map.
@@ -94,6 +98,10 @@ def _rebased_plan(
     than shared: the plan handed in is the process-global configured one, and
     mutating its steps would corrupt every later send.
 
+    Every step kind the plan carries is enumerated here by hand, so a step kind
+    added to :class:`~app.sep.bundle_upload.plan.DeliveryPlan` without a branch
+    below is silently carried through unrebased and issued at the wrong URL.
+
     :param plan: The configured plan to rebase.
     :param path: The endpoint's path.
     :param query: The endpoint's query pairs.
@@ -114,7 +122,19 @@ def _rebased_plan(
             "query": _rebase_query(plan.upload.query, query),
         }
     )
-    return plan.model_copy(update={"resolution_steps": steps, "upload": upload})
+    probe = (
+        None
+        if plan.probe is None
+        else plan.probe.model_copy(
+            update={
+                "path": _rebase_path(path, plan.probe.path),
+                "query": _rebase_query(plan.probe.query, query),
+            }
+        )
+    )
+    return plan.model_copy(
+        update={"resolution_steps": steps, "upload": upload, "probe": probe}
+    )
 
 
 @asynccontextmanager