From c6f4387cb4e962f5e43d432169afad2c0738a658 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Fri, 10 Jul 2026 02:23:38 +0530 Subject: [PATCH 1/5] feat(operator): auto-create RBAC for spark_application batch engine Reconcile server/driver Roles from an embedded YAML template so FeatureStore materialization works without manual SA/Role setup. Widen the operator ClusterRole only as needed to grant those permissions (including deletecollection for Spark driver cleanup). Signed-off-by: Aniket Paluskar --- .../feast-operator.clusterserviceversion.yaml | 49 ++- infra/feast-operator/config/rbac/role.yaml | 39 ++- infra/feast-operator/dist/install.yaml | 39 ++- .../controller/featurestore_controller.go | 8 +- .../controller/services/batch_engine_rbac.go | 301 ++++++++++++++++++ .../services/batch_engine_rbac_test.go | 168 ++++++++++ .../rbac_templates/spark_application.yaml | 26 ++ .../internal/controller/services/services.go | 3 + 8 files changed, 624 insertions(+), 9 deletions(-) create mode 100644 infra/feast-operator/internal/controller/services/batch_engine_rbac.go create mode 100644 infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go create mode 100644 infra/feast-operator/internal/controller/services/rbac_templates/spark_application.yaml diff --git a/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml b/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml index 82e1bc57b36..a4ed7ba70fe 100644 --- a/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml +++ b/infra/feast-operator/bundle/manifests/feast-operator.clusterserviceversion.yaml @@ -147,7 +147,7 @@ metadata: } ] capabilities: Basic Install - createdAt: "2026-06-13T11:22:06Z" + createdAt: "2026-07-09T20:32:45Z" operators.operatorframework.io/builder: operator-sdk-v1.41.0 operators.operatorframework.io/project_layout: go.kubebuilder.io/v4 name: feast-operator.v0.64.0 @@ -180,11 +180,11 @@ spec: resources: - configmaps - persistentvolumeclaims - - serviceaccounts - services verbs: - create - delete + - deletecollection - get - list - update @@ -193,18 +193,45 @@ spec: - "" resources: - namespaces - - pods - secrets verbs: - get - list - watch + - apiGroups: + - "" + resources: + - pods + verbs: + - create + - delete + - deletecollection + - get + - list + - watch - apiGroups: - "" resources: - pods/exec verbs: - create + - apiGroups: + - "" + resources: + - pods/log + verbs: + - get + - apiGroups: + - "" + resources: + - serviceaccounts + verbs: + - create + - delete + - get + - list + - update + - watch - apiGroups: - apps resources: @@ -246,6 +273,14 @@ spec: - patch - update - watch + - apiGroups: + - config.openshift.io + resources: + - apiservers + verbs: + - get + - list + - watch - apiGroups: - feast.dev resources: @@ -321,6 +356,14 @@ spec: - list - update - watch + - apiGroups: + - sparkoperator.k8s.io + resources: + - sparkapplications + verbs: + - create + - delete + - get - apiGroups: - authentication.k8s.io resources: diff --git a/infra/feast-operator/config/rbac/role.yaml b/infra/feast-operator/config/rbac/role.yaml index f6e6801dfa8..a79dca283ed 100644 --- a/infra/feast-operator/config/rbac/role.yaml +++ b/infra/feast-operator/config/rbac/role.yaml @@ -9,11 +9,11 @@ rules: resources: - configmaps - persistentvolumeclaims - - serviceaccounts - services verbs: - create - delete + - deletecollection - get - list - update @@ -22,18 +22,45 @@ rules: - "" resources: - namespaces - - pods - secrets verbs: - get - list - watch +- apiGroups: + - "" + resources: + - pods + verbs: + - create + - delete + - deletecollection + - get + - list + - watch - apiGroups: - "" resources: - pods/exec verbs: - create +- apiGroups: + - "" + resources: + - pods/log + verbs: + - get +- apiGroups: + - "" + resources: + - serviceaccounts + verbs: + - create + - delete + - get + - list + - update + - watch - apiGroups: - apps resources: @@ -158,3 +185,11 @@ rules: - list - update - watch +- apiGroups: + - sparkoperator.k8s.io + resources: + - sparkapplications + verbs: + - create + - delete + - get diff --git a/infra/feast-operator/dist/install.yaml b/infra/feast-operator/dist/install.yaml index 57853e4b83f..22614beed74 100644 --- a/infra/feast-operator/dist/install.yaml +++ b/infra/feast-operator/dist/install.yaml @@ -22033,11 +22033,11 @@ rules: resources: - configmaps - persistentvolumeclaims - - serviceaccounts - services verbs: - create - delete + - deletecollection - get - list - update @@ -22046,18 +22046,45 @@ rules: - "" resources: - namespaces - - pods - secrets verbs: - get - list - watch +- apiGroups: + - "" + resources: + - pods + verbs: + - create + - delete + - deletecollection + - get + - list + - watch - apiGroups: - "" resources: - pods/exec verbs: - create +- apiGroups: + - "" + resources: + - pods/log + verbs: + - get +- apiGroups: + - "" + resources: + - serviceaccounts + verbs: + - create + - delete + - get + - list + - update + - watch - apiGroups: - apps resources: @@ -22182,6 +22209,14 @@ rules: - list - update - watch +- apiGroups: + - sparkoperator.k8s.io + resources: + - sparkapplications + verbs: + - create + - delete + - get --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole diff --git a/infra/feast-operator/internal/controller/featurestore_controller.go b/infra/feast-operator/internal/controller/featurestore_controller.go index 1980fb1f089..b94808c0df5 100644 --- a/infra/feast-operator/internal/controller/featurestore_controller.go +++ b/infra/feast-operator/internal/controller/featurestore_controller.go @@ -65,10 +65,14 @@ type FeatureStoreReconciler struct { // +kubebuilder:rbac:groups=feast.dev,resources=featurestores/status,verbs=get;update;patch // +kubebuilder:rbac:groups=feast.dev,resources=featurestores/finalizers,verbs=update // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;create;update;watch;delete -// +kubebuilder:rbac:groups=core,resources=services;configmaps;persistentvolumeclaims;serviceaccounts,verbs=get;list;create;update;watch;delete +// +kubebuilder:rbac:groups=core,resources=services;configmaps;persistentvolumeclaims,verbs=get;list;create;update;watch;delete;deletecollection +// +kubebuilder:rbac:groups=core,resources=serviceaccounts,verbs=get;list;create;update;watch;delete // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles;rolebindings;clusterroles;clusterrolebindings;subjectaccessreviews,verbs=get;list;create;update;watch;delete -// +kubebuilder:rbac:groups=core,resources=secrets;pods;namespaces,verbs=get;list;watch +// +kubebuilder:rbac:groups=core,resources=secrets;namespaces,verbs=get;list;watch +// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;create;delete;deletecollection // +kubebuilder:rbac:groups=core,resources=pods/exec,verbs=create +// +kubebuilder:rbac:groups=core,resources=pods/log,verbs=get +// +kubebuilder:rbac:groups=sparkoperator.k8s.io,resources=sparkapplications,verbs=create;get;delete // +kubebuilder:rbac:groups=authentication.k8s.io,resources=tokenreviews,verbs=create // +kubebuilder:rbac:groups=route.openshift.io,resources=routes,verbs=get;list;create;update;watch;delete // +kubebuilder:rbac:groups=batch,resources=cronjobs,verbs=get;list;watch;create;update;patch;delete diff --git a/infra/feast-operator/internal/controller/services/batch_engine_rbac.go b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go new file mode 100644 index 00000000000..2d522db8b6b --- /dev/null +++ b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go @@ -0,0 +1,301 @@ +/* +Copyright 2024 Feast Community. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package services + +import ( + "embed" + "fmt" + + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/yaml" +) + +const ( + BatchEngineFeastType FeastServiceType = "batch-engine" + BatchDriverFeastType FeastServiceType = "batch-driver" +) + +//go:embed rbac_templates/*.yaml +var batchEngineRBACTemplates embed.FS + +// BatchEngineRBACTemplate declares RBAC requirements for a batch compute engine. +type BatchEngineRBACTemplate struct { + EngineType string `json:"engine_type" yaml:"engine_type"` + Server *RBACRoleSpec `json:"server,omitempty" yaml:"server,omitempty"` + Driver *DriverRBACSpec `json:"driver,omitempty" yaml:"driver,omitempty"` +} + +// RBACRoleSpec defines policy rules for a Role. +type RBACRoleSpec struct { + Rules []rbacv1.PolicyRule `json:"rules" yaml:"rules"` +} + +// DriverRBACSpec defines policy rules and optional SA creation for a driver Role. +type DriverRBACSpec struct { + CreateServiceAccount bool `json:"create_service_account" yaml:"create_service_account"` + Rules []rbacv1.PolicyRule `json:"rules" yaml:"rules"` +} + +func loadBatchEngineTemplate(engineType string) (*BatchEngineRBACTemplate, error) { + data, err := batchEngineRBACTemplates.ReadFile( + "rbac_templates/" + engineType + ".yaml", + ) + if err != nil { + return nil, nil + } + var tmpl BatchEngineRBACTemplate + if err := yaml.Unmarshal(data, &tmpl); err != nil { + return nil, fmt.Errorf("failed to parse RBAC template for engine %q: %w", engineType, err) + } + return &tmpl, nil +} + +func (feast *FeastServices) reconcileBatchEngineRBAC() error { + engineType := feast.getBatchEngineType() + if engineType == "" { + return feast.deleteBatchEngineRBAC() + } + + tmpl, err := loadBatchEngineTemplate(engineType) + if err != nil { + return err + } + if tmpl == nil { + return feast.deleteBatchEngineRBAC() + } + + if tmpl.Server != nil { + if err := feast.ensureBatchEngineRole(BatchEngineFeastType, tmpl.Server.Rules); err != nil { + return err + } + if err := feast.ensureBatchEngineRoleBinding(BatchEngineFeastType, feast.initFeastSA().Name); err != nil { + return err + } + } + + if tmpl.Driver != nil { + driverSAName := feast.getBatchDriverSAName() + if tmpl.Driver.CreateServiceAccount { + if err := feast.ensureBatchDriverServiceAccount(driverSAName); err != nil { + return err + } + } + if err := feast.ensureBatchEngineRole(BatchDriverFeastType, tmpl.Driver.Rules); err != nil { + return err + } + if err := feast.ensureBatchEngineRoleBinding(BatchDriverFeastType, driverSAName); err != nil { + return err + } + } + + return nil +} + +func (feast *FeastServices) getBatchEngineType() string { + appliedSpec := feast.Handler.FeatureStore.Status.Applied + if appliedSpec.BatchEngine == nil || appliedSpec.BatchEngine.ConfigMapRef == nil { + return "" + } + + configMapKey := appliedSpec.BatchEngine.ConfigMapKey + if configMapKey == "" { + configMapKey = "config" + } + + cm, err := feast.getConfigMap(appliedSpec.BatchEngine.ConfigMapRef.Name) + if err != nil { + return "" + } + + data, ok := cm.Data[configMapKey] + if !ok { + return "" + } + + var config map[string]interface{} + if err := yaml.Unmarshal([]byte(data), &config); err != nil { + return "" + } + + engineType, _ := config["type"].(string) + return engineType +} + +func (feast *FeastServices) getBatchDriverSAName() string { + appliedSpec := feast.Handler.FeatureStore.Status.Applied + if appliedSpec.BatchEngine == nil || appliedSpec.BatchEngine.ConfigMapRef == nil { + return feast.GetFeastServiceName(BatchDriverFeastType) + } + + configMapKey := appliedSpec.BatchEngine.ConfigMapKey + if configMapKey == "" { + configMapKey = "config" + } + + cm, err := feast.getConfigMap(appliedSpec.BatchEngine.ConfigMapRef.Name) + if err != nil { + return feast.GetFeastServiceName(BatchDriverFeastType) + } + + data, ok := cm.Data[configMapKey] + if !ok { + return feast.GetFeastServiceName(BatchDriverFeastType) + } + + var config map[string]interface{} + if err := yaml.Unmarshal([]byte(data), &config); err != nil { + return feast.GetFeastServiceName(BatchDriverFeastType) + } + + if sa, ok := config["service_account"].(string); ok && sa != "" { + return sa + } + return feast.GetFeastServiceName(BatchDriverFeastType) +} + +func (feast *FeastServices) ensureBatchEngineRole(feastType FeastServiceType, rules []rbacv1.PolicyRule) error { + logger := log.FromContext(feast.Handler.Context) + role := &rbacv1.Role{ + ObjectMeta: metav1.ObjectMeta{ + Name: feast.GetFeastServiceName(feastType), + Namespace: feast.Handler.FeatureStore.Namespace, + }, + } + role.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("Role")) + + op, err := controllerutil.CreateOrUpdate(feast.Handler.Context, feast.Handler.Client, role, func() error { + role.Labels = feast.getFeastTypeLabels(feastType) + role.Rules = rules + return controllerutil.SetControllerReference(feast.Handler.FeatureStore, role, feast.Handler.Scheme) + }) + if err != nil { + return err + } + if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated { + logger.Info("Successfully reconciled", "Role", role.Name, "operation", op) + } + return nil +} + +func (feast *FeastServices) ensureBatchEngineRoleBinding(feastType FeastServiceType, saName string) error { + logger := log.FromContext(feast.Handler.Context) + roleBinding := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: feast.GetFeastServiceName(feastType), + Namespace: feast.Handler.FeatureStore.Namespace, + }, + } + roleBinding.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("RoleBinding")) + + op, err := controllerutil.CreateOrUpdate(feast.Handler.Context, feast.Handler.Client, roleBinding, func() error { + roleBinding.Labels = feast.getFeastTypeLabels(feastType) + roleBinding.Subjects = []rbacv1.Subject{{ + Kind: rbacv1.ServiceAccountKind, + Name: saName, + Namespace: feast.Handler.FeatureStore.Namespace, + }} + roleBinding.RoleRef = rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "Role", + Name: feast.GetFeastServiceName(feastType), + } + return controllerutil.SetControllerReference(feast.Handler.FeatureStore, roleBinding, feast.Handler.Scheme) + }) + if err != nil { + return err + } + if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated { + logger.Info("Successfully reconciled", "RoleBinding", roleBinding.Name, "operation", op) + } + return nil +} + +func (feast *FeastServices) ensureBatchDriverServiceAccount(saName string) error { + logger := log.FromContext(feast.Handler.Context) + sa := &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{ + Name: saName, + Namespace: feast.Handler.FeatureStore.Namespace, + }, + } + sa.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("ServiceAccount")) + + op, err := controllerutil.CreateOrUpdate(feast.Handler.Context, feast.Handler.Client, sa, func() error { + if sa.Labels == nil { + sa.Labels = map[string]string{} + } + for k, v := range feast.getFeastTypeLabels(BatchDriverFeastType) { + sa.Labels[k] = v + } + return controllerutil.SetControllerReference(feast.Handler.FeatureStore, sa, feast.Handler.Scheme) + }) + if err != nil { + return err + } + if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated { + logger.Info("Successfully reconciled", "ServiceAccount", sa.Name, "operation", op) + } + return nil +} + +func (feast *FeastServices) deleteBatchEngineRBAC() error { + serverRoleName := feast.GetFeastServiceName(BatchEngineFeastType) + driverRoleName := feast.GetFeastServiceName(BatchDriverFeastType) + ns := feast.Handler.FeatureStore.Namespace + + serverRoleBinding := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{Name: serverRoleName, Namespace: ns}, + } + serverRoleBinding.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("RoleBinding")) + if err := feast.Handler.DeleteOwnedFeastObj(serverRoleBinding); err != nil { + return err + } + + serverRole := &rbacv1.Role{ + ObjectMeta: metav1.ObjectMeta{Name: serverRoleName, Namespace: ns}, + } + serverRole.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("Role")) + if err := feast.Handler.DeleteOwnedFeastObj(serverRole); err != nil { + return err + } + + driverRoleBinding := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{Name: driverRoleName, Namespace: ns}, + } + driverRoleBinding.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("RoleBinding")) + if err := feast.Handler.DeleteOwnedFeastObj(driverRoleBinding); err != nil { + return err + } + + driverRole := &rbacv1.Role{ + ObjectMeta: metav1.ObjectMeta{Name: driverRoleName, Namespace: ns}, + } + driverRole.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("Role")) + if err := feast.Handler.DeleteOwnedFeastObj(driverRole); err != nil { + return err + } + + driverSA := &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{Name: driverRoleName, Namespace: ns}, + } + driverSA.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("ServiceAccount")) + return feast.Handler.DeleteOwnedFeastObj(driverSA) +} diff --git a/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go b/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go new file mode 100644 index 00000000000..ff73a75976c --- /dev/null +++ b/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go @@ -0,0 +1,168 @@ +/* +Copyright 2024 Feast Community. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package services + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + rbacv1 "k8s.io/api/rbac/v1" + "sigs.k8s.io/yaml" +) + +var _ = Describe("Batch Engine RBAC", func() { + + Describe("loadBatchEngineTemplate", func() { + It("should load spark_application template", func() { + tmpl, err := loadBatchEngineTemplate("spark_application") + Expect(err).NotTo(HaveOccurred()) + Expect(tmpl).NotTo(BeNil()) + Expect(tmpl.EngineType).To(Equal("spark_application")) + Expect(tmpl.Server).NotTo(BeNil()) + Expect(tmpl.Server.Rules).NotTo(BeEmpty()) + Expect(tmpl.Driver).NotTo(BeNil()) + Expect(tmpl.Driver.CreateServiceAccount).To(BeTrue()) + Expect(tmpl.Driver.Rules).NotTo(BeEmpty()) + }) + + It("should return nil for unknown engine type", func() { + tmpl, err := loadBatchEngineTemplate("nonexistent_engine") + Expect(err).NotTo(HaveOccurred()) + Expect(tmpl).To(BeNil()) + }) + + It("should contain correct server rules for spark_application", func() { + tmpl, err := loadBatchEngineTemplate("spark_application") + Expect(err).NotTo(HaveOccurred()) + + serverRules := tmpl.Server.Rules + Expect(serverRules).To(HaveLen(4)) + + hasConfigMapRule := false + hasSparkAppRule := false + hasPodListRule := false + hasPodLogRule := false + + for _, rule := range serverRules { + if containsResource(rule, "configmaps") && containsVerb(rule, "create") && containsVerb(rule, "delete") { + hasConfigMapRule = true + } + if containsResource(rule, "sparkapplications") && containsVerb(rule, "create") && containsVerb(rule, "get") && containsVerb(rule, "delete") { + hasSparkAppRule = true + } + if containsResource(rule, "pods") && containsVerb(rule, "list") { + hasPodListRule = true + } + if containsResource(rule, "pods/log") && containsVerb(rule, "get") { + hasPodLogRule = true + } + } + + Expect(hasConfigMapRule).To(BeTrue(), "should have configmaps create/delete rule") + Expect(hasSparkAppRule).To(BeTrue(), "should have sparkapplications create/get/delete rule") + Expect(hasPodListRule).To(BeTrue(), "should have pods list rule") + Expect(hasPodLogRule).To(BeTrue(), "should have pods/log get rule") + }) + + It("should contain correct driver rules for spark_application", func() { + tmpl, err := loadBatchEngineTemplate("spark_application") + Expect(err).NotTo(HaveOccurred()) + + driverRules := tmpl.Driver.Rules + Expect(driverRules).To(HaveLen(2)) + + hasPodRule := false + hasResourceRule := false + for _, rule := range driverRules { + if containsResource(rule, "pods") && + containsVerb(rule, "create") && + containsVerb(rule, "deletecollection") { + hasPodRule = true + } + if containsResource(rule, "services") && + containsResource(rule, "configmaps") && + containsResource(rule, "persistentvolumeclaims") && + containsVerb(rule, "deletecollection") { + hasResourceRule = true + } + } + Expect(hasPodRule).To(BeTrue(), "should have pods CRUD + deletecollection rule") + Expect(hasResourceRule).To(BeTrue(), "should have services/configmaps/PVCs CRUD + deletecollection rule") + }) + }) + + Describe("BatchEngineRBACTemplate YAML parsing", func() { + It("should correctly unmarshal a template", func() { + yamlData := ` +engine_type: test_engine +server: + rules: + - apiGroups: [""] + resources: ["secrets"] + verbs: ["get", "list"] +driver: + create_service_account: true + rules: + - apiGroups: [""] + resources: ["pods"] + verbs: ["get"] +` + var tmpl BatchEngineRBACTemplate + err := yaml.Unmarshal([]byte(yamlData), &tmpl) + Expect(err).NotTo(HaveOccurred()) + Expect(tmpl.EngineType).To(Equal("test_engine")) + Expect(tmpl.Server).NotTo(BeNil()) + Expect(tmpl.Server.Rules).To(HaveLen(1)) + Expect(tmpl.Driver).NotTo(BeNil()) + Expect(tmpl.Driver.CreateServiceAccount).To(BeTrue()) + Expect(tmpl.Driver.Rules).To(HaveLen(1)) + }) + + It("should handle server-only template (no driver)", func() { + yamlData := ` +engine_type: server_only +server: + rules: + - apiGroups: ["batch"] + resources: ["jobs"] + verbs: ["create", "delete"] +` + var tmpl BatchEngineRBACTemplate + err := yaml.Unmarshal([]byte(yamlData), &tmpl) + Expect(err).NotTo(HaveOccurred()) + Expect(tmpl.Server).NotTo(BeNil()) + Expect(tmpl.Driver).To(BeNil()) + }) + }) +}) + +func containsResource(rule rbacv1.PolicyRule, resource string) bool { + for _, r := range rule.Resources { + if r == resource { + return true + } + } + return false +} + +func containsVerb(rule rbacv1.PolicyRule, verb string) bool { + for _, v := range rule.Verbs { + if v == verb { + return true + } + } + return false +} diff --git a/infra/feast-operator/internal/controller/services/rbac_templates/spark_application.yaml b/infra/feast-operator/internal/controller/services/rbac_templates/spark_application.yaml new file mode 100644 index 00000000000..c03e0f3db57 --- /dev/null +++ b/infra/feast-operator/internal/controller/services/rbac_templates/spark_application.yaml @@ -0,0 +1,26 @@ +engine_type: spark_application + +server: + rules: + - apiGroups: [""] + resources: ["configmaps"] + verbs: ["create", "delete"] + - apiGroups: ["sparkoperator.k8s.io"] + resources: ["sparkapplications"] + verbs: ["create", "get", "delete"] + - apiGroups: [""] + resources: ["pods"] + verbs: ["list"] + - apiGroups: [""] + resources: ["pods/log"] + verbs: ["get"] + +driver: + create_service_account: true + rules: + - apiGroups: [""] + resources: ["pods"] + verbs: ["create", "get", "list", "watch", "delete", "deletecollection"] + - apiGroups: [""] + resources: ["services", "configmaps", "persistentvolumeclaims"] + verbs: ["create", "get", "list", "watch", "delete", "deletecollection"] diff --git a/infra/feast-operator/internal/controller/services/services.go b/infra/feast-operator/internal/controller/services/services.go index c6271c28d3a..4d5429bbbde 100644 --- a/infra/feast-operator/internal/controller/services/services.go +++ b/infra/feast-operator/internal/controller/services/services.go @@ -79,6 +79,9 @@ func (feast *FeastServices) Deploy() error { if err := feast.createServiceAccount(); err != nil { return err } + if err := feast.reconcileBatchEngineRBAC(); err != nil { + return err + } if err := feast.createDeployment(); err != nil { return err } From d1b3f509616baa3fe9c602c9262701d597ec9aa7 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Wed, 15 Jul 2026 00:11:16 +0530 Subject: [PATCH 2/5] fix(operator): bake default batch-driver SA into feature_store.yaml When batch_engine.service_account is omitted, inject feast--batch-driver so the SparkApplication matches the SA/RBAC the operator already creates (R2-2). Co-authored-by: Cursor Signed-off-by: Aniket Paluskar --- .../controller/services/batch_engine_rbac.go | 16 +++++-- .../services/batch_engine_rbac_test.go | 27 ++++++++++++ .../controller/services/repo_config.go | 6 ++- .../controller/services/repo_config_test.go | 43 +++++++++++++++++++ 4 files changed, 87 insertions(+), 5 deletions(-) diff --git a/infra/feast-operator/internal/controller/services/batch_engine_rbac.go b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go index 2d522db8b6b..ae241e6af25 100644 --- a/infra/feast-operator/internal/controller/services/batch_engine_rbac.go +++ b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go @@ -20,6 +20,7 @@ import ( "embed" "fmt" + feastdevv1 "github.com/feast-dev/feast/infra/feast-operator/api/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -139,6 +140,16 @@ func (feast *FeastServices) getBatchEngineType() string { return engineType } +// resolveBatchDriverSAName returns the ServiceAccount name for the Spark driver. +// If batch engine config sets a non-empty service_account, that value wins. +// Otherwise defaults to feast--batch-driver (same name used for RBAC). +func resolveBatchDriverSAName(featureStore *feastdevv1.FeatureStore, config map[string]interface{}) string { + if sa, ok := config["service_account"].(string); ok && sa != "" { + return sa + } + return GetFeastServiceName(featureStore, BatchDriverFeastType) +} + func (feast *FeastServices) getBatchDriverSAName() string { appliedSpec := feast.Handler.FeatureStore.Status.Applied if appliedSpec.BatchEngine == nil || appliedSpec.BatchEngine.ConfigMapRef == nil { @@ -165,10 +176,7 @@ func (feast *FeastServices) getBatchDriverSAName() string { return feast.GetFeastServiceName(BatchDriverFeastType) } - if sa, ok := config["service_account"].(string); ok && sa != "" { - return sa - } - return feast.GetFeastServiceName(BatchDriverFeastType) + return resolveBatchDriverSAName(feast.Handler.FeatureStore, config) } func (feast *FeastServices) ensureBatchEngineRole(feastType FeastServiceType, rules []rbacv1.PolicyRule) error { diff --git a/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go b/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go index ff73a75976c..9d792c6f923 100644 --- a/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go +++ b/infra/feast-operator/internal/controller/services/batch_engine_rbac_test.go @@ -20,7 +20,10 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/yaml" + + feastdevv1 "github.com/feast-dev/feast/infra/feast-operator/api/v1" ) var _ = Describe("Batch Engine RBAC", func() { @@ -149,6 +152,30 @@ server: }) }) +var _ = Describe("resolveBatchDriverSAName", func() { + It("defaults to feast--batch-driver when service_account is omitted", func() { + fs := &feastdevv1.FeatureStore{ObjectMeta: metav1.ObjectMeta{Name: "spark-pg-e2e", Namespace: "feast-spark"}} + Expect(resolveBatchDriverSAName(fs, map[string]interface{}{ + "type": "spark_application", + "image": "quay.io/example/driver:v1", + })).To(Equal("feast-spark-pg-e2e-batch-driver")) + }) + + It("defaults when service_account is empty string", func() { + fs := &feastdevv1.FeatureStore{ObjectMeta: metav1.ObjectMeta{Name: "spark-pg-e2e"}} + Expect(resolveBatchDriverSAName(fs, map[string]interface{}{ + "service_account": "", + })).To(Equal("feast-spark-pg-e2e-batch-driver")) + }) + + It("keeps an explicit service_account override", func() { + fs := &feastdevv1.FeatureStore{ObjectMeta: metav1.ObjectMeta{Name: "spark-pg-e2e"}} + Expect(resolveBatchDriverSAName(fs, map[string]interface{}{ + "service_account": "my-custom-driver", + })).To(Equal("my-custom-driver")) + }) +}) + func containsResource(rule rbacv1.PolicyRule, resource string) bool { for _, r := range rule.Resources { if r == resource { diff --git a/infra/feast-operator/internal/controller/services/repo_config.go b/infra/feast-operator/internal/controller/services/repo_config.go index 21d04db7c0e..24bbf08efae 100644 --- a/infra/feast-operator/internal/controller/services/repo_config.go +++ b/infra/feast-operator/internal/controller/services/repo_config.go @@ -85,7 +85,7 @@ func getServiceRepoConfig( } if appliedSpec.BatchEngine != nil { - err := setRepoConfigBatchEngine(appliedSpec.BatchEngine, configMapExtractionFunc, &repoConfig) + err := setRepoConfigBatchEngine(featureStore, appliedSpec.BatchEngine, configMapExtractionFunc, &repoConfig) if err != nil { return repoConfig, err } @@ -342,6 +342,7 @@ func setRepoConfigOffline(services *feastdevv1.FeatureStoreServices, secretExtra } func setRepoConfigBatchEngine( + featureStore *feastdevv1.FeatureStore, batchEngineConfig *feastdevv1.BatchEngineConfig, configMapExtractionFunc func(configMapRef string, configMapKey string) (map[string]interface{}, error), repoConfig *RepoConfig) error { @@ -362,6 +363,9 @@ func setRepoConfigBatchEngine( return fmt.Errorf("batch engine config must contain 'type' field") } delete(config, "type") + // Inject service_account when omitted so baked feature_store.yaml matches the + // SA/RoleBinding created by reconcileBatchEngineRBAC (R2-2). + config["service_account"] = resolveBatchDriverSAName(featureStore, config) repoConfig.BatchEngine = &ComputeEngineConfig{ Type: engineType, Parameters: config, diff --git a/infra/feast-operator/internal/controller/services/repo_config_test.go b/infra/feast-operator/internal/controller/services/repo_config_test.go index 5ae22a795f6..2a2f806a3ef 100644 --- a/infra/feast-operator/internal/controller/services/repo_config_test.go +++ b/infra/feast-operator/internal/controller/services/repo_config_test.go @@ -673,6 +673,49 @@ var _ = Describe("Repo Config", func() { Expect(repoConfig.Materialization).To(BeNil()) Expect(repoConfig.OpenLineage).To(BeNil()) }) + + It("should inject default batch_engine.service_account when ConfigMap omits it", func() { + featureStore := minimalFeatureStore() + featureStore.Name = "spark-pg-e2e" + featureStore.Spec.BatchEngine = &feastdevv1.BatchEngineConfig{ + ConfigMapRef: &corev1.LocalObjectReference{Name: "spark-pg-batch-engine"}, + } + ApplyDefaultsToStatus(featureStore) + + extractCM := func(configMapRef string, configMapKey string) (map[string]interface{}, error) { + return map[string]interface{}{ + "type": "spark_application", + "image": "quay.io/example/feast-spark-driver:v6", + // service_account intentionally omitted + }, nil + } + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, extractCM, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.BatchEngine).NotTo(BeNil()) + Expect(repoConfig.BatchEngine.Type).To(Equal("spark_application")) + Expect(repoConfig.BatchEngine.Parameters["service_account"]).To(Equal("feast-spark-pg-e2e-batch-driver")) + }) + + It("should preserve explicit batch_engine.service_account from ConfigMap", func() { + featureStore := minimalFeatureStore() + featureStore.Name = "spark-pg-e2e" + featureStore.Spec.BatchEngine = &feastdevv1.BatchEngineConfig{ + ConfigMapRef: &corev1.LocalObjectReference{Name: "spark-pg-batch-engine"}, + } + ApplyDefaultsToStatus(featureStore) + + extractCM := func(configMapRef string, configMapKey string) (map[string]interface{}, error) { + return map[string]interface{}{ + "type": "spark_application", + "service_account": "my-custom-driver", + }, nil + } + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, extractCM, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.BatchEngine.Parameters["service_account"]).To(Equal("my-custom-driver")) + }) }) It("should fail to create the repo configs", func() { featureStore := minimalFeatureStore() From a769e0489ea1b665a42872bf24e62fd6bbaa17d4 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Thu, 16 Jul 2026 15:28:18 +0530 Subject: [PATCH 3/5] fix(operator): inject batch-driver SA only for spark_application Limit service_account baking to spark_application so other batch engines are not given an unexpected parameter. Signed-off-by: Aniket Paluskar --- .../controller/services/repo_config.go | 9 +++++--- .../controller/services/repo_config_test.go | 23 +++++++++++++++++++ 2 files changed, 29 insertions(+), 3 deletions(-) diff --git a/infra/feast-operator/internal/controller/services/repo_config.go b/infra/feast-operator/internal/controller/services/repo_config.go index 24bbf08efae..5df6bba3fbc 100644 --- a/infra/feast-operator/internal/controller/services/repo_config.go +++ b/infra/feast-operator/internal/controller/services/repo_config.go @@ -363,9 +363,12 @@ func setRepoConfigBatchEngine( return fmt.Errorf("batch engine config must contain 'type' field") } delete(config, "type") - // Inject service_account when omitted so baked feature_store.yaml matches the - // SA/RoleBinding created by reconcileBatchEngineRBAC (R2-2). - config["service_account"] = resolveBatchDriverSAName(featureStore, config) + // Inject service_account only for spark_application so baked feature_store.yaml + // matches the SA/RoleBinding created by reconcileBatchEngineRBAC. + // Other batch engines are left unchanged. + if engineType == "spark_application" { + config["service_account"] = resolveBatchDriverSAName(featureStore, config) + } repoConfig.BatchEngine = &ComputeEngineConfig{ Type: engineType, Parameters: config, diff --git a/infra/feast-operator/internal/controller/services/repo_config_test.go b/infra/feast-operator/internal/controller/services/repo_config_test.go index 2a2f806a3ef..333584beeab 100644 --- a/infra/feast-operator/internal/controller/services/repo_config_test.go +++ b/infra/feast-operator/internal/controller/services/repo_config_test.go @@ -716,6 +716,29 @@ var _ = Describe("Repo Config", func() { Expect(err).NotTo(HaveOccurred()) Expect(repoConfig.BatchEngine.Parameters["service_account"]).To(Equal("my-custom-driver")) }) + + It("should not inject service_account for non-spark_application batch engines", func() { + featureStore := minimalFeatureStore() + featureStore.Name = "spark-pg-e2e" + featureStore.Spec.BatchEngine = &feastdevv1.BatchEngineConfig{ + ConfigMapRef: &corev1.LocalObjectReference{Name: "other-batch-engine"}, + } + ApplyDefaultsToStatus(featureStore) + + extractCM := func(configMapRef string, configMapKey string) (map[string]interface{}, error) { + return map[string]interface{}{ + "type": "spark", + // no service_account — must stay omitted for non-spark_application + }, nil + } + + repoConfig, err := getServiceRepoConfig(featureStore, emptyMockExtractConfigFromSecret, extractCM, false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.BatchEngine).NotTo(BeNil()) + Expect(repoConfig.BatchEngine.Type).To(Equal("spark")) + _, hasSA := repoConfig.BatchEngine.Parameters["service_account"] + Expect(hasSA).To(BeFalse()) + }) }) It("should fail to create the repo configs", func() { featureStore := minimalFeatureStore() From f8495699d16f0686568d2f646dea45ff4210c438 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Fri, 17 Jul 2026 18:48:48 +0530 Subject: [PATCH 4/5] chore(operator): fix CI fmt and secrets baseline gofmt repo_config_test map literals and refresh .secrets.baseline line numbers after batch-engine RBAC wiring in services.go. Signed-off-by: Aniket Paluskar --- .secrets.baseline | 4 ++-- .../internal/controller/services/repo_config_test.go | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index f6a5bbb4b0a..57fa3236f27 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1172,7 +1172,7 @@ "filename": "infra/feast-operator/internal/controller/services/services.go", "hashed_secret": "36dc326eb15c7bdd8d91a6b87905bcea20b637d1", "is_verified": false, - "line_number": 180 + "line_number": 183 } ], "infra/feast-operator/internal/controller/services/tls_test.go": [ @@ -1555,5 +1555,5 @@ } ] }, - "generated_at": "2026-07-13T15:26:35Z" + "generated_at": "2026-07-17T12:34:13Z" } diff --git a/infra/feast-operator/internal/controller/services/repo_config_test.go b/infra/feast-operator/internal/controller/services/repo_config_test.go index 333584beeab..89941c6beb5 100644 --- a/infra/feast-operator/internal/controller/services/repo_config_test.go +++ b/infra/feast-operator/internal/controller/services/repo_config_test.go @@ -707,8 +707,8 @@ var _ = Describe("Repo Config", func() { extractCM := func(configMapRef string, configMapKey string) (map[string]interface{}, error) { return map[string]interface{}{ - "type": "spark_application", - "service_account": "my-custom-driver", + "type": "spark_application", + "service_account": "my-custom-driver", }, nil } From 3c00206f2b06837b44f27db871c925c96973aeed Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Sat, 18 Jul 2026 14:20:11 +0530 Subject: [PATCH 5/5] refactor(operator): parse batch-engine ConfigMap once per reconcile Share getBatchEngineConfig() for engine type and driver SA resolution instead of fetching/parsing the same ConfigMap twice. Signed-off-by: Aniket Paluskar --- .../controller/services/batch_engine_rbac.go | 58 ++++++------------- 1 file changed, 17 insertions(+), 41 deletions(-) diff --git a/infra/feast-operator/internal/controller/services/batch_engine_rbac.go b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go index ae241e6af25..97afcd3b48d 100644 --- a/infra/feast-operator/internal/controller/services/batch_engine_rbac.go +++ b/infra/feast-operator/internal/controller/services/batch_engine_rbac.go @@ -70,7 +70,12 @@ func loadBatchEngineTemplate(engineType string) (*BatchEngineRBACTemplate, error } func (feast *FeastServices) reconcileBatchEngineRBAC() error { - engineType := feast.getBatchEngineType() + config, ok := feast.getBatchEngineConfig() + if !ok { + return feast.deleteBatchEngineRBAC() + } + + engineType, _ := config["type"].(string) if engineType == "" { return feast.deleteBatchEngineRBAC() } @@ -93,7 +98,7 @@ func (feast *FeastServices) reconcileBatchEngineRBAC() error { } if tmpl.Driver != nil { - driverSAName := feast.getBatchDriverSAName() + driverSAName := resolveBatchDriverSAName(feast.Handler.FeatureStore, config) if tmpl.Driver.CreateServiceAccount { if err := feast.ensureBatchDriverServiceAccount(driverSAName); err != nil { return err @@ -110,10 +115,12 @@ func (feast *FeastServices) reconcileBatchEngineRBAC() error { return nil } -func (feast *FeastServices) getBatchEngineType() string { +// getBatchEngineConfig returns the parsed batch-engine ConfigMap data. +// ok=false means no batch engine is configured or the ConfigMap is unreadable. +func (feast *FeastServices) getBatchEngineConfig() (map[string]interface{}, bool) { appliedSpec := feast.Handler.FeatureStore.Status.Applied if appliedSpec.BatchEngine == nil || appliedSpec.BatchEngine.ConfigMapRef == nil { - return "" + return nil, false } configMapKey := appliedSpec.BatchEngine.ConfigMapKey @@ -123,21 +130,19 @@ func (feast *FeastServices) getBatchEngineType() string { cm, err := feast.getConfigMap(appliedSpec.BatchEngine.ConfigMapRef.Name) if err != nil { - return "" + return nil, false } - data, ok := cm.Data[configMapKey] - if !ok { - return "" + data, found := cm.Data[configMapKey] + if !found { + return nil, false } var config map[string]interface{} if err := yaml.Unmarshal([]byte(data), &config); err != nil { - return "" + return nil, false } - - engineType, _ := config["type"].(string) - return engineType + return config, true } // resolveBatchDriverSAName returns the ServiceAccount name for the Spark driver. @@ -150,35 +155,6 @@ func resolveBatchDriverSAName(featureStore *feastdevv1.FeatureStore, config map[ return GetFeastServiceName(featureStore, BatchDriverFeastType) } -func (feast *FeastServices) getBatchDriverSAName() string { - appliedSpec := feast.Handler.FeatureStore.Status.Applied - if appliedSpec.BatchEngine == nil || appliedSpec.BatchEngine.ConfigMapRef == nil { - return feast.GetFeastServiceName(BatchDriverFeastType) - } - - configMapKey := appliedSpec.BatchEngine.ConfigMapKey - if configMapKey == "" { - configMapKey = "config" - } - - cm, err := feast.getConfigMap(appliedSpec.BatchEngine.ConfigMapRef.Name) - if err != nil { - return feast.GetFeastServiceName(BatchDriverFeastType) - } - - data, ok := cm.Data[configMapKey] - if !ok { - return feast.GetFeastServiceName(BatchDriverFeastType) - } - - var config map[string]interface{} - if err := yaml.Unmarshal([]byte(data), &config); err != nil { - return feast.GetFeastServiceName(BatchDriverFeastType) - } - - return resolveBatchDriverSAName(feast.Handler.FeatureStore, config) -} - func (feast *FeastServices) ensureBatchEngineRole(feastType FeastServiceType, rules []rbacv1.PolicyRule) error { logger := log.FromContext(feast.Handler.Context) role := &rbacv1.Role{