diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 594f9595..acea6296 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,6 +24,7 @@ on: - feat/ar1-metadata-fabric-gravitino-identity - feat/ar1-metadata-fabric-identity-readiness-gate - feat/ar1-metadata-fabric-jdbc-catalog-restart + - feat/ar1-metadata-fabric-spark-iceberg-rest-interoperability env: PYTHON_VERSION: "3.13" @@ -145,6 +146,9 @@ jobs: - name: Validate metadata fabric Gravitino JDBC restart evidence run: python -m data_agent.metadata_fabric_gravitino_jdbc_restart validate + - name: Validate metadata fabric Spark/Iceberg REST interoperability evidence + run: python -m data_agent.metadata_fabric_spark_iceberg_rest_interoperability validate + - name: Validate DolphinScheduler adapter boundary run: python -m data_agent.dolphinscheduler_adapter validate @@ -204,6 +208,7 @@ jobs: data_agent/test_metadata_fabric_gravitino_identity.py \ data_agent/test_metadata_fabric_identity_gate.py \ data_agent/test_metadata_fabric_gravitino_jdbc_restart.py \ + data_agent/test_metadata_fabric_spark_iceberg_rest_interoperability.py \ data_agent/test_metadata_fabric_otel_failure_rehearsal.py \ data_agent/test_metadata_fabric_otel_metrics.py \ data_agent/test_metadata_fabric_provider_metrics.py \ diff --git a/config/metadata-fabric-spark-iceberg-rest-interoperability.local.yaml b/config/metadata-fabric-spark-iceberg-rest-interoperability.local.yaml new file mode 100644 index 00000000..94ddde85 --- /dev/null +++ b/config/metadata-fabric-spark-iceberg-rest-interoperability.local.yaml @@ -0,0 +1,95 @@ +schema: gda.metadata_fabric_spark_iceberg_rest_interoperability_profile.v1 +environment: local_docker_desktop + +cluster: + context: docker-desktop + source_namespace: gda-metadata-sandbox + rehearsal_namespace: gda-metadata-spark-interop + source_schema_configmap: metadata-gravitino-schema-1-3-0 + storage_class: standard + node: desktop-worker + +runtime: + manifest: k8s/metadata-fabric-spark-iceberg-rest-interoperability + gravitino_version: 1.3.0 + gravitino_image: gda/gravitino:1.3.0-local-arm64 + gravitino_host_image_id: sha256:d355dc7e92f9e3545d717f3eab2cbdf412115f2b82e1e544d7f6235c1eacd5a5 + gravitino_kubernetes_image_id: sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3 + iceberg_rest_version: 1.11.0 + postgresql_version: 16.10-bookworm + postgresql_image: postgres:16.10-bookworm + postgresql_image_digest: sha256:38471f330eb885e04de130b768d6db4e10469e2311879c7e5c699f6d2d8a1c74 + spark_version: 3.5.0 + iceberg_spark_runtime_version: 1.6.1 + spark_image: gisdataagent/mmfe-spark-runtime:local + spark_host_image_id: sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb + spark_kubernetes_image_id: sha256:4a4522bfd4e6d1c6c90a244d0145841fbfbbf21ed16ee29ca8b681b5cec60058 + service: gravitino-persistence + gravitino_service_port: 8090 + iceberg_rest_service_port: 9001 + iceberg_rest_path: /iceberg + spark_job: spark-iceberg-rest-probe + authenticator: basic + access_control_enabled: true + transport: local_cluster_http + +dependency: + evidence_path: docs/evidence/metadata-fabric-gravitino-jdbc-restart-2026-07-29.json + evidence_fingerprint: 34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa + required_claim: local_gravitino_jdbc_catalog_restart_verified + +identity: + service_admin: gda-interop-admin + user: gda-metadata-projection + role: gda-table-projection + material_delivery: runtime_generated_ephemeral_kubernetes_object + +catalog: + provider: lakehouse-iceberg + backend: jdbc + uri: jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg + jdbc_driver: org.postgresql.Driver + jdbc_driver_source: /opt/gravitino/libs/postgresql-42.7.0.jar + gravitino_jdbc_driver_mount: /opt/gravitino/catalogs/lakehouse-iceberg/libs/postgresql-42.7.0.jar + rest_jdbc_driver_mount: /opt/gravitino/iceberg-rest-server/libs/postgresql-42.7.0.jar + jdbc_initialize: true + warehouse: file:///var/lib/gravitino/warehouse + postgresql_pvc: data-gravitino-persistence-postgresql-0 + warehouse_pvc: warehouse-gravitino-persistence-0 + interoperability_scope: local_same_node_shared_rwo_pvc + +scope: + metalake: gda_interop + catalog: lakehouse + schema: published + table: gda_spark_interop_probe + denied_catalog: unauthorized_catalog + role_securable_objects: + - full_name: lakehouse + type: CATALOG + privileges: + - name: USE_CATALOG + condition: ALLOW + - full_name: lakehouse.published + type: SCHEMA + privileges: + - name: CREATE_TABLE + condition: ALLOW + - name: USE_SCHEMA + condition: ALLOW + +claims: + local_spark_iceberg_rest_interoperability_verified: false + local_spark_create_read_write_verified: false + local_spark_schema_evolution_verified: false + local_spark_snapshot_time_travel_verified: false + gravitino_api_metadata_readback_verified: false + local_same_node_shared_pvc_verified: false + persistent_catalog_identity_binding_verified: false + protected_workload_identity_verified: false + oidc_verified: false + tls_verified: false + spark_conformance_verified: false + flink_conformance_verified: false + production_ingestion_verified: false + production_ready: false diff --git a/data_agent/metadata_fabric_spark_iceberg_rest_interoperability.py b/data_agent/metadata_fabric_spark_iceberg_rest_interoperability.py new file mode 100644 index 00000000..e1cd473e --- /dev/null +++ b/data_agent/metadata_fabric_spark_iceberg_rest_interoperability.py @@ -0,0 +1,1514 @@ +"""Verify local Spark interoperability through Gravitino's Iceberg REST server. + +The rehearsal creates an Iceberg table through a bounded Gravitino Basic user, +then uses Spark 3.5 through the standard Iceberg REST protocol to read, append, +evolve, snapshot and time-travel the same table. Gravitino must read back the +evolved metadata and the bounded user must remain unable to create a catalog. +All workloads share one Docker Desktop node and one local RWO warehouse PVC, so +the result is local interoperability evidence, never production conformance. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import secrets +import subprocess +import sys +import time +from collections.abc import Mapping +from datetime import UTC, datetime +from pathlib import Path +from typing import Annotated, Any, Literal +from urllib.parse import quote +from uuid import UUID + +import yaml +from pydantic import BaseModel, ConfigDict, Field, SecretStr, StringConstraints + +from . import metadata_fabric_gravitino_identity as identity +from . import metadata_fabric_gravitino_jdbc_restart as jdbc_restart +from . import metadata_fabric_ingestion_replay as ingestion_replay +from . import metadata_fabric_provider_metrics as provider_metrics +from . import metadata_fabric_recovery_rehearsal as recovery + + +PROFILE_SCHEMA = ( + "gda.metadata_fabric_spark_iceberg_rest_interoperability_profile.v1" +) +CONTRACT_SCHEMA = ( + "gda.metadata_fabric_spark_iceberg_rest_interoperability_contract.v1" +) +OBSERVATION_SCHEMA = ( + "gda.metadata_fabric_spark_iceberg_rest_interoperability_observation.v1" +) +EVIDENCE_SCHEMA = ( + "gda.metadata_fabric_spark_iceberg_rest_interoperability_evidence.v1" +) +VALIDATION_SCHEMA = ( + "gda.metadata_fabric_spark_iceberg_rest_interoperability_validation.v1" +) + +CONTEXT = "docker-desktop" +SOURCE_NAMESPACE = "gda-metadata-sandbox" +REHEARSAL_NAMESPACE = "gda-metadata-spark-interop" +NODE_NAME = "desktop-worker" +GRAVITINO_SCHEMA_SHA256 = identity.GRAVITINO_SCHEMA_SHA256 +GRAVITINO_HOST_IMAGE_ID = jdbc_restart.GRAVITINO_HOST_IMAGE_ID +GRAVITINO_KUBERNETES_IMAGE_ID = jdbc_restart.GRAVITINO_KUBERNETES_IMAGE_ID +POSTGRESQL_IMAGE_DIGEST = jdbc_restart.POSTGRESQL_IMAGE_DIGEST +SPARK_HOST_IMAGE_ID = ( + "sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb" +) +SPARK_KUBERNETES_IMAGE_ID = ( + "sha256:4a4522bfd4e6d1c6c90a244d0145841fbfbbf21ed16ee29ca8b681b5cec60058" +) +JDBC_RESTART_EVIDENCE_FINGERPRINT = ( + "34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa" +) + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_PROFILE_PATH = ( + REPO_ROOT + / "config/metadata-fabric-spark-iceberg-rest-interoperability.local.yaml" +) +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-spark-iceberg-rest-interoperability-2026-07-29.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-spark-iceberg-rest-interoperability.sh" +) +MANIFEST_DIR = REPO_ROOT / "k8s/metadata-fabric-spark-iceberg-rest-interoperability" + +NonEmptyText = Annotated[ + str, + StringConstraints(strip_whitespace=True, min_length=1, max_length=1024), +] + + +class MetadataFabricSparkIcebergRestInteroperabilityError(RuntimeError): + """The local Spark/Iceberg REST interoperability contract failed closed.""" + + +class _FrozenModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class ClusterProfile(_FrozenModel): + context: Literal["docker-desktop"] + source_namespace: Literal["gda-metadata-sandbox"] + rehearsal_namespace: Literal["gda-metadata-spark-interop"] + source_schema_configmap: Literal["metadata-gravitino-schema-1-3-0"] + storage_class: Literal["standard"] + node: Literal["desktop-worker"] + + +class RuntimeProfile(_FrozenModel): + manifest: Literal[ + "k8s/metadata-fabric-spark-iceberg-rest-interoperability" + ] + gravitino_version: Literal["1.3.0"] + gravitino_image: Literal["gda/gravitino:1.3.0-local-arm64"] + gravitino_host_image_id: Literal[GRAVITINO_HOST_IMAGE_ID] + gravitino_kubernetes_image_id: Literal[GRAVITINO_KUBERNETES_IMAGE_ID] + iceberg_rest_version: Literal["1.11.0"] + postgresql_version: Literal["16.10-bookworm"] + postgresql_image: Literal["postgres:16.10-bookworm"] + postgresql_image_digest: Literal[POSTGRESQL_IMAGE_DIGEST] + spark_version: Literal["3.5.0"] + iceberg_spark_runtime_version: Literal["1.6.1"] + spark_image: Literal["gisdataagent/mmfe-spark-runtime:local"] + spark_host_image_id: Literal[SPARK_HOST_IMAGE_ID] + spark_kubernetes_image_id: Literal[SPARK_KUBERNETES_IMAGE_ID] + service: Literal["gravitino-persistence"] + gravitino_service_port: Literal[8090] + iceberg_rest_service_port: Literal[9001] + iceberg_rest_path: Literal["/iceberg"] + spark_job: Literal["spark-iceberg-rest-probe"] + authenticator: Literal["basic"] + access_control_enabled: Literal[True] + transport: Literal["local_cluster_http"] + + +class DependencyProfile(_FrozenModel): + evidence_path: Literal[ + "docs/evidence/metadata-fabric-gravitino-jdbc-restart-2026-07-29.json" + ] + evidence_fingerprint: Literal[JDBC_RESTART_EVIDENCE_FINGERPRINT] + required_claim: Literal["local_gravitino_jdbc_catalog_restart_verified"] + + +class IdentityProfile(_FrozenModel): + service_admin: Literal["gda-interop-admin"] + user: Literal["gda-metadata-projection"] + role: Literal["gda-table-projection"] + material_delivery: Literal["runtime_generated_ephemeral_kubernetes_object"] + + +class CatalogProfile(_FrozenModel): + provider: Literal["lakehouse-iceberg"] + backend: Literal["jdbc"] + uri: Literal[ + "jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg" + ] + jdbc_driver: Literal["org.postgresql.Driver"] + jdbc_driver_source: Literal["/opt/gravitino/libs/postgresql-42.7.0.jar"] + gravitino_jdbc_driver_mount: Literal[ + "/opt/gravitino/catalogs/lakehouse-iceberg/libs/postgresql-42.7.0.jar" + ] + rest_jdbc_driver_mount: Literal[ + "/opt/gravitino/iceberg-rest-server/libs/postgresql-42.7.0.jar" + ] + jdbc_initialize: Literal[True] + warehouse: Literal["file:///var/lib/gravitino/warehouse"] + postgresql_pvc: Literal["data-gravitino-persistence-postgresql-0"] + warehouse_pvc: Literal["warehouse-gravitino-persistence-0"] + interoperability_scope: Literal["local_same_node_shared_rwo_pvc"] + + +class PrivilegeProfile(_FrozenModel): + name: Literal["USE_CATALOG", "USE_SCHEMA", "CREATE_TABLE"] + condition: Literal["ALLOW"] + + +class SecurableObjectProfile(_FrozenModel): + full_name: Literal["lakehouse", "lakehouse.published"] + type: Literal["CATALOG", "SCHEMA"] + privileges: tuple[PrivilegeProfile, ...] + + +class ScopeProfile(_FrozenModel): + metalake: Literal["gda_interop"] + catalog: Literal["lakehouse"] + schema_name: Literal["published"] = Field(alias="schema") + table: Literal["gda_spark_interop_probe"] + denied_catalog: Literal["unauthorized_catalog"] + role_securable_objects: tuple[SecurableObjectProfile, ...] + + +class ClaimProfile(_FrozenModel): + local_spark_iceberg_rest_interoperability_verified: Literal[False] + local_spark_create_read_write_verified: Literal[False] + local_spark_schema_evolution_verified: Literal[False] + local_spark_snapshot_time_travel_verified: Literal[False] + gravitino_api_metadata_readback_verified: Literal[False] + local_same_node_shared_pvc_verified: Literal[False] + persistent_catalog_identity_binding_verified: Literal[False] + protected_workload_identity_verified: Literal[False] + oidc_verified: Literal[False] + tls_verified: Literal[False] + spark_conformance_verified: Literal[False] + flink_conformance_verified: Literal[False] + production_ingestion_verified: Literal[False] + production_ready: Literal[False] + + +class SparkIcebergRestInteroperabilityProfile(_FrozenModel): + schema_name: Literal[PROFILE_SCHEMA] = Field(alias="schema") + environment: Literal["local_docker_desktop"] + cluster: ClusterProfile + runtime: RuntimeProfile + dependency: DependencyProfile + identity: IdentityProfile + catalog: CatalogProfile + scope: ScopeProfile + claims: ClaimProfile + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _valid_uuid(value: Any) -> bool: + try: + UUID(str(value)) + except (TypeError, ValueError): + return False + return True + + +def _valid_sha256(value: Any) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _profile_securable_objects( + profile: SparkIcebergRestInteroperabilityProfile, +) -> list[dict[str, Any]]: + result: list[dict[str, Any]] = [] + for item in profile.scope.role_securable_objects: + result.append( + { + "fullName": item.full_name, + "type": item.type, + "privileges": sorted( + [entry.model_dump(mode="json") for entry in item.privileges], + key=lambda entry: entry["name"], + ), + } + ) + return sorted(result, key=lambda item: item["fullName"]) + + +def _load_dependency( + profile: SparkIcebergRestInteroperabilityProfile, +) -> dict[str, Any]: + path = (REPO_ROOT / profile.dependency.evidence_path).resolve() + try: + path.relative_to(REPO_ROOT) + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError) as exc: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Gravitino JDBC restart dependency is unavailable" + ) from exc + if not isinstance(value, dict): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Gravitino JDBC restart dependency is not an object" + ) + if ( + value.get("evidence_fingerprint") + != profile.dependency.evidence_fingerprint + or value.get(profile.dependency.required_claim) is not True + or jdbc_restart.verify_evidence_integrity(value) + ): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Gravitino JDBC restart dependency does not match" + ) + return value + + +def load_profile( + path: Path = DEFAULT_PROFILE_PATH, +) -> SparkIcebergRestInteroperabilityProfile: + try: + value = yaml.safe_load(path.resolve().read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("Spark interoperability profile must be an object") + ingestion_replay._reject_sensitive_fields(value) + profile = SparkIcebergRestInteroperabilityProfile.model_validate(value) + except (OSError, TypeError, ValueError, yaml.YAMLError) as exc: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Spark/Iceberg REST interoperability profile is invalid" + ) from exc + if _profile_securable_objects(profile) != identity._expected_securable_objects(): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Spark interoperability role exceeds the bounded table-create scope" + ) + _load_dependency(profile) + return profile + + +def _manifest_documents() -> list[dict[str, Any]]: + documents: list[dict[str, Any]] = [] + for path in sorted(MANIFEST_DIR.glob("*.yaml")): + if path.name == "kustomization.yaml": + continue + for value in yaml.safe_load_all(path.read_text(encoding="utf-8")): + if isinstance(value, dict): + documents.append(value) + return documents + + +def _validate_manifest() -> list[str]: + errors: list[str] = [] + try: + documents = _manifest_documents() + except (OSError, yaml.YAMLError) as exc: + return [f"Spark interoperability manifest is invalid: {type(exc).__name__}"] + if any(document.get("kind") == "Secret" for document in documents): + errors.append("Spark interoperability manifest may not commit Secret values") + kinds = {str(document.get("kind")) for document in documents} + required = { + "Namespace", + "ResourceQuota", + "ServiceAccount", + "ConfigMap", + "Service", + "StatefulSet", + "Job", + } + if not required.issubset(kinds): + errors.append("Spark interoperability manifest is incomplete") + rendered = json.dumps(documents, ensure_ascii=True, sort_keys=True) + markers = ( + "gravitino.authenticators = basic", + "gravitino.authorization.enable = true", + "gravitino.iceberg-rest.catalog-backend = jdbc", + "gravitino.iceberg-rest.httpPort = 9001", + "gravitino.iceberg-rest.warehouse = file:///var/lib/gravitino/warehouse", + "stage-postgresql-jdbc-driver", + "iceberg_runtime", + "http://gravitino-persistence:9001/iceberg", + "VERSION AS OF", + "snapshot_history_verified", + "automountServiceAccountToken", + "warehouse-gravitino-persistence-0", + "desktop-worker", + ) + for marker in markers: + if marker not in rendered: + errors.append(f"Spark interoperability manifest is missing marker: {marker}") + if "gravitino.authenticators = simple" in rendered: + errors.append("Spark interoperability may not enable simple authentication") + + job = next( + ( + document + for document in documents + if document.get("kind") == "Job" + and _mapping(document.get("metadata")).get("name") + == "spark-iceberg-rest-probe" + ), + {}, + ) + job_spec = _mapping(job.get("spec")) + pod_spec = _mapping(_mapping(job_spec.get("template")).get("spec")) + containers = pod_spec.get("containers") + container_items = containers if isinstance(containers, list) else [] + spark = next( + ( + _mapping(item) + for item in container_items + if _mapping(item).get("name") == "spark" + ), + {}, + ) + resources = _mapping(spark.get("resources")) + security = _mapping(spark.get("securityContext")) + if job_spec.get("suspend") is not True: + errors.append("Spark interoperability Job must start suspended") + if pod_spec.get("automountServiceAccountToken") is not False: + errors.append("Spark interoperability Job must disable token automount") + if not {"cpu", "memory"}.issubset(_mapping(resources.get("requests"))) or not { + "cpu", + "memory", + }.issubset(_mapping(resources.get("limits"))): + errors.append("Spark interoperability Job resources are incomplete") + if ( + security.get("allowPrivilegeEscalation") is not False + or security.get("readOnlyRootFilesystem") is not True + ): + errors.append("Spark interoperability Job security context is incomplete") + return errors + + +def build_contract_report( + profile_path: Path = DEFAULT_PROFILE_PATH, + wrapper_path: Path = DEFAULT_WRAPPER_PATH, +) -> dict[str, Any]: + errors: list[str] = [] + profile: SparkIcebergRestInteroperabilityProfile | None = None + try: + profile = load_profile(profile_path) + except MetadataFabricSparkIcebergRestInteroperabilityError as exc: + errors.append(str(exc)) + errors.extend(_validate_manifest()) + try: + wrapper = wrapper_path.resolve().read_text(encoding="utf-8") + for marker in ( + "set -euo pipefail", + "metadata_fabric_spark_iceberg_rest_interoperability", + ): + if marker not in wrapper: + errors.append(f"Spark interoperability wrapper is missing: {marker}") + except OSError as exc: + errors.append(f"Spark interoperability wrapper is invalid: {type(exc).__name__}") + + files: dict[str, dict[str, str]] = {} + paths = [Path(__file__).resolve(), profile_path.resolve(), wrapper_path.resolve()] + paths.extend(sorted(MANIFEST_DIR.glob("*.yaml"))) + for path in paths: + if not path.is_file(): + continue + try: + relative = path.relative_to(REPO_ROOT).as_posix() + except ValueError: + relative = path.name + files[relative] = {"path": relative, "sha256": recovery._file_sha256(path)} + + stable = { + "schema": CONTRACT_SCHEMA, + "context": CONTEXT, + "source_namespace": SOURCE_NAMESPACE, + "rehearsal_namespace": REHEARSAL_NAMESPACE, + "node": NODE_NAME, + "runtime_image_identity": { + "gravitino_host_image_id": ( + profile.runtime.gravitino_host_image_id if profile else None + ), + "gravitino_kubernetes_image_id": ( + profile.runtime.gravitino_kubernetes_image_id if profile else None + ), + "postgresql_image_digest": ( + profile.runtime.postgresql_image_digest if profile else None + ), + "spark_host_image_id": ( + profile.runtime.spark_host_image_id if profile else None + ), + "spark_kubernetes_image_id": ( + profile.runtime.spark_kubernetes_image_id if profile else None + ), + }, + "versions": { + "gravitino": profile.runtime.gravitino_version if profile else None, + "iceberg_rest": ( + profile.runtime.iceberg_rest_version if profile else None + ), + "spark": profile.runtime.spark_version if profile else None, + "iceberg_spark_runtime": ( + profile.runtime.iceberg_spark_runtime_version if profile else None + ), + }, + "jdbc_restart_evidence_fingerprint": JDBC_RESTART_EVIDENCE_FINGERPRINT, + "catalog": { + "provider": profile.catalog.provider if profile else None, + "backend": profile.catalog.backend if profile else None, + "uri": profile.catalog.uri if profile else None, + "warehouse": profile.catalog.warehouse if profile else None, + "interoperability_scope": ( + profile.catalog.interoperability_scope if profile else None + ), + }, + "role_securable_objects": ( + _profile_securable_objects(profile) if profile else None + ), + "local_static_contract_verified": not errors, + "local_spark_iceberg_rest_interoperability_verified": False, + "spark_conformance_verified": False, + "flink_conformance_verified": False, + "persistent_catalog_identity_binding_verified": False, + "protected_workload_identity_verified": False, + "oidc_verified": False, + "tls_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + "files": files, + "errors": errors, + } + return {**stable, "contract_fingerprint": recovery._canonical_sha256(stable)} + + +def _single_list_item(value: Mapping[str, Any], label: str) -> dict[str, Any]: + items = value.get("items") + if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + f"Kubernetes observation is not singular: {label}" + ) + return items[0] + + +def _container_status(pod: Mapping[str, Any], name: str) -> Mapping[str, Any]: + statuses = _mapping(pod.get("status")).get("containerStatuses") + if not isinstance(statuses, list): + return {} + return next( + (_mapping(item) for item in statuses if _mapping(item).get("name") == name), + {}, + ) + + +class IsolatedSparkInteroperabilityRuntime: + """Own the temporary namespace, shared PVC and suspended Spark Job.""" + + def __init__(self, profile: SparkIcebergRestInteroperabilityProfile) -> None: + self.profile = profile + self.kubectl = identity._Kubectl(profile.cluster.context) + self.owned_namespace = False + self.schema_sha256: str | None = None + self.gravitino_host_image_id: str | None = None + self.spark_host_image_id: str | None = None + + @staticmethod + def _inspect_host_image(image: str, expected_id: str, label: str) -> str: + try: + completed = subprocess.run( + ["docker", "image", "inspect", image, "--format", "{{.Id}}"], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + f"{label} host image identity is unavailable" + ) from exc + image_id = completed.stdout.strip() + if completed.returncode != 0 or image_id != expected_id: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + f"{label} host image identity does not match" + ) + return image_id + + def _runtime_inputs( + self, admin_material: SecretStr, database_material: SecretStr + ) -> str: + source = self.kubectl.get_json( + [ + "-n", + self.profile.cluster.source_namespace, + "get", + "configmap", + self.profile.cluster.source_schema_configmap, + ], + label="source schema lookup", + ) + assert source is not None + schema_sql = _mapping(source.get("data")).get("001-schema.sql") + if not isinstance(schema_sql, str): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "verified Gravitino PostgreSQL schema is unavailable" + ) + self.schema_sha256 = identity._sha256_text(schema_sql) + if self.schema_sha256 != GRAVITINO_SCHEMA_SHA256: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Gravitino PostgreSQL schema checksum drift" + ) + resources = { + "apiVersion": "v1", + "kind": "List", + "items": [ + { + "apiVersion": "v1", + "kind": "Secret", + "metadata": { + "name": "gravitino-persistence-runtime", + "namespace": self.profile.cluster.rehearsal_namespace, + }, + "type": "Opaque", + "stringData": { + "admin-password": admin_material.get_secret_value(), + "database-password": database_material.get_secret_value(), + }, + }, + { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { + "name": "gravitino-persistence-schema", + "namespace": self.profile.cluster.rehearsal_namespace, + }, + "data": {"001-schema.sql": schema_sql}, + }, + ], + } + return json.dumps(resources, ensure_ascii=True, separators=(",", ":")) + + def start( + self, *, admin_material: SecretStr, database_material: SecretStr + ) -> dict[str, Any]: + self.gravitino_host_image_id = self._inspect_host_image( + self.profile.runtime.gravitino_image, + self.profile.runtime.gravitino_host_image_id, + "Gravitino", + ) + self.spark_host_image_id = self._inspect_host_image( + self.profile.runtime.spark_image, + self.profile.runtime.spark_host_image_id, + "Spark", + ) + existing = self.kubectl.get_json( + ["get", "namespace", self.profile.cluster.rehearsal_namespace], + allow_not_found=True, + label="Spark interoperability namespace preflight", + ) + if existing is not None: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Spark interoperability namespace already exists" + ) + self.kubectl.run( + ["apply", "-f", str(MANIFEST_DIR / "namespace.yaml")], + label="Spark interoperability namespace apply", + ) + self.owned_namespace = True + self.kubectl.run( + ["apply", "-f", "-"], + input_text=self._runtime_inputs(admin_material, database_material), + label="ephemeral Spark interoperability inputs apply", + ) + self.kubectl.run( + ["apply", "-k", str(MANIFEST_DIR)], + label="Spark interoperability runtime apply", + ) + for workload in ( + "statefulset/gravitino-persistence-postgresql", + "statefulset/gravitino-persistence", + ): + self.kubectl.run( + [ + "-n", + self.profile.cluster.rehearsal_namespace, + "rollout", + "status", + workload, + "--timeout=10m", + ], + timeout=660, + label=f"{workload} rollout", + ) + return self.observe_runtime() + + def _workload( + self, + *, + statefulset_name: str, + label_name: str, + container_name: str, + pvc_name: str, + ) -> dict[str, Any]: + namespace = self.profile.cluster.rehearsal_namespace + statefulset = self.kubectl.get_json( + ["-n", namespace, "get", "statefulset", statefulset_name], + label=f"{statefulset_name} observation", + ) + pod_list = self.kubectl.get_json( + [ + "-n", + namespace, + "get", + "pods", + "-l", + f"app.kubernetes.io/name={label_name}", + ], + label=f"{statefulset_name} pod observation", + ) + pvc = self.kubectl.get_json( + ["-n", namespace, "get", "pvc", pvc_name], + label=f"{pvc_name} observation", + ) + assert statefulset is not None and pod_list is not None and pvc is not None + pod = _single_list_item(pod_list, statefulset_name) + pod_spec = _mapping(pod.get("spec")) + status = _container_status(pod, container_name) + pvc_spec = _mapping(pvc.get("spec")) + return { + "statefulset_uid": _mapping(statefulset.get("metadata")).get("uid"), + "pod_uid": _mapping(pod.get("metadata")).get("uid"), + "pod_name": _mapping(pod.get("metadata")).get("name"), + "node_name": pod_spec.get("nodeName"), + "ready_replicas": _mapping(statefulset.get("status")).get( + "readyReplicas", 0 + ), + "service_account": pod_spec.get("serviceAccountName"), + "service_account_automount_disabled": ( + pod_spec.get("automountServiceAccountToken") is False + ), + "image": status.get("image"), + "image_id": status.get("imageID"), + "pvc": { + "name": _mapping(pvc.get("metadata")).get("name"), + "uid": _mapping(pvc.get("metadata")).get("uid"), + "storage_class": pvc_spec.get("storageClassName"), + "volume_name": pvc_spec.get("volumeName"), + "phase": _mapping(pvc.get("status")).get("phase"), + }, + } + + def observe_runtime(self) -> dict[str, Any]: + namespace_name = self.profile.cluster.rehearsal_namespace + namespace = self.kubectl.get_json( + ["get", "namespace", namespace_name], label="namespace observation" + ) + service = self.kubectl.get_json( + ["-n", namespace_name, "get", "service", self.profile.runtime.service], + label="Gravitino interoperability service observation", + ) + pod = self.kubectl.get_json( + ["-n", namespace_name, "get", "pod", "gravitino-persistence-0"], + label="Gravitino interoperability pod observation", + ) + assert namespace is not None and service is not None and pod is not None + main_driver = self.kubectl.run( + [ + "-n", + namespace_name, + "exec", + "gravitino-persistence-0", + "-c", + "gravitino", + "--", + "test", + "-r", + self.profile.catalog.gravitino_jdbc_driver_mount, + ], + expected=frozenset({0, 1}), + timeout=60, + label="Gravitino JDBC driver mount probe", + ) + rest_driver = self.kubectl.run( + [ + "-n", + namespace_name, + "exec", + "gravitino-persistence-0", + "-c", + "iceberg-rest", + "--", + "test", + "-r", + self.profile.catalog.rest_jdbc_driver_mount, + ], + expected=frozenset({0, 1}), + timeout=60, + label="Iceberg REST JDBC driver mount probe", + ) + rest_status = _container_status(pod, "iceberg-rest") + ports = _mapping(service.get("spec")).get("ports") + service_ports = sorted( + [ + {"name": _mapping(item).get("name"), "port": _mapping(item).get("port")} + for item in ports + ], + key=lambda item: str(item["name"]), + ) if isinstance(ports, list) else [] + return { + "context": self.profile.cluster.context, + "gravitino_host_image_id": self.gravitino_host_image_id, + "spark_host_image_id": self.spark_host_image_id, + "namespace": { + "name": _mapping(namespace.get("metadata")).get("name"), + "uid": _mapping(namespace.get("metadata")).get("uid"), + }, + "service": { + "name": _mapping(service.get("metadata")).get("name"), + "uid": _mapping(service.get("metadata")).get("uid"), + "type": _mapping(service.get("spec")).get("type"), + "ports": service_ports, + }, + "postgresql": self._workload( + statefulset_name="gravitino-persistence-postgresql", + label_name="gravitino-persistence-postgresql", + container_name="postgresql", + pvc_name=self.profile.catalog.postgresql_pvc, + ), + "gravitino": self._workload( + statefulset_name="gravitino-persistence", + label_name="gravitino-persistence", + container_name="gravitino", + pvc_name=self.profile.catalog.warehouse_pvc, + ), + "iceberg_rest": { + "image": rest_status.get("image"), + "image_id": rest_status.get("imageID"), + "ready": _mapping(rest_status.get("ready")).get("value", False) + if isinstance(rest_status.get("ready"), Mapping) + else rest_status.get("ready"), + "jdbc_driver_mounted": rest_driver.returncode == 0, + "path": self.profile.runtime.iceberg_rest_path, + }, + "gravitino_jdbc_driver_mounted": main_driver.returncode == 0, + "source_schema_sha256": self.schema_sha256, + } + + def run_spark_probe(self) -> dict[str, Any]: + namespace = self.profile.cluster.rehearsal_namespace + job_name = self.profile.runtime.spark_job + self.kubectl.run( + [ + "-n", + namespace, + "patch", + "job", + job_name, + "--type=merge", + "-p", + '{"spec":{"suspend":false}}', + ], + label="Spark interoperability Job release", + ) + deadline = time.monotonic() + 900 + terminal_condition: str | None = None + while time.monotonic() < deadline: + current_job = self.kubectl.get_json( + ["-n", namespace, "get", "job", job_name], + label="Spark interoperability Job wait", + ) + assert current_job is not None + conditions = _mapping(current_job.get("status")).get("conditions") + if isinstance(conditions, list): + for condition in conditions: + item = _mapping(condition) + if item.get("status") == "True" and item.get("type") in { + "Complete", + "Failed", + }: + terminal_condition = str(item["type"]) + break + if terminal_condition is not None: + break + time.sleep(2) + job = self.kubectl.get_json( + ["-n", namespace, "get", "job", job_name], + label="Spark interoperability Job observation", + ) + pod_list = self.kubectl.get_json( + [ + "-n", + namespace, + "get", + "pods", + "-l", + "job-name=spark-iceberg-rest-probe", + ], + label="Spark interoperability pod observation", + ) + assert job is not None and pod_list is not None + pod = _single_list_item(pod_list, "Spark interoperability Job") + pod_name = _mapping(pod.get("metadata")).get("name") + logs = self.kubectl.run( + ["-n", namespace, "logs", str(pod_name), "-c", "spark"], + expected=frozenset({0, 1}), + timeout=120, + label="Spark interoperability result collection", + ) + result_lines = [ + line.removeprefix("GDA_SPARK_RESULT=") + for line in logs.stdout.splitlines() + if line.startswith("GDA_SPARK_RESULT=") + ] + result: dict[str, Any] | None = None + if len(result_lines) == 1: + try: + candidate = json.loads(result_lines[0]) + if isinstance(candidate, dict): + result = candidate + except json.JSONDecodeError: + result = None + pod_spec = _mapping(pod.get("spec")) + volumes = pod_spec.get("volumes") + warehouse_claim = None + if isinstance(volumes, list): + for volume in volumes: + item = _mapping(volume) + if item.get("name") == "warehouse": + warehouse_claim = _mapping(item.get("persistentVolumeClaim")).get( + "claimName" + ) + container = _container_status(pod, "spark") + job_status = _mapping(job.get("status")) + return { + "wait_completed": terminal_condition == "Complete", + "terminal_condition": terminal_condition, + "job": { + "name": _mapping(job.get("metadata")).get("name"), + "uid": _mapping(job.get("metadata")).get("uid"), + "succeeded": job_status.get("succeeded", 0), + "failed": job_status.get("failed", 0), + "completion_time": job_status.get("completionTime"), + }, + "pod": { + "name": pod_name, + "uid": _mapping(pod.get("metadata")).get("uid"), + "phase": _mapping(pod.get("status")).get("phase"), + "node_name": pod_spec.get("nodeName"), + "service_account": pod_spec.get("serviceAccountName"), + "service_account_automount_disabled": ( + pod_spec.get("automountServiceAccountToken") is False + ), + "image": container.get("image"), + "image_id": container.get("imageID"), + "warehouse_pvc": warehouse_claim, + }, + "result_line_count": len(result_lines), + "log_sha256": hashlib.sha256(logs.stdout.encode("utf-8")).hexdigest(), + "log_recorded": False, + "result": result, + } + + def cleanup(self) -> dict[str, Any]: + deleted = False + if self.owned_namespace: + try: + self.kubectl.run( + [ + "delete", + "namespace", + self.profile.cluster.rehearsal_namespace, + "--wait=true", + "--timeout=5m", + ], + timeout=330, + label="Spark interoperability namespace cleanup", + ) + deleted = True + finally: + self.owned_namespace = False + absent = ( + self.kubectl.get_json( + ["get", "namespace", self.profile.cluster.rehearsal_namespace], + allow_not_found=True, + label="Spark interoperability cleanup verification", + ) + is None + ) + return { + "namespace_delete_completed": deleted, + "namespace_absent": absent, + "provider_objects_retained": False, + "persistent_volumes_retained": False, + } + + +def _post_spark_readback( + rehearsal: jdbc_restart.PersistentCatalogRehearsal, + profile: SparkIcebergRestInteroperabilityProfile, + user_material: SecretStr, +) -> dict[str, Any]: + bounded = rehearsal._user(profile, user_material) + authentication_status, _payload = bounded.request( + "GET", "version", label="post-Spark bounded authentication" + ) + read_status, read_payload = bounded.request( + "GET", + rehearsal._table_path(profile), + label="post-Spark Gravitino table readback", + ) + table = identity._response_entity( + read_payload, "table", "post-Spark Gravitino table readback" + ) + denied_status, _payload = bounded.request( + "POST", + f"metalakes/{quote(profile.scope.metalake)}/catalogs", + json_body=rehearsal._denied_catalog_body(profile), + expected=frozenset({200, 403}), + label="post-Spark administrative denial", + ) + projection = jdbc_restart._table_projection(table) + return { + "authentication_status": authentication_status, + "read_status": read_status, + "table": { + "name": table.get("name"), + "projection": projection, + "fingerprint": recovery._canonical_sha256(projection), + }, + "denied_catalog_create_status": denied_status, + } + + +def _runtime_errors(runtime: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + namespace = _mapping(runtime.get("namespace")) + service = _mapping(runtime.get("service")) + postgresql = _mapping(runtime.get("postgresql")) + gravitino = _mapping(runtime.get("gravitino")) + iceberg_rest = _mapping(runtime.get("iceberg_rest")) + if ( + runtime.get("context") != CONTEXT + or runtime.get("gravitino_host_image_id") != GRAVITINO_HOST_IMAGE_ID + or runtime.get("spark_host_image_id") != SPARK_HOST_IMAGE_ID + or namespace.get("name") != REHEARSAL_NAMESPACE + or not _valid_uuid(namespace.get("uid")) + or service.get("name") != "gravitino-persistence" + or service.get("type") != "ClusterIP" + or service.get("ports") + != [{"name": "http", "port": 8090}, {"name": "iceberg-rest", "port": 9001}] + or runtime.get("source_schema_sha256") != GRAVITINO_SCHEMA_SHA256 + ): + errors.append("Spark interoperability runtime boundary does not match") + for name, value, image_id, pvc_name, account in ( + ( + "postgresql", + postgresql, + POSTGRESQL_IMAGE_DIGEST, + "data-gravitino-persistence-postgresql-0", + "gravitino-persistence-postgresql", + ), + ( + "gravitino", + gravitino, + GRAVITINO_KUBERNETES_IMAGE_ID, + "warehouse-gravitino-persistence-0", + "gravitino-persistence", + ), + ): + pvc = _mapping(value.get("pvc")) + if ( + not _valid_uuid(value.get("statefulset_uid")) + or not _valid_uuid(value.get("pod_uid")) + or value.get("ready_replicas") != 1 + or value.get("node_name") != NODE_NAME + or not str(value.get("image_id") or "").endswith(image_id) + or value.get("service_account") != account + or value.get("service_account_automount_disabled") is not True + or pvc.get("name") != pvc_name + or not _valid_uuid(pvc.get("uid")) + or pvc.get("storage_class") != "standard" + or pvc.get("phase") != "Bound" + ): + errors.append(f"{name} runtime observation does not match") + if ( + not str(iceberg_rest.get("image_id") or "").endswith( + GRAVITINO_KUBERNETES_IMAGE_ID + ) + or iceberg_rest.get("ready") is not True + or iceberg_rest.get("jdbc_driver_mounted") is not True + or iceberg_rest.get("path") != "/iceberg" + or runtime.get("gravitino_jdbc_driver_mounted") is not True + ): + errors.append("Iceberg REST sidecar boundary does not match") + return errors + + +def _spark_errors(spark: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + job = _mapping(spark.get("job")) + pod = _mapping(spark.get("pod")) + result = _mapping(spark.get("result")) + if ( + spark.get("wait_completed") is not True + or job.get("name") != "spark-iceberg-rest-probe" + or not _valid_uuid(job.get("uid")) + or job.get("succeeded") != 1 + or job.get("failed") != 0 + or not job.get("completion_time") + or pod.get("phase") != "Succeeded" + or not _valid_uuid(pod.get("uid")) + ): + errors.append("Spark interoperability Job did not complete exactly once") + if ( + pod.get("node_name") != NODE_NAME + or pod.get("service_account") != "spark-iceberg-rest-probe" + or pod.get("service_account_automount_disabled") is not True + or not str(pod.get("image_id") or "").endswith( + SPARK_KUBERNETES_IMAGE_ID + ) + or pod.get("warehouse_pvc") != "warehouse-gravitino-persistence-0" + ): + errors.append("Spark interoperability runtime boundary does not match") + if ( + spark.get("result_line_count") != 1 + or not _valid_sha256(spark.get("log_sha256")) + or spark.get("log_recorded") is not False + or result.get("schema") != "gda.spark_iceberg_rest_probe_result.v1" + or result.get("spark_version") != "3.5.0" + or result.get("iceberg_runtime") != "1.6.1" + or result.get("catalog_uri") + != "http://gravitino-persistence:9001/iceberg" + or result.get("table") != "rest.published.gda_spark_interop_probe" + ): + errors.append("Spark interoperability result envelope does not match") + if ( + result.get("initial_columns") != ["probe_id"] + or result.get("initial_row_count") != 0 + or result.get("current_columns") != ["probe_id", "quality"] + or result.get("current_rows") + != [["spark-a", None], ["spark-b", None], ["spark-c", "verified"]] + or result.get("create_read_write_verified") is not True + ): + errors.append("Spark create/read/write result does not match") + snapshot_ids = result.get("snapshot_ids") + if ( + not isinstance(snapshot_ids, list) + or len(snapshot_ids) != 2 + or len(set(snapshot_ids)) != 2 + or not all(isinstance(value, int) and value > 0 for value in snapshot_ids) + or result.get("snapshot_operations") != ["append", "append"] + or result.get("time_travel_snapshot_id") != snapshot_ids[0] + or result.get("time_travel_rows") != ["spark-a", "spark-b"] + or result.get("snapshot_history_verified") is not True + or result.get("time_travel_verified") is not True + ): + errors.append("Spark snapshot/time-travel result does not match") + if result.get("schema_evolution_verified") is not True: + errors.append("Spark schema evolution result does not match") + return errors + + +def _expected_post_spark_projection() -> dict[str, Any]: + return { + "name": "gda_spark_interop_probe", + "columns": [ + {"name": "probe_id", "type": "string", "nullable": False}, + {"name": "quality", "type": "string", "nullable": True}, + ], + "probe_property": "true", + } + + +def build_evidence(observation: Mapping[str, Any]) -> dict[str, Any]: + errors: list[str] = [] + try: + ingestion_replay._reject_sensitive_fields(observation) + except ValueError: + errors.append("Spark interoperability observation contains sensitive material") + if observation.get("schema") != OBSERVATION_SCHEMA: + errors.append("Spark interoperability observation schema does not match") + contract = _mapping(observation.get("contract")) + if ( + contract.get("local_static_contract_verified") is not True + or not _valid_sha256(contract.get("contract_fingerprint")) + or contract.get("jdbc_restart_evidence_fingerprint") + != JDBC_RESTART_EVIDENCE_FINGERPRINT + ): + errors.append("Spark interoperability contract binding does not match") + + runtime = _mapping(observation.get("runtime")) + errors.extend(_runtime_errors(runtime)) + pre_spark = _mapping(observation.get("pre_spark")) + pre_table = _mapping(pre_spark.get("table")) + expected_pre_projection = { + "name": "gda_spark_interop_probe", + "columns": [{"name": "probe_id", "type": "string", "nullable": False}], + "probe_property": "true", + } + if ( + _mapping(pre_spark.get("authentication")).get("admin_status") != 200 + or _mapping(pre_spark.get("authentication")).get("bounded_status") != 200 + or pre_table.get("create_status") != 200 + or pre_table.get("read_status") != 200 + or pre_table.get("projection") != expected_pre_projection + or pre_table.get("fingerprint") + != recovery._canonical_sha256(expected_pre_projection) + or pre_spark.get("denied_catalog_create_status") != 403 + ): + errors.append("Gravitino bounded pre-Spark table boundary does not match") + + spark = _mapping(observation.get("spark")) + spark_errors = _spark_errors(spark) + errors.extend(spark_errors) + post_spark = _mapping(observation.get("post_spark")) + post_table = _mapping(post_spark.get("table")) + expected_post_projection = _expected_post_spark_projection() + api_readback_verified = ( + post_spark.get("authentication_status") == 200 + and post_spark.get("read_status") == 200 + and post_table.get("projection") == expected_post_projection + and post_table.get("fingerprint") + == recovery._canonical_sha256(expected_post_projection) + ) + if not api_readback_verified: + errors.append("Gravitino API did not read back Spark schema evolution") + if post_spark.get("denied_catalog_create_status") != 403: + errors.append("Post-Spark administrative catalog mutation was not denied") + + spark_pod = _mapping(spark.get("pod")) + same_node_shared_pvc_verified = ( + _mapping(runtime.get("postgresql")).get("node_name") == NODE_NAME + and _mapping(runtime.get("gravitino")).get("node_name") == NODE_NAME + and spark_pod.get("node_name") == NODE_NAME + and _mapping(_mapping(runtime.get("gravitino")).get("pvc")).get("name") + == "warehouse-gravitino-persistence-0" + and spark_pod.get("warehouse_pvc") + == "warehouse-gravitino-persistence-0" + ) + if not same_node_shared_pvc_verified: + errors.append("Spark and Gravitino did not share the bounded local PVC scope") + + runtime_checks = _mapping(observation.get("runtime_checks")) + if ( + runtime_checks.get("namespace_delete_completed") is not True + or runtime_checks.get("namespace_absent") is not True + or runtime_checks.get("provider_objects_retained") is not False + or runtime_checks.get("persistent_volumes_retained") is not False + or runtime_checks.get("all_port_forwards_stopped") is not True + or runtime_checks.get("material_recorded") is not False + or runtime_checks.get("kubernetes_service_account_used_for_provider_login") + is not False + ): + errors.append("Spark interoperability cleanup is incomplete") + + result = _mapping(spark.get("result")) + spark_runtime_verified = not any( + error + in { + "Spark interoperability Job did not complete exactly once", + "Spark interoperability runtime boundary does not match", + "Spark interoperability result envelope does not match", + } + for error in spark_errors + ) + create_read_write_verified = ( + spark_runtime_verified + and not any("create/read/write" in error for error in spark_errors) + and result.get("create_read_write_verified") is True + ) + schema_evolution_verified = ( + spark_runtime_verified + and not any("schema evolution" in error for error in spark_errors) + and result.get("schema_evolution_verified") is True + ) + snapshot_time_travel_verified = ( + spark_runtime_verified + and not any("snapshot/time-travel" in error for error in spark_errors) + and result.get("snapshot_history_verified") is True + and result.get("time_travel_verified") is True + ) + verified = not errors + stable = { + "schema": EVIDENCE_SCHEMA, + "observed_at": observation.get("observed_at"), + "local_static_contract_verified": ( + contract.get("local_static_contract_verified") is True + ), + "local_spark_iceberg_rest_interoperability_verified": verified, + "local_spark_create_read_write_verified": create_read_write_verified, + "local_spark_schema_evolution_verified": schema_evolution_verified, + "local_spark_snapshot_time_travel_verified": ( + snapshot_time_travel_verified + ), + "gravitino_api_metadata_readback_verified": api_readback_verified, + "local_same_node_shared_pvc_verified": same_node_shared_pvc_verified, + "persistent_catalog_identity_binding_verified": False, + "protected_workload_identity_verified": False, + "oidc_verified": False, + "tls_verified": False, + "spark_conformance_verified": False, + "flink_conformance_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + "observation": dict(observation), + "errors": errors, + } + return {**stable, "evidence_fingerprint": recovery._canonical_sha256(stable)} + + +def verify_evidence_integrity(evidence: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + observation = _mapping(evidence.get("observation")) + rebuilt = build_evidence(observation) + if evidence.get("evidence_fingerprint") != rebuilt.get("evidence_fingerprint"): + errors.append("Spark interoperability evidence fingerprint does not match") + for key, expected in rebuilt.items(): + if key == "evidence_fingerprint": + continue + if evidence.get(key) != expected: + errors.append(f"Spark interoperability evidence field drift: {key}") + for claim in ( + "persistent_catalog_identity_binding_verified", + "protected_workload_identity_verified", + "oidc_verified", + "tls_verified", + "spark_conformance_verified", + "flink_conformance_verified", + "production_ingestion_verified", + "production_ready", + ): + if evidence.get(claim) is not False: + errors.append(f"Spark interoperability evidence may not claim {claim}") + return errors + + +def run_live_rehearsal( + profile_path: Path = DEFAULT_PROFILE_PATH, +) -> dict[str, Any]: + profile = load_profile(profile_path) + contract = build_contract_report(profile_path) + if contract.get("local_static_contract_verified") is not True: + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Spark interoperability static contract is invalid" + ) + + admin_material = SecretStr(secrets.token_urlsafe(24)) + database_material = SecretStr(secrets.token_urlsafe(24)) + user_material = SecretStr(secrets.token_urlsafe(24)) + runtime = IsolatedSparkInteroperabilityRuntime(profile) + forward: provider_metrics._PortForward | None = None + rehearsal: jdbc_restart.PersistentCatalogRehearsal | None = None + runtime_observation: dict[str, Any] | None = None + pre_spark: dict[str, Any] | None = None + spark: dict[str, Any] | None = None + post_spark: dict[str, Any] | None = None + forward_stopped = False + cleanup: dict[str, Any] = { + "namespace_delete_completed": False, + "namespace_absent": False, + "provider_objects_retained": True, + "persistent_volumes_retained": True, + } + try: + runtime_observation = runtime.start( + admin_material=admin_material, + database_material=database_material, + ) + forward = provider_metrics._PortForward( + kubectl="kubectl", + context=profile.cluster.context, + namespace=profile.cluster.rehearsal_namespace, + service=profile.runtime.service, + target_port=profile.runtime.gravitino_service_port, + ) + forward.start() + rehearsal = jdbc_restart.PersistentCatalogRehearsal( + base_url=f"http://127.0.0.1:{forward.local_port}/api", + admin_name=profile.identity.service_admin, + admin_material=admin_material, + ) + pre_spark = rehearsal.bootstrap( + profile, + database_material=database_material, + user_material=user_material, + ) + spark = runtime.run_spark_probe() + post_spark = _post_spark_readback(rehearsal, profile, user_material) + finally: + if rehearsal is not None: + rehearsal.close() + if forward is not None: + forward_stopped = forward.stop() + cleanup = runtime.cleanup() + + if ( + runtime_observation is None + or pre_spark is None + or spark is None + or post_spark is None + ): + raise MetadataFabricSparkIcebergRestInteroperabilityError( + "Spark interoperability rehearsal did not produce an outcome" + ) + observation = { + "schema": OBSERVATION_SCHEMA, + "observed_at": datetime.now(UTC).isoformat(), + "contract": { + "contract_fingerprint": contract["contract_fingerprint"], + "local_static_contract_verified": True, + "jdbc_restart_evidence_fingerprint": ( + JDBC_RESTART_EVIDENCE_FINGERPRINT + ), + }, + "runtime": runtime_observation, + "pre_spark": pre_spark, + "spark": spark, + "post_spark": post_spark, + "runtime_checks": { + **cleanup, + "all_port_forwards_stopped": forward_stopped, + "material_recorded": False, + "kubernetes_service_account_used_for_provider_login": False, + }, + } + return build_evidence(observation) + + +def build_validation_report( + *, + profile_path: Path = DEFAULT_PROFILE_PATH, + evidence_path: Path = DEFAULT_EVIDENCE_PATH, +) -> dict[str, Any]: + contract = build_contract_report(profile_path) + errors = list(contract["errors"]) + evidence: dict[str, Any] | None = None + try: + value = json.loads(evidence_path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("Spark interoperability evidence must be an object") + evidence = value + errors.extend(verify_evidence_integrity(evidence)) + observed_contract = _mapping( + _mapping(evidence.get("observation")).get("contract") + ).get("contract_fingerprint") + if observed_contract != contract.get("contract_fingerprint"): + errors.append("Spark interoperability evidence contract fingerprint drift") + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + errors.append(f"Spark interoperability evidence is invalid: {type(exc).__name__}") + verified = not errors + return { + "schema": VALIDATION_SCHEMA, + "local_static_contract_verified": contract["local_static_contract_verified"], + "local_spark_iceberg_rest_interoperability_verified": ( + verified + and evidence is not None + and evidence.get("local_spark_iceberg_rest_interoperability_verified") + is True + ), + "local_spark_create_read_write_verified": ( + verified + and evidence is not None + and evidence.get("local_spark_create_read_write_verified") is True + ), + "local_spark_schema_evolution_verified": ( + verified + and evidence is not None + and evidence.get("local_spark_schema_evolution_verified") is True + ), + "local_spark_snapshot_time_travel_verified": ( + verified + and evidence is not None + and evidence.get("local_spark_snapshot_time_travel_verified") is True + ), + "gravitino_api_metadata_readback_verified": ( + verified + and evidence is not None + and evidence.get("gravitino_api_metadata_readback_verified") is True + ), + "local_same_node_shared_pvc_verified": ( + verified + and evidence is not None + and evidence.get("local_same_node_shared_pvc_verified") is True + ), + "persistent_catalog_identity_binding_verified": False, + "protected_workload_identity_verified": False, + "oidc_verified": False, + "tls_verified": False, + "spark_conformance_verified": False, + "flink_conformance_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + "contract_fingerprint": contract["contract_fingerprint"], + "evidence_fingerprint": ( + evidence.get("evidence_fingerprint") if evidence else None + ), + "errors": errors, + } + + +def _write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + validate = subparsers.add_parser("validate") + validate.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + validate.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + rehearse = subparsers.add_parser("rehearse") + rehearse.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + rehearse.add_argument("--evidence-out", type=Path, required=True) + verify = subparsers.add_parser("verify") + verify.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + args = parser.parse_args(argv) + try: + if args.command == "validate": + report = build_validation_report( + profile_path=args.profile, evidence_path=args.evidence + ) + print(json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not report["errors"] else 1 + if args.command == "verify": + value = json.loads(args.evidence.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("Spark interoperability evidence must be an object") + errors = verify_evidence_integrity(value) + print(json.dumps({"verified": not errors, "errors": errors}, indent=2)) + return 0 if not errors else 1 + evidence = run_live_rehearsal(args.profile) + _write_json(args.evidence_out, evidence) + print(json.dumps(evidence, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not evidence["errors"] else 1 + except ( + KeyError, + OSError, + TypeError, + ValueError, + json.JSONDecodeError, + identity.MetadataFabricGravitinoIdentityError, + jdbc_restart.MetadataFabricGravitinoJdbcRestartError, + MetadataFabricSparkIcebergRestInteroperabilityError, + KeyboardInterrupt, + ) as exc: + print(f"metadata fabric Spark/Iceberg REST interoperability: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/platform_truth.py b/data_agent/platform_truth.py index 1f63d3ef..ebf93b80 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -668,6 +668,26 @@ def _config( ), "Protected identity and production catalog durability/conformance gate", ), + RuntimeSpec( + "metadata_spark_iceberg_rest_interoperability_rehearsal", + "spark_iceberg_rest_interoperability_rehearsal", + "governed", + "evidence_durable", + "committed local Spark/Iceberg REST interoperability evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_spark_iceberg_rest_interoperability.py", + "scripts/metadata-fabric-spark-iceberg-rest-interoperability.sh", + ), + ( + ( + "data_agent/metadata_fabric_spark_iceberg_rest_interoperability.py", + "subprocess.run", + ), + ), + "Protected Spark/Flink conformance and production object-store catalog gate", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_spark_iceberg_rest_interoperability.py b/data_agent/test_metadata_fabric_spark_iceberg_rest_interoperability.py new file mode 100644 index 00000000..2685019f --- /dev/null +++ b/data_agent/test_metadata_fabric_spark_iceberg_rest_interoperability.py @@ -0,0 +1,280 @@ +import json +from copy import deepcopy +from pathlib import Path + +import pytest +import yaml + +from data_agent import metadata_fabric_spark_iceberg_rest_interoperability as interop + + +def _checked_evidence() -> dict: + return json.loads(interop.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + + +def _observation() -> dict: + return deepcopy(_checked_evidence()["observation"]) + + +def _write_profile(tmp_path: Path, value: dict) -> Path: + path = tmp_path / "profile.yaml" + path.write_text(yaml.safe_dump(value, sort_keys=False), encoding="utf-8") + return path + + +def test_checked_in_contract_and_evidence_verify_the_local_boundary(): + contract = interop.build_contract_report() + validation = interop.build_validation_report() + evidence = _checked_evidence() + + assert contract["contract_fingerprint"] == ( + "a78b95d36a6a5d5f4b5e303be21263d00fdd7102c3a70ce282ba69f2d8cdcd2e" + ) + assert contract["local_static_contract_verified"] is True + assert contract["runtime_image_identity"] == { + "gravitino_host_image_id": interop.GRAVITINO_HOST_IMAGE_ID, + "gravitino_kubernetes_image_id": interop.GRAVITINO_KUBERNETES_IMAGE_ID, + "postgresql_image_digest": interop.POSTGRESQL_IMAGE_DIGEST, + "spark_host_image_id": interop.SPARK_HOST_IMAGE_ID, + "spark_kubernetes_image_id": interop.SPARK_KUBERNETES_IMAGE_ID, + } + assert interop.verify_evidence_integrity(evidence) == [] + assert evidence["evidence_fingerprint"] == ( + "50f9d0021db11e22364697d1ad8928ee068d28dc8046556bbca1a4e1c819f8e0" + ) + assert validation["errors"] == [] + assert validation["local_spark_iceberg_rest_interoperability_verified"] is True + assert validation["local_spark_create_read_write_verified"] is True + assert validation["local_spark_schema_evolution_verified"] is True + assert validation["local_spark_snapshot_time_travel_verified"] is True + assert validation["gravitino_api_metadata_readback_verified"] is True + assert validation["local_same_node_shared_pvc_verified"] is True + assert validation["spark_conformance_verified"] is False + assert validation["flink_conformance_verified"] is False + assert validation["persistent_catalog_identity_binding_verified"] is False + assert validation["protected_workload_identity_verified"] is False + assert validation["oidc_verified"] is False + assert validation["tls_verified"] is False + assert validation["production_ingestion_verified"] is False + assert validation["production_ready"] is False + + +def test_profile_rejects_privilege_expansion_and_sensitive_fields(tmp_path): + profile = yaml.safe_load(interop.DEFAULT_PROFILE_PATH.read_text(encoding="utf-8")) + profile["scope"]["role_securable_objects"][1]["privileges"].append( + {"name": "MODIFY_TABLE", "condition": "ALLOW"} + ) + profile["catalog"]["jdbc_password"] = "must-not-enter-profile" + + with pytest.raises( + interop.MetadataFabricSparkIcebergRestInteroperabilityError, + match="profile is invalid", + ): + interop.load_profile(_write_profile(tmp_path, profile)) + + +def test_profile_rejects_tampered_jdbc_restart_dependency(tmp_path, monkeypatch): + profile = yaml.safe_load(interop.DEFAULT_PROFILE_PATH.read_text(encoding="utf-8")) + dependency = json.loads( + ( + interop.REPO_ROOT + / "docs/evidence/metadata-fabric-gravitino-jdbc-restart-2026-07-29.json" + ).read_text(encoding="utf-8") + ) + dependency["production_ready"] = True + dependency_path = ( + tmp_path + / "docs/evidence/metadata-fabric-gravitino-jdbc-restart-2026-07-29.json" + ) + dependency_path.parent.mkdir(parents=True) + dependency_path.write_text(json.dumps(dependency), encoding="utf-8") + monkeypatch.setattr(interop, "REPO_ROOT", tmp_path) + + with pytest.raises( + interop.MetadataFabricSparkIcebergRestInteroperabilityError, + match="dependency does not match", + ): + interop.load_profile(_write_profile(tmp_path, profile)) + + +def test_manifest_rejects_committed_secret_and_incomplete_runtime( + tmp_path, monkeypatch +): + (tmp_path / "runtime.yaml").write_text( + """ +apiVersion: v1 +kind: Namespace +metadata: + name: gda-metadata-spark-interop +--- +apiVersion: v1 +kind: Secret +metadata: + name: forbidden +stringData: + password: forbidden +""".strip(), + encoding="utf-8", + ) + monkeypatch.setattr(interop, "MANIFEST_DIR", tmp_path) + + errors = interop._validate_manifest() + + assert "Spark interoperability manifest may not commit Secret values" in errors + assert "Spark interoperability manifest is incomplete" in errors + + +def test_manifest_requires_suspended_tokenless_job_with_resources(monkeypatch): + documents = interop._manifest_documents() + job = next(document for document in documents if document.get("kind") == "Job") + job["spec"]["suspend"] = False + pod_spec = job["spec"]["template"]["spec"] + pod_spec["automountServiceAccountToken"] = True + pod_spec["containers"][0].pop("resources") + monkeypatch.setattr(interop, "_manifest_documents", lambda: documents) + + errors = interop._validate_manifest() + + assert "Spark interoperability Job must start suspended" in errors + assert "Spark interoperability Job must disable token automount" in errors + assert "Spark interoperability Job resources are incomplete" in errors + + +@pytest.mark.parametrize( + ("mutate", "expected"), + [ + ( + lambda value: value["spark"]["job"].update( + {"succeeded": 0, "failed": 1} + ), + "Spark interoperability Job did not complete exactly once", + ), + ( + lambda value: value["spark"]["pod"].update( + {"image_id": "sha256:" + "0" * 64} + ), + "Spark interoperability runtime boundary does not match", + ), + ( + lambda value: value["spark"]["pod"].update( + {"warehouse_pvc": "unrelated-pvc"} + ), + "Spark interoperability runtime boundary does not match", + ), + ( + lambda value: value["runtime"]["iceberg_rest"].update( + {"ready": False} + ), + "Iceberg REST sidecar boundary does not match", + ), + ], +) +def test_evidence_rejects_job_image_pvc_or_rest_runtime_drift(mutate, expected): + observation = _observation() + mutate(observation) + + evidence = interop.build_evidence(observation) + + assert expected in evidence["errors"] + assert evidence["local_spark_iceberg_rest_interoperability_verified"] is False + if expected.startswith("Spark interoperability"): + assert evidence["local_spark_create_read_write_verified"] is False + assert evidence["local_spark_schema_evolution_verified"] is False + assert evidence["local_spark_snapshot_time_travel_verified"] is False + + +@pytest.mark.parametrize( + ("mutate", "expected"), + [ + ( + lambda value: value["spark"]["result"].update( + {"current_rows": [["forged", None]]} + ), + "Spark create/read/write result does not match", + ), + ( + lambda value: value["spark"]["result"].update( + {"snapshot_operations": ["append", "overwrite"]} + ), + "Spark snapshot/time-travel result does not match", + ), + ( + lambda value: value["post_spark"]["table"]["projection"][ + "columns" + ].pop(), + "Gravitino API did not read back Spark schema evolution", + ), + ( + lambda value: value["post_spark"].update( + {"denied_catalog_create_status": 200} + ), + "Post-Spark administrative catalog mutation was not denied", + ), + ], +) +def test_evidence_rejects_data_snapshot_schema_or_denial_drift(mutate, expected): + observation = _observation() + mutate(observation) + + evidence = interop.build_evidence(observation) + + assert expected in evidence["errors"] + assert evidence["local_spark_iceberg_rest_interoperability_verified"] is False + + +def test_evidence_recomputes_api_fingerprint_instead_of_trusting_it(): + observation = _observation() + projection = observation["post_spark"]["table"]["projection"] + projection["columns"][1]["nullable"] = False + observation["post_spark"]["table"]["fingerprint"] = ( + interop.recovery._canonical_sha256(projection) + ) + + evidence = interop.build_evidence(observation) + + assert "Gravitino API did not read back Spark schema evolution" in evidence["errors"] + assert evidence["gravitino_api_metadata_readback_verified"] is False + + +def test_evidence_rejects_incomplete_cleanup_and_sensitive_material(): + observation = _observation() + observation["runtime_checks"]["namespace_absent"] = False + observation["database_password"] = "must-not-enter-evidence" + + evidence = interop.build_evidence(observation) + + assert "Spark interoperability observation contains sensitive material" in evidence[ + "errors" + ] + assert "Spark interoperability cleanup is incomplete" in evidence["errors"] + assert evidence["local_spark_iceberg_rest_interoperability_verified"] is False + + +def test_evidence_integrity_rejects_tampering_and_production_overclaim(): + evidence = _checked_evidence() + evidence["observation"]["spark"]["result"]["current_rows"][0][0] = "tampered" + + errors = interop.verify_evidence_integrity(evidence) + + assert "Spark interoperability evidence fingerprint does not match" in errors + + forged = _checked_evidence() + forged["spark_conformance_verified"] = True + forged["production_ready"] = True + stable = { + key: value for key, value in forged.items() if key != "evidence_fingerprint" + } + forged["evidence_fingerprint"] = interop.recovery._canonical_sha256(stable) + errors = interop.verify_evidence_integrity(forged) + assert ( + "Spark interoperability evidence may not claim spark_conformance_verified" + in errors + ) + assert "Spark interoperability evidence may not claim production_ready" in errors + + +def test_wrapper_is_fail_closed(): + wrapper = interop.DEFAULT_WRAPPER_PATH.read_text(encoding="utf-8") + + assert "set -euo pipefail" in wrapper + assert "metadata_fabric_spark_iceberg_rest_interoperability" in wrapper diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index a7e33ab1..b1dad665 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -218,6 +218,12 @@ def test_repository_source_access_and_runtime_baselines_match(): and item["production_role"] == "local_verification_only" for item in static_report["runtime"]["inventory"] ) + assert any( + item["runtime_id"] + == "metadata_spark_iceberg_rest_interoperability_rehearsal" + and item["production_role"] == "local_verification_only" + for item in static_report["runtime"]["inventory"] + ) def test_runtime_report_detects_unregistered_background_mechanism(tmp_path): diff --git a/docs/architecture-decisions/adr-055-local-spark-iceberg-rest-interoperability.md b/docs/architecture-decisions/adr-055-local-spark-iceberg-rest-interoperability.md new file mode 100644 index 00000000..8fa8335f --- /dev/null +++ b/docs/architecture-decisions/adr-055-local-spark-iceberg-rest-interoperability.md @@ -0,0 +1,73 @@ +# ADR-055: Local Spark and Gravitino Iceberg REST Interoperability + +**Status**: Accepted + +**Date**: 2026-07-29 + +**Decision owners**: Metadata Platform, Data Engineering, SRE, Platform Architecture + +**Related decisions**: [ADR-006](adr-006-openmetadata-governance-and-active-metadata-platform.md) · [ADR-052](adr-052-local-gravitino-basic-bounded-provider-identity.md) · [ADR-054](adr-054-local-gravitino-jdbc-catalog-restart-continuity.md) + +## Context + +M3-8 proved authenticated Gravitino JDBC catalog continuity across controlled PostgreSQL and Gravitino restarts. It did not prove that Spark can use that catalog through a standard engine protocol. ADR-006 requires real Spark/Flink create, read, write, schema evolution, snapshot, cancellation, reconciliation and lineage conformance before Gravitino can become the sole production catalog for those engines. Trino success or provider documentation cannot substitute for that evidence. + +The next bounded step is real Spark 3.5 interoperability through Gravitino's standard Iceberg REST server. It must reuse the same JDBC database and warehouse as the Gravitino API, preserve the M3-6 authorization denial, and prove bidirectional metadata visibility. It is not the complete ADR-006 conformance suite. + +## Decision + +### 1. Use one isolated same-node catalog boundary + +The rehearsal creates only `gda-metadata-spark-interop` in Docker Desktop Kubernetes. PostgreSQL stores Gravitino entities and the separate Iceberg JDBC catalog. A Gravitino `1.3.0` StatefulSet runs the authenticated API and the bundled Iceberg REST `1.11.0` server as two containers. They mount the same local warehouse PVC and use the same PostgreSQL `iceberg` database. + +Spark `3.5.0` with Iceberg runtime `1.6.1` runs as a suspended Job that is released only after Gravitino creates the table. The Job, Gravitino and PostgreSQL are pinned to `desktop-worker`; Spark and Gravitino mount the same `ReadWriteOnce` warehouse PVC. This deliberately avoids adding object storage to the first interoperability slice, so the result applies only to one local node and failure domain. + +No Secret is committed. Administrator, bounded-user and database materials are generated for one run and excluded from the observation. Dedicated ServiceAccounts disable token automount. The namespace and its dynamically provisioned volumes must be absent before evidence can pass. + +### 2. Keep authenticated control and engine protocol claims separate + +The Gravitino API uses the built-in Basic IdP and authorization. The bounded role contains exactly `USE_CATALOG` on `lakehouse`, plus `USE_SCHEMA` and `CREATE_TABLE` on `lakehouse.published`. That user creates and reads `gda_spark_interop_probe`; catalog creation must return 403 before Spark runs and again afterward. + +Spark connects to `http://gravitino-persistence:9001/iceberg`, the standard Iceberg REST endpoint. This local endpoint is unauthenticated HTTP. The authenticated Gravitino create/read and denial checks do not turn Spark's connection into protected workload identity, OIDC, TLS or production provider authorization evidence. + +### 3. Require cross-client data and metadata verification + +The Spark Job must: + +- read the zero-row table and required `probe_id` column created by Gravitino; +- append two deterministic rows through Iceberg REST; +- add nullable `quality`, then append a third deterministic row; +- read the exact evolved schema and three current rows; +- observe two append snapshots and time-travel to the first snapshot; +- emit one structured result whose Spark, Iceberg, catalog URI and table identity match the checked profile. + +After the Job succeeds, the bounded Gravitino user must read back the new `quality` column from the same JDBC catalog. Evidence binds the Docker host image index and the Kubernetes ARM64 manifest identity separately, because those runtimes report different but legitimate identifiers for the Spark image. + +### 4. Do not promote the result to full conformance + +Evidence may set these local claims to true: `local_spark_iceberg_rest_interoperability_verified`, create/read/write, schema evolution, snapshot/time-travel, Gravitino API metadata readback and same-node shared-PVC verification. + +`spark_conformance_verified` remains false because cancellation, reconcile and production lineage were not exercised. Flink conformance, persistent production identity binding, protected workload identity, OIDC, TLS, object-store durability, production ingestion and `production_ready` also remain false. + +## Verification + +The final Docker Desktop rehearsal produced contract fingerprint `a78b95d36a6a5d5f4b5e303be21263d00fdd7102c3a70ce282ba69f2d8cdcd2e` and evidence fingerprint `50f9d0021db11e22364697d1ad8928ee068d28dc8046556bbca1a4e1c819f8e0`. + +- the bounded Gravitino user created/read the table and received 403 for catalog creation before and after Spark; +- Spark read the Gravitino-created table, committed two append snapshots, evolved the schema and read three exact rows; +- time travel to the first snapshot returned only `spark-a` and `spark-b`; +- Gravitino read back the Spark-added nullable `quality` column; +- Spark, Gravitino and PostgreSQL ran on `desktop-worker`, and Spark/Gravitino referenced the same warehouse PVC; +- the Spark Job completed once using the checked ARM64 manifest, all ServiceAccounts had token automount disabled, the port-forward stopped, and the namespace/PVs were deleted. + +Focused tests cover profile and dependency drift, committed Secret rejection, suspended Job/security/resource requirements, engine image and PVC identity, REST readiness, deterministic rows, snapshot operations, time travel, API schema readback, authorization denial, cleanup, sensitive material, evidence tampering and production overclaim. CI validates the checked evidence without running Kubernetes; the live rehearsal remains an explicit local operator action. + +## Consequences + +**Positive**: the catalog path now has real Spark 3.5 create/read/write, schema evolution, snapshot and bidirectional metadata evidence through the standard Iceberg REST protocol. Trino or documentation is not used as a proxy. + +**Negative**: all components share one Docker Desktop node, one local RWO PVC, Basic credentials and unauthenticated HTTP. This does not test object storage, multiple Spark executors, cancellation, reconcile, OpenLineage, tenant isolation or failures during commits. + +**Mitigation**: keep the independent authenticated Iceberg REST provider as the production fallback required by ADR-006. The next conformance slices must move the warehouse to the selected object store, use protected identity/TLS, add multi-node Spark/Sedona and Flink, and test cancel/reconcile/lineage plus failure injection. + +**Revisit trigger**: Gravitino, Spark or Iceberg version changes; catalog or warehouse implementation changes; production authentication is selected; or the full ADR-006 conformance suite is introduced. diff --git a/docs/evidence/metadata-fabric-spark-iceberg-rest-interoperability-2026-07-29.json b/docs/evidence/metadata-fabric-spark-iceberg-rest-interoperability-2026-07-29.json new file mode 100644 index 00000000..a5cd1404 --- /dev/null +++ b/docs/evidence/metadata-fabric-spark-iceberg-rest-interoperability-2026-07-29.json @@ -0,0 +1,269 @@ +{ + "errors": [], + "evidence_fingerprint": "50f9d0021db11e22364697d1ad8928ee068d28dc8046556bbca1a4e1c819f8e0", + "flink_conformance_verified": false, + "gravitino_api_metadata_readback_verified": true, + "local_same_node_shared_pvc_verified": true, + "local_spark_create_read_write_verified": true, + "local_spark_iceberg_rest_interoperability_verified": true, + "local_spark_schema_evolution_verified": true, + "local_spark_snapshot_time_travel_verified": true, + "local_static_contract_verified": true, + "observation": { + "contract": { + "contract_fingerprint": "a78b95d36a6a5d5f4b5e303be21263d00fdd7102c3a70ce282ba69f2d8cdcd2e", + "jdbc_restart_evidence_fingerprint": "34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa", + "local_static_contract_verified": true + }, + "observed_at": "2026-07-29T10:45:18.802719+00:00", + "post_spark": { + "authentication_status": 200, + "denied_catalog_create_status": 403, + "read_status": 200, + "table": { + "fingerprint": "598d9308cce90331df3f52fe6e524c16012e0aee59150392f112948a73f024ed", + "name": "gda_spark_interop_probe", + "projection": { + "columns": [ + { + "name": "probe_id", + "nullable": false, + "type": "string" + }, + { + "name": "quality", + "nullable": true, + "type": "string" + } + ], + "name": "gda_spark_interop_probe", + "probe_property": "true" + } + } + }, + "pre_spark": { + "authentication": { + "admin_status": 200, + "bounded_status": 200, + "material_recorded": false + }, + "catalog": { + "backend": "jdbc", + "catalog": "lakehouse", + "jdbc_initialize": true, + "material_recorded": false, + "metalake": "gda_interop", + "provider": "lakehouse-iceberg", + "schema": "published", + "uri": "jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg", + "warehouse": "file:///var/lib/gravitino/warehouse" + }, + "denied_catalog_create_status": 403, + "role": { + "name": "gda-table-projection", + "securable_objects": [ + { + "fullName": "lakehouse", + "privileges": [ + { + "condition": "ALLOW", + "name": "USE_CATALOG" + } + ], + "type": "CATALOG" + }, + { + "fullName": "lakehouse.published", + "privileges": [ + { + "condition": "ALLOW", + "name": "CREATE_TABLE" + }, + { + "condition": "ALLOW", + "name": "USE_SCHEMA" + } + ], + "type": "SCHEMA" + } + ] + }, + "table": { + "create_status": 200, + "fingerprint": "d4294e37113ecca08ed893244650859f65dac6bdea5b544028e573e06d613144", + "name": "gda_spark_interop_probe", + "projection": { + "columns": [ + { + "name": "probe_id", + "nullable": false, + "type": "string" + } + ], + "name": "gda_spark_interop_probe", + "probe_property": "true" + }, + "read_status": 200 + } + }, + "runtime": { + "context": "docker-desktop", + "gravitino": { + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "node_name": "desktop-worker", + "pod_name": "gravitino-persistence-0", + "pod_uid": "f9830b05-1516-466f-af7c-88866ac54ce3", + "pvc": { + "name": "warehouse-gravitino-persistence-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "94d822c3-3c55-4c05-8055-a7f263154dfe", + "volume_name": "pvc-94d822c3-3c55-4c05-8055-a7f263154dfe" + }, + "ready_replicas": 1, + "service_account": "gravitino-persistence", + "service_account_automount_disabled": true, + "statefulset_uid": "04e83aa3-d069-42b9-bcaf-b303e041ac55" + }, + "gravitino_host_image_id": "sha256:d355dc7e92f9e3545d717f3eab2cbdf412115f2b82e1e544d7f6235c1eacd5a5", + "gravitino_jdbc_driver_mounted": true, + "iceberg_rest": { + "image": "docker.io/gda/gravitino:1.3.0-local-arm64", + "image_id": "sha256:18e24b43be854dabdc13e96b1019eb3dc691d59cc64e411aa6a3cc49225fe2d3", + "jdbc_driver_mounted": true, + "path": "/iceberg", + "ready": true + }, + "namespace": { + "name": "gda-metadata-spark-interop", + "uid": "78605ecc-c13f-4d25-9bd6-e272ba15ed67" + }, + "postgresql": { + "image": "docker.io/library/postgres:16.10-bookworm", + "image_id": "docker.io/library/postgres@sha256:38471f330eb885e04de130b768d6db4e10469e2311879c7e5c699f6d2d8a1c74", + "node_name": "desktop-worker", + "pod_name": "gravitino-persistence-postgresql-0", + "pod_uid": "b16189d2-96c7-4721-9036-f548c8ff3244", + "pvc": { + "name": "data-gravitino-persistence-postgresql-0", + "phase": "Bound", + "storage_class": "standard", + "uid": "7118fb78-68b5-45ad-8c9e-fa40d9adaa01", + "volume_name": "pvc-7118fb78-68b5-45ad-8c9e-fa40d9adaa01" + }, + "ready_replicas": 1, + "service_account": "gravitino-persistence-postgresql", + "service_account_automount_disabled": true, + "statefulset_uid": "3ce1b9a8-b733-4021-a96e-4ff84e318916" + }, + "service": { + "name": "gravitino-persistence", + "ports": [ + { + "name": "http", + "port": 8090 + }, + { + "name": "iceberg-rest", + "port": 9001 + } + ], + "type": "ClusterIP", + "uid": "e7983504-3357-4a20-982e-2ac9a4f05257" + }, + "source_schema_sha256": "7a2d605a677a462ca619dba594ce7ebcf500358345560ad084c1b67a25c722df", + "spark_host_image_id": "sha256:f201367640c7583add224796a629150e63d3859ddd7fe9fd47741662a6d415bb" + }, + "runtime_checks": { + "all_port_forwards_stopped": true, + "kubernetes_service_account_used_for_provider_login": false, + "material_recorded": false, + "namespace_absent": true, + "namespace_delete_completed": true, + "persistent_volumes_retained": false, + "provider_objects_retained": false + }, + "schema": "gda.metadata_fabric_spark_iceberg_rest_interoperability_observation.v1", + "spark": { + "job": { + "completion_time": "2026-07-29T10:44:00Z", + "failed": 0, + "name": "spark-iceberg-rest-probe", + "succeeded": 1, + "uid": "1c6ebd13-6655-4083-9e92-a23989d53f06" + }, + "log_recorded": false, + "log_sha256": "76b83d4cffed47d74abe1e3fda54fb29a56e847d105691fa309fb2e655152f3a", + "pod": { + "image": "docker.io/gisdataagent/mmfe-spark-runtime:local", + "image_id": "sha256:4a4522bfd4e6d1c6c90a244d0145841fbfbbf21ed16ee29ca8b681b5cec60058", + "name": "spark-iceberg-rest-probe-cq2gt", + "node_name": "desktop-worker", + "phase": "Succeeded", + "service_account": "spark-iceberg-rest-probe", + "service_account_automount_disabled": true, + "uid": "b338906f-167f-4ccd-abd7-d41199f27062", + "warehouse_pvc": "warehouse-gravitino-persistence-0" + }, + "result": { + "catalog_uri": "http://gravitino-persistence:9001/iceberg", + "create_read_write_verified": true, + "current_columns": [ + "probe_id", + "quality" + ], + "current_rows": [ + [ + "spark-a", + null + ], + [ + "spark-b", + null + ], + [ + "spark-c", + "verified" + ] + ], + "iceberg_runtime": "1.6.1", + "initial_columns": [ + "probe_id" + ], + "initial_row_count": 0, + "schema": "gda.spark_iceberg_rest_probe_result.v1", + "schema_evolution_verified": true, + "snapshot_history_verified": true, + "snapshot_ids": [ + 8331041238302716169, + 6506991433109985360 + ], + "snapshot_operations": [ + "append", + "append" + ], + "spark_version": "3.5.0", + "table": "rest.published.gda_spark_interop_probe", + "time_travel_rows": [ + "spark-a", + "spark-b" + ], + "time_travel_snapshot_id": 8331041238302716169, + "time_travel_verified": true + }, + "result_line_count": 1, + "terminal_condition": "Complete", + "wait_completed": true + } + }, + "observed_at": "2026-07-29T10:45:18.802719+00:00", + "oidc_verified": false, + "persistent_catalog_identity_binding_verified": false, + "production_ingestion_verified": false, + "production_ready": false, + "protected_workload_identity_verified": false, + "schema": "gda.metadata_fabric_spark_iceberg_rest_interoperability_evidence.v1", + "spark_conformance_verified": false, + "tls_verified": false +} diff --git a/docs/roadmap-ar0-platform-truth-2026-07-24.md b/docs/roadmap-ar0-platform-truth-2026-07-24.md index 4527d915..1cc8aa62 100644 --- a/docs/roadmap-ar0-platform-truth-2026-07-24.md +++ b/docs/roadmap-ar0-platform-truth-2026-07-24.md @@ -199,7 +199,7 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 当前完成仅指本地合同、授权 evidence、outbox/callback 代码、数据库成功终局门、托管 worker 代码、默认关闭的部署模板及离线 activation/release preflight、candidate/registry/provenance/artifact-release/live observation evidence gate、合成 golden slice、定向测试、真实 PostgreSQL 16 事务边界和 canonical mainline 治理。`candidate_validated`、`registry_subject_bound`、本地合成 `provenance_verified`、`ready_for_activation`、`ready_for_staging_apply`、`verified_for_staging_apply` 和本地 live collection 都不等于真实镜像已 attested 或 staging 已部署；真实 IAM/OIDC 与 service token 生命周期、首次 GHCR publish/verify、真实 provenance artifact verify、registry-backed live staging revision、worker/callback 扩容运行、golden slice staging 运行链、受保护 release/live evidence provenance、独立 DolphinScheduler metadata PostgreSQL 和真实数据终局证据仍属于 4.7 后续切片。 -### 4.8 Metadata Fabric Bridge M1 + M2 + M3-8（本地认证 JDBC catalog 重启连续性已验证） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-9（本地 Spark/Iceberg REST 互操作已验证） 第八块回到 AR-1 的 metadata control plane，以 [ADR-036](architecture-decisions/adr-036-read-only-metadata-fabric-bridge-contract.md) 固定 OpenMetadata + Gravitino + GDA Control Ledger 的首条 table slice： @@ -228,8 +228,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 23. [ADR-052](architecture-decisions/adr-052-local-gravitino-basic-bounded-provider-identity.md) 已在隔离 Gravitino `1.3.0` namespace 启用 Basic IdP 与 authorization，bounded user 只获得 `lakehouse` 的 `USE_CATALOG` 以及 `lakehouse.published` 的 `USE_SCHEMA`/`CREATE_TABLE`；table create/read 为 200/200，越权 catalog create 为 403，密码轮换使旧值返回 401，IdP 用户删除使替换值返回 401。临时 metalake/catalog/schema/table/user/role、namespace 与 loopback port-forward 均已清理，证据 SHA 为 `f0b0de1f80f079d43318937e0a0cc151a8546e9e307bef204738b1367f9b29fd`。这只是 Gravitino Basic local POC；probe catalog 使用 memory backend，不包含 OIDC、TLS、Kubernetes-to-provider identity exchange、持久生产 catalog 或双 provider production identity。 24. [ADR-053](architecture-decisions/adr-053-production-metadata-fabric-identity-readiness-gate.md) 已将 OIDC federation、双 provider integration、digest-pinned authentication component、workload/tenant claim、Kubernetes ServiceAccount、禁止 direct bypass、short-lived token、M3-5/M3-6 精确 allow/deny contract、TLS/mTLS、持久 Gravitino catalog、tenant isolation、owner/audit/SLO/runbook 与 18 项 protected attestation check 冻结为 fail-closed profile。Gravitino `1.3.0` 镜像只发现 Basic IdP jar，因此生产只允许明确选择并证明 `custom_oidc_authenticator` 或 `identity_aware_proxy`，不假设 native OIDC。当前 profile fingerprint 为 `2e9d5cac3560b853820f923669f6794ead63bcb36a528639fc0e9539e148ee2f`，合同有效但 40 项外部输入 blocked，没有真实 attestation，全部 production identity claims 仍为 `false`。 25. [ADR-054](architecture-decisions/adr-054-local-gravitino-jdbc-catalog-restart-continuity.md) 已在隔离 namespace 中将 Gravitino Iceberg catalog metadata 落到 PostgreSQL JDBC、warehouse 落到独立 PVC，并复用 M3-6 精确 `USE_CATALOG`/`USE_SCHEMA`/`CREATE_TABLE` 角色。依次重启 PostgreSQL 与 Gravitino 后，两者 Pod UID 均变化而 StatefulSet/PVC UID 保持；同一 bounded user 重新认证并读取相同 table fingerprint，catalog create 前后均为 403，namespace/PV 完整清理。evidence fingerprint 为 `34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa`。该结果仅证明 Docker Desktop 单集群、本地 Basic/HTTP/file warehouse 的 restart continuity，不等于 production persistent identity binding、OIDC、TLS、备份恢复、Spark/Flink conformance 或生产 ingestion。 +26. [ADR-055](architecture-decisions/adr-055-local-spark-iceberg-rest-interoperability.md) 已将 Gravitino API 与 bundled Iceberg REST `1.11.0` 连接到同一 PostgreSQL JDBC catalog 和 file warehouse PVC，并在 `desktop-worker` 运行 Spark `3.5.0` + Iceberg `1.6.1`。bounded Basic user 先创建零行表且 catalog create 返回 403；Spark 经标准 `/iceberg` REST 读取该表、两次 append、增加 nullable `quality`、验证三行 current state、两个 snapshot 与 first-snapshot time travel；随后 Gravitino API 回读相同演进 schema，catalog create 仍为 403，namespace/PV 完整清理。该结果只证明本地同节点共享 RWO PVC 的 engine interoperability；Spark REST 路径仍是无认证 HTTP，cancel/reconcile/lineage、Flink、对象存储、生产身份/TLS 和完整 `spark_conformance_verified` 均未证明。 -此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续，不证明生产 failure domain、持久 identity binding 或 engine conformance。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产 OpenLineage、生产 ingest/conformance、三项 production gate 和 `production_ready` 仍为 `false`。 +此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明 Spark 经无认证本地 Iceberg REST 在同节点共享 file warehouse PVC 上完成 read/write/schema evolution/snapshot/time travel，并由 Gravitino 回读 schema，不证明生产 failure domain、持久 identity binding、Flink 或完整 engine conformance。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产 OpenLineage、生产 ingest/conformance、三项 production gate 和 `production_ready` 仍为 `false`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index 8dcf150e..dc20bc55 100644 --- a/docs/system-of-record-matrix-2026-07-24.md +++ b/docs/system-of-record-matrix-2026-07-24.md @@ -2,9 +2,9 @@ 日期：2026-07-29 -阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract 与 M3-8 local Gravitino JDBC restart continuity 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity attestation 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity 与 M3-9 local Spark/Iceberg REST interoperability 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity attestation 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-jdbc-catalog-restart` +适用分支：`feat/ar1-metadata-fabric-spark-iceberg-rest-interoperability` ## 判定规则 @@ -20,12 +20,12 @@ | SQL schema 历史 | PostgreSQL `schema_migrations`，以完整 migration ID + checksum 为权威 | migration CLI 的 JSON 报告 | 保持现有 ledger；任何 drift fail closed | Data Platform | AR-0，已验证 | | 部署配置策略 | Compose/K8s/进程环境；`platform_truth.CONFIG_SPECS` 定义关键类型与策略；DolphinScheduler worker 有默认零副本、外部 ConfigMap/Secret 驱动的 Kustomize 模板、静态 validator 和 staging activation preflight | `.env` 仅补默认；脱敏 snapshot、Secret key attestation、未扩容 Deployment 和 `ready_for_activation` 都是观测/模板 | 版本化 DeploymentProfile + secret reference；部署环境始终优先；模板或 preflight 通过都不等于环境已启用 | Platform/SRE/Security | AR-0，部分实现；worker 模板/preflight 本地已验证 | | 环境发布与晋级 | 本地 candidate/registry/provenance/release/live 合同已绑定 publisher、verifier、OCI 和 manifest identity；canonical `main@0182406`、archive refs、三组 active ruleset 与 `staging-provenance` protected environment 已建立，但尚无成功 publisher/verifier 或 deployment | 旧 mainline、feature branch、CI artifact、JSON、离线 report 和合成 `verified_for_staging_apply` 都不能单独成为发布权威；publisher SHA、verifier SHA 与 branch lineage 必须分别验证 | 由受保护 environment 的 DeploymentRevision 绑定 OCI、provenance artifact、release manifest 与全部 live verdict | Platform/SRE/Security/Repository Owner | AR-1 mainline 治理已恢复 -> 首次 GHCR publish/verify -> 真实 staging | -| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；M2b recovery runner、M2c-1 provider probe、M2c-2 `_OtelPortForward`、M2c-3 failure rehearsal、M2d-1 NetworkPolicy rehearsal 与 M3-8 JDBC restart runner 均登记为 `local_verification_only`，不是 scheduler、worker、持续监控、生产 policy/catalog controller 或状态权威 | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog-restart evidence | PlatformRun ledger 唯一登记最终状态；framework/provider attempt 只能回报观测；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker 本地已验证；metadata recovery/metrics/policy/catalog runner 仅本地验证 -> staging 控制链待接入 | +| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；M2b recovery runner、M2c-1 provider probe、M2c-2 `_OtelPortForward`、M2c-3 failure rehearsal、M2d-1 NetworkPolicy rehearsal、M3-8 JDBC restart runner 与 M3-9 Spark interoperability runner 均登记为 `local_verification_only`，不是 scheduler、worker、持续监控、生产 policy/catalog controller 或状态权威 | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability evidence | PlatformRun ledger 唯一登记最终状态；framework/provider attempt 只能回报观测；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker 本地已验证；metadata recovery/metrics/policy/catalog/interoperability runner 仅本地验证 -> staging 控制链待接入 | | 原始文件/对象 | 当前 local uploads、S3/MinIO/OBS 均可能被直接写入，权威边界未统一 | 临时上传、下载缓存、预览文件 | Landing object 以 immutable URI + checksum + retention 为权威；本地 scratch 可删除 | Data Platform | AR-2 | | 湖仓表与 snapshot | Iceberg/STAC/S3A 有局部实现，尚无通用发布权威 | STAC item、GeoParquet export | Iceberg catalog snapshot 是分析表版本权威；对象是物理内容，STAC 是发现投影 | Data Platform | AR-2 | | 在线空间数据 | PostGIS 业务表是当前编辑/查询事实，部分临时表混入 | Martin MVT、API JSON、导出文件 | 已批准 DataProductVersion 物化到 PostGIS；不能由瓦片或临时表反向定义产品版本 | GIS/Data Platform | AR-2 -> AR-4 | | 数据资产身份与版本 | `gda_control.resource/resource_version` 已实现 identity、hash、predecessor、tenant FK 和幂等 gateway 写入；`agent_data_assets`、`agent_asset_versions` 仍是兼容写路径 | UI catalog、search index、STAC | GDA ledger 管身份与版本绑定；旧行只有在 tenant、authority identity、checksum 和 version evidence 完整时才可形成 eligible plan；OpenMetadata 管治理目录，Gravitino 管技术对象映射 | Metadata Platform | AR-1 gateway 已验证 -> 生产切换待验收 | -| 技术元数据 | M1 已冻结 Gravitino table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-1 固定 technical projection intent，M3-2 已在 Gravitino memory catalog 创建/read-back，M3-3 将验证后的 ref 追加到 tenant-scoped 本地 binding ledger；M3-6 又在隔离 Gravitino Basic IdP 中验证 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 已冻结 production identity profile/attestation gate；M3-8 已验证同一 Basic role 与 Iceberg JDBC table 在 PostgreSQL/Gravitino Pod restart 后保持 | harvester 结果、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity/JDBC restart observation、projection plan、provider evidence、binding ledger 与 readiness report | 源系统技术对象是原始证据；Gravitino 映射并联邦，不能覆盖业务 ResourceVersion；GDA binding ledger 只记录已验证关系，本地 memory/JDBC catalog evidence 不得冒充生产持久技术权威；Basic IdP、loopback HTTP、file warehouse、pending profile 和合成 attestation 都不是生产身份或生产 storage | Metadata Platform | AR-1 M1/M2 + M3-6 identity + M3-7 gate + M3-8 local persistence 已验证 -> 受保护身份/生产 catalog/attestation/conformance 待执行 | +| 技术元数据 | M1 已冻结 Gravitino table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-1 固定 technical projection intent，M3-2 已在 Gravitino memory catalog 创建/read-back，M3-3 将验证后的 ref 追加到 tenant-scoped 本地 binding ledger；M3-6 又在隔离 Gravitino Basic IdP 中验证 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 已冻结 production identity profile/attestation gate；M3-8 已验证同一 Basic role 与 Iceberg JDBC table 在 PostgreSQL/Gravitino Pod restart 后保持；M3-9 已验证 Spark 经标准 Iceberg REST 对同一 JDBC catalog 做 read/write/schema evolution/snapshot/time travel，并由 Gravitino API 回读 schema | harvester 结果、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity/JDBC restart/Spark interoperability observation、projection plan、provider evidence、binding ledger 与 readiness report | 源系统技术对象是原始证据；Gravitino 映射并联邦，不能覆盖业务 ResourceVersion；GDA binding ledger 只记录已验证关系，本地 memory/JDBC/file-PVC catalog evidence 不得冒充生产持久技术权威；Basic IdP、无认证 REST/HTTP、file warehouse、pending profile 和合成 attestation 都不是生产身份或生产 storage | Metadata Platform | AR-1 M1/M2 + M3-6 identity + M3-7 gate + M3-8 local persistence + M3-9 local Spark interoperability 已验证 -> 受保护身份/对象存储/完整 Spark-Flink conformance 待执行 | | 治理目录 | M1 已冻结 OpenMetadata table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-2 已用 bootstrap admin 创建目标并回读真实 UUID；M3-3 将该 UUID 经 evidence gate 追加到本地 GDA binding ledger；M3-4 将精确 OpenLineage candidate 经 outbox 投递到本地 HTTP receiver；M3-5 已验证临时非管理员 bot 的 scoped `table/Create` grant、policy-create 拒绝及 JWT 轮换/吊销；M3-7 将其 allow/deny 范围纳入双 provider production identity gate | 搜索/页面视图、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity observation、projection/provider evidence、binding ledger、lineage outbox/receipt、OpenLineage event 与 readiness report | OpenMetadata 为 owner/glossary/classification/quality discoverability 权威；GDA ledger 保留审批/provider identity，outbox 只拥有投递状态，receiver 拥有接收状态；pending identity profile 和合成 attestation 均不反写 ResourceVersion 或建立生产权威 | Governance | AR-1 M1/M2 + M3-5 local bounded identity + M3-7 readiness contract 已验证 -> protected identity ingestion/生产持久 binding/受保护 production receiver 待执行 | | 血缘 | `gda_control.lineage_event` 已实现 immutable version edge 和幂等 gateway ingest；`agent_asset_lineage` 旧记录仍是可变 asset edge | OpenMetadata lineage graph、UI DAG | 只有 source/target ResourceVersion 与 event checksum 证据完整的旧记录可形成 eligible plan；目录图只作可重建投影 | Data Platform | AR-1 gateway 已验证 -> adapter 待接入 | | Definition | `gda_control.platform_definition_version` 已绑定 definition ResourceVersion、完整逻辑 hash 和原子 gateway registration；3.4.2 adapter 可编译、创建并上线 provider DAG；binding 已以 append-only `execution_plan` Artifact 持久化并可按 tenant + artifact UUID 读取，旧 workflow/template/YAML 仍在写入 | 编辑器状态、DolphinScheduler DAG/definition | 旧 workflow 必须规范化并完整 hash 后才可形成 PlatformDefinitionVersion；provider binding 作为 ExecutionPlanArtifact/evidence，不可反写 definition | DataOps | AR-1 binding persistence 代码已验证 -> staging 调用链待验收 | @@ -56,7 +56,7 @@ 11. `platform_command_outbox` 只拥有投递状态；callback 只触发 reconcile，不能把 provider payload 直接写成 PlatformRun 状态或平台终局。 12. QualityResult evaluator 必须是 workload，且成功终局中的 evaluator 不能等于 Run workload；该代码级职责分离不替代生产 IAM。 13. `candidate_validated`、`registry_subject_bound`、GitHub provenance action 成功、CI artifact、离线 preflight、未独立 attested 的 live observation JSON 或人工批准都不能单独授权 production；缺少同一 source revision 的 OCI subject 独立验证、registry/live revision/identity/health/golden-slice 绑定及受保护 provenance 时，promotion 必须失败。 -14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback HTTP、memory/file-backed JDBC catalog、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 +14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明 Spark 经无认证本地 Iceberg REST 在同节点共享 RWO PVC 上完成 read/write/schema evolution/snapshot/time travel，并由 Gravitino 回读 schema，不覆盖 cancel/reconcile/lineage 或 Flink。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback/cluster HTTP、memory/file-backed JDBC catalog、同节点共享 PVC、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -88,6 +88,7 @@ - Metadata Fabric M3-6 已在隔离 Gravitino `1.3.0` Basic IdP 中将 bounded user 限定为 `lakehouse` 的 `USE_CATALOG` 与 `lakehouse.published` 的 `USE_SCHEMA`/`CREATE_TABLE`；table create/read 为 200/200，catalog create 为 403，旧密码轮换后为 401，用户删除后替换密码为 401，临时 provider 对象、namespace 和 port-forward 全部清理。evidence SHA 为 `f0b0de1f80f079d43318937e0a0cc151a8546e9e307bef204738b1367f9b29fd`。`local_gravitino_minimum_privilege_verified=true` 只描述本地 Basic rehearsal；OIDC、TLS、持久 catalog 和 production identity 均仍为 `false`。 - Metadata Fabric M3-7 已建立 production identity profile/attestation gate；checked-in profile fingerprint 为 `2e9d5cac3560b853820f923669f6794ead63bcb36a528639fc0e9539e148ee2f`，report fingerprint 为 `c607589ee25a87acc8a1ab71372618a9a4c10c1e8ebff15b8db7e78b37600b9f`，`profile_valid=true`，40 项 federation/provider/TLS/catalog/tenancy/operations 外部输入以 blockers 暴露，`ready_for_protected_verification=false`、`production_identity_gate_passed=false`、`production_ready=false`。合成完整 attestation 只验证门禁逻辑，不计入生产证据。 - Metadata Fabric M3-8 已在隔离 namespace 将 Iceberg catalog metadata 落到 PostgreSQL JDBC、warehouse 落到独立 PVC；PostgreSQL/Gravitino Pod UID 在顺序 restart 后均变化，StatefulSet/PVC UID 保持，同一 bounded Basic user 重认证并读取相同 table fingerprint，catalog create 前后均为 403，临时 namespace/PV 清理完成。contract fingerprint 为 `f622d8a61bae49171bc76a16bfe64280c616c028bddf88479f1ad04acb1dadf0`，evidence fingerprint 为 `34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa`。这不证明 protected workload identity、OIDC/TLS、生产 failure domain、backup/PITR、Spark/Flink conformance、生产 ingestion 或 `production_ready`。 +- Metadata Fabric M3-9 已在隔离 namespace 让 bounded Gravitino Basic user 先创建 Iceberg JDBC table，再由 Spark `3.5.0` + Iceberg `1.6.1` 经 Gravitino Iceberg REST `1.11.0` 读取、两次 append、增加 nullable column、验证三行 current state、两个 snapshot 与 first-snapshot time travel；Gravitino API 随后回读演进 schema，catalog create 前后均为 403，Spark/Gravitino 共用 `desktop-worker` 上的 warehouse PVC，namespace/PV 完整清理。contract fingerprint 为 `a78b95d36a6a5d5f4b5e303be21263d00fdd7102c3a70ce282ba69f2d8cdcd2e`，evidence fingerprint 为 `50f9d0021db11e22364697d1ad8928ee068d28dc8046556bbca1a4e1c819f8e0`。这只证明本地同节点共享 RWO PVC 互操作；无认证 HTTP REST、Basic IdP、file warehouse 不证明 protected identity、OIDC/TLS、对象存储、cancel/reconcile/lineage、Flink、完整 `spark_conformance_verified`、生产 ingestion 或 `production_ready`。 ## 下一验收证据 @@ -96,5 +97,5 @@ - staging 的 migration role、应用 login membership、连接池 role/tenant 复位、双租户 API 和 success finalization 运行产物； - DolphinScheduler adapter 的真实 IAM/OIDC、service token provisioning/轮换、provider 最小权限、binding artifact staging 接入、managed outbox worker/provider callback 实际扩容部署、唯一 worker ID、status/lease 故障恢复和无双写证据； - 首条真实图斑链对 golden slice 的 output hash、独立质量结果/evidence、血缘、发布 revision 和 rollback 演练； -- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back 和 conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation 与 M3-8 本地 JDBC restart continuity 均不计入生产退出门； +- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity 与 M3-9 本地同节点 Spark interoperability 均不计入生产退出门； - DolphinScheduler/Temporal sandbox 的独立数据库、备份恢复、身份、版本和升级责任证明；DolphinScheduler standalone/H2 不计入此退出门。 diff --git a/k8s/metadata-fabric-spark-iceberg-rest-interoperability/kustomization.yaml b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/kustomization.yaml new file mode 100644 index 00000000..c120f673 --- /dev/null +++ b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/kustomization.yaml @@ -0,0 +1,12 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: gda-metadata-spark-interop +resources: + - namespace.yaml + - runtime.yaml + - spark-job.yaml +labels: + - pairs: + app.kubernetes.io/part-of: gda-metadata-fabric-spark-iceberg-rest-interoperability + gda.openai.com/environment: spark-iceberg-rest-interoperability-rehearsal + includeSelectors: false diff --git a/k8s/metadata-fabric-spark-iceberg-rest-interoperability/namespace.yaml b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/namespace.yaml new file mode 100644 index 00000000..626bc9f7 --- /dev/null +++ b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/namespace.yaml @@ -0,0 +1,25 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: gda-metadata-spark-interop + labels: + pod-security.kubernetes.io/enforce: baseline + pod-security.kubernetes.io/audit: restricted + pod-security.kubernetes.io/warn: restricted +--- +apiVersion: v1 +kind: ResourceQuota +metadata: + name: spark-iceberg-rest-interoperability + namespace: gda-metadata-spark-interop +spec: + hard: + requests.cpu: "6" + requests.memory: 8Gi + limits.cpu: "12" + limits.memory: 14Gi + pods: "8" + persistentvolumeclaims: "4" + requests.storage: 6Gi + secrets: "4" + configmaps: "8" diff --git a/k8s/metadata-fabric-spark-iceberg-rest-interoperability/runtime.yaml b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/runtime.yaml new file mode 100644 index 00000000..ad4a00a3 --- /dev/null +++ b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/runtime.yaml @@ -0,0 +1,429 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: gravitino-persistence-postgresql + namespace: gda-metadata-spark-interop +automountServiceAccountToken: false +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: gravitino-persistence + namespace: gda-metadata-spark-interop +automountServiceAccountToken: false +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: spark-iceberg-rest-probe + namespace: gda-metadata-spark-interop +automountServiceAccountToken: false +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: gravitino-persistence-iceberg-init + namespace: gda-metadata-spark-interop +data: + 002-iceberg.sql: | + CREATE DATABASE iceberg OWNER gravitino; +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: gravitino-persistence-config + namespace: gda-metadata-spark-interop +data: + gravitino.conf: | + gravitino.server.shutdown.timeout = 30000 + gravitino.server.webserver.host = 0.0.0.0 + gravitino.server.webserver.httpPort = 8090 + gravitino.server.health.entityStore.probeTimeoutMs = 5000 + gravitino.entity.store = relational + gravitino.entity.store.relational = JDBCBackend + gravitino.entity.store.relational.jdbcUrl = jdbc:postgresql://gravitino-persistence-postgresql:5432/gravitino + gravitino.entity.store.relational.jdbcDriver = org.postgresql.Driver + gravitino.entity.store.relational.jdbcUser = gravitino + gravitino.authenticators = basic + gravitino.server.rest.extensionPackages = org.apache.gravitino.idp.web.rest.feature + gravitino.authorization.enable = true + gravitino.authorization.serviceAdmins = gda-interop-admin + gravitino.cache.enabled = false + gravitino.auxService.names = +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: gravitino-iceberg-rest-config + namespace: gda-metadata-spark-interop +data: + gravitino-iceberg-rest-server.conf: | + gravitino.iceberg-rest.shutdown.timeout = 3000 + gravitino.iceberg-rest.host = 0.0.0.0 + gravitino.iceberg-rest.httpPort = 9001 + gravitino.iceberg-rest.minThreads = 4 + gravitino.iceberg-rest.maxThreads = 32 + gravitino.iceberg-rest.stopTimeout = 30000 + gravitino.iceberg-rest.idleTimeout = 30000 + gravitino.iceberg-rest.threadPoolWorkQueueSize = 100 + gravitino.iceberg-rest.requestHeaderSize = 131072 + gravitino.iceberg-rest.responseHeaderSize = 131072 + gravitino.iceberg-rest.catalog-backend = jdbc + gravitino.iceberg-rest.jdbc-driver = org.postgresql.Driver + gravitino.iceberg-rest.uri = jdbc:postgresql://gravitino-persistence-postgresql:5432/iceberg + gravitino.iceberg-rest.jdbc-user = gravitino + gravitino.iceberg-rest.jdbc-initialize = true + gravitino.iceberg-rest.warehouse = file:///var/lib/gravitino/warehouse +--- +apiVersion: v1 +kind: Service +metadata: + name: gravitino-persistence-postgresql + namespace: gda-metadata-spark-interop +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: gravitino-persistence-postgresql + ports: + - name: postgresql + port: 5432 + targetPort: postgresql +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: gravitino-persistence-postgresql + namespace: gda-metadata-spark-interop +spec: + serviceName: gravitino-persistence-postgresql + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: gravitino-persistence-postgresql + template: + metadata: + labels: + app.kubernetes.io/name: gravitino-persistence-postgresql + app.kubernetes.io/component: database + spec: + serviceAccountName: gravitino-persistence-postgresql + automountServiceAccountToken: false + nodeSelector: + kubernetes.io/hostname: desktop-worker + terminationGracePeriodSeconds: 30 + securityContext: + fsGroup: 999 + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + containers: + - name: postgresql + image: postgres:16.10-bookworm + imagePullPolicy: IfNotPresent + env: + - name: POSTGRES_USER + value: gravitino + - name: POSTGRES_DB + value: gravitino + - name: POSTGRES_PASSWORD + valueFrom: + secretKeyRef: + name: gravitino-persistence-runtime + key: database-password + - name: PGDATA + value: /var/lib/postgresql/data/pgdata + ports: + - name: postgresql + containerPort: 5432 + startupProbe: + exec: + command: ["pg_isready", "-U", "gravitino", "-d", "gravitino"] + periodSeconds: 3 + failureThreshold: 60 + readinessProbe: + exec: + command: ["pg_isready", "-U", "gravitino", "-d", "gravitino"] + periodSeconds: 5 + failureThreshold: 12 + resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "1" + memory: 1Gi + securityContext: + runAsNonRoot: true + runAsUser: 999 + runAsGroup: 999 + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: data + mountPath: /var/lib/postgresql/data + - name: runtime + mountPath: /var/run/postgresql + - name: tmp + mountPath: /tmp + - name: schema + mountPath: /docker-entrypoint-initdb.d/001-schema.sql + subPath: 001-schema.sql + readOnly: true + - name: iceberg-init + mountPath: /docker-entrypoint-initdb.d/002-iceberg.sql + subPath: 002-iceberg.sql + readOnly: true + volumes: + - name: runtime + emptyDir: {} + - name: tmp + emptyDir: {} + - name: schema + configMap: + name: gravitino-persistence-schema + - name: iceberg-init + configMap: + name: gravitino-persistence-iceberg-init + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: ["ReadWriteOnce"] + storageClassName: standard + resources: + requests: + storage: 2Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: gravitino-persistence + namespace: gda-metadata-spark-interop +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: gravitino-persistence + ports: + - name: http + port: 8090 + targetPort: http + - name: iceberg-rest + port: 9001 + targetPort: iceberg-rest +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: gravitino-persistence + namespace: gda-metadata-spark-interop +spec: + serviceName: gravitino-persistence + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: gravitino-persistence + template: + metadata: + labels: + app.kubernetes.io/name: gravitino-persistence + app.kubernetes.io/component: technical-metadata + spec: + serviceAccountName: gravitino-persistence + automountServiceAccountToken: false + nodeSelector: + kubernetes.io/hostname: desktop-worker + terminationGracePeriodSeconds: 60 + securityContext: + fsGroup: 1000 + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + initContainers: + - name: stage-postgresql-jdbc-driver + image: gda/gravitino:1.3.0-local-arm64 + imagePullPolicy: Never + command: ["/bin/sh", "-ec"] + args: + - install -m 0444 /opt/gravitino/libs/postgresql-42.7.0.jar /driver/postgresql-42.7.0.jar + resources: + requests: + cpu: 25m + memory: 64Mi + limits: + cpu: 100m + memory: 128Mi + securityContext: + runAsNonRoot: true + runAsUser: 1000 + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: jdbc-driver + mountPath: /driver + containers: + - name: gravitino + image: gda/gravitino:1.3.0-local-arm64 + imagePullPolicy: Never + command: ["/bin/bash", "-ec"] + args: + - | + install -d -m 0700 /var/run/gravitino/conf + install -m 0600 /etc/gravitino/gravitino.conf /var/run/gravitino/conf/gravitino.conf + install -m 0444 /opt/gravitino/conf/gravitino-env.sh /var/run/gravitino/conf/gravitino-env.sh + printf '\ngravitino.entity.store.relational.jdbcPassword = ' >> /var/run/gravitino/conf/gravitino.conf + cat /var/run/secrets/gravitino/database-password >> /var/run/gravitino/conf/gravitino.conf + printf '\n' >> /var/run/gravitino/conf/gravitino.conf + exec /opt/gravitino/bin/gravitino.sh --config /var/run/gravitino/conf run + env: + - name: GRAVITINO_MEM + value: -Xms384m -Xmx1024m -XX:MaxMetaspaceSize=384m + - name: GRAVITINO_INITIAL_ADMIN_PASSWORD + valueFrom: + secretKeyRef: + name: gravitino-persistence-runtime + key: admin-password + ports: + - name: http + containerPort: 8090 + startupProbe: + httpGet: + path: /api/health/ready + port: http + periodSeconds: 5 + failureThreshold: 120 + readinessProbe: + httpGet: + path: /api/health/ready + port: http + periodSeconds: 10 + failureThreshold: 12 + resources: + requests: + cpu: 250m + memory: 768Mi + limits: + cpu: "2" + memory: 2Gi + securityContext: + runAsNonRoot: true + runAsUser: 1000 + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: config + mountPath: /etc/gravitino + readOnly: true + - name: secrets + mountPath: /var/run/secrets/gravitino + readOnly: true + - name: runtime + mountPath: /var/run/gravitino + - name: logs + mountPath: /opt/gravitino/logs + - name: tmp + mountPath: /tmp + - name: warehouse + mountPath: /var/lib/gravitino/warehouse + - name: jdbc-driver + mountPath: /opt/gravitino/catalogs/lakehouse-iceberg/libs/postgresql-42.7.0.jar + subPath: postgresql-42.7.0.jar + readOnly: true + - name: iceberg-rest + image: gda/gravitino:1.3.0-local-arm64 + imagePullPolicy: Never + command: ["/bin/bash", "-ec"] + args: + - | + install -d -m 0700 /var/run/gravitino/iceberg-rest-conf + install -m 0600 /etc/gravitino-iceberg-rest/gravitino-iceberg-rest-server.conf /var/run/gravitino/iceberg-rest-conf/gravitino-iceberg-rest-server.conf + install -m 0444 /opt/gravitino/conf/gravitino-iceberg-rest-log4j2.properties /var/run/gravitino/iceberg-rest-conf/gravitino-iceberg-rest-log4j2.properties + install -m 0444 /opt/gravitino/conf/gravitino-env.sh /var/run/gravitino/iceberg-rest-conf/gravitino-env.sh + printf '\ngravitino.iceberg-rest.jdbc-password = ' >> /var/run/gravitino/iceberg-rest-conf/gravitino-iceberg-rest-server.conf + cat /var/run/secrets/gravitino/database-password >> /var/run/gravitino/iceberg-rest-conf/gravitino-iceberg-rest-server.conf + printf '\n' >> /var/run/gravitino/iceberg-rest-conf/gravitino-iceberg-rest-server.conf + exec /opt/gravitino/bin/gravitino-iceberg-rest-server.sh --config /var/run/gravitino/iceberg-rest-conf run + env: + - name: GRAVITINO_MEM + value: -Xms256m -Xmx768m -XX:MaxMetaspaceSize=256m + ports: + - name: iceberg-rest + containerPort: 9001 + startupProbe: + httpGet: + path: /iceberg/v1/config + port: iceberg-rest + periodSeconds: 5 + failureThreshold: 120 + readinessProbe: + httpGet: + path: /iceberg/v1/config + port: iceberg-rest + periodSeconds: 10 + failureThreshold: 12 + resources: + requests: + cpu: 200m + memory: 512Mi + limits: + cpu: "2" + memory: 1536Mi + securityContext: + runAsNonRoot: true + runAsUser: 1000 + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: iceberg-rest-config + mountPath: /etc/gravitino-iceberg-rest + readOnly: true + - name: secrets + mountPath: /var/run/secrets/gravitino + readOnly: true + - name: runtime + mountPath: /var/run/gravitino + - name: logs + mountPath: /opt/gravitino/logs + - name: tmp + mountPath: /tmp + - name: warehouse + mountPath: /var/lib/gravitino/warehouse + - name: jdbc-driver + mountPath: /opt/gravitino/iceberg-rest-server/libs/postgresql-42.7.0.jar + subPath: postgresql-42.7.0.jar + readOnly: true + volumes: + - name: config + configMap: + name: gravitino-persistence-config + - name: iceberg-rest-config + configMap: + name: gravitino-iceberg-rest-config + - name: secrets + secret: + secretName: gravitino-persistence-runtime + defaultMode: 0400 + - name: runtime + emptyDir: {} + - name: logs + emptyDir: {} + - name: tmp + emptyDir: {} + - name: jdbc-driver + emptyDir: {} + volumeClaimTemplates: + - metadata: + name: warehouse + spec: + accessModes: ["ReadWriteOnce"] + storageClassName: standard + resources: + requests: + storage: 1Gi diff --git a/k8s/metadata-fabric-spark-iceberg-rest-interoperability/spark-job.yaml b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/spark-job.yaml new file mode 100644 index 00000000..a88e6082 --- /dev/null +++ b/k8s/metadata-fabric-spark-iceberg-rest-interoperability/spark-job.yaml @@ -0,0 +1,179 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: spark-iceberg-rest-probe + namespace: gda-metadata-spark-interop +data: + probe.py: | + import json + + from pyspark.sql import SparkSession + + + TABLE = "rest.published.gda_spark_interop_probe" + EXPECTED_INITIAL_COLUMNS = ["probe_id"] + EXPECTED_CURRENT_COLUMNS = ["probe_id", "quality"] + EXPECTED_CURRENT_ROWS = [ + ["spark-a", None], + ["spark-b", None], + ["spark-c", "verified"], + ] + + spark = ( + SparkSession.builder.appName("gda-spark-iceberg-rest-interoperability") + .master("local[2]") + .config( + "spark.sql.extensions", + "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions", + ) + .config("spark.sql.catalog.rest", "org.apache.iceberg.spark.SparkCatalog") + .config("spark.sql.catalog.rest.type", "rest") + .config( + "spark.sql.catalog.rest.uri", + "http://gravitino-persistence:9001/iceberg", + ) + .config("spark.sql.catalog.rest.cache-enabled", "false") + .config("spark.sql.shuffle.partitions", "2") + .getOrCreate() + ) + spark.sparkContext.setLogLevel("WARN") + + initial = spark.table(TABLE) + initial_columns = initial.columns + initial_count = initial.count() + if initial_columns != EXPECTED_INITIAL_COLUMNS or initial_count != 0: + raise RuntimeError( + f"unexpected Gravitino-created table: {initial_columns=} {initial_count=}" + ) + + spark.createDataFrame([("spark-a",), ("spark-b",)], ["probe_id"]).writeTo( + TABLE + ).append() + first_snapshots = spark.sql( + f"SELECT snapshot_id, committed_at FROM {TABLE}.snapshots ORDER BY committed_at" + ).collect() + if len(first_snapshots) != 1: + raise RuntimeError(f"expected one snapshot after first append: {first_snapshots}") + first_snapshot_id = first_snapshots[0]["snapshot_id"] + + spark.sql(f"ALTER TABLE {TABLE} ADD COLUMN quality STRING") + spark.createDataFrame( + [("spark-c", "verified")], ["probe_id", "quality"] + ).writeTo(TABLE).append() + + current = spark.table(TABLE) + current_columns = current.columns + current_rows = [list(row) for row in current.orderBy("probe_id").collect()] + snapshots = spark.sql( + f"SELECT snapshot_id, parent_id, operation FROM {TABLE}.snapshots " + "ORDER BY committed_at" + ).collect() + historical_rows = [ + row["probe_id"] + for row in spark.sql( + f"SELECT probe_id FROM {TABLE} VERSION AS OF {first_snapshot_id} " + "ORDER BY probe_id" + ).collect() + ] + + if current_columns != EXPECTED_CURRENT_COLUMNS: + raise RuntimeError(f"unexpected evolved schema: {current_columns}") + if current_rows != EXPECTED_CURRENT_ROWS: + raise RuntimeError(f"unexpected current rows: {current_rows}") + if len(snapshots) != 2: + raise RuntimeError(f"expected two snapshots: {snapshots}") + if historical_rows != ["spark-a", "spark-b"]: + raise RuntimeError(f"unexpected time-travel rows: {historical_rows}") + + result = { + "schema": "gda.spark_iceberg_rest_probe_result.v1", + "spark_version": spark.version, + "iceberg_runtime": "1.6.1", + "catalog_uri": "http://gravitino-persistence:9001/iceberg", + "table": TABLE, + "initial_columns": initial_columns, + "initial_row_count": initial_count, + "current_columns": current_columns, + "current_rows": current_rows, + "snapshot_ids": [row["snapshot_id"] for row in snapshots], + "snapshot_operations": [row["operation"] for row in snapshots], + "time_travel_snapshot_id": first_snapshot_id, + "time_travel_rows": historical_rows, + "create_read_write_verified": True, + "schema_evolution_verified": True, + "snapshot_history_verified": True, + "time_travel_verified": True, + } + print("GDA_SPARK_RESULT=" + json.dumps(result, sort_keys=True)) + spark.stop() +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: spark-iceberg-rest-probe + namespace: gda-metadata-spark-interop +spec: + suspend: true + backoffLimit: 0 + activeDeadlineSeconds: 900 + template: + metadata: + labels: + app.kubernetes.io/name: spark-iceberg-rest-probe + app.kubernetes.io/component: interoperability-probe + spec: + serviceAccountName: spark-iceberg-rest-probe + automountServiceAccountToken: false + nodeSelector: + kubernetes.io/hostname: desktop-worker + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 100 + fsGroup: 100 + seccompProfile: + type: RuntimeDefault + containers: + - name: spark + image: gisdataagent/mmfe-spark-runtime:local + imagePullPolicy: Never + command: ["python", "/opt/gda/probe.py"] + env: + - name: JAVA_HOME + value: /usr/lib/jvm/java-17-openjdk-arm64 + - name: HOME + value: /tmp/spark-home + - name: SPARK_LOCAL_DIRS + value: /tmp/spark-local + - name: PYSPARK_PYTHON + value: python + resources: + requests: + cpu: "1" + memory: 2Gi + limits: + cpu: "4" + memory: 5Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: probe + mountPath: /opt/gda + readOnly: true + - name: warehouse + mountPath: /var/lib/gravitino/warehouse + - name: tmp + mountPath: /tmp + volumes: + - name: probe + configMap: + name: spark-iceberg-rest-probe + - name: warehouse + persistentVolumeClaim: + claimName: warehouse-gravitino-persistence-0 + - name: tmp + emptyDir: {} diff --git a/scripts/metadata-fabric-spark-iceberg-rest-interoperability.sh b/scripts/metadata-fabric-spark-iceberg-rest-interoperability.sh new file mode 100755 index 00000000..6a8af284 --- /dev/null +++ b/scripts/metadata-fabric-spark-iceberg-rest-interoperability.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +COMMON_GIT_DIR="$(git -C "$ROOT" rev-parse --path-format=absolute --git-common-dir 2>/dev/null || true)" +SHARED_ROOT="" +if [ -n "$COMMON_GIT_DIR" ]; then + SHARED_ROOT="$(cd "$COMMON_GIT_DIR/.." && pwd)" +fi + +if [ -n "${PYTHON:-}" ]; then + : +elif [ -x "$ROOT/.venv/bin/python" ]; then + PYTHON="$ROOT/.venv/bin/python" +elif [ -n "$SHARED_ROOT" ] && [ -x "$SHARED_ROOT/.venv/bin/python" ]; then + PYTHON="$SHARED_ROOT/.venv/bin/python" +else + PYTHON="python" +fi + +cd "$ROOT" +exec "$PYTHON" -m data_agent.metadata_fabric_spark_iceberg_rest_interoperability "$@"