Parse the pipeline deployment descriptor and generate k8s resources manifests

.pre-commit-config.yaml

@@ -40,15 +40,27 @@ repos:
           types-requests,
           responses,
           "sentry-arroyo>=2.18.2",
+          types-pyYAML,
         ]
         files: ^sentry_streams/.+
+    -   id: mypy
+        args: [--config-file, sentry_streams_operator/mypy.ini, --strict]
+        additional_dependencies: [
+          pytest==7.1.2,
+          types-pyYAML,
+        ]
+        files: ^sentry_streams_operator/.+
   - repo: https://github.com/pycqa/isort
     rev: 6.0.0
     hooks:
       - id: isort
         name: isort (python)
         files: ^sentry_streams/.+
         args: [--settings-path=./sentry_streams/pyproject.toml]
+      - id: isort
+        name: isort (python)
+        files: ^sentry_streams_operator/.+
+        args: [--settings-path=./sentry_streams_operator/pyproject.toml]
       - id: isort
         name: isort (python)
         files: ^sentry_flink/.+

Makefile

@@ -3,10 +3,12 @@ install-dev:
 	which uv || (curl -LsSf https://astral.sh/uv/install.sh | sh)
 	uv sync --project ./sentry_streams
 	PROJECT_ROOT=`pwd`/sentry_flink uv sync --project ./sentry_flink
+	PROJECT_ROOT=`pwd`/sentry_streams_operator uv sync --project ./sentry_streams_operator
 .PHONY: install-dev
 
 install-pre-commit-hook:
 	./sentry_streams/.venv/bin/pre-commit install --install-hooks
+	./sentry_streams_operator/.venv/bin/pre-commit install --install-hooks
 .PHONY: install-pre-commit-hook
 
 tests-streams:
@@ -17,6 +19,10 @@ tests-flink:
 	./sentry_flink/.venv/bin/pytest -vv sentry_flink/tests
 .PHONY: tests-flink
 
+tests-operator:
+	./sentry_streams_operator/.venv/bin/pytest -vv sentry_streams_operator/tests
+.PHONY: tests-operator
+
 typecheck:
 	./sentry_streams/.venv/bin/mypy --config-file sentry_streams/mypy.ini --strict sentry_streams/sentry_streams/
 	./sentry_flink/.venv/bin/mypy --config-file sentry_flink/mypy.ini --strict sentry_flink/sentry_flink/

sentry_streams/sentry_streams/examples/sample_configs/deployment_config.yaml

@@ -0,0 +1,18 @@
+env:
+  topics:
+    events: events
+    transformed-events: transformed-events
+    transformed-events-2: transformed-events-2
+
+pipeline:
+  name: "example-pipeline"
+  segments:
+    - parallelism: 2
+      steps_config:
+        myinput:
+          starts_segment: true
+          bootstrap_servers: 'kafka:9093'
+    - parallelism: 3
+      steps_config:
+        kafkasink:
+          bootstrap_servers: 'kafka:9093'

sentry_streams_operator/.envrc

@@ -0,0 +1 @@
+source .venv/bin/activate

sentry_streams_operator/Makefile

@@ -0,0 +1,38 @@
+
+.PHONY: buld-operator-image
+build-operator-image:
+	cp pyproject.toml docker/src/
+	cp -r sentry_streams docker/src/
+	find docker/src/ -type f -name "*.py[co]" -delete
+	find docker/src/ -type d -name "__pycache__" -delete
+	cd docker && docker buildx build -f Dockerfile.operator -t streams-k8s-operator:latest .
+	kind load docker-image streams-k8s-operator:latest
+
+.PHONY: deploy
+deploy:
+	kubectl apply -f k8s/operator.deploy.yaml
+	kubectl -n streams-operator rollout restart deploy streams-operator
+	kubectl -n streams-operator get pod
+
+
+.PHONY: logs
+logs:
+	kubectl -n streams-operator logs deployment/streams-operator
+
+.PHONY: k8s-delete
+k8s-delete:
+	kubectl -n streams-operator delete cm transactions-pipeline
+	kubectl -n streams-operator delete cm events-pipeline
+	kubectl -n streams-operator delete deploy events-pipeline-1
+	kubectl -n streams-operator delete deploy events-pipeline-0
+	kubectl -n streams-operator delete deploy transactions-pipeline-0
+
+.PHONY: k8s-list
+k8s-list:
+	kubectl -n streams-operator get cm
+	kubectl -n streams-operator get deploy
+	kubectl -n streams-operator get pod
+
+.PHONY: single-run
+single-run:
+	SENTRY_STREAMS_OPERATOR_INCLUSTER=false python -m sentry_streams_operator.runenv

sentry_streams_operator/docker/Dockerfile.operator

@@ -0,0 +1,11 @@
+FROM python:3.13
+
+RUN pip install uv
+
+ADD src /src
+
+RUN uv venv
+RUN uv pip install -e /src/
+RUN uv pip install kopf kubernetes
+
+# CMD kopf run /src/handlers.py --verbose

sentry_streams_operator/k8s/kind.cluster.yaml

@@ -0,0 +1,6 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+- role: control-plane
+- role: worker
+- role: worker

sentry_streams_operator/k8s/operator.deploy.yaml

@@ -0,0 +1,132 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: streams-operator-pipelines
+  namespace: streams-operator
+  labels:
+    service: streams-operator
+data:
+  events.yaml: |
+    env:
+      topics:
+        events: events
+        transformed-events: transformed-events
+        transformed-events-2: transformed-events-2
+
+    pipeline:
+      name: "events-pipeline"
+      segments:
+        - parallelism: 2
+          steps_config:
+            myinput:
+              starts_segment: true
+              bootstrap_servers: 'kafka:9093'
+        - parallelism: 3
+          steps_config:
+            kafkasink:
+              bootstrap_servers: 'kafka:9093'
+  transactions.yaml: |
+    env:
+      topics:
+        events: events
+        transformed-events: transformed-events
+        transformed-events-2: transformed-events-2
+
+    pipeline:
+      name: "transactions-pipeline"
+      segments:
+        - parallelism: 2
+          steps_config:
+            myinput:
+              starts_segment: true
+              bootstrap_servers: 'kafka:9093'
+  crons.yaml: |
+    env:
+      topics:
+        events: events
+        transformed-events: transformed-events
+        transformed-events-2: transformed-events-2
+
+    pipeline:
+      name: "crons-pipeline"
+      segments:
+        - parallelism: 2
+          steps_config:
+            myinput:
+              starts_segment: true
+              bootstrap_servers: 'kafka:9093'
+  issues.yaml: |
+    env:
+      topics:
+        events: events
+        transformed-events: transformed-events
+        transformed-events-2: transformed-events-2
+
+    pipeline:
+      name: "issues-pipeline"
+      segments:
+        - parallelism: 2
+          steps_config:
+            myinput:
+              starts_segment: true
+              bootstrap_servers: 'kafka:9093'
+  replay.yaml: |
+    env:
+      topics:
+        events: events
+        transformed-events: transformed-events
+        transformed-events-2: transformed-events-2
+
+    pipeline:
+      name: "replay-pipeline"
+      segments:
+        - parallelism: 2
+          steps_config:
+            myinput:
+              starts_segment: true
+              bootstrap_servers: 'kafka:9093'
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: streams-operator
+  namespace: streams-operator
+  labels:
+    service: streams-operator
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      service: streams-operator
+  template:
+    metadata:
+      labels:
+        service: streams-operator
+    spec:
+      serviceAccountName: streams-operator-serviceaccount
+      containers:
+        - name: streams-operator
+          image: "streams-k8s-operator:latest"
+          imagePullPolicy: Never
+          command:
+            - "/.venv/bin/kopf"
+            - "run"
+            - "/src/sentry_streams/k8s/operator/handlers.py"
+            - "--verbose"
+          volumeMounts:
+            - name: pipelines
+              mountPath: "/etc/streams/pipelines/"
+          env:
+            - name: STREAMS_PIPELINES_CONFIGMAP
+              value: streams-operator-pipelines
+            - name: STREAMS_PIPELINE_CONFIGS_DIR
+              value: "/etc/streams/pipelines/"
+            - name: STREAMS_DEPLOYMENT_TEMPLATES_DIR
+              value: "/etc/streams/deployment_templates/"
+            - name: STREAMS_CONFIGMAP_TEMPLATES_DIR
+              value: "/etc/streams/configmap_templates/"
+      volumes:
+        - name: pipelines
+          configMap:
+            name: streams-operator-pipelines

sentry_streams_operator/mypy.ini

@@ -0,0 +1,3 @@
+[mypy]
+ignore_missing_imports = False
+python_version = 3.11

sentry_streams_operator/pyproject.toml

@@ -0,0 +1,41 @@
+[build-system]
+requires = ["setuptools>=70", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "sentry-streams-operator"
+version = "0.0.1"
+description = "The python Sentry Streaming k8s Operator"
+readme = "README.md"
+
+requires-python = ">=3.11"
+dependencies = [
+    "kopf",
+    "kubernetes",
+    "sentry-streams",
+]
+
+[dependency-groups]
+dev = [
+    "mypy>=1.14.1",
+    "pre-commit>=4.1.0",
+    "pytest>=8.3.4",
+]
+
+[tool.setuptools]
+packages = ["sentry_streams_operator"]
+
+[tool.setuptools.package-data]
+sentry_streams = ["py.typed"]
+
+[tool.pytest.ini_options]
+minversion = "0.0.1"
+addopts = "-ra -q"
+testpaths = ["tests"]
+
+[tool.black]
+# File filtering is taken care of in pre-commit.
+line-length = 100
+
+[tool.isort]
+profile = "black"