feat: add support for customizing connect service (#2709)

* Add new field .spec.server.service to SparkConnect

Signed-off-by: Yi Chen <github@chenyicn.net>

* Add support for customing Spark connect server service

Signed-off-by: Yi Chen <github@chenyicn.net>

* Set Kubernetes client to suppress warnings in REST config

Signed-off-by: Yi Chen <github@chenyicn.net>

---------

Signed-off-by: Yi Chen <github@chenyicn.net>

Author: ChenYi015

api/v1alpha1/sparkconnect_types.go

@@ -87,6 +87,10 @@ type SparkConnectSpec struct {
 // ServerSpec is specification of the Spark connect server.
 type ServerSpec struct {
 	SparkPodSpec `json:",inline"`
+
+	// Service exposes the Spark connect server.
+	// +optional
+	Service *corev1.Service `json:"service,omitempty"`
 }
 
 // ExecutorSpec is specification of the executor.

api/v1alpha1/zz_generated.deepcopy.go

@@ -28,6 +28,7 @@ import (
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
+	"k8s.io/client-go/rest"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -204,6 +205,7 @@ func start() {
 
 	// Create the client rest config. Use kubeConfig if given, otherwise assume in-cluster.
 	cfg, err := ctrl.GetConfig()
+	cfg.WarningHandler = rest.NoWarnings{}
 	if err != nil {
 		logger.Error(err, "failed to get kube config")
 		os.Exit(1)

charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml

@@ -447,16 +447,18 @@ func (r *Reconciler) mutateServerPod(_ context.Context, conn *v1alpha1.SparkConn
 // createOrUpdateServerService creates or updates the server service for the SparkConnect resource.
 func (r *Reconciler) createOrUpdateServerService(ctx context.Context, conn *v1alpha1.SparkConnect) error {
 	logger := ctrl.LoggerFrom(ctx)
-	logger.V(1).Info("Create or update server service")
 
-	svc := &corev1.Service{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      GetServerServiceName(conn),
-			Namespace: conn.Namespace,
-		},
+	// Use the service specified in the server spec if provided.
+	svc := conn.Spec.Server.Service
+	if svc == nil {
+		svc = &corev1.Service{}
 	}
+	svc.Name = GetServerServiceName(conn)
+	// Namespace provided by user will be ignored.
+	svc.Namespace = conn.Namespace
 
-	_, err := controllerutil.CreateOrUpdate(ctx, r.client, svc, func() error {
+	// Create or update server service.
+	opResult, err := controllerutil.CreateOrUpdate(ctx, r.client, svc, func() error {
 		if err := r.mutateServerService(ctx, conn, svc); err != nil {
 			return fmt.Errorf("failed to mutate server service: %v", err)
 		}
@@ -465,6 +467,12 @@ func (r *Reconciler) createOrUpdateServerService(ctx context.Context, conn *v1al
 	if err != nil {
 		return fmt.Errorf("failed to create or update server service: %v", err)
 	}
+	switch opResult {
+	case controllerutil.OperationResultCreated:
+		logger.Info("Server service created")
+	case controllerutil.OperationResultUpdated:
+		logger.Info("Server service updated")
+	}
 
 	// Update SparkConnect status.
 	conn.Status.Server.ServiceName = svc.Name
@@ -475,7 +483,6 @@ func (r *Reconciler) createOrUpdateServerService(ctx context.Context, conn *v1al
 // mutateServerService mutates the server service for the SparkConnect resource.
 func (r *Reconciler) mutateServerService(_ context.Context, conn *v1alpha1.SparkConnect, svc *corev1.Service) error {
 	if svc.CreationTimestamp.IsZero() {
-		svc.Spec.Type = corev1.ServiceTypeClusterIP
 		svc.Spec.Ports = []corev1.ServicePort{
 			{
 				Name:       "driver-rpc",

cmd/operator/controller/start.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2025 The Kubeflow authors.
+
+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
+
+    https://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 sparkconnect
+
+import (
+	"fmt"
+	"path/filepath"
+	"runtime"
+	"testing"
+
+	"github.com/kubeflow/spark-operator/v2/api/v1alpha1"
+	"github.com/kubeflow/spark-operator/v2/api/v1beta2"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	// +kubebuilder:scaffold:imports
+)
+
+// These tests use Ginkgo (BDD-style Go testing framework). Refer to
+// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
+
+var cfg *rest.Config
+var k8sClient client.Client
+var testEnv *envtest.Environment
+
+func TestSparkConnectController(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "SparkConnect Controller Suite")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "..", "config", "crd", "bases")},
+		ErrorIfCRDPathMissing: true,
+
+		// The BinaryAssetsDirectory is only required if you want to run the tests directly
+		// without call the makefile target test. If not informed it will look for the
+		// default path defined in controller-runtime which is /usr/local/kubebuilder/.
+		// Note that you must have the required binaries setup under the bin directory to perform
+		// the tests directly. When we run make test it will be setup and used automatically.
+		BinaryAssetsDirectory: filepath.Join("..", "..", "..", "bin", "k8s",
+			fmt.Sprintf("1.32.0-%s-%s", runtime.GOOS, runtime.GOARCH)),
+	}
+
+	var err error
+	// cfg is defined in this file globally.
+	cfg, err = testEnv.Start()
+	Expect(err).NotTo(HaveOccurred())
+	Expect(cfg).NotTo(BeNil())
+
+	Expect(v1alpha1.AddToScheme(scheme.Scheme)).Should(Succeed())
+	Expect(v1beta2.AddToScheme(scheme.Scheme)).Should(Succeed())
+	// +kubebuilder:scaffold:scheme
+
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	Expect(err).NotTo(HaveOccurred())
+	Expect(k8sClient).NotTo(BeNil())
+
+})
+
+var _ = AfterSuite(func() {
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).NotTo(HaveOccurred())
+})

config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml

@@ -65,6 +65,13 @@ func GetServerPodName(conn *v1alpha1.SparkConnect) string {
 
 // GetServerServiceName returns the name of the server service for SparkConnect.
 func GetServerServiceName(conn *v1alpha1.SparkConnect) string {
+	// Use the service specified in the server spec if provided.
+	svc := conn.Spec.Server.Service
+	if svc != nil {
+		return svc.Name
+	}
+
+	// Otherwise, use the default service name.
 	return fmt.Sprintf("%s-server", conn.Name)
 }
 

internal/controller/sparkconnect/reconciler.go

@@ -0,0 +1,125 @@
+/*
+Copyright 2025 The kubeflow authors.
+
+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
+
+    https://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 sparkconnect
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/kubeflow/spark-operator/v2/api/v1alpha1"
+	"github.com/kubeflow/spark-operator/v2/pkg/common"
+)
+
+var _ = Describe("Util functions", func() {
+	var conn *v1alpha1.SparkConnect
+
+	BeforeEach(func() {
+		conn = &v1alpha1.SparkConnect{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-spark-connect",
+				Namespace: "test-namespace",
+			},
+			Spec: v1alpha1.SparkConnectSpec{
+				SparkVersion: "4.0.0",
+				Server: v1alpha1.ServerSpec{
+					SparkPodSpec: v1alpha1.SparkPodSpec{},
+				},
+				Executor: v1alpha1.ExecutorSpec{
+					SparkPodSpec: v1alpha1.SparkPodSpec{},
+				},
+			},
+		}
+	})
+
+	Context("GetCommonLabels", func() {
+		It("should return correct common labels", func() {
+			labels := GetCommonLabels(conn)
+			Expect(labels).To(HaveLen(2))
+			Expect(labels).To(HaveKeyWithValue(common.LabelCreatedBySparkOperator, "true"))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkConnectName, "test-spark-connect"))
+		})
+	})
+
+	Context("GetServerSelectorLabels", func() {
+		It("should return correct server selector labels", func() {
+			labels := GetServerSelectorLabels(conn)
+			Expect(labels).To(HaveLen(4))
+			Expect(labels).To(HaveKeyWithValue(common.LabelLaunchedBySparkOperator, "true"))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkConnectName, "test-spark-connect"))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkRole, common.SparkRoleConnectServer))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkVersion, "4.0.0"))
+		})
+	})
+
+	Context("GetExecutorSelectorLabels", func() {
+		It("should return correct executor selector labels", func() {
+			labels := GetExecutorSelectorLabels(conn)
+			Expect(labels).To(HaveLen(3))
+			Expect(labels).To(HaveKeyWithValue(common.LabelLaunchedBySparkOperator, "true"))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkConnectName, "test-spark-connect"))
+			Expect(labels).To(HaveKeyWithValue(common.LabelSparkRole, common.SparkRoleExecutor))
+		})
+	})
+
+	Context("GetConfigMapName", func() {
+		It("should return correct config map name", func() {
+			name := GetConfigMapName(conn)
+			Expect(name).To(Equal("test-spark-connect-conf"))
+		})
+	})
+
+	Context("GetServerPodName", func() {
+		It("should return correct server pod name", func() {
+			name := GetServerPodName(conn)
+			Expect(name).To(Equal("test-spark-connect-server"))
+		})
+	})
+
+	Context("GetServerServiceName", func() {
+		When("service is not specified in server spec", func() {
+			It("should return default server service name", func() {
+				name := GetServerServiceName(conn)
+				Expect(name).To(Equal("test-spark-connect-server"))
+			})
+		})
+
+		When("service is specified in server spec", func() {
+			BeforeEach(func() {
+				conn.Spec.Server.Service = &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "custom-service-name",
+					},
+				}
+			})
+
+			It("should return the specified service name", func() {
+				name := GetServerServiceName(conn)
+				Expect(name).To(Equal("custom-service-name"))
+			})
+		})
+	})
+
+	Context("GetServerServiceHost", func() {
+		It("should return correct server service host", func() {
+			host := GetServerServiceHost(conn)
+			Expect(host).To(Equal("test-spark-connect-server.test-namespace.svc.cluster.local"))
+		})
+	})
+})