fix(central): read storage and telemetry from their own instance metadata attributes

docs/configuration.md

@@ -213,7 +213,7 @@ email_recipients = ["ops@example.com"]
 
 Shared tables are MariaDB, Postgres, Let's Encrypt, Central, the edge proxy, telemetry, resource limits, and the admin JWKS issuer. A bench exposes these values through its own `BenchConfig`; the model merges shared values on read and writes them back to the common file.
 
-Central endpoint and authentication data come from instance metadata. While Central is enabled and bootstrap is pending, remote-token verification uses only the current staged JWKS URL, audience, and initial key set from instance metadata. Pilot caches that staged issuer in each Admin process until shared config changes or bootstrap completes. After bootstrap, it clears the staged entry and uses only the issuer saved in shared config. The metadata can include `initial_jwks_cache`, the issuer's JWK set. Pilot uses it to initialize an empty JWKS cache before it marks the host bootstrapped, so the first remote token does not wait for an issuer fetch. It does not replace keys already fetched from the issuer. `central.hostname_aliases` maps a VM hostname pattern to its current local target. The VM ID is assigned at runtime, so use `*` for that part. Pilot creates redirect rules only for aliases whose targets exist on the bench. Renaming a site or moving the admin domain re-points the matching alias automatically; remove one when the rule is no longer needed. `pilot setup central` writes these settings - see [Setup Commands](commands.md#setup-commands).
+Central endpoint and authentication data come from instance metadata. While Central is enabled and bootstrap is pending, remote-token verification uses only the current staged JWKS URL, audience, and initial key set from instance metadata. Pilot caches that staged issuer in each Admin process until shared config changes or bootstrap completes. After bootstrap, it clears the staged entry and uses only the issuer saved in shared config. The metadata can include `initial_jwks_cache`, the issuer's JWK set. Pilot uses it to initialize an empty JWKS cache before it marks the host bootstrapped, so the first remote token does not wait for an issuer fetch. It does not replace keys already fetched from the issuer. The Central configuration is in the `pilot-central` attribute. The team's backup bucket is in `pilot-storage`, and the Datum endpoint and token are in `pilot-telemetry`. Both are optional. Each has its own attribute because the cloud caps a metadata value at 1 KiB. `central.hostname_aliases` maps a VM hostname pattern to its current local target. The VM ID is assigned at runtime, so use `*` for that part. Pilot creates redirect rules only for aliases whose targets exist on the bench. Renaming a site or moving the admin domain re-points the matching alias automatically; remove one when the rule is no longer needed. `pilot setup central` writes these settings - see [Setup Commands](commands.md#setup-commands).
 
 The domain provider controls the edge route for each hostname. Its route policy gives Pilot the public scheme, origin scheme, and client IP source.
 

pilot/integrations/central/metadata.py

@@ -3,7 +3,7 @@
 import json
 import os
 import urllib.request
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, TypedDict
 
 from pilot.integrations.central.client import CentralClientError
 
@@ -20,21 +20,39 @@
 TELEMETRY_KEYS = ("endpoint", "token")
 
 
+class MetadataBlock(TypedDict):
+    attribute: str
+    keys: tuple[str, ...]
+    url_key: str
+
+
+# The cloud caps each metadata value at 1 KiB, so each optional block has its own attribute.
+BLOCKS: dict[str, MetadataBlock] = {
+    "s3": {"attribute": "pilot-storage", "keys": S3_KEYS, "url_key": "endpoint_url"},
+    "telemetry": {"attribute": "pilot-telemetry", "keys": TELEMETRY_KEYS, "url_key": "endpoint"},
+}
+
+
 def attribute_name() -> str:
     return os.environ.get("PILOT_METADATA_KEY", "pilot-central")
 
 
-def _parse_credentials(raw: str, name: str) -> dict[str, Any]:
-    """Anything rejected here is a provisioning bug, not a host still waiting."""
-    from pilot.internal.validators import validate_external_url
-
-    source = f"Instance metadata '{name}'"
+def _parse_object(raw: str, source: str) -> dict[str, Any]:
     try:
         payload = json.loads(raw)
     except ValueError as exc:
         raise CentralClientError(f"{source} is not JSON: {exc}") from exc
     if not isinstance(payload, dict):
         raise CentralClientError(f"{source} is not a JSON object.")
+    return payload
+
+
+def _parse_credentials(raw: str, name: str) -> dict[str, Any]:
+    """Anything rejected here is a provisioning bug, not a host still waiting."""
+    from pilot.internal.validators import validate_external_url
+
+    source = f"Instance metadata '{name}'"
+    payload = _parse_object(raw, source)
 
     if missing := [key for key in REQUIRED_KEYS if not payload.get(key)]:
         raise CentralClientError(f"{source} is missing: {', '.join(missing)}")
@@ -49,28 +67,19 @@ def _parse_credentials(raw: str, name: str) -> dict[str, Any]:
             raise CentralClientError(f"{source}: initial_jwks_cache is not a JSON object.")
         credentials["initial_jwks_cache"] = initial_jwks_cache
 
-    for name, keys, url_key in (("s3", S3_KEYS, "endpoint_url"), ("telemetry", TELEMETRY_KEYS, "endpoint")):
-        if (block := _parse_block(payload.get(name), source, name, keys, url_key)) is not None:
-            credentials[name] = block
-
     return credentials
 
 
-def _parse_block(
-    block: Any, source: str, name: str, keys: tuple[str, ...], url_key: str
-) -> dict[str, str] | None:
-    """An optional block of the attribute: every key set, and its URL safe to call."""
+def _parse_block(raw: str, attribute: str, keys: tuple[str, ...], url_key: str) -> dict[str, str]:
     from pilot.internal.validators import validate_external_url
 
-    if block is None:
-        return None
-    if not isinstance(block, dict):
-        raise CentralClientError(f"{source}: {name} is not a JSON object.")
+    source = f"Instance metadata '{attribute}'"
+    block = _parse_object(raw, source)
     if missing := [key for key in keys if not block.get(key)]:
-        raise CentralClientError(f"{source}: {name} is missing: {', '.join(missing)}")
+        raise CentralClientError(f"{source} is missing: {', '.join(missing)}")
 
     values = {key: str(block[key]) for key in keys}
-    if error := validate_external_url(values[url_key], f"{name}.{url_key}"):
+    if error := validate_external_url(values[url_key], url_key):
         raise CentralClientError(f"{source}: {error}")
     return values
 
@@ -86,7 +95,14 @@ def get_credentials(self) -> dict[str, Any] | None:
         """The staged credential, or None until the cloud writes it. Malformed raises."""
         name = attribute_name()
         raw = self.get_attribute(name)
-        return _parse_credentials(raw, name) if raw else None
+        if not raw:
+            return None
+
+        credentials = _parse_credentials(raw, name)
+        for block, spec in BLOCKS.items():
+            if value := self.get_attribute(spec["attribute"]):
+                credentials[block] = _parse_block(value, spec["attribute"], spec["keys"], spec["url_key"])
+        return credentials
 
     def get_attribute(self, name: str) -> str | None:
         token = self._token()

tests/admin/backend/test_central_bootstrap.py

@@ -34,7 +34,7 @@ def _staged(value: str | None):
     """The raw attribute, so the real parsing still runs."""
     return patch(
         "pilot.integrations.central.metadata.InstanceMetadata.get_attribute",
-        return_value=value,
+        side_effect=lambda name: value if name == "pilot-central" else None,
     )
 
 

tests/pilot/integrations/test_central_metadata.py

@@ -33,22 +33,23 @@
 class _FakeMetadata(InstanceMetadata):
     """The metadata service. A None value means the attribute is unset."""
 
-    def __init__(self, value: str | None) -> None:
+    def __init__(self, value: str | None, storage: str | None = None, telemetry: str | None = None) -> None:
         super().__init__()
         self.value = value
+        self.blocks = {"pilot-storage": storage, "pilot-telemetry": telemetry}
         self.requested: list[str] = []
 
     def get_attribute(self, name: str) -> str | None:
         self.requested.append(name)
-        return self.value
+        return self.blocks.get(name, self.value)
 
 
 def test_credentials_come_back_from_the_configured_attribute(monkeypatch) -> None:
     monkeypatch.setenv("PILOT_METADATA_KEY", "my-key")
     metadata = _FakeMetadata(json.dumps(_ATTRIBUTE))
 
     assert metadata.get_credentials() == _ATTRIBUTE
-    assert metadata.requested == ["my-key"]
+    assert metadata.requested == ["my-key", "pilot-storage", "pilot-telemetry"]
 
 
 def test_an_unset_attribute_is_not_an_error() -> None:
@@ -140,7 +141,7 @@ def test_the_initial_jwks_cache_comes_back_with_the_credentials() -> None:
 
 
 def test_storage_configuration_comes_back_with_the_credentials() -> None:
-    credentials = _FakeMetadata(json.dumps({**_ATTRIBUTE, "s3": _S3})).get_credentials()
+    credentials = _FakeMetadata(json.dumps(_ATTRIBUTE), storage=json.dumps(_S3)).get_credentials()
 
     assert credentials["s3"] == _S3
 
@@ -172,7 +173,7 @@ def on_credentials(credentials) -> None:
 def test_apply_saves_central_storage_as_the_default_s3_config(tmp_path: Path) -> None:
     bench = _awaiting_bench(tmp_path)
 
-    apply_central_config(bench, _FakeMetadata(json.dumps({**_ATTRIBUTE, "s3": _S3})))
+    apply_central_config(bench, _FakeMetadata(json.dumps(_ATTRIBUTE), storage=json.dumps(_S3)))
 
     saved = BenchConfig.read(bench.path)
     assert saved.s3.access_key == "garage-access"
@@ -192,39 +193,48 @@ def test_apply_preserves_an_existing_provider_config(tmp_path: Path) -> None:
         config.s3.provider = "aws"
         config.s3.region = "us-east-1"
 
-    apply_central_config(bench, _FakeMetadata(json.dumps({**_ATTRIBUTE, "s3": _S3})))
+    apply_central_config(bench, _FakeMetadata(json.dumps(_ATTRIBUTE), storage=json.dumps(_S3)))
 
     assert BenchConfig.read(bench.path).s3.provider == "aws"
 
 
 def test_telemetry_comes_back_with_the_credentials() -> None:
-    credentials = _FakeMetadata(json.dumps({**_ATTRIBUTE, "telemetry": _TELEMETRY})).get_credentials()
+    credentials = _FakeMetadata(json.dumps(_ATTRIBUTE), telemetry=json.dumps(_TELEMETRY)).get_credentials()
 
     assert credentials["telemetry"] == _TELEMETRY
 
 
 def test_apply_saves_the_datum_credential(tmp_path: Path) -> None:
     bench = _awaiting_bench(tmp_path)
 
-    apply_central_config(bench, _FakeMetadata(json.dumps({**_ATTRIBUTE, "telemetry": _TELEMETRY})))
+    apply_central_config(bench, _FakeMetadata(json.dumps(_ATTRIBUTE), telemetry=json.dumps(_TELEMETRY)))
 
     saved = BenchConfig.read(bench.path).telemetry
     assert saved.endpoint == "https://datum.in-mumbai.example.test"
     assert saved.token == "datum-token"
 
 
 def test_an_incomplete_telemetry_block_raises() -> None:
-    incomplete = json.dumps({**_ATTRIBUTE, "telemetry": {"endpoint": _TELEMETRY["endpoint"]}})
+    incomplete = json.dumps({"endpoint": _TELEMETRY["endpoint"]})
 
-    with pytest.raises(CentralClientError, match="telemetry is missing: token"):
-        _FakeMetadata(incomplete).get_credentials()
+    with pytest.raises(CentralClientError, match="pilot-telemetry' is missing: token"):
+        _FakeMetadata(json.dumps(_ATTRIBUTE), telemetry=incomplete).get_credentials()
 
 
 def test_a_metadata_flavoured_telemetry_endpoint_is_rejected() -> None:
-    hostile = json.dumps({**_ATTRIBUTE, "telemetry": {**_TELEMETRY, "endpoint": "http://169.254.169.254/"}})
+    hostile = json.dumps({**_TELEMETRY, "endpoint": "http://169.254.169.254/"})
 
-    with pytest.raises(CentralClientError):
-        _FakeMetadata(hostile).get_credentials()
+    with pytest.raises(CentralClientError, match="pilot-telemetry"):
+        _FakeMetadata(json.dumps(_ATTRIBUTE), telemetry=hostile).get_credentials()
+
+
+def test_blocks_inside_the_central_attribute_are_ignored() -> None:
+    attribute = json.dumps({**_ATTRIBUTE, "s3": _S3, "telemetry": _TELEMETRY})
+
+    credentials = _FakeMetadata(attribute).get_credentials()
+
+    assert "s3" not in credentials
+    assert "telemetry" not in credentials
 
 
 def test_a_metadata_flavoured_endpoint_is_rejected() -> None: