ref(k8s): build operator and push to gar

.github/workflows/build-operator.yaml

@@ -0,0 +1,39 @@
+name: Build & Push Streaming Operator Image
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - sentry_streams_k8s/**
+      - .github/workflows/build-operator.yaml
+
+jobs:
+  build-and-push:
+    name: Build and push image to ${{ matrix.registry }} Google Artifact Registry
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+      id-token: write
+    strategy:
+      matrix:
+        include:
+          - registry: single-region
+            image: us-central1-docker.pkg.dev/sentryio/streaming-operator/image
+          - registry: multi-region
+            image: us-docker.pkg.dev/sentryio/streaming-operator-mr/image
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - uses: getsentry/action-build-and-push-images@4852d671d747d7c0268b2a3fc429fee9d4a16f78
+        with:
+          image_name: 'streaming-operator'
+          platforms: linux/amd64
+          dockerfile_path: './sentry_streams_k8s/Dockerfile'
+          build_context: './sentry_streams_k8s'
+          ghcr: false
+          tag_nightly: false
+          google_ar: true
+          google_ar_image_name: ${{ matrix.image }}
+          google_workload_identity_provider: projects/868781662168/locations/global/workloadIdentityPools/prod-github/providers/github-oidc-pool
+          google_service_account: gha-gcr-push@sac-prod-sa.iam.gserviceaccount.com

sentry_streams_k8s/.dockerignore

@@ -0,0 +1,62 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Virtual environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Git
+.git/
+.gitignore
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+
+# Logs
+*.log
+
+# Temporary files
+*.tmp
+*.temp
+
+# Project-specific (not needed in the image)
+tests/
+uv.lock
+.mypy_cache/

sentry_streams_k8s/Dockerfile

@@ -0,0 +1,41 @@
+# syntax=docker/dockerfile:1
+
+FROM python:3.11-slim
+
+ENV PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# Create non-root user and group
+RUN set -ex; \
+    groupadd -r streaming-operator --gid 1000; \
+    useradd -r -m -g streaming-operator --uid 1000 streaming-operator
+
+# Install minimal OS deps
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install uv
+RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \
+    pip install --no-cache-dir uv
+
+# Install Python dependencies declared in pyproject.toml, including the
+# operator extra (kopf and kubernetes).
+COPY pyproject.toml pyproject.toml
+RUN uv pip install --system -r pyproject.toml --extra operator
+
+# Copy application source (owned by non-root user)
+COPY --chown=streaming-operator:streaming-operator README.md README.md
+COPY --chown=streaming-operator:streaming-operator sentry_streams_k8s/ sentry_streams_k8s/
+
+# Install the package
+RUN uv pip install --system --no-deps .
+
+# Ensure workspace ownership
+RUN chown -R streaming-operator:streaming-operator /app
+
+# Switch to non-root user
+USER streaming-operator
+
+ENTRYPOINT ["streams-k8s-operator"]

sentry_streams_k8s/mypy.ini

@@ -1,3 +1,6 @@
 [mypy]
 ignore_missing_imports = False
 python_version = 3.11
+
+[mypy-kubernetes.*]
+ignore_missing_imports = True

sentry_streams_k8s/pyproject.toml

@@ -16,25 +16,38 @@ classifiers = [
 ]
 dependencies = [
     "jsonschema>=4.23.0",
-    "requests>=2.32.3",
-    "sentry-infra-tools>=1.24.0",
     "sentry-streams>=0.0.40",
     "pyyaml>=6.0.2",
 ]
 
+[project.optional-dependencies]
+macro = [
+    "sentry-infra-tools>=1.24.0",
+]
+operator = [
+    "kopf>=1.37",
+    "kubernetes>=29",
+]
+
+[project.scripts]
+streams-k8s-operator = "sentry_streams_k8s.operator.operator:main"
+
 [dependency-groups]
 dev = [
+    "kopf>=1.37",
+    "kubernetes>=29",
     "mypy>=1.14.1",
     "pre-commit>=4.1.0",
     "pytest>=8.3.4",
     "responses>=0.25.6",
+    "sentry-infra-tools>=1.24.0",
     "types-jsonschema>=4.23.0",
     "types-pyyaml>=6.0.2",
-    "types-requests>=2.32.0",
 ]
 
 [tool.setuptools.packages.find]
 where = ["."]
+include = ["sentry_streams_k8s*"]
 
 [tool.setuptools.package-data]
 sentry_streams_k8s = ["py.typed", "templates/deployment.yaml", "templates/container.yaml"]

sentry_streams_k8s/sentry_streams_k8s/__init__.py

@@ -1,9 +1,6 @@
 from sentry_streams_k8s.merge import ScalarOverwriteError, TypeMismatchError
-from sentry_streams_k8s.pipeline_step import PipelineStep, PipelineStepContext
 
 __all__ = [
-    "PipelineStep",
-    "PipelineStepContext",
     "ScalarOverwriteError",
     "TypeMismatchError",
 ]

sentry_streams_k8s/sentry_streams_k8s/operator/__init__.py

@@ -0,0 +1,13 @@
+from sentry_streams_k8s.operator.streaming_pipeline import (
+    StreamingPipelineSpec,
+    from_crd_spec,
+    render,
+    validate,
+)
+
+__all__ = [
+    "StreamingPipelineSpec",
+    "from_crd_spec",
+    "render",
+    "validate",
+]

sentry_streams_k8s/sentry_streams_k8s/operator/operator.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import kopf
+from kubernetes import client, dynamic
+
+from sentry_streams_k8s.consumer_builder import compute_config_version, make_k8s_name
+from sentry_streams_k8s.operator.streaming_pipeline import (
+    from_crd_spec,
+    render,
+    validate,
+)
+
+logger = logging.getLogger(__name__)
+
+GROUP = "streams.sentry.io"
+VERSION = "v1alpha1"
+PLURAL = "streamingpipelines"
+FIELD_MANAGER = "streaming-operator"
+
+
+def _apply(dyn: dynamic.DynamicClient, manifest: dict[str, Any], namespace: str) -> None:
+    resource = dyn.resources.get(api_version=manifest["apiVersion"], kind=manifest["kind"])
+    dyn.server_side_apply(
+        resource,
+        body=manifest,
+        namespace=namespace,
+        field_manager=FIELD_MANAGER,
+        force_conflicts=True,
+    )
+
+
+def _prune_stale_deployments(
+    *,
+    namespace: str,
+    owner_uid: str,
+    service_name: str,
+    pipeline_name: str,
+    desired_names: set[str],
+) -> None:
+    apps = client.AppsV1Api()
+    selector = f"service={make_k8s_name(service_name)},pipeline={make_k8s_name(pipeline_name)}"
+    existing = apps.list_namespaced_deployment(namespace=namespace, label_selector=selector)
+    for item in existing.items:
+        owner_refs = item.metadata.owner_references or []
+        if not any(ref.uid == owner_uid for ref in owner_refs):
+            continue
+        if item.metadata.name not in desired_names:
+            logger.info("Pruning stale deployment %s/%s", namespace, item.metadata.name)
+            apps.delete_namespaced_deployment(name=item.metadata.name, namespace=namespace)
+
+
+def _condition(type_: str, status: bool, reason: str, message: str = "") -> dict[str, Any]:
+    return {
+        "type": type_,
+        "status": "True" if status else "False",
+        "reason": reason,
+        "message": message,
+    }
+
+
+@kopf.on.create(GROUP, VERSION, PLURAL)
+@kopf.on.update(GROUP, VERSION, PLURAL)
+@kopf.on.resume(GROUP, VERSION, PLURAL)
+def reconcile(
+    spec: kopf.Spec,
+    name: str,
+    namespace: str | None,
+    uid: str,
+    patch: kopf.Patch,
+    **_: Any,
+) -> None:
+    assert namespace is not None
+
+    consumer = from_crd_spec(dict(spec), name=name)
+    try:
+        validate(consumer)
+        result = render(consumer)
+    except Exception as e:
+        patch.status["conditions"] = [_condition("Rendered", False, type(e).__name__, str(e))]
+        raise kopf.PermanentError(f"StreamingPipeline {namespace}/{name} failed to render: {e}")
+
+    manifests = [result["configmap"], result["deployment"]]
+    if "canary_deployment" in result:
+        manifests.append(result["canary_deployment"])
+
+    dyn = dynamic.DynamicClient(client.ApiClient())
+    for manifest in manifests:
+        kopf.adopt(manifest)
+        _apply(dyn, manifest, namespace)
+
+    _prune_stale_deployments(
+        namespace=namespace,
+        owner_uid=uid,
+        service_name=consumer["service_name"],
+        pipeline_name=consumer["pipeline_name"],
+        desired_names={m["metadata"]["name"] for m in manifests if m["kind"] == "Deployment"},
+    )
+
+    replicas = consumer.get("replicas", 1)
+    canary = 1 if "canary_deployment" in result else 0
+    patch.status["conditions"] = [
+        _condition("Rendered", True, "Rendered"),
+        _condition("Applied", True, "Applied"),
+    ]
+    patch.status["config_version"] = compute_config_version(consumer["pipeline_config"])
+    patch.status["replicas"] = {"primary": replicas - canary, "canary": canary}
+
+
+def main() -> None:
+    kopf.run(standalone=True, clusterwide=True)
+
+
+if __name__ == "__main__":
+    main()