mirror of
https://github.com/hanzo-ml/spark-operator.git
synced 2026-10-30 20:39:07 +00:00
shruthi sankepellyandGitHub
70d5ba061e
* Fix driver ingress annotations and add deep copy support Use driverIngressOptions annotations and TLS instead of sparkUIOptions, and prevent map mutation by deep copying before modifications. - Pass driverIngressConfiguration.IngressAnnotations and IngressTLS instead of empty maps to fix ignored user annotations - Resolve precedence in createDriverIngress: per-entry values take priority, fall back to operator defaults when empty - Deep copy annotations in V1/Legacy to prevent mutation of source maps (either user's SparkApplication spec or operator defaults) - Precedence resolution happens once in createDriverIngress, not duplicated in V1/Legacy (they receive already-resolved values) Fixes mutation bug where rewrite-target annotation for subpath routing would leak into the user's SparkApplication or operator's shared defaults. Signed-off-by: Shruthi Sankepelly <[email protected]> * Add tests for driver ingress annotation precedence and mutation prevention Add comprehensive tests to verify the annotation precedence logic and deep copy implementation: - TestDriverIngressAnnotationPrecedence: Verifies that user annotations take precedence over operator defaults, with proper fallback behavior - TestDriverIngressAnnotationMutationPrevention: Verifies that the deep copy prevents mutation of source maps (user spec and operator defaults) - Fix existing test to pass new defaultIngressTLS and defaultIngressAnnotations parameters Tests query the fake Kubernetes client to verify the actual created Ingress resources, ensuring the precedence logic and mutation prevention work correctly in practice. Signed-off-by: Shruthi Sankepelly <[email protected]> * Run go fmt on test file Signed-off-by: Shruthi Sankepelly <[email protected]> --------- Signed-off-by: Shruthi Sankepelly <[email protected]>
128 lines
4.3 KiB
Go
128 lines
4.3 KiB
Go
/*
|
|
Copyright 2017 Google LLC
|
|
|
|
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 sparkapplication
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/url"
|
|
"strconv"
|
|
|
|
networkingv1 "k8s.io/api/networking/v1"
|
|
|
|
"github.com/kubeflow/spark-operator/v2/api/v1beta2"
|
|
"github.com/kubeflow/spark-operator/v2/pkg/common"
|
|
"github.com/kubeflow/spark-operator/v2/pkg/util"
|
|
)
|
|
|
|
func (r *Reconciler) configWebUI(_ context.Context, app *v1beta2.SparkApplication) error {
|
|
if !r.options.EnableUIService || r.options.IngressURLFormat == "" {
|
|
return nil
|
|
}
|
|
|
|
ingressURL, err := getDriverIngressURL(r.options.IngressURLFormat, app)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get ingress url: %v", err)
|
|
}
|
|
if ingressURL.Path == "" {
|
|
return nil
|
|
}
|
|
|
|
if app.Spec.SparkConf == nil {
|
|
app.Spec.SparkConf = make(map[string]string)
|
|
}
|
|
util.SetIfNotExists(app.Spec.SparkConf, common.SparkUIProxyBase, ingressURL.Path)
|
|
util.SetIfNotExists(app.Spec.SparkConf, common.SparkUIProxyRedirectURI, "/")
|
|
return nil
|
|
}
|
|
|
|
func (r *Reconciler) createWebUIService(ctx context.Context, app *v1beta2.SparkApplication) (*SparkService, error) {
|
|
portName := getWebUIServicePortName(app)
|
|
port, err := getWebUIServicePort(app)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid Spark UI servicePort: %d", port)
|
|
}
|
|
|
|
targetPort, err := getWebUITargetPort(app)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid Spark UI targetPort: %d", targetPort)
|
|
}
|
|
|
|
serviceName := util.GetDefaultUIServiceName(app)
|
|
serviceType := util.GetWebUIServiceType(app)
|
|
serviceLabels := util.GetWebUIServiceLabels(app)
|
|
serviceAnnotations := util.GetWebUIServiceAnnotations(app)
|
|
|
|
return r.createDriverIngressService(ctx, app, portName, port, targetPort, serviceName, serviceType, serviceAnnotations, serviceLabels)
|
|
}
|
|
|
|
func (r *Reconciler) createWebUIIngress(ctx context.Context, app *v1beta2.SparkApplication, service SparkService, ingressURL *url.URL, ingressClassName string, defaultIngressTLS []networkingv1.IngressTLS, defaultIngressAnnotations map[string]string) (*SparkIngress, error) {
|
|
ingressName := util.GetDefaultUIIngressName(app)
|
|
|
|
// Resolve annotations and TLS: per-entry values take precedence, falling back to defaults
|
|
ingressAnnotations := util.GetWebUIIngressAnnotations(app)
|
|
if len(ingressAnnotations) == 0 {
|
|
ingressAnnotations = defaultIngressAnnotations
|
|
}
|
|
ingressTLS := util.GetWebUIIngressTLS(app)
|
|
if len(ingressTLS) == 0 {
|
|
ingressTLS = defaultIngressTLS
|
|
}
|
|
|
|
if util.IngressCapabilities.Has("networking.k8s.io/v1") {
|
|
return r.createDriverIngressV1(ctx, app, service, ingressName, ingressURL, ingressClassName, ingressTLS, ingressAnnotations)
|
|
}
|
|
return r.createDriverIngressLegacy(ctx, app, service, ingressName, ingressURL, ingressTLS, ingressAnnotations)
|
|
}
|
|
|
|
func getWebUIServicePortName(app *v1beta2.SparkApplication) string {
|
|
if app.Spec.SparkUIOptions == nil {
|
|
return common.DefaultSparkWebUIPortName
|
|
}
|
|
portName := app.Spec.SparkUIOptions.ServicePortName
|
|
if portName != nil {
|
|
return *portName
|
|
}
|
|
return common.DefaultSparkWebUIPortName
|
|
}
|
|
|
|
func getWebUIServicePort(app *v1beta2.SparkApplication) (int32, error) {
|
|
if app.Spec.SparkUIOptions == nil {
|
|
return getWebUITargetPort(app)
|
|
}
|
|
port := app.Spec.SparkUIOptions.ServicePort
|
|
if port != nil {
|
|
return *port, nil
|
|
}
|
|
return common.DefaultSparkWebUIPort, nil
|
|
}
|
|
|
|
// getWebUITargetPort attempts to get the Spark web UI port from configuration property spark.ui.port
|
|
// in Spec.SparkConf if it is present, otherwise the default port is returned.
|
|
// Note that we don't attempt to get the port from Spec.SparkConfigMap.
|
|
func getWebUITargetPort(app *v1beta2.SparkApplication) (int32, error) {
|
|
portStr, ok := app.Spec.SparkConf[common.SparkUIPortKey]
|
|
if !ok {
|
|
return common.DefaultSparkWebUIPort, nil
|
|
}
|
|
port, err := strconv.ParseInt(portStr, 10, 32)
|
|
if err != nil {
|
|
return common.DefaultSparkWebUIPort, nil
|
|
}
|
|
return int32(port), nil
|
|
}
|