operator: Connect Pipeline CRD + controller (revives #1337 with enterprise-branch enhancements)

[david-yu]

Note: This PR revives #1337 (closed when the Pipeline feature moved to the enterprise repo) and adds the enhancements the feature accumulated during its enterprise development. The enterprise repo won't be ready to ship this for a while, so the public operator carries it in the meantime. The original #1337 description follows, updated where behavior changed; the enhancements are summarized at the end.

Summary

Adds a Pipeline CRD (cluster.redpanda.com/v1alpha2) that manages Redpanda Connect pipelines as first-class Kubernetes resources. The spec is statically typed against the same Kubernetes-native primitives the rest of the v1alpha2 CRDs already use:

• cluster (*ClusterSource) — same primitive Topic/User use. Either point at a Redpanda CR (clusterRef) or supply brokers/TLS/SASL inline (staticConfiguration). When set, the operator inline-merges seed_brokers, tls, and sasl into any input.redpanda / output.redpanda blocks the user wrote in configYaml, and additionally renders the resolved connection as the top-level redpanda: shared client so redpanda_common (and any shared-client plugin) works without inline credentials. User-set keys win on conflict.

• userRef — optional alongside cluster.clusterRef. Binds the pipeline to a User CR; the operator reads the User's Secret-backed password + SCRAM mechanism and uses User.metadata.name as the SASL username. The User CR stays user-managed so ACL scoping is auditable (operator does not auto-create or modify it). Omit for unauthenticated clusters.

  Why userRef is flat while cluster.clusterRef is wrapped. cluster is a ClusterSource — a discriminated union of two sources: point at a Redpanda CR (clusterRef: { name }) OR supply brokers/TLS/SASL inline (staticConfiguration: {...}). The wrapping exists to express that union, and it matches the existing Topic/User CRDs (spec.cluster: *ClusterSource). userRef has only one source today — point at a User CR by name — so it stays flat: userRef: { name }. The inline-SASL counterpart already lives at cluster.staticConfiguration.kafka.sasl.{mechanism,username,password} (also flat — also a single source), and CEL forbids combining it with userRef. If a future use case needs a second user-identity source (e.g., inline SASL without a User CR backing it), we can promote userRef into a user: UserSource wrapper at that point.

• serviceAccountName — the ServiceAccount bound to the pipeline pod. When unset, the namespace's default SA is used. Set this to scope cloud-IAM trust (IRSA on EKS, Workload Identity on GKE, Pod Identity on AKS) per-pipeline rather than sharing the namespace's default SA across every pipeline. The operator does not create the SA — provision it (with the cloud-IAM annotations) out-of-band.

• valueSources — typed list of named env-var projections (one named pull per entry) backed by inline / configMapKeyRef / secretKeyRef / externalSecretRef. Replaces the earlier secretRef[] env-splat and env[] raw corev1.EnvVar approaches.

• image — per-pipeline Connect runtime override. Three-tier precedence: Pipeline.spec.image wins, then the chart-level connectController.image.{repository,tag} default (plumbed in via the operator's --connect-default-image flag), then the binary-baked PipelineDefaultImage constant.

• configYaml — the user's Connect pipeline YAML. Stays the inline catch-all for anything the typed fields don't cover.

CEL on PipelineSpec enforces the contract: userRef is forbidden alongside cluster.staticConfiguration (the static path carries its own inline SASL), and userRef is forbidden without cluster.clusterRef (no cluster context to authenticate against otherwise). userRef is otherwise an opt-in for SASL-enabled clusters.

Worked examples

A. Cluster-bound — Pipeline points at a Redpanda CR on the same Kubernetes cluster

The user provisions a SCRAM User CR with ACLs scoped to what the pipeline reads/writes; the Pipeline then references both the cluster and the user.

User CR (separate manifest, owns the SCRAM identity + ACLs):

apiVersion: cluster.redpanda.com/v1alpha2
kind: User
metadata:
  name: orders-to-warehouse
  namespace: redpanda
spec:
  cluster:
    clusterRef:
      name: redpanda
  authentication:
    type: scram-sha-512
    password:
      valueFrom:
        secretKeyRef:
          name: orders-to-warehouse-password
          key: password
  authorization:
    acls:
      - type: allow
        resource: { type: topic, name: orders, patternType: literal }
        operations: [Read, Describe]
      - type: allow
        resource: { type: group, name: orders-to-warehouse-ingest, patternType: literal }
        operations: [Read, Describe]

ACL scoping tip (from e2e testing): the literal grants above fit this single, static pipeline. Pipelines whose config creates topics (e.g. CDC producing mysql.*) or consumes on multiple/dynamic groups (e.g. a redpanda_common fanout where each pipeline runs its own group) need patternType: prefixed grants instead — topics mysql. with [Create, Write, Read, Describe], groups iceberg-orders- with [Read, Describe]. With a non-superuser pipeline user, a missing group ACL only surfaces as a consumer authorization error at runtime, so scope these up front.

Pipeline CR:

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: orders-to-warehouse
  namespace: redpanda
spec:
  cluster:
    clusterRef:
      name: redpanda             # operator resolves brokers + TLS
  userRef:
    name: orders-to-warehouse    # operator reads password Secret + mechanism

  valueSources:
    - name: S3_SECRET_KEY
      source:
        secretKeyRef: { name: s3-creds, key: secret_access_key }

  configYaml: |
    input:
      redpanda:
        # Only the per-plugin fields — seed_brokers, tls, sasl are
        # filled in by the operator from clusterRef + userRef.
        topics: [orders]
        consumer_group: orders-to-warehouse-ingest
    pipeline:
      processors:
        - mapping: |
            root = this
            root.ingested_at = now()
    output:
      aws_s3:
        bucket: warehouse-orders
        region: us-east-2
        credentials:
          secret: ${S3_SECRET_KEY}

What the operator renders into the pod's /config/connect.yaml:

input:
  redpanda:
    seed_brokers:                                      # injected
      - redpanda-0.redpanda.redpanda.svc.cluster.local.:9093
      - redpanda-1.redpanda.redpanda.svc.cluster.local.:9093
      - redpanda-2.redpanda.redpanda.svc.cluster.local.:9093
    tls:                                               # injected
      enabled: true
      root_cas_file: /etc/tls/certs/ca/ca.crt
    sasl:                                              # injected
      - mechanism: SCRAM-SHA-512
        username: ${REDPANDA_SASL_USERNAME}
        password: ${REDPANDA_SASL_PASSWORD}
    topics: [orders]                                   # user-supplied
    consumer_group: orders-to-warehouse-ingest         # user-supplied
pipeline:
  processors:
    - mapping: |
        root = this
        root.ingested_at = now()
output:
  aws_s3:
    bucket: warehouse-orders
    region: us-east-2
    credentials:
      secret: ${S3_SECRET_KEY}

Pod env (auto-derived):

• REDPANDA_SASL_USERNAME = "orders-to-warehouse" (literal, from User.metadata.name)
• REDPANDA_SASL_MECHANISM = "SCRAM-SHA-512" (literal, from User.spec.authentication.type)
• REDPANDA_SASL_PASSWORD ← secretKeyRef: { orders-to-warehouse-password, password } (from User.spec.authentication.password.valueFrom.secretKeyRef)
• S3_SECRET_KEY ← secretKeyRef: { s3-creds, secret_access_key } (from valueSources)

User keys win on conflict. If the user had written seed_brokers: [external.example.com:9093] inside input.redpanda, the operator would have left that value untouched and only injected the missing tls and sasl keys. That's the escape hatch for cluster-bound pipelines that need to point a specific input/output at a different cluster.

B. External Kafka / BYOC — static configuration

For pipelines reaching an external Redpanda or Kafka the operator doesn't run (Redpanda Cloud BYOC, cross-region tap, Confluent Cloud, MSK, etc.). No userRef; SASL credentials live inline on the static config and the password is itself a ValueSource.

Pipeline CR:

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: cross-region-mirror
  namespace: redpanda
spec:
  cluster:
    staticConfiguration:
      kafka:
        brokers:
          - kafka.us-east.example.com:9094
        tls:
          enabled: true
          caCertSecretRef:
            name: external-kafka-ca
            key: ca.crt
        sasl:
          mechanism: SCRAM-SHA-512
          username: pipeline-mirror-svc
          password:
            secretKeyRef:
              name: external-kafka-creds
              key: password
  # No userRef — staticConfiguration carries its own SASL identity.
  configYaml: |
    input:
      redpanda:
        topics: [orders]
    output:
      redpanda:
        topic: orders.mirrored

What the operator renders:

input:
  redpanda:
    seed_brokers: [kafka.us-east.example.com:9094]     # injected
    tls:                                               # injected
      enabled: true
      root_cas_file: /etc/tls/certs/ca/ca.crt
    sasl:                                              # injected
      - mechanism: SCRAM-SHA-512
        username: ${REDPANDA_SASL_USERNAME}
        password: ${REDPANDA_SASL_PASSWORD}
    topics: [orders]                                   # user-supplied
output:
  redpanda:
    seed_brokers: [kafka.us-east.example.com:9094]     # injected
    tls: { enabled: true, root_cas_file: /etc/tls/certs/ca/ca.crt }
    sasl:                                              # injected
      - mechanism: SCRAM-SHA-512
        username: ${REDPANDA_SASL_USERNAME}
        password: ${REDPANDA_SASL_PASSWORD}
    topic: orders.mirrored                             # user-supplied

staticConfiguration and clusterRef produce the same inline-merge contract — only the source-of-truth for the connection fields differs. No User CR involved.

C. Per-pipeline IRSA — native RDS IAM database authentication via serviceAccountName

Pipeline binds to a Redpanda cluster for its output and to RDS for its CDC input. The pipeline pod itself calls AWS APIs (rds:GenerateDBAuthToken) using an IAM role assumed via IRSA. serviceAccountName scopes that trust to this one Pipeline — no other workload in the redpanda namespace can assume the role.

Out-of-band: ServiceAccount with the IRSA annotation (terraform / pulumi / a separate manifest — the operator does not create it):

apiVersion: v1
kind: ServiceAccount
metadata:
  name: mysql-cdc-orders-rds
  namespace: redpanda
  annotations:
    # The role's trust policy is scoped to
    # system:serviceaccount:redpanda:mysql-cdc-orders-rds — i.e. this
    # exact SA. Its inline policy grants rds-db:connect on
    # arn:aws:rds-db:us-east-2:<acct>:dbuser:<DbiResourceId>/cdc.
    eks.amazonaws.com/role-arn: arn:aws:iam::605419575229:role/mysql-cdc-orders-rds

Pipeline CR:

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: mysql-cdc-orders
  namespace: redpanda
spec:
  cluster:
    clusterRef: { name: redpanda }
  userRef:
    name: mysql-cdc-orders-svc       # SASL identity for output
  serviceAccountName: mysql-cdc-orders-rds   # IRSA boundary for AWS

  valueSources:
    # No MYSQL_PASSWORD — IAM auth supplies it on the fly.
    - name: MYSQL_HOST
      source:
        secretKeyRef: { name: mysql-cdc-creds, key: host }
    - name: MYSQL_USER
      source:
        secretKeyRef: { name: mysql-cdc-creds, key: username }

  configYaml: |
    input:
      mysql_cdc:
        # IAM auth: Connect calls rds:GenerateDBAuthToken using the
        # pod's IRSA-assumed role and uses the 15-min token as the
        # MySQL password. allowCleartextPasswords=1 is required so the
        # Go MySQL driver sends the token (which it sees as a plaintext
        # password) over the TLS-protected connection.
        dsn: "${MYSQL_USER}@tcp(${MYSQL_HOST}:3306)/shop?tls=skip-verify&allowCleartextPasswords=1"
        aws:
          enabled: true
          region: us-east-2
          endpoint: "${MYSQL_HOST}:3306"
        stream_snapshot: true
        tables: [orders]
        flavor: mysql
    pipeline:
      processors:
        - mapping: |
            root = this
            root.cdc_received_at = now()
    output:
      redpanda:                       # only the per-plugin fields
        topic: mysql.shop.orders
        key: '${! @table }'

The pipeline pod assumes mysql-cdc-orders-rds (and only that role) via the projected service-account token at /var/run/secrets/eks.amazonaws.com/serviceaccount/token. The MySQL connection uses an IAM-generated token; no MySQL password lives anywhere in the pod, the Pipeline CR, or a Secret.

This is the production K8s-RDS pattern: IRSA gates AWS API access (so the pipeline can mint MySQL tokens); Pipeline userRef gates Redpanda access (so the pipeline can write its output topic). The two trust boundaries are orthogonal.

D. Inline — pipeline references multiple external sources via valueSources

For pipelines whose primary connection isn't Redpanda at all, or that fan out to multiple non-Kafka backends. valueSources is destination-agnostic: each entry is a typed pull from inline / Secret / ConfigMap / ExternalSecret and projects to an env var the YAML references via ${NAME}. Connect plugins read those env vars however they expose credentials.

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: orders-fanout
  namespace: redpanda
spec:
  cluster:
    clusterRef: { name: redpanda }
  userRef:
    name: orders-fanout-svc

  valueSources:
    # MongoDB Atlas — full connection URI from a Secret managed by the
    # team that provisions the cluster.
    - name: MONGO_URI
      source:
        secretKeyRef:
          name: mongo-orders-atlas
          key: connection_uri          # mongodb+srv://user:pass@cluster.../...

    # Snowflake key-pair auth — account/user inline; private key from a
    # Secret; passphrase via external-secrets.io (1Password / Vault / etc.).
    - name: SNOWFLAKE_ACCOUNT
      source: { inline: "ab12345.us-east-1" }
    - name: SNOWFLAKE_USER
      source: { inline: "PIPELINE_SVC" }
    - name: SNOWFLAKE_PRIVATE_KEY
      source:
        secretKeyRef:
          name: snowflake-pipeline-svc
          key: rsa_key.p8
    - name: SNOWFLAKE_PRIVATE_KEY_PASSPHRASE
      source:
        externalSecretRef:
          name: snowflake-pipeline-svc-passphrase

    # MySQL — DSN composed from a Secret-backed password and a ConfigMap-
    # provided host so app teams can rotate the read-replica endpoint
    # without touching Pipeline CRs.
    - name: MYSQL_USER
      source: { inline: "warehouse_writer" }
    - name: MYSQL_PASSWORD
      source:
        secretKeyRef:
          name: mysql-warehouse-creds
          key: password
    - name: MYSQL_HOST
      source:
        configMapKeyRef:
          name: warehouse-env
          key: mysql_replica_host

  configYaml: |
    input:
      redpanda:
        # Only the per-plugin fields — seed_brokers / tls / sasl
        # injected by the operator from clusterRef + userRef.
        topics: [orders]
        consumer_group: orders-fanout

    output:
      broker:
        outputs:
          - mongodb:
              url: ${MONGO_URI}
              database: orders
              collection: ingested
              operation: insert-one

          - snowflake_put:
              account: ${SNOWFLAKE_ACCOUNT}
              user: ${SNOWFLAKE_USER}
              private_key: ${SNOWFLAKE_PRIVATE_KEY}
              private_key_pass: ${SNOWFLAKE_PRIVATE_KEY_PASSPHRASE}
              database: WAREHOUSE
              schema: PUBLIC
              stage: "@PIPELINE_STAGE"

          - sql_insert:
              driver: mysql
              dsn: "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:3306)/warehouse?parseTime=true"
              table: orders
              columns: [order_id, ingested_at, payload]
              args_mapping: |
                root = [ this.id, this.ingested_at, this.format_json() ]

Properties this design intentionally preserves:

• Plugin-agnostic: the inline-merge only touches input.redpanda and output.redpanda blocks. New Connect plugins ship in future Connect releases without any operator change; non-redpanda blocks (mongodb, snowflake_put, sql_insert, aws_s3, mysql_cdc, etc.) pass through untouched.
• Mixed sources per pipeline: inline / secretKeyRef / configMapKeyRef / externalSecretRef can mix freely across entries in the same valueSources list.
• Per-key, not per-Secret: unlike the earlier secretRef[] env-splat, every value is a named pull. Unused keys in a Secret don't leak into the pod env, the env name and the Secret's key can differ, and multiple pipelines can pull non-overlapping keys from the same Secret.
• Orthogonal trust boundaries: cluster + userRef gates Redpanda access; serviceAccountName gates cloud-IAM access. Either can be set without the other.

Status conditions

Condition	True	False (examples)
ClusterRef	cluster.clusterRef resolved → broker list + TLS material loaded	ClusterRefInvalid (cluster not found / not Ready)
UserRef	userRef.name exists, has password.valueFrom.secretKeyRef set, mechanism resolved	UserInvalid (User CR not found, missing Secret-backed password)
ConfigValid	redpanda-connect lint passes	ConfigInvalid
Ready	all of the above + Deployment Ready	otherwise false

Tests

• TestRender_InlineMergesRedpandaPlugins — six subtests covering the cluster-binding render path: merges into output.redpanda, merges into input.redpanda, user-supplied keys win on conflict, no *.redpanda block in user config → no injection, output.redpanda_common is intentionally not auto-configured (regression guard against re-emitting a top-level redpanda: block), fully-inline pipeline (no cluster binding) passes through unchanged.
• TestRender_Deployment_ServiceAccountName — propagation of spec.serviceAccountName to Deployment.Spec.Template.Spec.ServiceAccountName; empty when unset.
• TestRender_Deployment_ImagePrecedence — three subtests, one per image precedence tier (per-pipeline spec.image > chart-level connectController.image > binary constant).
• TestRender_Deployment_ValueSources — asserts EnvFrom is empty on both the lint init and the connect container, and that each ValueSource entry projects as exactly one typed EnvVar (inline → Value; secretKeyRef/configMapKeyRef → ValueFrom.{Secret,ConfigMap}KeyRef).
• TestReconcile_InvalidClusterRefCleansUpManagedResources — cluster.clusterRef resolution failure short-circuits before user resolution; status surfaces ClusterRefInvalid and managed resources are torn down.
• Existing TestRender_* cases cover Replicas, Paused, Zones, Resources, Annotations, Topology, Budget, ConfigFiles, MonitoringPodMonitor.
• task lint (helm lint + golangci-lint + actionlint) clean.
• task generate clean (no further diff).

End-to-end validation

The branch was exercised against real AWS infrastructure (EKS 1.34 + RDS MySQL 8.0 with iam_database_authentication_enabled=true) using the Example C scenario (per-pipeline IRSA + native RDS IAM auth + Pipeline CR writing to Redpanda). MySQL snapshot rows reached the mysql.shop.orders topic with no MySQL password anywhere in the Pipeline CR, the pod env, or any Secret. See the e2e comment thread for the full run + reproduction steps.

Enhancements on top of #1337 (ported from the enterprise branch)

• Top-level redpanda: shared client rendered from the resolved cluster connection (in addition to the input.redpanda/output.redpanda merge) — makes redpanda_common and other shared-client plugins work without inline credentials; user-set keys still win. (This supersedes Add Pipeline CRD for Redpanda Connect pipeline management #1337's deliberate non-configuration of redpanda_common; the description above has been updated accordingly.)
• Config-checksum pod-template annotation (cluster.redpanda.com/config-checksum) — configYaml/configFiles changes roll the Deployment. Connect reads its config once at startup from the mounted ConfigMap, so without this a config change never reached running pods.
• spec.extraInitContainers — ordered init-container passthrough that runs ahead of the operator's built-in lint init container, so anything staged into a shared volume (certs, warmed caches, dependency waits) is visible to lint and the runtime.
• spec.affinity — pod affinity/anti-affinity for node-pool scheduling of pipeline pods.
• Workload survival on reconcile failures — clusterRef/userRef resolution failures and license-validation failures now surface on .status (with requeue) without tearing down running pipelines; the last-known-good children keep processing data. Only deleting the Pipeline CR removes them. Previously a transient API hiccup or a briefly-unreadable license Secret would delete the pipeline Deployment mid-stream.
• clusterRef render memoization + bounded concurrency — the expensive chart-render behind clusterRef resolution is cached per Redpanda cluster spec generation, and MaxConcurrentReconciles bounds parallel reconciles for large pipeline fleets.
• staticConfiguration fixes — CA path and SASL fallback corrections in the rendered config.
• Connect telemetry — anonymous pipeline fleet stats (counts, running/paused, desired/ready replicas, distinct images, node spread) collected by the operator's existing central telemetry runnable, with a connectController feature flag reported from cmd/run.
• Default Connect image bumped to 4.100.0.

Behavior-contract test updates

• TestReconcile_Invalid{License,ClusterRef}CleansUpManagedResources → …KeepsManagedResources (children survive failed reconciles).
• TestRender_InlineMergesRedpandaPlugins now asserts the top-level shared client is rendered (previously asserted its absence).
• Render goldens + chart schema/partials/template goldens regenerated.

Testing

operator/internal/controller/pipeline (envtest + goldens), operator/internal/telemetry, operator/chart (template goldens), and operator/cmd/... all pass; golangci-lint clean. The underlying features were validated end-to-end on EKS + RDS MySQL (mysql_cdc snapshot + live binlog via a redpanda_common output, config-change auto-roll with no manual pod delete).

🤖 Generated with Claude Code

[secpanda]

✅ Snyk checks have passed. No issues have been found so far.

Status	Scan Engine	Critical	High	Medium	Low	Total (0)
✅	Open Source Security	0	0	0	0	0 issues
✅	Licenses	0	0	0	0	0 issues

💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse.

[david-yu]

E2E evidence: mysql_cdc from RDS + Iceberg/Glue fanout on EKS (IRSA, zero static credentials)

Ran this branch (65d57b68, current head) end to end on EKS (us-east-2) with --enable-connect, exercising most of the new spec surface in one realistic flow: RDS MySQL → CDC pipeline → topic → two-pipeline fanout → two Glue-cataloged Iceberg tables in S3. Certified twice (initial run + a re-certification ~7 hours later against the same live pipelines).

What the flow exercised

• cluster.clusterRef + userRef — CDC pipeline bound to the Redpanda CR and a SCRAM User CR (Secret-backed password, metadata.name as username). The User carries explicit ACLs (mysql.* topics Create/Write/Read/Describe + iceberg-orders-* groups) rather than running as superuser.
• valueSources — RDS credentials projected from a Secret materialized by external-secrets (ESO), no inline credentials in configYaml.
• serviceAccountName — the Iceberg pipelines run under an IRSA-annotated ServiceAccount scoped to Glue catalog/table ops + the warehouse bucket. Glue REST catalog auths via catalog.auth.aws_sigv4 and S3 via the ambient IAM chain — zero static credentials anywhere in the specs.
• Top-level redpanda: shared client — both fanout pipelines use redpanda_common input/output on independent consumer groups; the operator-rendered shared client from cluster made that work with no inline connection config.
• Config-checksum rollouts — every configYaml iteration during bring-up rolled the pipeline Deployments automatically; observed repeatedly, no manual restarts.
• Default Connect image — pipelines ran the branch's new default (connect:4.100.0), nothing pinned.

Results

• mysql_cdc snapshotted shop.orders and streamed live binlog inserts into mysql.shop.orders.
• Both Iceberg tables (orders_raw, orders_flat) auto-created on first write with the full metadata chain (metadata.json → manifest avro → snapshot avro → parquet).
• Row-level proof: live-inserted rows read back from the orders_flat parquet with the flat mapping's columns + CDC timestamps.
• All three Pipelines Ready/Running continuously across the ~8h session.

Operational notes potentially worth folding into docs

1. schema_evolution.table_location needs a trailing slash — the writer concatenates table_location + namespace/table without a separator (s3://bucket → nonexistent-bucket error on CreateTable; s3://bucket/db/table → doubled paths). s3://bucket/ works.
2. Glue's Iceberg REST endpoint is SigV4, not OAuth2 — the iceberg output's catalog.auth.oauth2 block is for Polaris/Unity-style catalogs; use catalog.auth.aws_sigv4 for Glue.
3. The CDC produce path does not implicitly create topics even with a Create ACL — pre-create or auto_create_topics_enabled=true.
4. Fanout consumers need group ACLs alongside topic ACLs when the pipeline user isn't a superuser (easy to miss if earlier testing ran superuser).

Nice work — clusterRef/userRef/valueSources/serviceAccountName compose exactly as designed, and the checksum-driven rollout makes iteration on configYaml genuinely pleasant.

[RafalKorepta]

4-Reviewer Panel Review

Independent review panel over the full PR range (8e90bff..847e46e): staff engineer (correctness/design), documentation, security, and a Codex adversarial pass. All four returned request-changes. Findings below are deduplicated and severity-sorted; each names the reviewer(s) that flagged it.

Reviewer	Verdict	Overall take
Staff engineer	request-changes	Well-structured port (clean controller/renderer split, deliberate workload-survival design), but verified correctness defects in TLS resolution, cache invalidation, and the PodMonitor watch probe.
Documentation	request-changes	Field-level docs are thorough and accurate, but the pipeline README's operational core is stale enterprise-branch content contradicting this PR's actual wiring.
Security	request-changes	Well-gated by default (off in chart and flag, no net-new RBAC), but the SSA force-ownership sync enables resource hijack, and the confused-deputy surface is under-documented.
Codex (adversarial)	request-changes	Several public fields are accepted then ignored or lossily translated at runtime: wrong-cluster binding, broken TLS/mTLS, credentials that neither validate nor roll pods.

Blocker

1. Pipeline README documents a different repository (docs) — operator/internal/controller/pipeline/README.md:44-58, 350, 396
   Ported from the enterprise branch without updating: it directs users to clone a personal external repo (david-yu/redpanda-setup), export a REDPANDA_LICENSE_FILE env var nothing in the codebase reads (the real source is --license-file-path via enterprise.licenseSecretRef), and names a --enable-connect-controller flag that doesn't exist (the real flag is --enable-connect, wired in this PR). A user following the README never finds the real enablement path and every pipeline stays LicenseInvalid.

Major

2. clusterRef TLS silently drops ConfigMap truststores and mTLS client certs (staff-eng, Codex) — operator/internal/controller/pipeline/cluster.go:287
   Only CaCert.SecretKeyRef is mapped. The chart's InternalTLS.ToCommonTLS legitimately emits ConfigMapKeyRef truststores, and Cert/Key when requireClientAuth: true. A pipeline against such a listener gets conn.TLS == nil → RPK_TLS_ENABLED=false → plaintext connection to a TLS listener, with ClusterRefResolved=True giving no hint. The clusterTLS struct already has CACertConfigMapRef — the static path populates it, the clusterRef path never does.

3. staticConfiguration tls: {enabled: false} still renders TLS on (Codex, staff-eng) — cluster.go:323
   Any non-nil kafka.tls block forces tls.enabled: true into connect.yaml, violating the documented CommonTLS contract (common.go:152). A pipeline pointed at plaintext brokers with an explicit enabled: false attempts TLS and fails. mTLS cert/key on the static path are also silently ignored.

4. clusterRef namespace/group/kind accepted by the schema but silently ignored (security, Codex) — cluster.go:254-262
   Resolution always does Get{Name: ref.Name, Namespace: pipeline.Namespace}. Setting clusterRef: {name: main, namespace: prod} from namespace dev silently binds to dev's same-named cluster — with that cluster's bootstrap-superuser SASL credentials — and produces/consumes against the wrong cluster with no error. The watch/index path is also name-only, so the actually-referenced namespace never enqueues. Reject non-default namespace/group/kind (ideally via CEL) or honor them.

5. clusterConnCache serves stale connections after Redpanda delete/recreate (staff-eng) — cluster.go:56-78
   Keyed namespace/name, validated only by generation. A recreated CR restarts at generation 1, so delete/recreate with a different spec (TLS on, new port) serves the old brokers/TLS/SASL until a spec edit or operator restart. Include the UID in the key or evict on delete events. (Also: no eviction for deleted clusters — slow unbounded growth.)

6. Referenced Secrets/Users are never validated; ValueSourcesResolved condition documented but never set (Codex, docs) — cluster.go:210-230, render.go, pipeline_types.go:87-102
resolveUserRef reads only the User CR, then marks UserRef=True even when the password Secret doesn't exist — pods fail later with CreateContainerConfigError, and the README's promised ValueSourcesResolved condition never appears (constants exist, controller never references them). The config checksum also hashes only ConfigMap data, so Secret rotations never roll pods. Resolve refs at reconcile time, set the conditions, and cover secret content (hash or resourceVersion) in the roll trigger.

7. extraInitContainers documented shared-volume pattern produces an invalid Deployment (Codex) — pipeline_types.go:187-207, render.go:460-499
   The API example mounts a volume name: shared, but the renderer only defines the config and TLS volumes and the spec exposes no extra volumes. Following the docs yields a Deployment referencing a nonexistent volume. Add extra-volume support or narrow the docs.

8. SSA ForceOwnership lets pipelines-create hijack and delete pre-existing same-named ConfigMaps/Secrets (security) — controller.go:330
   Children are named after the Pipeline (ConfigMap = name, Secret = name+"-license") and force-applied with no prior-ownership check. A principal with only pipelines create/delete can name a Pipeline after another team's ConfigMap; the operator adopts and overwrites it, and Pipeline deletion then deletes the hijacked object. Guard adoption: refuse (Ready=False, NameConflict) unless the existing object already carries the Pipeline's ownership markers.

9. PodMonitor watch is never registered in production (staff-eng) — controller.go:580-590
skipPodMonitorWatchIfNotInstalled does a cached List from SetupWithManager — before mgr.Start(), so it always gets ErrCacheNotStarted (a pitfall this repo documents at run.go:733) and skips the watch on every real deployment. It also hardcodes the default namespace. Use the RESTMapper or mgr.GetAPIReader() for CRD detection. Invisible to tests because envtest never exercises SetupWithManager.

10. Malformed + duplicate changie entry will break the release cut (docs, staff-eng) — .changes/unreleased/operator-Added-20260323-connect-crd.yaml
    Missing project: and time: keys (changie can't attribute or order it) and duplicates the complete operator-Added-20260715-120000.yaml entry. Delete it.

11. README default-image claims are stale (docs) — README says connect:4.96.0 in three places; the code ships 4.100.0 (pipeline_types.go:22). The values.yaml commented example pins 4.92.0 — uncommenting it silently downgrades every pipeline.

12. README claims replicas: 0 → phase Stopped; code reports Provisioning/Ready=False forever (docs) — deriveStatus returns Stopped only for paused: true; replicas-0 falls into the default branch with a 15-second requeue loop and "Pipeline is starting up". Fix either side.

Minor

• Transient clusterRef/userRef/license failures flip a healthy Running pipeline to Pending/Ready=False despite workload survival keeping it processing data — trips alerting and the acceptance step. Derive phase from the live Deployment and overlay only the failing condition. (staff-eng — controller.go:240,271,295)
• Comma in a commonAnnotations value crash-loops the operator: deployment.go:393-408 joins k=v,k2=v2 for pflag StringToStringVar, which splits on unquoted commas — a values-only change breaks binary startup. (staff-eng)
• --enable-connect=true silently ignored when --enable-redpanda-controllers=false — Pipeline setup is nested inside the v2-controllers block with no validation error, unlike NodePool/Console. (Codex — run.go:341-346)
• Telemetry misreports the Connect image two ways: GetImage() skips the --connect-default-image tier so fleets on the chart default report the baked constant (staff-eng, Codex), and it ships the verbatim .spec.image string (internal registry hostnames, team names) in a field documented as "Anonymous" (security — collector.go:279-283). Report only the version/tag and plumb the flag value in.
• clusterRef without userRef silently runs as the cluster's bootstrap superuser — pipelines-create becomes cluster-superuser-equivalent (arbitrary image + superuser creds in env, exfiltratable by the pipeline config itself). At minimum document it; ideally gate behind explicit opt-in. (security — cluster.go:300)
• Inline staticConfiguration SASL passwords render as plaintext EnvVar.Value in the Deployment (readable by anyone with deployments/pods get; lands in backups). Route inline passwords through the existing pipeline-owned Secret. (security — render.go:560)
• Enterprise license is mirrored into a plain Secret in every namespace hosting a Pipeline — intentional (Connect's runtime gate) but the widened exfiltration audience is documented nowhere. (security — controller.go:296)
• Itemized pipeline RBAC omits users get/list/watch that resolveUserRef needs — works only because the v2-manager role happens to include it; defeats the purpose of the itemized bundle. (staff-eng — controller.go:108-117)
• Test coverage misses exactly where the bugs live: no successful-clusterRef test (any TLS variant), no cache-invalidation test, no userRef reconcile test, SetupWithManager never exercised; the acceptance "Update a Pipeline config" scenario can pass vacuously (poll satisfied by pre-update status — gate on observedGeneration or the checksum roll). (staff-eng)
• Controller.Disabled doc comment is stale (wrong flag name, field never set by run.go, "no telemetry" claim wrong — the central collector reports pipelines regardless). (docs — controller.go:80-95)
• --license-file-path help says telemetry-only but now gates the entire Connect controller — a user reading --help will skip it and get fleet-wide LicenseInvalid. (docs — run.go:236)
• CRD/type docs call the resource "Connect" ("Connect defines…", "list of Connect resources") — leaks into kubectl explain pipeline and crd-docs. (docs — pipeline_types.go:384,407)

Nit

• values.yaml: the new connectController block orphans the additionalCmdFlags example comment (which also has a pre-existing typo), and the "license … on each CR" phrasing misstates the operator-level license model. (staff-eng, docs)
• run.go Connect-reconciler comment claims per-CR license support that doesn't exist. (staff-eng)
• Lint init-container log tail (running with REDPANDA_SASL_PASSWORD/REDPANDA_LICENSE in env) is copied verbatim into the ConfigValid condition message — visible to anyone with pipelines/get, no pod-log RBAC needed. Consider truncating/sanitizing. (security — controller.go:445)

---

One positive note from the adversarial pass: Codex specifically checked the finalizer path for a PodMonitor deletion deadlock and found none — the common syncer skips unknown API types safely.

Highest-leverage fixes before merge: the README rewrite (#1), the TLS resolution pair (#2, #3), the ignored clusterRef namespace (#4), and the SSA hijack guard (#8).

🤖 Generated with Claude Code — 4-reviewer panel (staff engineer, documentation, security, Codex adversarial)

[JakeSCahill]

Docs-side validation report (building this branch at head b861c263 and running it on kind): the feature works end-to-end, but the Pipeline CRD as generated on this head cannot be installed on a real Kubernetes API server — the panel-fix commit's new CEL rules push it over the validation-cost budget.

CRD install failure (blocker)

On kind v1.28.0, both redpanda-operator crd (the chart's CRD hook job, which crash-loops and fails the whole helm install when crds.enabled=true) and a direct kubectl apply reject pipelines.cluster.redpanda.com:

spec.validation.openAPIV3Schema.properties[spec].properties[valueSources].items.properties[source].x-kubernetes-validations[1..4].rule:
Forbidden: estimated rule cost exceeds budget by factor of 1.258291x
(try simplifying the rule, or adding maxItems, maxProperties, and maxLength where arrays, maps, and strings are declared)

The four source one-of rules need bounded inputs (maxLength on the string fields of ValueSource/refs) to fit the budget. Additionally, the crd installer path reported a CEL compile failure for the new namespace-rejection rule (undefined field 'namespace' on spec rule !has(self.cluster) || !has(self.cluster.clusterRef) || !has(self.cluster.clusterRef.namespace)), which did not reproduce via direct kubectl apply — possibly a difference in how the installer submits the schema; including it for completeness.

Suggestion: an acceptance/CI step that applies the generated CRDs against a real kube-apiserver (kind or envtest with validation enabled) would catch this class of issue — envtest may not enforce the CEL cost budget the way a real API server does.

What I validated (with the CRD locally patched to remove the over-budget rules)

Everything below worked as designed at head b861c263 on kind v1.28.0, connectController.enabled=true, Enterprise license:

• Inline pipeline: Ready=True/Running in ~30s, lint init container, stdout output.
• Invalid config: lint fails, ConfigValid=False with the lint error surfaced in status.
• valueSources secretKeyRef → ${VAR} interpolation.
• Config change → checksum annotation → Recreate rollout.
• Pause (Stopped) / resume / delete with full child-resource cleanup.
• clusterRef against a TLS-enabled Redpanda CR: ClusterRefResolved=True, CA correctly projected (/etc/tls/certs/ca/ca.crt), both the merged output.redpanda and top-level redpanda blocks rendered, messages verified in the topic.

Two behaviors worth a deliberate decision (I'll document them either way):

1. With auto_create_topics_enabled=false (chart default), a pipeline producing to a missing topic reports Ready=True/Running while every send fails (UNKNOWN_TOPIC_OR_PARTITION visible only in pod logs). Consider surfacing sustained delivery failure in status, or we'll document "Running means the process is up, not that messages are delivered."
2. kubectl apply of the huge CRD client-side also trips the 262144-byte annotation limit — worth mentioning server-side apply in any manual-install instructions.

Context: docs draft for this feature is redpanda-data/rp-connect-docs#457 (DOC-2275).

[david-yu]

Thanks @JakeSCahill will take a look.

[david-yu]

Validation follow-up: CRD install is version-dependent; supported range is k8s 1.33–1.36

Thanks @JakeSCahill — your finding reproduces exactly, and I dug into the version boundary. Summary: this operator release supports k8s API server 1.33–1.36, and the Pipeline CRD installs cleanly across that entire range. The rejection you hit is confined to k8s ≤1.29, which is below the supported range.

CRD install vs. k8s version

Bracketed with a direct server-side apply of the generated CRD (head b861c263):

k8s API server	Pipeline CRD install
1.28.15	❌ cost-budget overrun on valueSources[].source rules [2,3,4] (1.258291x) + undefined field 'namespace' compile error
1.29.14	❌ rejected
1.33.7 (support floor)	✅ accepted + Established
1.35.0	✅ accepted + Established
1.36.2-eks (support ceiling)	✅ accepted + Established

Root cause is the k8s CEL cost estimator, which became more accurate in newer apiextensions-apiservers — the same schema is over budget on ≤1.29 and under budget on ≥1.33. The undefined field 'namespace' is an older-CEL-compiler artifact: clusterRef.namespace is declared in the schema (shared ClusterRef type; the rule intentionally rejects it for pipelines), so the rule is valid CEL and compiles on ≥1.33 — which is why it didn't reproduce via kubectl apply for you.

EKS 1.36 end-to-end (the support ceiling)

On EKS v1.36.2-eks-8f14419, operator built from b861c263, enterprise license:

• CRD: kubectl apply --server-side → accepted + Established, no cost-budget rejection.
• Chart hook job: helm install with crds.enabled=true + connectController.enabled=true succeeded — the CRD hook job did not crash-loop (contrast your 1.28 result). Operator runs with --enable-connect=true.
• Pipeline: inline generate → stdout reached phase=Running / Ready=True / ConfigValid=True / License=True in ~20s; the Connect pod emits {"message":"hello world"}.

(One toolchain note so it doesn't trip others: the operator chart's CRD hook stalls under helm v4.2.1 — the pre-install-hook --wait regression helm/helm#32214, fixed in v4.2.2. Install with helm 3.19.1 or ≥4.2.2. Not a chart issue.)

[JakeSCahill]

Correction to my earlier report: the CRD install blocker does not reproduce on a supported Kubernetes version — it was an artifact of my test environment running K8s 1.28, below the 26.2 support range.

Retested today at head b861c263 on kind v1.34.0:

• The unpatched pipelines.cluster.redpanda.com CRD applies cleanly (server-side apply, no CEL cost-budget errors, no compile errors).
• The CEL rules enforce as intended: a Pipeline with cluster.clusterRef.namespace set is rejected at admission with the "not supported for pipelines" message.

So the only real finding there is informational: the CRD cannot install on much older API servers (1.28) because their CEL cost estimator rejects the valueSources rules — irrelevant within the supported range, but worth keeping in mind if a minimum-Kubernetes-version statement lands in the docs.

Everything in the functional-validation section of my earlier comment stands (that testing exercised the controller, not the API server): lint, lifecycle, valueSources, config rollout, clusterRef + TLS injection, and the two documented behaviors (missing-topic pipelines report Running while sends fail; huge CRD needs server-side apply for manual installs).

Docs side: our stretch guide stated "Kubernetes 1.28 or later" as a prerequisite, which is what put my test rig on 1.28 in the first place — fixing that now to point at the supported-versions compatibility matrix instead.

[david-yu]

8359de25 — HPA/KEDA autoscaling via the scale subresource

The Pipeline CRD now exposes the Kubernetes scale subresource, wired to .spec.replicas / .status.replicas / a new .status.selector (stamped on every status write with the pipeline pod selector). Anything that speaks /scale can drive pipeline replicas — kubectl scale, autoscaling/v2 HPA, and KEDA — with no fighting against the operator: autoscalers write the Pipeline's replicas and the controller propagates to the Deployment as usual.

Semantics:

• spec.paused: true wins over autoscaler-written counts; unpausing resumes at the autoscaler-managed count.
• An explicit replicas: 0 via /scale (KEDA scale-to-zero) is preserved — the CRD default of 1 only applies when the field is absent.
• Strictly opt-in: without an autoscaler nothing changes, and no metrics-server/Prometheus/KEDA is required to run Pipelines.
• One autoscaler per Pipeline (don't point an HPA and a ScaledObject at the same one).
• When an autoscaler owns replicas, omit the field from applied manifests (or have your GitOps tool ignore it) so syncs don't undo it.

How to: manual scaling (nothing to install)

kubectl scale pipeline/<name> --replicas=4
kubectl get --raw /apis/cluster.redpanda.com/v1alpha2/namespaces/<ns>/pipelines/<name>/scale

How to: HPA on CPU/memory

1. Ensure metrics-server is installed (kubectl top pods works). Pipeline pods always carry resource requests (operator defaults 100m/256Mi when spec.resources is unset), so utilization metrics work out of the box.
2. Create the HPA targeting the Pipeline, not its Deployment:

   apiVersion: autoscaling/v2
   kind: HorizontalPodAutoscaler
   metadata:
     name: my-pipeline
   spec:
     scaleTargetRef:
       apiVersion: cluster.redpanda.com/v1alpha2
       kind: Pipeline
       name: my-pipeline
     minReplicas: 1
     maxReplicas: 10
     metrics:
       - type: Resource
         resource:
           name: cpu
           target: { type: Utilization, averageUtilization: 80 }

3. Verify: kubectl get hpa my-pipeline shows live targets (e.g. cpu: 300%/80%) and kubectl get pipeline my-pipeline shows REPLICAS moving.

How to: KEDA on consumer-group lag (the fit for consumer pipelines)

1. Install KEDA: helm install keda kedacore/keda -n keda --create-namespace.
2. Create a ScaledObject with a kafka trigger pointed at the pipeline's consumer group — no Prometheus or metrics-server needed, KEDA reads lag from the brokers:

   apiVersion: keda.sh/v1alpha1
   kind: ScaledObject
   metadata:
     name: my-pipeline
   spec:
     scaleTargetRef:
       apiVersion: cluster.redpanda.com/v1alpha2
       kind: Pipeline
       name: my-pipeline
     minReplicaCount: 1        # 0 enables scale-to-zero
     maxReplicaCount: 12       # capped at partition count when allowIdleConsumers=false
     triggers:
       - type: kafka
         metadata:
           bootstrapServers: <cluster>.<ns>.svc.cluster.local:9093
           consumerGroup: <group from input.redpanda.consumer_group>
           topic: <topic>
           lagThreshold: "500"
           allowIdleConsumers: "false"

3. Verify: KEDA materializes hpa/keda-hpa-my-pipeline; watch lag with rpk group describe <group> and replicas with kubectl get pipeline -w.

How to: KEDA/HPA on metrics Connect emits

1. Get the pods' metrics (:4195/metrics — input_received, output_sent, …) into Prometheus: set connectController.monitoring.enabled=true in the operator chart (renders a PodMonitor per pipeline; requires the Prometheus Operator CRDs), or scrape them yourself.
2. Either feed HPA through prometheus-adapter (type: Pods metric), or use a KEDA prometheus trigger:

   triggers:
     - type: prometheus
       metadata:
         serverAddress: http://prometheus-operated.monitoring.svc:9090
         query: sum(rate(input_received{namespace="<ns>", pod=~"my-pipeline-.*"}[1m]))
         threshold: "1000"     # desired msg/s per replica

3. Pick the threshold from the pipeline's real per-replica throughput (query Prometheus first) — a threshold above what one pod can push means the autoscaler correctly never fires.

E2E (EKS 1.36.2, operator image at this commit)

Scenario	Result
kubectl scale + raw GET /scale	proper Scale object w/ selector; Deployment follows in seconds
HPA cpu+memory	cpu: 300%/50% → scaled 1→4 in ~30s
KEDA kafka-lag: 12-partition topic → consumer-group pipeline → RDS Postgres sql_insert	lag ~500k → 1→6, partitions rebalanced across all 6 pods; 596,075 rows, 7 distinct writer pods, 0 duplicate ids; after load stop, drained to lag=0 and stepped down 6→4→1
KEDA prometheus on input_received (via operator PodMonitor)	scaled 1→3 in ~90s

Pipelines stayed Ready=True through every rebalance and scale transition. Full details in the README's new "Autoscaling with HPA or KEDA" section.

🤖 Generated with Claude Code