feat: automate GCP Workload Identity via Config Connector IAMPolicyMember

charts/restate-operator-helm/templates/rbac.yaml

@@ -103,6 +103,19 @@ rules:
     apiGroups:
       - eks.services.k8s.aws
   {{- end }}
+  {{- if .Values.gcpConfigConnector }}
+  - resources:
+      - iampolicymembers
+    verbs:
+      - get
+      - list
+      - watch
+      - create
+      - patch
+      - delete
+    apiGroups:
+      - iam.cnrm.cloud.google.com
+  {{- end }}
   - resources:
       - configurations
       - routes

charts/restate-operator-helm/values.yaml

@@ -14,6 +14,7 @@ serviceAccount:
 podAnnotations: {}
 
 awsPodIdentityAssociationCluster: null
+gcpConfigConnector: null
 clusterDns: null # defaults to "cluster.local" in the operator binary
 
 podSecurityContext:

release-notes/unreleased/gcp-workload-identity.md

@@ -0,0 +1,49 @@
+# Release Notes: GCP Workload Identity automation via Config Connector
+
+## New Feature
+
+### What Changed
+The operator now automatically creates Config Connector `IAMPolicyMember`
+resources to bind Kubernetes service accounts to GCP service accounts via
+Workload Identity. This is triggered when a RestateCluster has the
+`iam.gke.io/gcp-service-account` annotation in `serviceAccountAnnotations`.
+
+The GCP project ID is extracted from the service account email
+(`name@PROJECT.iam.gserviceaccount.com`), so no additional configuration
+is needed beyond the annotation that the control plane already sets.
+
+A canary job validates that Workload Identity credentials are available
+before allowing the StatefulSet to proceed, preventing Restate from starting
+without GCS access.
+
+This mirrors the existing AWS Pod Identity Association pattern.
+
+### Why This Matters
+Previously, the IAM binding for Workload Identity had to be created manually
+via `gcloud` commands each time a new environment was provisioned. This
+automates that step, bringing GCP to parity with the existing AWS automation.
+
+### Impact on Users
+- **Existing deployments**: No impact. The feature only activates when
+  `iam.gke.io/gcp-service-account` is present in `serviceAccountAnnotations`.
+- **GCP deployments using this annotation**: Config Connector must be installed
+  on the GKE cluster. If the `IAMPolicyMember` CRD is not available, the
+  operator sets a `NotReady` status condition with a clear message rather
+  than crashing.
+- **AWS deployments**: No impact. The canary job infrastructure was refactored
+  to share code between AWS and GCP, but behavior is unchanged.
+
+### Migration Guidance
+No migration needed. To use this feature:
+
+1. Install Config Connector on the GKE cluster
+2. Ensure the Config Connector service account has `roles/iam.serviceAccountAdmin`
+   on the target GCP service account
+3. Set `serviceAccountAnnotations` on the RestateCluster:
+
+```yaml
+spec:
+  security:
+    serviceAccountAnnotations:
+      iam.gke.io/gcp-service-account: restate@my-project.iam.gserviceaccount.com
+```

src/controllers/restatecluster/controller.rs

@@ -35,6 +35,7 @@ use tokio::{sync::RwLock, time::Duration};
 use tracing::*;
 
 use crate::controllers::{Diagnostics, State};
+use crate::resources::iampolicymembers::IAMPolicyMember;
 use crate::resources::podidentityassociations::PodIdentityAssociation;
 use crate::resources::restateclusters::{
     RESTATE_CLUSTER_FINALIZER, RestateCluster, RestateClusterCondition, RestateClusterStatus,
@@ -396,10 +397,10 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
         security_group_policy_installed,
         pod_identity_association_installed,
         secret_provider_class_installed,
-    ) = api_groups
-        .groups
-        .iter()
-        .fold((false, false, false), |(sgp, pia, spc), group| {
+        iam_policy_member_installed,
+    ) = api_groups.groups.iter().fold(
+        (false, false, false, false),
+        |(sgp, pia, spc, ipm), group| {
             fn group_matches<R: Resource<DynamicType = ()>>(group: &APIGroup) -> bool {
                 group.name == R::group(&())
                     && group.versions.iter().any(|v| v.version == R::version(&()))
@@ -408,8 +409,10 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
                 sgp || group_matches::<SecurityGroupPolicy>(group),
                 pia || group_matches::<PodIdentityAssociation>(group),
                 spc || group_matches::<SecretProviderClass>(group),
+                ipm || group_matches::<IAMPolicyMember>(group),
             )
-        });
+        },
+    );
 
     let rc_api = Api::<RestateCluster>::all(client.clone());
     let ns_api = Api::<Namespace>::all(client.clone());
@@ -421,6 +424,7 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
     let cm_api = Api::<ConfigMap>::all(client.clone());
     let np_api = Api::<NetworkPolicy>::all(client.clone());
     let pia_api = Api::<PodIdentityAssociation>::all(client.clone());
+    let ipm_api = Api::<IAMPolicyMember>::all(client.clone());
     let sgp_api = Api::<SecurityGroupPolicy>::all(client.clone());
     let spc_api = Api::<SecretProviderClass>::all(client.clone());
 
@@ -551,6 +555,27 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
     } else {
         controller
     };
+    let controller = if iam_policy_member_installed {
+        let ipm_watcher = watcher(ipm_api, cfg.clone())
+            .map(ensure_deletion_change)
+            .touched_objects()
+            .predicate_filter(changed_predicate.combine(status_predicate));
+
+        let wi_job_api = Api::<Job>::all(client.clone());
+        let wi_job_watcher = metadata_watcher(
+            wi_job_api,
+            Config::default().labels("app.kubernetes.io/name=restate-wi-canary"),
+        )
+        .map(ensure_deletion_change)
+        .touched_objects()
+        .predicate_filter(changed_predicate);
+
+        controller
+            .owns_stream(ipm_watcher)
+            .owns_stream(wi_job_watcher)
+    } else {
+        controller
+    };
     controller
         .run(
             reconcile,