Cloud Native Geospatial Resource Registration¶

Note — Relationship to the EOEPCA end-to-end demonstration

This notebook is based on the Data Cube Access end-to-end guide. It deliberately narrows that workflow to the cloud-native geospatial patterns used by several stages of the end-to-end demonstration.

Warning — Scope and scientific caveat

This is a technical reproduction of the registration pattern, not an evaluation of the domain-specific scientific workflow. In particular, a STAC search bounding box selects intersecting source items; it does not crop global raster assets. The source projection, footprint, shape, and transform therefore remain unchanged unless assets are physically subset and rewritten.

The Workspace Building Block provides the capabilities needed for this cloud-native geospatial pattern:

an OCI registry, to publish portable, content-addressed artifacts;
Datalab compute connected to a dedicated vCluster, to run packaging work and dynamically deploy services close to the data;
object storage, to hold relocated source assets when the workflow is extended beyond metadata registration.

This notebook currently exercises the OCI registry and compute/vCluster capabilities. Workspace object storage is deliberately not used yet; it becomes relevant when the source asset bytes are relocated, as described in the upcoming-work section.

For a concrete, reproducible deployment, the example uses a Datalab session named e2e started in the ws-alice workspace. Those names are demonstration defaults rather than requirements of the pattern.

Here, resource registration means turning selected external STAC records into a stable, team-owned resource: the metadata is materialized, published to the Workspace OCI registry under an immutable digest, and exposed through a standards-based STAC API for other team workloads. It does not mean uploading the external science assets or registering them in a central platform catalogue.

The notebook demonstrates:

selecting a small set of externally hosted STAC Items without rewriting their spatial metadata;
materializing them as STAC GeoParquet;
packaging and publishing them with stacpkg as a content-addressed OCI artifact;
resolving the immutable OCI manifest digest;
pulling the package into a vCluster Pod with an init container;
serving the local GeoParquet with stac-fastapi-geoparquet.

This notebook computes a SHA-256 checksum for every package file before publication. It then resolves the OCI manifest digest, pulls the package by that immutable digest, recomputes the file checksums, and asserts that they match; the vCluster stage reports the same checksums again. This verifies byte-level integrity for the demonstrated round trip. It does not authenticate the publisher, prevent deletion, or guarantee retention, so durable provenance still requires registry access control, retention policy, and—where needed—artifact signing.

Runtime and data flow¶

The full workflow is run from a Workspace Datalab because it provides the registry connection and a kubectl context for the session's dedicated vCluster. In the tested demonstration, this is the e2e session in ws-alice; the notebook exposes these as EOEPCA_SESSION and EOEPCA_WORKSPACE defaults so another Workspace deployment can override them.

flowchart LR
    DLR[DLR STAC API] -->|selected Items| MEMORY[ItemCollection in memory]
    MEMORY -->|JSON stdin + Arrow stream| PKG[stacpkg package with items.parquet]
    PKG -->|OCI layers + SHA-256 digests| OCI[(Workspace OCI registry)]
    OCI -->|init container pulls by manifest digest| VOL[(Pod emptyDir)]
    VOL -->|local items.parquet| API[stac-fastapi-geoparquet]
    API -->|STAC API| CLIENT[other vCluster workloads]

OCI is the single source of truth for the materialized catalog. The source Cloud Fraction raster bytes remain at DLR in this version.

1. Configure the Workspace resources¶

The defaults below are intentionally specific to the demonstration requested here; override the non-secret EOEPCA_* values when reusing the notebook elsewhere.

Registry access is loaded explicitly from the Datalab's REGISTRY_HOST, REGISTRY_USERNAME, and REGISTRY_PASSWORD environment variables.

In [1]:

Copied!





from __future__ import annotations

import hashlib
import json
import os
import shutil
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path

import pystac
import requests
import stacpkg.oci as stacpkg_oci
from IPython.display import Markdown, display
from pystac_client import Client

WORKSPACE_NAME = os.getenv("EOEPCA_WORKSPACE", "ws-alice")
SESSION_NAME = os.getenv("EOEPCA_SESSION", "e2e")

REGISTRY_HOST = os.getenv("REGISTRY_HOST")
REGISTRY_USERNAME = os.getenv("REGISTRY_USERNAME")
REGISTRY_PASSWORD = os.getenv("REGISTRY_PASSWORD")

WORK_DIR = Path(os.getenv("EOEPCA_WORK_DIR", "/tmp/cng-resource-registration"))
PACKAGE_DIR = WORK_DIR / "package"
PULLED_DIR = WORK_DIR / "pulled-package"
API_NAME = "cng-stac-api"
REGISTRY_SECRET = f"{API_NAME}-registry"
INIT_CONFIG = f"{API_NAME}-init"
MANIFEST_DIR = Path(
    os.getenv(
        "EOEPCA_MANIFEST_DIR",
        "docs/getting-started/cng-resource-registration",
    )
)
if not MANIFEST_DIR.is_dir() and Path("cng-resource-registration").is_dir():
    MANIFEST_DIR = Path("cng-resource-registration")

WORK_DIR.mkdir(parents=True, exist_ok=True)

print(
    {
        "workspace": WORKSPACE_NAME,
        "session": SESSION_NAME,
        "registry_host": REGISTRY_HOST,
        "registry_credentials_loaded": bool(
            REGISTRY_HOST and REGISTRY_USERNAME and REGISTRY_PASSWORD
        ),
    }
)
from __future__ import annotations

import hashlib
import json
import os
import shutil
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path

import pystac
import requests
import stacpkg.oci as stacpkg_oci
from IPython.display import Markdown, display
from pystac_client import Client

WORKSPACE_NAME = os.getenv("EOEPCA_WORKSPACE", "ws-alice")
SESSION_NAME = os.getenv("EOEPCA_SESSION", "e2e")

REGISTRY_HOST = os.getenv("REGISTRY_HOST")
REGISTRY_USERNAME = os.getenv("REGISTRY_USERNAME")
REGISTRY_PASSWORD = os.getenv("REGISTRY_PASSWORD")

WORK_DIR = Path(os.getenv("EOEPCA_WORK_DIR", "/tmp/cng-resource-registration"))
PACKAGE_DIR = WORK_DIR / "package"
PULLED_DIR = WORK_DIR / "pulled-package"
API_NAME = "cng-stac-api"
REGISTRY_SECRET = f"{API_NAME}-registry"
INIT_CONFIG = f"{API_NAME}-init"
MANIFEST_DIR = Path(
    os.getenv(
        "EOEPCA_MANIFEST_DIR",
        "docs/getting-started/cng-resource-registration",
    )
)
if not MANIFEST_DIR.is_dir() and Path("cng-resource-registration").is_dir():
    MANIFEST_DIR = Path("cng-resource-registration")

WORK_DIR.mkdir(parents=True, exist_ok=True)

print(
    {
        "workspace": WORKSPACE_NAME,
        "session": SESSION_NAME,
        "registry_host": REGISTRY_HOST,
        "registry_credentials_loaded": bool(
            REGISTRY_HOST and REGISTRY_USERNAME and REGISTRY_PASSWORD
        ),
    }
)

{'workspace': 'ws-alice', 'session': 'e2e', 'registry_host': 'registry-ws-alice-e2e.lab.develop.eoepca.org', 'registry_credentials_loaded': True}

In [2]:

Copied!





def require_environment(*names: str) -> None:
    missing = [name for name in names if not os.getenv(name)]
    if missing:
        raise RuntimeError(
            "This cell must run in the Workspace Datalab; missing environment "
            f"variables: {', '.join(missing)}"
        )


def run(
    command: list[str],
    *,
    input_text: str | None = None,
    attempts: int = 1,
) -> subprocess.CompletedProcess:
    for attempt in range(1, attempts + 1):
        try:
            completed = subprocess.run(
                command,
                input=input_text,
                text=True,
                check=True,
                capture_output=True,
            )
            break
        except subprocess.CalledProcessError as error:
            if attempt < attempts:
                time.sleep(2 ** (attempt - 1))
                continue
            detail = (error.stderr or error.stdout or "").strip()
            raise RuntimeError(
                f"Command failed ({' '.join(command)}): {detail}"
            ) from error
    if completed.stdout.strip():
        print(completed.stdout.strip())
    return completed


def pipe(
    first: list[str],
    second: list[str],
    *,
    input_text: str | None = None,
) -> None:
    with subprocess.Popen(
        first,
        stdin=subprocess.PIPE if input_text is not None else None,
        stdout=subprocess.PIPE,
    ) as producer:
        assert producer.stdout is not None
        consumer = subprocess.Popen(
            second,
            stdin=producer.stdout,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        producer.stdout.close()
        if input_text is not None:
            assert producer.stdin is not None
            try:
                producer.stdin.write(input_text.encode("utf-8"))
            except BrokenPipeError:
                pass
            finally:
                producer.stdin.close()
        consumer_stdout, consumer_stderr = consumer.communicate()
        producer_status = producer.wait()
    if producer_status:
        raise subprocess.CalledProcessError(producer_status, first)
    if consumer.returncode:
        detail = consumer_stderr.decode().strip()
        raise RuntimeError(f"Command failed ({' '.join(second)}): {detail}")
    if consumer_stdout:
        print(consumer_stdout.decode().strip())


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def kubectl_apply(manifest: str, *, attempts: int = 5) -> None:
    for attempt in range(1, attempts + 1):
        completed = subprocess.run(
            ["kubectl", "apply", "-f", "-"],
            input=manifest,
            text=True,
            capture_output=True,
        )
        if completed.returncode == 0:
            print(completed.stdout.strip())
            return
        if attempt < attempts:
            time.sleep(2 ** (attempt - 1))
    raise RuntimeError(
        f"kubectl apply failed after {attempts} attempts: "
        f"{completed.stderr.strip()}"
    )


def render_manifest(template: str, values: dict[str, str]) -> str:
    rendered = template
    for name, value in values.items():
        rendered = rendered.replace(f"{{{{{name}}}}}", json.dumps(value))
    if "{{" in rendered or "}}" in rendered:
        raise ValueError("Unresolved manifest template value")
    return rendered


require_environment(
    "KUBECONFIG",
    "REGISTRY_HOST",
    "REGISTRY_USERNAME",
    "REGISTRY_PASSWORD",
)

# stacpkg/oras reads these standard aliases; the source credentials remain
# the values injected into the Datalab by the Workspace.
os.environ["ORAS_USER"] = REGISTRY_USERNAME
os.environ["ORAS_PASS"] = REGISTRY_PASSWORD
def require_environment(*names: str) -> None:
    missing = [name for name in names if not os.getenv(name)]
    if missing:
        raise RuntimeError(
            "This cell must run in the Workspace Datalab; missing environment "
            f"variables: {', '.join(missing)}"
        )


def run(
    command: list[str],
    *,
    input_text: str | None = None,
    attempts: int = 1,
) -> subprocess.CompletedProcess:
    for attempt in range(1, attempts + 1):
        try:
            completed = subprocess.run(
                command,
                input=input_text,
                text=True,
                check=True,
                capture_output=True,
            )
            break
        except subprocess.CalledProcessError as error:
            if attempt < attempts:
                time.sleep(2 ** (attempt - 1))
                continue
            detail = (error.stderr or error.stdout or "").strip()
            raise RuntimeError(
                f"Command failed ({' '.join(command)}): {detail}"
            ) from error
    if completed.stdout.strip():
        print(completed.stdout.strip())
    return completed


def pipe(
    first: list[str],
    second: list[str],
    *,
    input_text: str | None = None,
) -> None:
    with subprocess.Popen(
        first,
        stdin=subprocess.PIPE if input_text is not None else None,
        stdout=subprocess.PIPE,
    ) as producer:
        assert producer.stdout is not None
        consumer = subprocess.Popen(
            second,
            stdin=producer.stdout,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        producer.stdout.close()
        if input_text is not None:
            assert producer.stdin is not None
            try:
                producer.stdin.write(input_text.encode("utf-8"))
            except BrokenPipeError:
                pass
            finally:
                producer.stdin.close()
        consumer_stdout, consumer_stderr = consumer.communicate()
        producer_status = producer.wait()
    if producer_status:
        raise subprocess.CalledProcessError(producer_status, first)
    if consumer.returncode:
        detail = consumer_stderr.decode().strip()
        raise RuntimeError(f"Command failed ({' '.join(second)}): {detail}")
    if consumer_stdout:
        print(consumer_stdout.decode().strip())


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def kubectl_apply(manifest: str, *, attempts: int = 5) -> None:
    for attempt in range(1, attempts + 1):
        completed = subprocess.run(
            ["kubectl", "apply", "-f", "-"],
            input=manifest,
            text=True,
            capture_output=True,
        )
        if completed.returncode == 0:
            print(completed.stdout.strip())
            return
        if attempt < attempts:
            time.sleep(2 ** (attempt - 1))
    raise RuntimeError(
        f"kubectl apply failed after {attempts} attempts: "
        f"{completed.stderr.strip()}"
    )


def render_manifest(template: str, values: dict[str, str]) -> str:
    rendered = template
    for name, value in values.items():
        rendered = rendered.replace(f"{{{{{name}}}}}", json.dumps(value))
    if "{{" in rendered or "}}" in rendered:
        raise ValueError("Unresolved manifest template value")
    return rendered


require_environment(
    "KUBECONFIG",
    "REGISTRY_HOST",
    "REGISTRY_USERNAME",
    "REGISTRY_PASSWORD",
)

# stacpkg/oras reads these standard aliases; the source credentials remain
# the values injected into the Datalab by the Workspace.
os.environ["ORAS_USER"] = REGISTRY_USERNAME
os.environ["ORAS_PASS"] = REGISTRY_PASSWORD

2. Select external STAC Items¶

The regional bounding box is used only as a search predicate. The DLR collection contains global daily rasters, so this notebook preserves every returned Item's original geometry and projection fields.

STAC temporal searches use interval intersection. A source Item beginning on 31 July can therefore match an interval beginning on 1 August if its observation interval crosses midnight. We explicitly post-filter on the Item's representative datetime and select three days to keep this infrastructure demonstration small.

In [3]:

Copied!





SOURCE_API = "https://geoservice.dlr.de/eoc/ogc/stac/v1/"
SOURCE_COLLECTION = "S5P_TROPOMI_L3_P1D_CF_v2"
EUROPE_BBOX = [-10.0, 35.0, 30.0, 70.0]
START = datetime(2023, 8, 1, tzinfo=timezone.utc)
END = datetime(2023, 8, 3, 23, 59, 59, tzinfo=timezone.utc)
SEARCH_INTERVAL = "2023-08-01T00:00:00Z/2023-08-03T23:59:59Z"

catalog = Client.open(SOURCE_API)
candidates = list(
    catalog.search(
        collections=[SOURCE_COLLECTION],
        bbox=EUROPE_BBOX,
        datetime=SEARCH_INTERVAL,
        limit=20,
    ).items()
)

items = sorted(
    [
        item
        for item in candidates
        if item.datetime is not None and START <= item.datetime <= END
    ],
    key=lambda item: item.datetime,
)[:3]

if len(items) != 3:
    raise RuntimeError(f"Expected three post-filtered source Items, got {len(items)}")

print(
    [
        {
            "id": item.id,
            "datetime": item.datetime.isoformat(),
            "bbox": item.bbox,
            "proj:shape": item.properties.get("proj:shape"),
        }
        for item in items
    ]
)
SOURCE_API = "https://geoservice.dlr.de/eoc/ogc/stac/v1/"
SOURCE_COLLECTION = "S5P_TROPOMI_L3_P1D_CF_v2"
EUROPE_BBOX = [-10.0, 35.0, 30.0, 70.0]
START = datetime(2023, 8, 1, tzinfo=timezone.utc)
END = datetime(2023, 8, 3, 23, 59, 59, tzinfo=timezone.utc)
SEARCH_INTERVAL = "2023-08-01T00:00:00Z/2023-08-03T23:59:59Z"

catalog = Client.open(SOURCE_API)
candidates = list(
    catalog.search(
        collections=[SOURCE_COLLECTION],
        bbox=EUROPE_BBOX,
        datetime=SEARCH_INTERVAL,
        limit=20,
    ).items()
)

items = sorted(
    [
        item
        for item in candidates
        if item.datetime is not None and START <= item.datetime <= END
    ],
    key=lambda item: item.datetime,
)[:3]

if len(items) != 3:
    raise RuntimeError(f"Expected three post-filtered source Items, got {len(items)}")

print(
    [
        {
            "id": item.id,
            "datetime": item.datetime.isoformat(),
            "bbox": item.bbox,
            "proj:shape": item.properties.get("proj:shape"),
        }
        for item in items
    ]
)

[{'id': 'S5P_DLR_OFFL_01_L3_CF_20230801', 'datetime': '2023-08-01T12:31:25.784500+00:00', 'bbox': [-180.0, -90.0, 180.0, 90.0], 'proj:shape': [1800, 3600]}, {'id': 'S5P_DLR_OFFL_01_L3_CF_20230802', 'datetime': '2023-08-02T12:12:17.578500+00:00', 'bbox': [-180.0, -90.0, 180.0, 90.0], 'proj:shape': [1800, 3600]}, {'id': 'S5P_DLR_OFFL_01_L3_CF_20230803', 'datetime': '2023-08-03T12:43:54.323000+00:00', 'bbox': [-180.0, -90.0, 180.0, 90.0], 'proj:shape': [1800, 3600]}]

In [4]:

Copied!





item_collection = pystac.ItemCollection(items)
# A technical collection document tells the API which local GeoParquet to serve.
# It does not rewrite any source Item or claim a new science-data license.
index_bbox = [
    min(item.bbox[0] for item in items),
    min(item.bbox[1] for item in items),
    max(item.bbox[2] for item in items),
    max(item.bbox[3] for item in items),
]
index_start = min(
    item.common_metadata.start_datetime or item.datetime for item in items
)
index_end = max(
    item.common_metadata.end_datetime or item.datetime for item in items
)
index_collection = pystac.Collection(
    id=SOURCE_COLLECTION,
    description="Materialized STAC GeoParquet index of selected source Items.",
    extent=pystac.Extent(
        spatial=pystac.SpatialExtent([index_bbox]),
        temporal=pystac.TemporalExtent([[index_start, index_end]]),
    ),
    license="other",
)
index_collection.add_asset(
    "items",
    pystac.Asset(
        href="items.parquet",
        media_type="application/vnd.apache.parquet",
        roles=["data"],
    ),
)
collections_document = [index_collection.to_dict()]
print(
    f"Prepared {len(items)} unmodified source Items and the technical "
    "API collection document in memory"
)
item_collection = pystac.ItemCollection(items)
# A technical collection document tells the API which local GeoParquet to serve.
# It does not rewrite any source Item or claim a new science-data license.
index_bbox = [
    min(item.bbox[0] for item in items),
    min(item.bbox[1] for item in items),
    max(item.bbox[2] for item in items),
    max(item.bbox[3] for item in items),
]
index_start = min(
    item.common_metadata.start_datetime or item.datetime for item in items
)
index_end = max(
    item.common_metadata.end_datetime or item.datetime for item in items
)
index_collection = pystac.Collection(
    id=SOURCE_COLLECTION,
    description="Materialized STAC GeoParquet index of selected source Items.",
    extent=pystac.Extent(
        spatial=pystac.SpatialExtent([index_bbox]),
        temporal=pystac.TemporalExtent([[index_start, index_end]]),
    ),
    license="other",
)
index_collection.add_asset(
    "items",
    pystac.Asset(
        href="items.parquet",
        media_type="application/vnd.apache.parquet",
        roles=["data"],
    ),
)
collections_document = [index_collection.to_dict()]
print(
    f"Prepared {len(items)} unmodified source Items and the technical "
    "API collection document in memory"
)

Prepared 3 unmodified source Items and the technical API collection document in memory

3. Materialize STAC GeoParquet and build a package¶

stacpkg items from-json reads the selected in-memory ItemCollection from standard input and emits an Arrow stream. stacpkg build consumes that stream and materializes the STAC GeoParquet directly as package/items.parquet; no raw STAC JSON file or standalone GeoParquet intermediate is written. The in-memory API collection document is then written directly into the package—there is no separate staging copy. The package contains:

items.parquet, the materialized Item records;
assets.lock.parquet, the external asset references and available object facts;
collections.json, a small API control document pointing to local items.parquet.

Asset bytes are deliberately not copied in this version. --no-probe-metadata also avoids turning upstream object headers into an accidental availability guarantee.

In [5]:

Copied!





if PACKAGE_DIR.exists():
    shutil.rmtree(PACKAGE_DIR)

pipe(
    ["stacpkg", "items", "from-json"],
    [
        "stacpkg",
        "build",
        "--no-probe-metadata",
        str(PACKAGE_DIR),
    ],
    input_text=json.dumps(item_collection.to_dict()),
)
(PACKAGE_DIR / "collections.json").write_text(
    json.dumps(collections_document, indent=2),
    encoding="utf-8",
)
run(["stacpkg", "inspect", "--format", "markdown", str(PACKAGE_DIR)])

local_hashes = {
    path.name: sha256(path)
    for path in sorted(PACKAGE_DIR.iterdir())
    if path.is_file() and path.name != "manifest.json"
}
print({"package_sha256": local_hashes})
if PACKAGE_DIR.exists():
    shutil.rmtree(PACKAGE_DIR)

pipe(
    ["stacpkg", "items", "from-json"],
    [
        "stacpkg",
        "build",
        "--no-probe-metadata",
        str(PACKAGE_DIR),
    ],
    input_text=json.dumps(item_collection.to_dict()),
)
(PACKAGE_DIR / "collections.json").write_text(
    json.dumps(collections_document, indent=2),
    encoding="utf-8",
)
run(["stacpkg", "inspect", "--format", "markdown", str(PACKAGE_DIR)])

local_hashes = {
    path.name: sha256(path)
    for path in sorted(PACKAGE_DIR.iterdir())
    if path.is_file() and path.name != "manifest.json"
}
print({"package_sha256": local_hashes})

# stacpkg Inspect

- Package: `/tmp/cng-resource-registration/package`
- Items: 3
- Collections: S5P_TROPOMI_L3_P1D_CF_v2
- Assets: 15
- Asset keys: cf, data, overview, quicklook, thumbnail
- Known asset bytes: 0

## Files

- `assets.lock.parquet` (application/vnd.apache.parquet, 3506 bytes)
- `collections.json` (application/json, 751 bytes)
- `items.parquet` (application/vnd.apache.parquet, 37262 bytes)
{'package_sha256': {'assets.lock.parquet': '3db0d383af51b22fd33ad19faf8813ac7943c887a50f29af8ff8a91f3ad5e72a', 'collections.json': 'aaf62165b18f1ee88c4c3b5c20ab1a9bdc511ee46f64abc0e2bca8948e074ba5', 'items.parquet': 'b623d81ceb0161b609b5d8f38442ed01be41e98cd11417688b1c0f254e54dec6'}}

4. Publish the package to the session OCI registry¶

The Datalab provides REGISTRY_HOST, REGISTRY_USERNAME, and REGISTRY_PASSWORD. Alice's Educates registry uses HTTP Basic authentication, selected through the public auth_backend="basic" option in stacpkg 0.1.4.

The setup maps those injected values to the standard ORAS_USER and ORAS_PASS environment variables consumed by stacpkg; credentials are never printed or embedded in the notebook. The first cell below performs only the push. The following cell independently resolves, inspects, pulls, and verifies the artifact.

The tag contains the leading bytes of the Item-table checksum for discoverability. Verification then resolves and uses the complete immutable manifest digest.

For observability, the output lists the repository's tags and the safe parts of the resolved OCI manifest: artifact type, layer names, media types, digests, and sizes. Registry credentials and layer content are not displayed.

In [6]:

Copied!





OCI_REPOSITORY = "cng-resource-registration/s5p-cloud-fraction"

content_tag = local_hashes["items.parquet"][:16]
tagged_reference = f"{REGISTRY_HOST}/{OCI_REPOSITORY}:sha256-{content_tag}"
stacpkg_oci.push_package(
    PACKAGE_DIR,
    tagged_reference,
    auth_backend="basic",
)
print({"published_reference": tagged_reference})
OCI_REPOSITORY = "cng-resource-registration/s5p-cloud-fraction"

content_tag = local_hashes["items.parquet"][:16]
tagged_reference = f"{REGISTRY_HOST}/{OCI_REPOSITORY}:sha256-{content_tag}"
stacpkg_oci.push_package(
    PACKAGE_DIR,
    tagged_reference,
    auth_backend="basic",
)
print({"published_reference": tagged_reference})

{'published_reference': 'registry-ws-alice-e2e.lab.develop.eoepca.org/cng-resource-registration/s5p-cloud-fraction:sha256-b623d81ceb0161b6'}

Verify the immutable artifact¶

A tag is convenient for discovery but can move. This cell resolves the tag to its immutable OCI manifest digest, shows the repository contents, pulls by digest, and checks every package file against the pre-push SHA-256 values.

In [7]:

Copied!





registry_auth = (REGISTRY_USERNAME, REGISTRY_PASSWORD)
manifest_headers = {
    "Accept": "application/vnd.oci.image.manifest.v1+json"
}
tagged_manifest_url = (
    f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/"
    f"manifests/sha256-{content_tag}"
)
manifest_response = requests.head(
    tagged_manifest_url,
    auth=registry_auth,
    headers=manifest_headers,
    timeout=30,
)
manifest_response.raise_for_status()
manifest_digest = manifest_response.headers["Docker-Content-Digest"]
oci_digest_reference = (
    f"{REGISTRY_HOST}/{OCI_REPOSITORY}@{manifest_digest}"
)

tags_response = requests.get(
    f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/tags/list",
    auth=registry_auth,
    timeout=30,
)
tags_response.raise_for_status()
manifest_document_response = requests.get(
    (
        f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/"
        f"manifests/{manifest_digest}"
    ),
    auth=registry_auth,
    headers=manifest_headers,
    timeout=30,
)
manifest_document_response.raise_for_status()
manifest_document = manifest_document_response.json()
registry_contents = {
    "repository": OCI_REPOSITORY,
    "tags": sorted(tags_response.json().get("tags") or []),
    "resolved_reference": oci_digest_reference,
    "artifact_type": manifest_document.get("artifactType"),
    "layers": [
        {
            "name": (layer.get("annotations") or {}).get(
                "org.opencontainers.image.title"
            ),
            "media_type": layer.get("mediaType"),
            "digest": layer.get("digest"),
            "size_bytes": layer.get("size"),
        }
        for layer in manifest_document.get("layers", [])
    ],
}
print(json.dumps({"registry_contents": registry_contents}, indent=2))

if PULLED_DIR.exists():
    shutil.rmtree(PULLED_DIR)
stacpkg_oci.pull_package(
    oci_digest_reference,
    PULLED_DIR,
    auth_backend="basic",
)

pulled_hashes = {
    path.name: sha256(path)
    for path in sorted(PULLED_DIR.iterdir())
    if path.is_file() and path.name != "manifest.json"
}
assert pulled_hashes == local_hashes
print(
    {
        "oci_reference": oci_digest_reference,
        "package_sha256": pulled_hashes,
        "round_trip_verified": True,
    }
)
registry_auth = (REGISTRY_USERNAME, REGISTRY_PASSWORD)
manifest_headers = {
    "Accept": "application/vnd.oci.image.manifest.v1+json"
}
tagged_manifest_url = (
    f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/"
    f"manifests/sha256-{content_tag}"
)
manifest_response = requests.head(
    tagged_manifest_url,
    auth=registry_auth,
    headers=manifest_headers,
    timeout=30,
)
manifest_response.raise_for_status()
manifest_digest = manifest_response.headers["Docker-Content-Digest"]
oci_digest_reference = (
    f"{REGISTRY_HOST}/{OCI_REPOSITORY}@{manifest_digest}"
)

tags_response = requests.get(
    f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/tags/list",
    auth=registry_auth,
    timeout=30,
)
tags_response.raise_for_status()
manifest_document_response = requests.get(
    (
        f"https://{REGISTRY_HOST}/v2/{OCI_REPOSITORY}/"
        f"manifests/{manifest_digest}"
    ),
    auth=registry_auth,
    headers=manifest_headers,
    timeout=30,
)
manifest_document_response.raise_for_status()
manifest_document = manifest_document_response.json()
registry_contents = {
    "repository": OCI_REPOSITORY,
    "tags": sorted(tags_response.json().get("tags") or []),
    "resolved_reference": oci_digest_reference,
    "artifact_type": manifest_document.get("artifactType"),
    "layers": [
        {
            "name": (layer.get("annotations") or {}).get(
                "org.opencontainers.image.title"
            ),
            "media_type": layer.get("mediaType"),
            "digest": layer.get("digest"),
            "size_bytes": layer.get("size"),
        }
        for layer in manifest_document.get("layers", [])
    ],
}
print(json.dumps({"registry_contents": registry_contents}, indent=2))

if PULLED_DIR.exists():
    shutil.rmtree(PULLED_DIR)
stacpkg_oci.pull_package(
    oci_digest_reference,
    PULLED_DIR,
    auth_backend="basic",
)

pulled_hashes = {
    path.name: sha256(path)
    for path in sorted(PULLED_DIR.iterdir())
    if path.is_file() and path.name != "manifest.json"
}
assert pulled_hashes == local_hashes
print(
    {
        "oci_reference": oci_digest_reference,
        "package_sha256": pulled_hashes,
        "round_trip_verified": True,
    }
)

{
  "registry_contents": {
    "repository": "cng-resource-registration/s5p-cloud-fraction",
    "tags": [
      "basic-auth-test",
      "sha256-2619064d21dfb606",
      "sha256-b623d81ceb0161b6",
      "sha256-f8649f978690e41d"
    ],
    "resolved_reference": "registry-ws-alice-e2e.lab.develop.eoepca.org/cng-resource-registration/s5p-cloud-fraction@sha256:476356f58549fd21b6a2f17613e29979237332868cb3d8dd5c0ceede3e2e449f",
    "artifact_type": "application/vnd.stacpkg.package.v1+json",
    "layers": [
      {
        "name": "items.parquet",
        "media_type": "application/vnd.stacpkg.items.v1.parquet",
        "digest": "sha256:b623d81ceb0161b609b5d8f38442ed01be41e98cd11417688b1c0f254e54dec6",
        "size_bytes": 37262
      },
      {
        "name": "assets.lock.parquet",
        "media_type": "application/vnd.stacpkg.asset-lock.v1.parquet",
        "digest": "sha256:3db0d383af51b22fd33ad19faf8813ac7943c887a50f29af8ff8a91f3ad5e72a",
        "size_bytes": 3506
      },
      {
        "name": "collections.json",
        "media_type": "application/json",
        "digest": "sha256:aaf62165b18f1ee88c4c3b5c20ab1a9bdc511ee46f64abc0e2bca8948e074ba5",
        "size_bytes": 751
      }
    ]
  }
}

{'oci_reference': 'registry-ws-alice-e2e.lab.develop.eoepca.org/cng-resource-registration/s5p-cloud-fraction@sha256:476356f58549fd21b6a2f17613e29979237332868cb3d8dd5c0ceede3e2e449f', 'package_sha256': {'assets.lock.parquet': '3db0d383af51b22fd33ad19faf8813ac7943c887a50f29af8ff8a91f3ad5e72a', 'collections.json': 'aaf62165b18f1ee88c4c3b5c20ab1a9bdc511ee46f64abc0e2bca8948e074ba5', 'items.parquet': 'b623d81ceb0161b609b5d8f38442ed01be41e98cd11417688b1c0f254e54dec6'}, 'round_trip_verified': True}

5. Stage the OCI package and start a STAC API in the vCluster¶

An init container pulls the stacpkg artifact by immutable digest into a shared emptyDir. The API container starts only after that succeeds, loads /data/collections.json, and serves the referenced local /data/items.parquet from the same Pod-local volume.

The registry credentials are loaded explicitly from REGISTRY_* and placed in the vCluster Secret cng-stac-api-registry. The Secret is needed because the init container is a different process in the dedicated vCluster; the optional cleanup cell removes it together with the workload.

The Kubernetes resources are maintained as YAML files under docs/getting-started/cng-resource-registration/. The next cell loads and displays those source manifests before applying them. Only the registry connection values and immutable OCI reference are substituted at runtime; the displayed Secret redacts its username and password.

Other vCluster workloads can consume the API at:

http://cng-stac-api.default.svc.cluster.local

For a production service, prebuild and scan dedicated puller and API images instead of installing pinned packages at container startup. Runtime installation keeps this infrastructure demonstration self-contained while there are no official application images for these Python packages. The STAC FastAPI components are explicitly pinned to the versions used by the stac-fastapi-geoparquet 0.0.6 project lockfile.

After rollout, the notebook shows the workload status and recomputes SHA-256 checksums inside the API container so the staged filesystem can be compared with the OCI manifest and the original package.

In [8]:

Copied!





manifest_paths = sorted(MANIFEST_DIR.glob("*.yaml"))
if not manifest_paths:
    raise FileNotFoundError(f"No Kubernetes manifests found in {MANIFEST_DIR}")

manifest_values = {
    "REGISTRY_HOST": REGISTRY_HOST,
    "REGISTRY_USERNAME": REGISTRY_USERNAME,
    "REGISTRY_PASSWORD": REGISTRY_PASSWORD,
    "OCI_REFERENCE": oci_digest_reference,
}
display_values = {
    **manifest_values,
    "REGISTRY_USERNAME": "<redacted>",
    "REGISTRY_PASSWORD": "<redacted>",
}

rendered_manifests = []
for manifest_path in manifest_paths:
    template = manifest_path.read_text(encoding="utf-8")
    display_manifest = render_manifest(template, display_values)
    display(
        Markdown(
            f"### `{manifest_path.name}`\n\n"
            f"```yaml\n{display_manifest.rstrip()}\n```"
        )
    )
    rendered_manifests.append(render_manifest(template, manifest_values))

kubectl_apply("\n---\n".join(rendered_manifests))

run(
    [
        "kubectl",
        "rollout",
        "status",
        f"deployment/{API_NAME}",
        "--timeout=5m",
    ],
    attempts=6,
)
run(
    [
        "kubectl",
        "get",
        "deployment,service,pod",
        "-l",
        f"app={API_NAME}",
        "-o",
        "wide",
    ]
)
staged_files_script = """
import hashlib
import json
from pathlib import Path

files = []
for path in sorted(Path('/data').iterdir()):
    if path.is_file():
        files.append(
            {
                'name': path.name,
                'size_bytes': path.stat().st_size,
                'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
            }
        )
print(json.dumps({'staged_files': files}, indent=2))
"""
staged_result = run(
    [
        "kubectl",
        "exec",
        f"deployment/{API_NAME}",
        "-c",
        "api",
        "--",
        "python",
        "-c",
        staged_files_script,
    ]
)
staged_files = json.loads(staged_result.stdout)["staged_files"]
staged_hashes = {
    entry["name"]: entry["sha256"] for entry in staged_files
}
assert staged_hashes == local_hashes
print({"staged_hashes_match_package": True})
manifest_paths = sorted(MANIFEST_DIR.glob("*.yaml"))
if not manifest_paths:
    raise FileNotFoundError(f"No Kubernetes manifests found in {MANIFEST_DIR}")

manifest_values = {
    "REGISTRY_HOST": REGISTRY_HOST,
    "REGISTRY_USERNAME": REGISTRY_USERNAME,
    "REGISTRY_PASSWORD": REGISTRY_PASSWORD,
    "OCI_REFERENCE": oci_digest_reference,
}
display_values = {
    **manifest_values,
    "REGISTRY_USERNAME": "<redacted>",
    "REGISTRY_PASSWORD": "<redacted>",
}

rendered_manifests = []
for manifest_path in manifest_paths:
    template = manifest_path.read_text(encoding="utf-8")
    display_manifest = render_manifest(template, display_values)
    display(
        Markdown(
            f"### `{manifest_path.name}`\n\n"
            f"```yaml\n{display_manifest.rstrip()}\n```"
        )
    )
    rendered_manifests.append(render_manifest(template, manifest_values))

kubectl_apply("\n---\n".join(rendered_manifests))

run(
    [
        "kubectl",
        "rollout",
        "status",
        f"deployment/{API_NAME}",
        "--timeout=5m",
    ],
    attempts=6,
)
run(
    [
        "kubectl",
        "get",
        "deployment,service,pod",
        "-l",
        f"app={API_NAME}",
        "-o",
        "wide",
    ]
)
staged_files_script = """
import hashlib
import json
from pathlib import Path

files = []
for path in sorted(Path('/data').iterdir()):
    if path.is_file():
        files.append(
            {
                'name': path.name,
                'size_bytes': path.stat().st_size,
                'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
            }
        )
print(json.dumps({'staged_files': files}, indent=2))
"""
staged_result = run(
    [
        "kubectl",
        "exec",
        f"deployment/{API_NAME}",
        "-c",
        "api",
        "--",
        "python",
        "-c",
        staged_files_script,
    ]
)
staged_files = json.loads(staged_result.stdout)["staged_files"]
staged_hashes = {
    entry["name"]: entry["sha256"] for entry in staged_files
}
assert staged_hashes == local_hashes
print({"staged_hashes_match_package": True})

`01-registry-secret.yaml`¶

apiVersion: v1
kind: Secret
metadata:
  name: cng-stac-api-registry
  labels:
    app: cng-stac-api
type: Opaque
stringData:
  host: "registry-ws-alice-e2e.lab.develop.eoepca.org"
  username: "<redacted>"
  password: "<redacted>"
  reference: "registry-ws-alice-e2e.lab.develop.eoepca.org/cng-resource-registration/s5p-cloud-fraction@sha256:476356f58549fd21b6a2f17613e29979237332868cb3d8dd5c0ceede3e2e449f"

`02-pull-script-configmap.yaml`¶

apiVersion: v1
kind: ConfigMap
metadata:
  name: cng-stac-api-init
  labels:
    app: cng-stac-api
data:
  pull.py: |
    import os

    import stacpkg.oci as stacpkg_oci


    os.environ["ORAS_USER"] = os.environ["REGISTRY_USERNAME"]
    os.environ["ORAS_PASS"] = os.environ["REGISTRY_PASSWORD"]
    stacpkg_oci.pull_package(
        os.environ["OCI_REFERENCE"],
        "/data",
        auth_backend="basic",
    )

`03-service.yaml`¶

apiVersion: v1
kind: Service
metadata:
  name: cng-stac-api
  labels:
    app: cng-stac-api
spec:
  selector:
    app: cng-stac-api
  ports:
    - name: http
      port: 80
      targetPort: http

`04-deployment.yaml`¶

apiVersion: apps/v1
kind: Deployment
metadata:
  name: cng-stac-api
  labels:
    app: cng-stac-api
spec:
  replicas: 1
  selector:
    matchLabels:
      app: cng-stac-api
  template:
    metadata:
      labels:
        app: cng-stac-api
    spec:
      initContainers:
        - name: pull-stac-package
          image: python@sha256:4c2cf9917bd1cbacc5e9b07320025bdb7cdf2df7b0ceaccb55e9dd7e30987419
          command:
            - /bin/sh
            - -c
          args:
            - >-
              python -m pip install --no-cache-dir
              'stacpkg==0.1.4'
              'oras==0.2.42'
              && exec python /scripts/pull.py
          env:
            - name: REGISTRY_HOST
              valueFrom:
                secretKeyRef:
                  name: cng-stac-api-registry
                  key: host
            - name: REGISTRY_USERNAME
              valueFrom:
                secretKeyRef:
                  name: cng-stac-api-registry
                  key: username
            - name: REGISTRY_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: cng-stac-api-registry
                  key: password
            - name: OCI_REFERENCE
              valueFrom:
                secretKeyRef:
                  name: cng-stac-api-registry
                  key: reference
          volumeMounts:
            - name: data
              mountPath: /data
            - name: init-script
              mountPath: /scripts
              readOnly: true
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: "1"
              memory: 512Mi
          securityContext:
            allowPrivilegeEscalation: false
            capabilities:
              drop:
                - ALL
      containers:
        - name: api
          image: python@sha256:4c2cf9917bd1cbacc5e9b07320025bdb7cdf2df7b0ceaccb55e9dd7e30987419
          command:
            - /bin/sh
            - -c
          args:
            - >-
              python -m pip install --no-cache-dir
              'stac-fastapi-geoparquet[serve]==0.0.6'
              'stac-fastapi-api==6.2.0'
              'stac-fastapi-extensions==6.2.0'
              'stac-fastapi-types==6.2.0'
              'fastapi==0.128.0'
              'starlette==0.50.0'
              'pydantic==2.12.5'
              'pydantic-settings==2.12.0'
              'uvicorn==0.40.0'
              'geojson-pydantic==2.1.0'
              'stac-pydantic==3.4.0'
              'rustac==0.9.3'
              'obstore==0.8.2'
              && exec python -m uvicorn
              stac_fastapi.geoparquet.main:app
              --host 0.0.0.0
              --port 8000
              --root-path "${ROOT_PATH}"
          env:
            - name: STAC_FASTAPI_COLLECTIONS_HREF
              value: /data/collections.json
            - name: ROOT_PATH
              value: /proxy/8000
          ports:
            - name: http
              containerPort: 8000
          readinessProbe:
            httpGet:
              path: /
              port: http
            periodSeconds: 5
            failureThreshold: 30
          volumeMounts:
            - name: data
              mountPath: /data
              readOnly: true
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: "1"
              memory: 1Gi
          securityContext:
            allowPrivilegeEscalation: false
            capabilities:
              drop:
                - ALL
      volumes:
        - name: data
          emptyDir: {}
        - name: init-script
          configMap:
            name: cng-stac-api-init
      securityContext:
        seccompProfile:
          type: RuntimeDefault

secret/cng-stac-api-registry configured
configmap/cng-stac-api-init unchanged
service/cng-stac-api unchanged
deployment.apps/cng-stac-api unchanged
deployment "cng-stac-api" successfully rolled out
NAME                           READY   UP-TO-DATE   AVAILABLE   AGE   CONTAINERS   IMAGES                                                                           SELECTOR
deployment.apps/cng-stac-api   1/1     1            1           41m   api          python@sha256:4c2cf9917bd1cbacc5e9b07320025bdb7cdf2df7b0ceaccb55e9dd7e30987419   app=cng-stac-api

NAME                   TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)   AGE   SELECTOR
service/cng-stac-api   ClusterIP   10.43.200.194   <none>        80/TCP    41m   app=cng-stac-api

NAME                               READY   STATUS    RESTARTS   AGE     IP           NODE            NOMINATED NODE   READINESS GATES
pod/cng-stac-api-594b85479-jrqlh   1/1     Running   0          2m40s   10.42.1.49   192.168.9.244   <none>           <none>

{
  "staged_files": [
    {
      "name": "assets.lock.parquet",
      "size_bytes": 3506,
      "sha256": "3db0d383af51b22fd33ad19faf8813ac7943c887a50f29af8ff8a91f3ad5e72a"
    },
    {
      "name": "collections.json",
      "size_bytes": 751,
      "sha256": "aaf62165b18f1ee88c4c3b5c20ab1a9bdc511ee46f64abc0e2bca8948e074ba5"
    },
    {
      "name": "items.parquet",
      "size_bytes": 37262,
      "sha256": "b623d81ceb0161b609b5d8f38442ed01be41e98cd11417688b1c0f254e54dec6"
    }
  ]
}
{'staged_hashes_match_package': True}

6. Inspect and query the dynamic STAC API¶

For notebook and browser access, a temporary port-forward is used. The Service remains cluster-local, so no dedicated public ingress is created.

The Datalab editor hostname still routes to the editor itself: appending /collections directly to that hostname does not select the forwarded port. VS Code exposes the forwarded application below /proxy/<port>/. This notebook keeps the port-forward running and prints exact browser URLs for the landing page, conformance declaration, collections, selected Collection, Items, and first Item after verifying those operations. The Jupyter kernel must remain active while those URLs are used.

The editor proxy strips /proxy/<port> before forwarding requests. Consequently, links generated by this version of stac-fastapi do not retain that inspection-only prefix; use the explicit URLs printed by the notebook rather than removing the prefix. Workloads inside the vCluster should use the cluster-local Service URL shown above, whose STAC links remain directly navigable.

In [9]:

Copied!





LOCAL_API_PORT = int(os.getenv("EOEPCA_API_PORT", "8000"))

if "port_forward" in globals() and port_forward.poll() is None:
    port_forward.terminate()
    port_forward.wait(timeout=10)

port_forward = subprocess.Popen(
    [
        "kubectl",
        "port-forward",
        f"service/{API_NAME}",
        f"{LOCAL_API_PORT}:80",
    ],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.PIPE,
    text=True,
)
api_root = f"http://127.0.0.1:{LOCAL_API_PORT}"
for attempt in range(60):
    try:
        root_response = requests.get(api_root, timeout=2)
        root_response.raise_for_status()
        break
    except requests.RequestException:
        if port_forward.poll() is not None:
            error = port_forward.stderr.read() if port_forward.stderr else ""
            raise RuntimeError(f"kubectl port-forward stopped: {error}")
        time.sleep(1)
else:
    raise TimeoutError("STAC API did not become ready within 60 seconds")

collection_path = f"/collections/{SOURCE_COLLECTION}"
first_item_id = items[0].id
requests_to_check = {
    "conformance": requests.get(f"{api_root}/conformance", timeout=10),
    "collections": requests.get(f"{api_root}/collections", timeout=10),
    "collection": requests.get(f"{api_root}{collection_path}", timeout=10),
    "collection_items": requests.get(
        f"{api_root}{collection_path}/items",
        params={"limit": len(items)},
        timeout=30,
    ),
    "item": requests.get(
        f"{api_root}{collection_path}/items/{first_item_id}",
        timeout=30,
    ),
    "search": requests.post(
        f"{api_root}/search",
        json={"collections": [SOURCE_COLLECTION], "limit": 2},
        timeout=30,
    ),
}
for response in requests_to_check.values():
    response.raise_for_status()

collections_document = requests_to_check["collections"].json()
collection_document = requests_to_check["collection"].json()
collection_items = requests_to_check["collection_items"].json()["features"]
item_document = requests_to_check["item"].json()
returned_features = requests_to_check["search"].json()["features"]
expected_item_ids = {item.id for item in items}

assert collection_document["id"] == SOURCE_COLLECTION
assert {feature["id"] for feature in collection_items} == expected_item_ids
assert item_document["id"] == first_item_id
assert len(returned_features) == 2
assert {feature["id"] for feature in returned_features} <= expected_item_ids

browser_proxy_root = None
if os.getenv("SESSION_NAME") and os.getenv("INGRESS_DOMAIN"):
    ingress_protocol = os.getenv("INGRESS_PROTOCOL", "https")
    browser_proxy_root = (
        f"{ingress_protocol}://editor-{os.environ['SESSION_NAME']}."
        f"{os.environ['INGRESS_DOMAIN']}/proxy/{LOCAL_API_PORT}"
    )

print(
    {
        "endpoint_status": {
            "root": root_response.status_code,
            **{
                name: response.status_code
                for name, response in requests_to_check.items()
            },
        },
        "stac_version": root_response.json()["stac_version"],
        "collection_ids": [
            collection["id"]
            for collection in collections_document["collections"]
        ],
        "collection_item_ids": [
            feature["id"] for feature in collection_items
        ],
        "search_item_ids": [feature["id"] for feature in returned_features],
        "browser_urls": (
            {
                "root": f"{browser_proxy_root}/",
                "conformance": f"{browser_proxy_root}/conformance",
                "collections": f"{browser_proxy_root}/collections",
                "collection": f"{browser_proxy_root}{collection_path}",
                "collection_items": (
                    f"{browser_proxy_root}{collection_path}/items"
                ),
                "first_item": (
                    f"{browser_proxy_root}{collection_path}/items/"
                    f"{first_item_id}"
                ),
            }
            if browser_proxy_root
            else "Use the VS Code Ports tab for the forwarded local port."
        ),
    }
)
LOCAL_API_PORT = int(os.getenv("EOEPCA_API_PORT", "8000"))

if "port_forward" in globals() and port_forward.poll() is None:
    port_forward.terminate()
    port_forward.wait(timeout=10)

port_forward = subprocess.Popen(
    [
        "kubectl",
        "port-forward",
        f"service/{API_NAME}",
        f"{LOCAL_API_PORT}:80",
    ],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.PIPE,
    text=True,
)
api_root = f"http://127.0.0.1:{LOCAL_API_PORT}"
for attempt in range(60):
    try:
        root_response = requests.get(api_root, timeout=2)
        root_response.raise_for_status()
        break
    except requests.RequestException:
        if port_forward.poll() is not None:
            error = port_forward.stderr.read() if port_forward.stderr else ""
            raise RuntimeError(f"kubectl port-forward stopped: {error}")
        time.sleep(1)
else:
    raise TimeoutError("STAC API did not become ready within 60 seconds")

collection_path = f"/collections/{SOURCE_COLLECTION}"
first_item_id = items[0].id
requests_to_check = {
    "conformance": requests.get(f"{api_root}/conformance", timeout=10),
    "collections": requests.get(f"{api_root}/collections", timeout=10),
    "collection": requests.get(f"{api_root}{collection_path}", timeout=10),
    "collection_items": requests.get(
        f"{api_root}{collection_path}/items",
        params={"limit": len(items)},
        timeout=30,
    ),
    "item": requests.get(
        f"{api_root}{collection_path}/items/{first_item_id}",
        timeout=30,
    ),
    "search": requests.post(
        f"{api_root}/search",
        json={"collections": [SOURCE_COLLECTION], "limit": 2},
        timeout=30,
    ),
}
for response in requests_to_check.values():
    response.raise_for_status()

collections_document = requests_to_check["collections"].json()
collection_document = requests_to_check["collection"].json()
collection_items = requests_to_check["collection_items"].json()["features"]
item_document = requests_to_check["item"].json()
returned_features = requests_to_check["search"].json()["features"]
expected_item_ids = {item.id for item in items}

assert collection_document["id"] == SOURCE_COLLECTION
assert {feature["id"] for feature in collection_items} == expected_item_ids
assert item_document["id"] == first_item_id
assert len(returned_features) == 2
assert {feature["id"] for feature in returned_features} <= expected_item_ids

browser_proxy_root = None
if os.getenv("SESSION_NAME") and os.getenv("INGRESS_DOMAIN"):
    ingress_protocol = os.getenv("INGRESS_PROTOCOL", "https")
    browser_proxy_root = (
        f"{ingress_protocol}://editor-{os.environ['SESSION_NAME']}."
        f"{os.environ['INGRESS_DOMAIN']}/proxy/{LOCAL_API_PORT}"
    )

print(
    {
        "endpoint_status": {
            "root": root_response.status_code,
            **{
                name: response.status_code
                for name, response in requests_to_check.items()
            },
        },
        "stac_version": root_response.json()["stac_version"],
        "collection_ids": [
            collection["id"]
            for collection in collections_document["collections"]
        ],
        "collection_item_ids": [
            feature["id"] for feature in collection_items
        ],
        "search_item_ids": [feature["id"] for feature in returned_features],
        "browser_urls": (
            {
                "root": f"{browser_proxy_root}/",
                "conformance": f"{browser_proxy_root}/conformance",
                "collections": f"{browser_proxy_root}/collections",
                "collection": f"{browser_proxy_root}{collection_path}",
                "collection_items": (
                    f"{browser_proxy_root}{collection_path}/items"
                ),
                "first_item": (
                    f"{browser_proxy_root}{collection_path}/items/"
                    f"{first_item_id}"
                ),
            }
            if browser_proxy_root
            else "Use the VS Code Ports tab for the forwarded local port."
        ),
    }
)

{'endpoint_status': {'root': 200, 'conformance': 200, 'collections': 200, 'collection': 200, 'collection_items': 200, 'item': 200, 'search': 200}, 'stac_version': '1.0.0', 'collection_ids': ['S5P_TROPOMI_L3_P1D_CF_v2'], 'collection_item_ids': ['S5P_DLR_OFFL_01_L3_CF_20230801', 'S5P_DLR_OFFL_01_L3_CF_20230802', 'S5P_DLR_OFFL_01_L3_CF_20230803'], 'search_item_ids': ['S5P_DLR_OFFL_01_L3_CF_20230801', 'S5P_DLR_OFFL_01_L3_CF_20230802'], 'browser_urls': {'root': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/', 'conformance': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/conformance', 'collections': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/collections', 'collection': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/collections/S5P_TROPOMI_L3_P1D_CF_v2', 'collection_items': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/collections/S5P_TROPOMI_L3_P1D_CF_v2/items', 'first_item': 'https://editor-ws-alice-e2e.lab.develop.eoepca.org/proxy/8000/collections/S5P_TROPOMI_L3_P1D_CF_v2/items/S5P_DLR_OFFL_01_L3_CF_20230801'}}

7. Optional cleanup¶

The OCI package is the registered resource and is intentionally retained. Set EOEPCA_CLEAN_UP=1 before running the next cell to stop the notebook's port-forward and remove only the temporary API workload, Service, init script, and registry credential Secret. With cleanup disabled, the API and browser proxy URLs remain available while the Jupyter kernel and Datalab session are running.

In [10]:

Copied!





if os.getenv("EOEPCA_CLEAN_UP", "0") == "1":
    if "port_forward" in globals() and port_forward.poll() is None:
        port_forward.terminate()
        try:
            port_forward.wait(timeout=10)
        except subprocess.TimeoutExpired:
            port_forward.kill()
    run(
        [
            "kubectl",
            "delete",
            f"deployment/{API_NAME}",
            f"service/{API_NAME}",
            f"configmap/{INIT_CONFIG}",
            f"secret/{REGISTRY_SECRET}",
            "--ignore-not-found",
        ],
        attempts=6,
    )
else:
    print(
        "Temporary API retained. The port-forward remains available while "
        "this Jupyter kernel is running; set EOEPCA_CLEAN_UP=1 to remove "
        "the temporary API resources."
    )
if os.getenv("EOEPCA_CLEAN_UP", "0") == "1":
    if "port_forward" in globals() and port_forward.poll() is None:
        port_forward.terminate()
        try:
            port_forward.wait(timeout=10)
        except subprocess.TimeoutExpired:
            port_forward.kill()
    run(
        [
            "kubectl",
            "delete",
            f"deployment/{API_NAME}",
            f"service/{API_NAME}",
            f"configmap/{INIT_CONFIG}",
            f"secret/{REGISTRY_SECRET}",
            "--ignore-not-found",
        ],
        attempts=6,
    )
else:
    print(
        "Temporary API retained. The port-forward remains available while "
        "this Jupyter kernel is running; set EOEPCA_CLEAN_UP=1 to remove "
        "the temporary API resources."
    )

Temporary API retained. The port-forward remains available while this Jupyter kernel is running; set EOEPCA_CLEAN_UP=1 to remove the temporary API resources.

Upcoming: relocate and lock source assets¶

Object storage is not used by the current metadata-registration path. It should be introduced only when the actual science asset bytes are relocated into a Workspace bucket.

With stacpkg it is easy to copy the actual asset bytes as well and rewrite the STAC Items to point to them, either by replacing the primary href or by adding the Workspace copy as an additional alternate. This is all upcoming and is not performed in this notebook.

A follow-up should also verify source and destination byte checksums, record size and media type in assets.lock.parquet, and publish a new immutable OCI package only after every copy has been validated. Asset relocation has different cost, licensing, retention, and scientific-validation concerns from materializing a small metadata index.

Result¶

The selected STAC Items are now registered as a shared resource for the team. Their materialized metadata is stored in the Workspace registry as a versioned, content-addressed OCI artifact and can be consumed through the temporary STAC API. The external asset references are captured in assets.lock.parquet, making them explicit and ready for verification and later relocation; this notebook does not yet copy or checksum the referenced asset bytes themselves.

The current pattern keeps responsibilities explicit:

OCI is the single source of truth for the portable, content-addressed STAC package;
the init container stages the package by immutable manifest digest into Pod-local storage;
the vCluster hosts a disposable standards-based STAC API;
Workspace object storage is reserved for the upcoming relocation of actual asset bytes;
source science metadata and assets remain authoritative until that relocation is implemented.