feat(operator): pod management by bmcquilkin-sentry · Pull Request #345 · getsentry/streams

bmcquilkin-sentry · 2026-07-20T21:21:46Z

Adds Pod management abilities to the operator. Instead of generating a Deployment, the operator will generate and manage Pods. Uses a ConfigMap per CR to track the Pod generations. Watches Pod events and uses a health classifier to determine how to respond to them and reconcile.

This is a different implementation than the original test we did and is based on Kopf's official suggestion for how to reconcile: https://docs.kopf.dev/en/latest/reconciliation/#level-based-triggering

For each StreamingPipeline CR, the operator starts a daemon in the same container that owns the reconciliation for that CR. Updates and events only set a flag that requests reconcile. The daemon runs a reconcile on one of these triggers or the 60s health check timer. Since we only set a flag, multiple events on the same CR will only trigger one reconcile (until it is processed). We also avoid the risk of multiple reconciles on the same CR.

cursor · 2026-07-20T21:24:42Z

+            )
+            core.delete_namespaced_config_map(
+                name=configmap.metadata.name, namespace=workload_namespace
+            )


Orphaned Deployments after upgrade

High Severity

The operator no longer deletes previously created Deployment workloads, and deployment RBAC was removed. On upgrade, old Deployments keep running their Pods while the new path also creates managed Pods for the same pipeline, so consumers can run twice and double-process traffic.

Additional Locations (1)

sentry_streams_k8s/chart/streaming-operator/templates/rbac.yaml#L49-L55

^{Reviewed by Cursor Bugbot for commit 50b3b68. Configure here.}

This is real if we test the operator with Deployments on s4s2 before merging

cursor · 2026-07-20T21:24:42Z

+        namespace=namespace,
+        field_manager=FIELD_MANAGER,
+        force_conflicts=True,
+    )


Missing workload ownership checks

Medium Severity

apply_pod and save_generations apply with force_conflicts and never verify streams.sentry.io/owner-uid. Unlike _apply for ConfigMaps, a second StreamingPipeline that renders the same base name can overwrite another pipeline’s Pods or generation ledger instead of failing permanently.

Additional Locations (2)

sentry_streams_k8s/sentry_streams_k8s/operator/generations.py#L86-L126

sentry_streams_k8s/sentry_streams_k8s/operator/reconcile.py#L71-L102

^{Reviewed by Cursor Bugbot for commit 50b3b68. Configure here.}

linear-code · 2026-07-20T22:14:39Z

STREAM-1609

cursor

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 4 total unresolved issues (including 3 from previous reviews).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit d2ee769. Configure here.}

cursor · 2026-07-21T17:22:14Z

+# Waiting states caused by an invalid workload specification.
+# Should raise a PermanentError and not trigger replacement:
+
+PERMANENT_WAITING_REASONS = frozenset({"InvalidImageName"})


Unhandled waiting reasons stay silent

Medium Severity

pod_health only classifies ErrImagePull, ImagePullBackOff, and InvalidImageName waiting states. Other common failures such as CreateContainerConfigError produce no reason, so Pods are neither replaced, marked permanent, nor listed under unhealthyPods, leaving pipelines stuck with readyReplicas at 0 and little signal why.

Additional Locations (1)

sentry_streams_k8s/sentry_streams_k8s/operator/pod_health.py#L148-L181

^{Reviewed by Cursor Bugbot for commit db11345. Configure here.}

cursor

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 4 total unresolved issues (including 3 from previous reviews).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit ca2e22e. Configure here.}

cursor · 2026-07-21T23:35:38Z

+                desired_configmaps=set(),
+                logger=logger,
+            )
+    scheduler.discard(uid)


Old Deployments left orphaned

High Severity

Deployment pruning was removed while switching to direct Pod management, and neither reconcile nor cleanup deletes owner-labeled Deployments. Existing Deployment-backed workloads can keep running beside newly created Pods after upgrade or deletion, risking duplicate consumer processing.

^{Reviewed by Cursor Bugbot for commit ca2e22e. Configure here.}

fpacifici

Please see my comments inline for the details. Overall:

I don't think you need a configMap for the generaiton state. If you can do that with the status of a CR, please avoid the configMap complexity.
Some high level design constraints: Untyped dictionaries, breakdown into modules, public/private interfaces. Comments are inline
Consider breaking this down to make the task of reviewing tractable. A suggestion: One PR for the pod spec manipulation logic, one PR for the reconcile code, one PR that wire things up together in the operator. Failing to do so will make the review process really long as it requires the reviewer to allocate a lot of time to review each iteration.

fpacifici · 2026-07-22T21:29:47Z

+never shares a name with the old one that is still terminating. This is what
+lets us recreate instantly instead of waiting for the old Pod to be deleted.
+We save the highest generation per ordinal in a ConfigMap (one per CR).


I am not sure you need a configMap to store the most recent generation.
It seems this could fit in the status of your CRD.
See details here https://kubernetes.io/docs/concepts/overview/working-with-objects/#object-spec-and-status

The spec field is what the manifest provides: what you want to achieve, the status field is managed by the operator/controller and represent data bout the current state of the resource.
Example:
this is the status field of a statefulset

status: availableReplicas: 39 collisionCount: 0 currentReplicas: 39 currentRevision: task-ingest-errors-broker-947979787 observedGeneration: 150 readyReplicas: 39 replicas: 39 updateRevision: task-ingest-errors-broker-947979787 updatedReplicas: 39

You may see it contains a current revision.

I am pretty sure kopf allows you to set the current status of an object during the callbacks:
https://docs.kopf.dev/en/latest/walkthrough/updates/#updating-the-objects

So you should not have to manage this state on your own via a configmap

fpacifici · 2026-07-22T21:35:35Z

+    return dyn.resources.get(api_version="v1", kind="Pod")
+
+
+def _to_dict(obj: Any) -> dict[str, Any]:


Why do you need to process the resources as dictionaries ?

fpacifici · 2026-07-22T21:36:36Z

+
+
+def list_owned_pods(
+    dyn: DynamicClient,


Why do you need to use a Dynamic client to manage standard resources like pods ?

fpacifici · 2026-07-22T21:37:47Z

+    dyn: DynamicClient,
+    namespace: str,
+    owner_uid: str,
+    workload_set: str | None = None,


What is a workload_set ?

fpacifici · 2026-07-22T21:44:29Z

+def apply_pod(dyn: DynamicClient, pod: Mapping[str, Any], namespace: str) -> None:
+    _pod_resource(dyn).server_side_apply(
+        body=pod,
+        name=pod_metadata(pod)["name"],


You have a pod_name method but you are not using here. Any reason?

fpacifici · 2026-07-22T23:32:32Z

+def pod_keep_key(pod: Mapping[str, Any], health: Mapping[str, Any]) -> tuple[bool, int, str]:
+    return (cast(bool, health["ready"]), pod_generation(pod), pod_name(pod))
+
+
+def pod_template_from_deployment(
+    deployment: Mapping[str, Any],
+) -> tuple[Mapping[str, Any], Mapping[str, Any]]:
+    template = cast(Mapping[str, Any], deployment["spec"]["template"])
+    return cast(Mapping[str, Any], template.get("metadata", {}) or {}), cast(
+        Mapping[str, Any], template["spec"]
+    )


A general note about this procedural design. If you feel this is the best option I would not push back as this is your coding style and it is a legit design. Though, since your system is doing a lot of manipulation of complex nested data structures, have you considered making this more object oriented by creating classes for the main concepts you are manipulating (Spec, Pod, ConfigMap, etc) and add methods to extract or transform them, plus modules with the application logic to reconcile and call the api ?
This would make a design emerge as it would:

force you to package the methods with the data types they manipulate

separate the data type manipulation from the execution of the application logic (calling api)

IF you like sticking with this style better I think you will have to do more work to properly modularize the system:

You will need properly structured types rather than relying on untyped dictionaries

You need to make the break down into modules more deliberate: examples could be having a pod module, a configmap module, maybe container or status. Each one of these contains the functions to manipulate these data structures. Then you'll need modules to perform the orchestration work: apply/delete/patch etc.

fpacifici · 2026-07-22T23:42:54Z

This test is only testing the pod_health function.
Please add one single test to test the various accessor functions that return data from the pod or make them private (pod_metadata, pod_labels, etc)

fpacifici · 2026-07-22T23:45:53Z

I do not see unit tests for this module. Am I missing something ?

fpacifici · 2026-07-22T23:47:04Z

+    monkeypatch.setattr(
+        "sentry_streams_k8s.operator.reconcile.list_owned_pods",
+        list_pods,
+    )
+    monkeypatch.setattr(
+        "sentry_streams_k8s.operator.reconcile.apply_pod",
+        lambda _dyn, pod, _namespace: applied.append(dict(pod)),
+    )
+    monkeypatch.setattr(
+        "sentry_streams_k8s.operator.reconcile.delete_pod",
+        lambda _dyn, name, _namespace, force=False: deleted.append((name, force)),
+    )


Please do not monkeypatch methods you built.
This test seems like an integration test. So you should avoid mocking or monkeypatching what is not strictly needed. Please mock just the k8s API object.

Also I think this is monkeypatching the wrong path as delete_pod, apply_pod and list_owned_pods are not in the reconcile module. Unless you need to monkeypatch them there because that's where they are used.

fpacifici · 2026-07-22T23:50:24Z

+    return applied, deleted, result, ledger
+
+
+def test_build_pipeline_pod_stamps_identity_without_mutating_template() -> None:


What does this mean ? (I guess this is AI generated, am I right ? )

feat(operator): pod management

50b3b68

bmcquilkin-sentry requested a review from a team as a code owner July 20, 2026 21:21

cursor Bot reviewed Jul 20, 2026

View reviewed changes

sentry-warden Bot reviewed Jul 20, 2026

View reviewed changes

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/generations.py

cursor Bot reviewed Jul 20, 2026

View reviewed changes

bmcquilkin-sentry force-pushed the bmcquilkin/operator/pods branch from d2ee769 to 50b3b68 Compare July 20, 2026 23:06

getsentry deleted a comment from sentry Bot Jul 20, 2026

getsentry deleted a comment from cursor Bot Jul 20, 2026

sentry-warden Bot reviewed Jul 20, 2026

View reviewed changes

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/reconcile.py

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/operator.py Outdated

feat(operator): level-based reconcile w/ daemon

399b877

cursor Bot reviewed Jul 20, 2026

View reviewed changes

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/operator.py

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/pod_resources.py

sentry-warden Bot reviewed Jul 21, 2026

View reviewed changes

Comment thread sentry_streams_k8s/sentry_streams_k8s/operator/operator.py

bmcquilkin-sentry added 2 commits July 21, 2026 10:05

fix(operator): pod name length

771ab6f

fix(operator): stale removed set bug

db11345

cursor Bot reviewed Jul 21, 2026

View reviewed changes

ref(k8s): remove rbac from here

ca2e22e

cursor Bot reviewed Jul 21, 2026

View reviewed changes

fpacifici requested changes Jul 22, 2026

View reviewed changes

		return dyn.resources.get(api_version="v1", kind="Pod")


		def _to_dict(obj: Any) -> dict[str, Any]:

		return applied, deleted, result, ledger


		def test_build_pipeline_pod_stamps_identity_without_mutating_template() -> None:

Uh oh!

Uh oh!

Conversation

bmcquilkin-sentry commented Jul 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cursor Bot Jul 20, 2026

Choose a reason for hiding this comment

Orphaned Deployments after upgrade

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cursor Bot Jul 20, 2026

Choose a reason for hiding this comment

Missing workload ownership checks

Uh oh!

Uh oh!

Uh oh!

linear-code Bot commented Jul 20, 2026

Uh oh!

cursor Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cursor Bot Jul 21, 2026

Choose a reason for hiding this comment

Unhandled waiting reasons stay silent

Uh oh!

cursor Bot left a comment

Choose a reason for hiding this comment

Uh oh!

cursor Bot Jul 21, 2026

Choose a reason for hiding this comment

Old Deployments left orphaned

Uh oh!

fpacifici left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

bmcquilkin-sentry commented Jul 20, 2026 •

edited

Loading