feat(operator): initial operator implementation (no deploy)

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,21 +16,33 @@ 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]

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()

sentry_streams_k8s/sentry_streams_k8s/operator/streaming_pipeline.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from typing import Any, Mapping, NotRequired, TypedDict
+
+from sentry_streams_k8s.consumer_builder import ConsumerBuilder, ConsumerSpec
+
+
+class StreamingPipelineSpec(TypedDict):
+    service_name: str
+    pipeline_name: str
+    pipeline_module: str
+    image_name: str
+    cpu_per_process: int
+    memory_per_process: int
+    deployment_template: dict[str, Any]
+    container_template: dict[str, Any]
+    pipeline_config: dict[str, Any]
+    replicas: NotRequired[int]
+    with_canary: NotRequired[bool]
+    segment_id: NotRequired[int]
+    log_level: NotRequired[str]
+    enable_liveness_probe: NotRequired[bool]
+    container_name: NotRequired[str]
+    emergency_patch: NotRequired[dict[str, Any]]
+
+
+REQUIRED_FIELDS = (
+    "service_name",
+    "pipeline_name",
+    "pipeline_module",
+    "image_name",
+    "cpu_per_process",
+    "memory_per_process",
+    "deployment_template",
+    "container_template",
+    "pipeline_config",
+)
+
+
+def from_crd_spec(crd_spec: Mapping[str, Any], *, name: str | None = None) -> StreamingPipelineSpec:
+    spec: dict[str, Any] = dict(crd_spec)
+    if "pipeline_name" not in spec and name is not None:
+        spec["pipeline_name"] = name
+    return spec  # type: ignore[return-value]
+
+
+def validate(spec: StreamingPipelineSpec) -> None:
+    missing = [f for f in REQUIRED_FIELDS if f not in spec]
+    if missing:
+        raise ValueError(
+            f"StreamingPipeline is missing required field(s): {', '.join(sorted(missing))}."
+        )
+
+
+def to_consumer_spec(spec: StreamingPipelineSpec) -> ConsumerSpec:
+    return ConsumerSpec(
+        service_name=spec["service_name"],
+        pipeline_name=spec["pipeline_name"],
+        pipeline_module=spec["pipeline_module"],
+        image=spec["image_name"],
+        cpu_per_process=spec["cpu_per_process"],
+        memory_per_process=spec["memory_per_process"],
+        segment_id=spec.get("segment_id", 0),
+        replicas=spec.get("replicas", 1),
+        log_level=spec.get("log_level", "INFO"),
+        enable_liveness_probe=spec.get("enable_liveness_probe", True),
+        with_canary=spec.get("with_canary", False),
+        container_name=spec.get("container_name", "pipeline-consumer"),
+        emergency_patch=spec.get("emergency_patch", {}),
+    )
+
+
+def render(spec: StreamingPipelineSpec) -> dict[str, Any]:
+    builder = ConsumerBuilder(spec["deployment_template"], spec["container_template"])
+    consumer = to_consumer_spec(spec)
+    builder.validate(consumer, spec["pipeline_config"])
+    return builder.build(consumer, spec["pipeline_config"])