fix: gate GCP worload ID cleanup

release v2.3.0 introduced gcp workload identity but included a
regression. if you weren't on GCP it would try to check for a CR for
which the CRD was absent.

in this commit we plumb through the helm conditional used for the RBAC
into the runtime and use it to gate all GCP WI related steps.

Author: lukebond

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

@@ -60,6 +60,10 @@ spec:
             - name: AWS_POD_IDENTITY_ASSOCIATION_CLUSTER
               value: {{ .Values.awsPodIdentityAssociationCluster }}
           {{- end }}
+          {{- if .Values.gcpWorkloadIdentity }}
+            - name: GCP_WORKLOAD_IDENTITY
+              value: "true"
+          {{- end }}
           {{- if .Values.clusterDns }}
             - name: CLUSTER_DNS
               value: {{ .Values.clusterDns }}

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

@@ -103,7 +103,7 @@ rules:
     apiGroups:
       - eks.services.k8s.aws
   {{- end }}
-  {{- if .Values.gcpConfigConnector }}
+  {{- if .Values.gcpWorkloadIdentity }}
   - resources:
       - iampolicymembers
     verbs:

charts/restate-operator-helm/values.yaml

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

src/controllers/mod.rs

@@ -38,6 +38,8 @@ pub struct State {
     pub registry: prometheus::Registry,
     /// If set, watch AWS PodIdentityAssociation resources, and if requested create them against this cluster
     aws_pod_identity_association_cluster: Option<String>,
+    /// If true, manage GCP Workload Identity via Config Connector IAMPolicyMember
+    gcp_workload_identity: bool,
 
     /// Our namespace, needed for network policies and reading secrets
     operator_namespace: String,
@@ -57,6 +59,7 @@ pub struct State {
 impl State {
     pub fn new(
         aws_pod_identity_association_cluster: Option<String>,
+        gcp_workload_identity: bool,
         operator_namespace: String,
         operator_label_name: Option<String>,
         operator_label_value: Option<String>,
@@ -67,6 +70,7 @@ impl State {
             diagnostics: Arc::new(RwLock::new(Diagnostics::default())),
             registry: prometheus::Registry::default(),
             aws_pod_identity_association_cluster,
+            gcp_workload_identity,
             operator_namespace,
             operator_label_name,
             operator_label_value,

src/controllers/restatecluster/controller.rs

@@ -72,6 +72,8 @@ pub(super) struct Context {
     pub security_group_policy_installed: bool,
     // Whether the SecretProviderClass CRD is installed
     pub secret_provider_class_installed: bool,
+    // Whether GCP Workload Identity management is enabled
+    pub gcp_workload_identity: bool,
     /// The cluster DNS suffix (e.g. "cluster.local")
     pub cluster_dns: String,
     /// Diagnostics read by the web server
@@ -81,6 +83,7 @@ pub(super) struct Context {
 }
 
 impl Context {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         client: Client,
         metrics: Metrics,
@@ -103,6 +106,7 @@ impl Context {
             operator_label_value: state.operator_label_value.clone(),
             security_group_policy_installed,
             secret_provider_class_installed,
+            gcp_workload_identity: state.gcp_workload_identity,
             cluster_dns: state.cluster_dns.clone(),
             diagnostics: state.diagnostics.clone(),
             metrics,
@@ -555,7 +559,7 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
     } else {
         controller
     };
-    let controller = if iam_policy_member_installed {
+    let controller = if state.gcp_workload_identity && iam_policy_member_installed {
         let ipm_watcher = watcher(ipm_api, cfg.clone())
             .map(ensure_deletion_change)
             .touched_objects()

src/controllers/restatecluster/reconcilers/compute.rs

@@ -624,63 +624,73 @@ pub async fn reconcile_compute(
         .and_then(|s| s.service_account_annotations.as_ref())
         .and_then(|a| a.get("iam.gke.io/gcp-service-account"));
 
-    if let Some(gcp_sa_email) = gcp_sa_email {
-        let gcp_project = match parse_gcp_project_from_sa_email(gcp_sa_email) {
-            Some(project) => project,
-            None => {
-                return Err(Error::InvalidRestateConfig(format!(
-                    "Cannot parse GCP project from service account email '{gcp_sa_email}'; expected format: name@PROJECT.iam.gserviceaccount.com"
-                )));
-            }
-        };
-
-        let ipm = match apply_iam_policy_member(
-            name,
-            &ipm_api,
-            restate_iam_policy_member(name, base_metadata, gcp_project, gcp_sa_email),
-        )
-        .await
-        {
-            Ok(ipm) => ipm,
-            Err(Error::KubeError(kube::Error::Api(kube::error::ErrorResponse {
-                code: 404,
-                ..
-            }))) => {
+    match (ctx.gcp_workload_identity, gcp_sa_email) {
+        (true, Some(gcp_sa_email)) => {
+            let gcp_project = match parse_gcp_project_from_sa_email(gcp_sa_email) {
+                Some(project) => project,
+                None => {
+                    return Err(Error::InvalidRestateConfig(format!(
+                        "Cannot parse GCP project from service account email '{gcp_sa_email}'; expected format: name@PROJECT.iam.gserviceaccount.com"
+                    )));
+                }
+            };
+
+            let ipm = match apply_iam_policy_member(
+                name,
+                &ipm_api,
+                restate_iam_policy_member(name, base_metadata, gcp_project, gcp_sa_email),
+            )
+            .await
+            {
+                Ok(ipm) => ipm,
+                Err(Error::KubeError(kube::Error::Api(kube::error::ErrorResponse {
+                    code: 404,
+                    ..
+                }))) => {
+                    return Err(Error::NotReady {
+                        reason: "IAMPolicyMemberCRDNotFound".into(),
+                        message: "IAMPolicyMember CRD not found - is Config Connector installed?"
+                            .into(),
+                        requeue_after: Some(Duration::from_secs(60)),
+                    });
+                }
+                Err(err) => return Err(err),
+            };
+
+            if !is_iam_policy_member_ready(&ipm) {
                 return Err(Error::NotReady {
-                    reason: "IAMPolicyMemberCRDNotFound".into(),
-                    message: "IAMPolicyMember CRD not found - is Config Connector installed?"
-                        .into(),
-                    requeue_after: Some(Duration::from_secs(60)),
+                    reason: "IAMPolicyMemberNotReady".into(),
+                    message:
+                        "Waiting for Config Connector to provision the IAM policy member binding"
+                            .into(),
+                    requeue_after: None,
                 });
             }
-            Err(err) => return Err(err),
-        };
 
-        if !is_iam_policy_member_ready(&ipm) {
-            return Err(Error::NotReady {
-                reason: "IAMPolicyMemberNotReady".into(),
-                message: "Waiting for Config Connector to provision the IAM policy member binding"
-                    .into(),
-                requeue_after: None,
-            });
-        }
-
-        check_workload_identity(
-            name,
-            base_metadata,
-            spec.compute.tolerations.as_ref(),
-            &job_api,
-        )
-        .await?;
+            check_workload_identity(
+                name,
+                base_metadata,
+                spec.compute.tolerations.as_ref(),
+                &job_api,
+            )
+            .await?;
 
-        let pod_annotations = pod_annotations.get_or_insert_with(Default::default);
-        pod_annotations.insert(
-            "restate.dev/gcp-service-account".into(),
-            gcp_sa_email.to_owned(),
-        );
-    } else {
-        delete_iam_policy_member(name, &ipm_api, "restate-workload-identity").await?;
-        delete_job(name, &job_api, "restate-wi-canary").await?;
+            let pod_annotations = pod_annotations.get_or_insert_with(Default::default);
+            pod_annotations.insert(
+                "restate.dev/gcp-service-account".into(),
+                gcp_sa_email.to_owned(),
+            );
+        }
+        (true, None) => {
+            delete_iam_policy_member(name, &ipm_api, "restate-workload-identity").await?;
+            delete_job(name, &job_api, "restate-wi-canary").await?;
+        }
+        (false, Some(gcp_sa_email)) => {
+            warn!(
+                "Ignoring iam.gke.io/gcp-service-account annotation {gcp_sa_email} as the operator is not configured with --gcp-workload-identity"
+            );
+        }
+        (false, None) => {}
     }
 
     let restate_service = restate_service(

src/main.rs

@@ -17,6 +17,14 @@ struct Arguments {
     )]
     aws_pod_identity_association_cluster: Option<String>,
 
+    #[arg(
+        long = "gcp-workload-identity",
+        env = "GCP_WORKLOAD_IDENTITY",
+        value_name = "ENABLED",
+        default_value = "false"
+    )]
+    gcp_workload_identity: bool,
+
     #[arg(
         long = "operator-namespace",
         env = "OPERATOR_NAMESPACE",
@@ -95,6 +103,7 @@ async fn main() -> anyhow::Result<()> {
     // Initialize Kubernetes controller state
     let state = State::new(
         args.aws_pod_identity_association_cluster,
+        args.gcp_workload_identity,
         args.operator_namespace,
         args.operator_label_name,
         args.operator_label_value,