diff --git a/flyteplugins/go/tasks/plugins/k8s/spark/config.go b/flyteplugins/go/tasks/plugins/k8s/spark/config.go index 413c0d03333..9c6c2f3c90c 100644 --- a/flyteplugins/go/tasks/plugins/k8s/spark/config.go +++ b/flyteplugins/go/tasks/plugins/k8s/spark/config.go @@ -9,6 +9,7 @@ import ( var ( defaultConfig = &Config{ + EnablePodTemplate: true, LogConfig: LogConfig{ Mixed: logs.LogConfig{ IsKubernetesEnabled: true, @@ -26,6 +27,7 @@ type Config struct { SparkHistoryServerURL string `json:"spark-history-server-url" pflag:",URL for SparkHistory Server that each job will publish the execution history to."` Features []Feature `json:"features" pflag:"-,List of optional features supported."` LogConfig LogConfig `json:"logs" pflag:",Config for log links for spark applications."` + EnablePodTemplate bool `json:"enable-pod-template" pflag:"-,Pass the full pod spec through as the driver/executor pod template on clusters whose SparkApplication CRD supports it. Disable as a kill switch."` } type LogConfig struct { diff --git a/flyteplugins/go/tasks/plugins/k8s/spark/podtemplate.go b/flyteplugins/go/tasks/plugins/k8s/spark/podtemplate.go new file mode 100644 index 00000000000..74f5e7f8813 --- /dev/null +++ b/flyteplugins/go/tasks/plugins/k8s/spark/podtemplate.go @@ -0,0 +1,84 @@ +package spark + +import ( + "context" + "sync" + "time" + + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + + "github.com/flyteorg/flyte/v2/flytestdlib/logger" +) + +const sparkApplicationCRDName = "sparkapplications.sparkoperator.k8s.io" + +var podTemplateCapability struct { + once sync.Once + supported bool +} + +// podTemplateSupported reports whether the cluster's SparkApplication CRD schema accepts the +// driver `template` field. The check runs once per process; a restart is required to pick up a +// CRD upgrade. Any failure to determine the answer (missing RBAC, API error) reports false, +// which keeps the plugin on the legacy fields only — safe against every operator version. +func podTemplateSupported(ctx context.Context) bool { + podTemplateCapability.once.Do(func() { + podTemplateCapability.supported = detectPodTemplateSupport(ctx) + logger.Infof(ctx, "SparkApplication CRD pod template support: %v", podTemplateCapability.supported) + }) + return podTemplateCapability.supported +} + +func detectPodTemplateSupport(ctx context.Context) bool { + restConfig, err := rest.InClusterConfig() + if err != nil { + restConfig, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + clientcmd.NewDefaultClientConfigLoadingRules(), &clientcmd.ConfigOverrides{}).ClientConfig() + if err != nil { + logger.Warnf(ctx, "Cannot load kube config to inspect the SparkApplication CRD, assuming no pod template support: %v", err) + return false + } + } + + clientset, err := apiextensionsclientset.NewForConfig(restConfig) + if err != nil { + logger.Warnf(ctx, "Cannot build apiextensions client to inspect the SparkApplication CRD, assuming no pod template support: %v", err) + return false + } + + getCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + crd, err := clientset.ApiextensionsV1().CustomResourceDefinitions().Get(getCtx, sparkApplicationCRDName, metav1.GetOptions{}) + if err != nil { + logger.Warnf(ctx, "Cannot get the SparkApplication CRD (needs get on customresourcedefinitions), assuming no pod template support: %v", err) + return false + } + + return crdHasDriverTemplate(crd) +} + +func crdHasDriverTemplate(crd *apiextensionsv1.CustomResourceDefinition) bool { + for _, version := range crd.Spec.Versions { + if version.Name != "v1beta2" || !version.Served { + continue + } + if version.Schema == nil || version.Schema.OpenAPIV3Schema == nil { + return false + } + spec, ok := version.Schema.OpenAPIV3Schema.Properties["spec"] + if !ok { + return false + } + driver, ok := spec.Properties["driver"] + if !ok { + return false + } + _, ok = driver.Properties["template"] + return ok + } + return false +} diff --git a/flyteplugins/go/tasks/plugins/k8s/spark/spark.go b/flyteplugins/go/tasks/plugins/k8s/spark/spark.go index 0f7b2d66706..326d43090a1 100644 --- a/flyteplugins/go/tasks/plugins/k8s/spark/spark.go +++ b/flyteplugins/go/tasks/plugins/k8s/spark/spark.go @@ -8,8 +8,8 @@ import ( "strings" "time" - sparkOp "github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2" - sparkOpConfig "github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/config" + sparkOp "github.com/kubeflow/spark-operator/v2/api/v1beta2" + sparkOpCommon "github.com/kubeflow/spark-operator/v2/pkg/common" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes/scheme" @@ -29,17 +29,22 @@ import ( "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/plugins" ) -const KindSparkApplication = "SparkApplication" -const sparkDriverUI = "sparkDriverUI" -const sparkHistoryUI = "sparkHistoryUI" -const defaultDriverPrimaryContainerName = "spark-kubernetes-driver" +const ( + KindSparkApplication = "SparkApplication" + sparkDriverUI = "sparkDriverUI" + sparkHistoryUI = "sparkHistoryUI" + defaultDriverPrimaryContainerName = "spark-kubernetes-driver" +) var featureRegex = regexp.MustCompile(`^spark.((flyteorg)|(flyte)).(.+).enabled$`) var sparkTaskType = "spark" -type sparkResourceHandler struct { -} +// applicationStatePendingSubmission was written by pre-2.x operators; the kubeflow 2.x client +// no longer defines it but a status carrying it must still map to "submitted". +const applicationStatePendingSubmission = sparkOp.ApplicationStateType("PENDING_SUBMISSION") + +type sparkResourceHandler struct{} func validateSparkJob(sparkJob *plugins.SparkJob) error { if sparkJob == nil { @@ -111,21 +116,21 @@ func getSparkConfig(taskCtx pluginsCore.TaskExecutionContext, sparkJob *plugins. } // Set pod limits. - if len(sparkConfig[sparkOpConfig.SparkDriverCoreLimitKey]) == 0 { + if len(sparkConfig[sparkOpCommon.SparkKubernetesDriverLimitCores]) == 0 { // spark.kubernetes.driver.request.cores takes precedence over spark.driver.cores - if len(sparkConfig[sparkOpConfig.SparkDriverCoreRequestKey]) != 0 { - sparkConfig[sparkOpConfig.SparkDriverCoreLimitKey] = sparkConfig[sparkOpConfig.SparkDriverCoreRequestKey] + if len(sparkConfig[sparkOpCommon.SparkKubernetesDriverRequestCores]) != 0 { + sparkConfig[sparkOpCommon.SparkKubernetesDriverLimitCores] = sparkConfig[sparkOpCommon.SparkKubernetesDriverRequestCores] } else if len(sparkConfig["spark.driver.cores"]) != 0 { - sparkConfig[sparkOpConfig.SparkDriverCoreLimitKey] = sparkConfig["spark.driver.cores"] + sparkConfig[sparkOpCommon.SparkKubernetesDriverLimitCores] = sparkConfig["spark.driver.cores"] } } - if len(sparkConfig[sparkOpConfig.SparkExecutorCoreLimitKey]) == 0 { + if len(sparkConfig[sparkOpCommon.SparkKubernetesExecutorLimitCores]) == 0 { // spark.kubernetes.executor.request.cores takes precedence over spark.executor.cores - if len(sparkConfig[sparkOpConfig.SparkExecutorCoreRequestKey]) != 0 { - sparkConfig[sparkOpConfig.SparkExecutorCoreLimitKey] = sparkConfig[sparkOpConfig.SparkExecutorCoreRequestKey] + if len(sparkConfig[sparkOpCommon.SparkKubernetesExecutorRequestCores]) != 0 { + sparkConfig[sparkOpCommon.SparkKubernetesExecutorLimitCores] = sparkConfig[sparkOpCommon.SparkKubernetesExecutorRequestCores] } else if len(sparkConfig["spark.executor.cores"]) != 0 { - sparkConfig[sparkOpConfig.SparkExecutorCoreLimitKey] = sparkConfig["spark.executor.cores"] + sparkConfig[sparkOpCommon.SparkKubernetesExecutorLimitCores] = sparkConfig["spark.executor.cores"] } } @@ -143,7 +148,7 @@ func serviceAccountName(metadata pluginsCore.TaskExecutionMetadata) string { return name } -func createSparkPodSpec(taskCtx pluginsCore.TaskExecutionContext, podSpec *v1.PodSpec, container *v1.Container) *sparkOp.SparkPodSpec { +func createSparkPodSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionContext, podSpec *v1.PodSpec, container *v1.Container) *sparkOp.SparkPodSpec { annotations := utils.UnionMaps(config.GetK8sPluginConfig().DefaultAnnotations, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetAnnotations())) labels := utils.UnionMaps(config.GetK8sPluginConfig().DefaultLabels, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetLabels())) @@ -154,17 +159,34 @@ func createSparkPodSpec(taskCtx pluginsCore.TaskExecutionContext, podSpec *v1.Po sparkEnv = append(sparkEnv, v1.EnvVar{Name: "FLYTE_MAX_ATTEMPTS", Value: strconv.Itoa(int(taskCtx.TaskExecutionMetadata().GetMaxAttempts()))}) spec := sparkOp.SparkPodSpec{ - Affinity: podSpec.Affinity, - Annotations: annotations, - Labels: labels, - Env: sparkEnv, - Image: &container.Image, - SecurityContenxt: podSpec.SecurityContext.DeepCopy(), - DNSConfig: podSpec.DNSConfig.DeepCopy(), - Tolerations: podSpec.Tolerations, - SchedulerName: &podSpec.SchedulerName, - NodeSelector: podSpec.NodeSelector, - HostNetwork: &podSpec.HostNetwork, + Affinity: podSpec.Affinity, + Annotations: annotations, + Labels: labels, + Env: sparkEnv, + Image: &container.Image, + + // In pre-2.x SparkApplication CRDs, this field was called SecurityContenxt (sic) and serialized to `securityContext`. + // This new field serializes to `podSecurityContext`, which is invisible to the old CRD. + // Users who set this field (whether via platform defaults, pod templates, or pod templates in user code) + // will see this setting getting dropped. Once they upgrade the CRD, things will start working again. + PodSecurityContext: podSpec.SecurityContext.DeepCopy(), + + DNSConfig: podSpec.DNSConfig.DeepCopy(), + Tolerations: podSpec.Tolerations, + SchedulerName: &podSpec.SchedulerName, + NodeSelector: podSpec.NodeSelector, + HostNetwork: &podSpec.HostNetwork, + } + + // The legacy fields above are always populated so the object stays valid on clusters whose + // CRD/operator predate pod-template support (unknown fields are pruned by the API server + // there). Where the CRD accepts it, additionally pass the full pod spec through as the pod + // template; the operator treats explicit fields as overrides of the template, and both are + // derived from the same pod spec. + if GetSparkConfig().EnablePodTemplate && podTemplateSupported(ctx) { + spec.Template = &v1.PodTemplateSpec{Spec: *podSpec.DeepCopy()} + sa := serviceAccountName(taskCtx.TaskExecutionMetadata()) + spec.ServiceAccount = &sa } return &spec } @@ -212,12 +234,12 @@ func createDriverSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCont if err != nil { return nil, err } - sparkPodSpec := createSparkPodSpec(nonInterruptibleTaskCtx, podSpec, primaryContainer) + sparkPodSpec := createSparkPodSpec(ctx, nonInterruptibleTaskCtx, podSpec, primaryContainer) serviceAccountName := serviceAccountName(nonInterruptibleTaskCtx.TaskExecutionMetadata()) + sparkPodSpec.ServiceAccount = &serviceAccountName spec := driverSpec{ &sparkOp.DriverSpec{ - SparkPodSpec: *sparkPodSpec, - ServiceAccount: &serviceAccountName, + SparkPodSpec: *sparkPodSpec, }, } if cores, err := strconv.ParseInt(sparkConfig["spark.driver.cores"], 10, 32); err == nil { @@ -228,9 +250,8 @@ func createDriverSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCont } type executorSpec struct { - container *v1.Container - sparkSpec *sparkOp.ExecutorSpec - serviceAccountName string + container *v1.Container + sparkSpec *sparkOp.ExecutorSpec } func createExecutorSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionContext, sparkConfig map[string]string, sparkJob *plugins.SparkJob) (*executorSpec, error) { @@ -269,14 +290,12 @@ func createExecutorSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCo if err != nil { return nil, err } - sparkPodSpec := createSparkPodSpec(taskCtx, podSpec, primaryContainer) - serviceAccountName := serviceAccountName(taskCtx.TaskExecutionMetadata()) + sparkPodSpec := createSparkPodSpec(ctx, taskCtx, podSpec, primaryContainer) spec := executorSpec{ primaryContainer, &sparkOp.ExecutorSpec{ SparkPodSpec: *sparkPodSpec, }, - serviceAccountName, } if execCores, err := strconv.ParseInt(sparkConfig["spark.executor.cores"], 10, 32); err == nil { spec.sparkSpec.Cores = intPtr(int32(execCores)) @@ -289,7 +308,8 @@ func createExecutorSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCo } func createSparkApplication(sparkJob *plugins.SparkJob, sparkConfig map[string]string, driverSpec *driverSpec, - executorSpec *executorSpec) *sparkOp.SparkApplication { + executorSpec *executorSpec, +) *sparkOp.SparkApplication { // Hack: Retry submit failures in-case of resource limits hit. submissionFailureRetries := int32(14) @@ -299,17 +319,16 @@ func createSparkApplication(sparkJob *plugins.SparkJob, sparkConfig map[string]s APIVersion: sparkOp.SchemeGroupVersion.String(), }, Spec: sparkOp.SparkApplicationSpec{ - ServiceAccount: &executorSpec.serviceAccountName, - Type: getApplicationType(sparkJob.GetApplicationType()), - Image: &executorSpec.container.Image, - Arguments: executorSpec.container.Args, - Driver: *driverSpec.sparkSpec, - Executor: *executorSpec.sparkSpec, - SparkConf: sparkConfig, - HadoopConf: sparkJob.GetHadoopConf(), + Type: getApplicationType(sparkJob.GetApplicationType()), + Image: &executorSpec.container.Image, + Arguments: executorSpec.container.Args, + Driver: *driverSpec.sparkSpec, + Executor: *executorSpec.sparkSpec, + SparkConf: sparkConfig, + HadoopConf: sparkJob.GetHadoopConf(), // SubmissionFailures handled here. Task Failures handled at Propeller/Job level. RestartPolicy: sparkOp.RestartPolicy{ - Type: sparkOp.OnFailure, + Type: sparkOp.RestartPolicyOnFailure, OnSubmissionFailureRetries: &submissionFailureRetries, }, }, @@ -329,7 +348,6 @@ func createSparkApplication(sparkJob *plugins.SparkJob, sparkConfig map[string]s } func addConfig(sparkConfig map[string]string, key string, value string) { - if strings.ToLower(strings.TrimSpace(value)) != "true" { sparkConfig[key] = value return @@ -358,15 +376,15 @@ func addConfig(sparkConfig map[string]string, key string, value string) { func getApplicationType(applicationType plugins.SparkApplication_Type) sparkOp.SparkApplicationType { switch applicationType { case plugins.SparkApplication_PYTHON: - return sparkOp.PythonApplicationType + return sparkOp.SparkApplicationTypePython case plugins.SparkApplication_JAVA: - return sparkOp.JavaApplicationType + return sparkOp.SparkApplicationTypeJava case plugins.SparkApplication_SCALA: - return sparkOp.ScalaApplicationType + return sparkOp.SparkApplicationTypeScala case plugins.SparkApplication_R: - return sparkOp.RApplicationType + return sparkOp.SparkApplicationTypeR } - return sparkOp.PythonApplicationType + return sparkOp.SparkApplicationTypePython } func (sparkResourceHandler) BuildIdentityResource(ctx context.Context, taskCtx pluginsCore.TaskExecutionMetadata) (client.Object, error) { @@ -402,7 +420,6 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, TaskExecutionID: taskExecID, EnableVscode: flytek8s.IsVscodeEnabled(ctx, sj.Spec.Driver.Env), }) - if err != nil { return nil, err } @@ -422,7 +439,6 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, Namespace: sj.Namespace, TaskExecutionID: taskExecID, }) - if err != nil { return nil, err } @@ -443,7 +459,7 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, }) for executorPodName, executorState := range sj.Status.ExecutorState { - if executorState != sparkOp.ExecutorPendingState && executorState != sparkOp.ExecutorUnknownState { + if executorState != sparkOp.ExecutorStatePending && executorState != sparkOp.ExecutorStateUnknown { logCtx.Pods = append(logCtx.Pods, &core.PodLogContext{ Namespace: sj.Namespace, PodName: executorPodName, @@ -466,7 +482,6 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, Namespace: sj.Namespace, TaskExecutionID: taskExecID, }) - if err != nil { return nil, err } @@ -485,7 +500,6 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, Namespace: sj.Namespace, TaskExecutionID: taskExecID, }) - if err != nil { return nil, err } @@ -501,7 +515,7 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, customInfoMap := make(map[string]string) // Spark UI. - if sj.Status.AppState.State == sparkOp.FailedState || sj.Status.AppState.State == sparkOp.CompletedState { + if sj.Status.AppState.State == sparkOp.ApplicationStateFailed || sj.Status.AppState.State == sparkOp.ApplicationStateCompleted { if sj.Status.SparkApplicationID != "" && GetSparkConfig().SparkHistoryServerURL != "" { customInfoMap[sparkHistoryUI] = fmt.Sprintf("%s/history/%s", GetSparkConfig().SparkHistoryServerURL, sj.Status.SparkApplicationID) // Custom doesn't work unless the UI has a custom plugin to parse this, hence add to Logs as well. @@ -513,7 +527,7 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, LinkType: core.TaskLog_DASHBOARD, }) } - } else if sj.Status.AppState.State == sparkOp.RunningState && sj.Status.DriverInfo.WebUIIngressAddress != "" { + } else if sj.Status.AppState.State == sparkOp.ApplicationStateRunning && sj.Status.DriverInfo.WebUIIngressAddress != "" { // Older versions of spark-operator does not append http:// but newer versions do. uri := sj.Status.DriverInfo.WebUIIngressAddress if !strings.HasPrefix(uri, "https://") && !strings.HasPrefix(uri, "http://") { @@ -544,7 +558,6 @@ func getEventInfoForSpark(ctx context.Context, pluginContext k8s.PluginContext, } func (sparkResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s.PluginContext, resource client.Object) (pluginsCore.PhaseInfo, error) { - app := resource.(*sparkOp.SparkApplication) info, err := getEventInfoForSpark(ctx, pluginContext, app) if err != nil { @@ -561,17 +574,17 @@ func (sparkResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s. } occurredAt := time.Now() switch app.Status.AppState.State { - case sparkOp.NewState: + case sparkOp.ApplicationStateNew: phaseInfo = pluginsCore.PhaseInfoQueuedWithTaskInfo(occurredAt, pluginsCore.DefaultPhaseVersion, "job queued", info) - case sparkOp.SubmittedState, sparkOp.PendingSubmissionState: + case sparkOp.ApplicationStateSubmitted, applicationStatePendingSubmission: phaseInfo = pluginsCore.PhaseInfoInitializing(occurredAt, pluginsCore.DefaultPhaseVersion, "job submitted", info) - case sparkOp.FailedSubmissionState: + case sparkOp.ApplicationStateFailedSubmission: reason := fmt.Sprintf("Spark Job Submission Failed with Error: %s", app.Status.AppState.ErrorMessage) phaseInfo = pluginsCore.PhaseInfoRetryableFailure(errors.DownstreamSystemError, reason, info) - case sparkOp.FailedState: + case sparkOp.ApplicationStateFailed: reason := fmt.Sprintf("Spark Job Failed with Error: %s", app.Status.AppState.ErrorMessage) phaseInfo = pluginsCore.PhaseInfoRetryableFailure(errors.DownstreamSystemError, reason, info) - case sparkOp.CompletedState: + case sparkOp.ApplicationStateCompleted: phaseInfo = pluginsCore.PhaseInfoSuccess(info) default: phaseInfo = pluginsCore.PhaseInfoRunning(pluginsCore.DefaultPhaseVersion, info) @@ -587,7 +600,6 @@ func (sparkResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s. tl.Ready = true phaseInfo.WithReason("Spark driver UI is ready") } - } else if tl != nil && tl.LinkType == core.TaskLog_IDE { if phaseInfo.Phase() != pluginsCore.PhaseRunning { phaseInfo.WithReason("Vscode server is not ready") @@ -612,7 +624,7 @@ func (sparkResourceHandler) IsTerminal(_ context.Context, resource client.Object return false, fmt.Errorf("unexpected resource type: expected *SparkApplication, got %T", resource) } state := app.Status.AppState.State - return state == sparkOp.CompletedState || state == sparkOp.FailedState || state == sparkOp.FailedSubmissionState, nil + return state == sparkOp.ApplicationStateCompleted || state == sparkOp.ApplicationStateFailed || state == sparkOp.ApplicationStateFailedSubmission, nil } // GetCompletionTime returns the termination time of the SparkApplication @@ -627,8 +639,8 @@ func (sparkResourceHandler) GetCompletionTime(resource client.Object) (time.Time } // Fallback to submission time or creation time - if !app.Status.SubmissionTime.IsZero() { - return app.Status.SubmissionTime.Time, nil + if !app.Status.LastSubmissionAttemptTime.IsZero() { + return app.Status.LastSubmissionAttemptTime.Time, nil } return app.CreationTimestamp.Time, nil diff --git a/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go b/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go index fa9c37c147e..b77cf09be57 100644 --- a/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go +++ b/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go @@ -8,9 +8,9 @@ import ( "testing" "time" - sj "github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2" - sparkOp "github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2" structpb "github.com/golang/protobuf/ptypes/struct" + sj "github.com/kubeflow/spark-operator/v2/api/v1beta2" + sparkOp "github.com/kubeflow/spark-operator/v2/api/v1beta2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -73,10 +73,10 @@ var ( ) func TestGetApplicationType(t *testing.T) { - assert.Equal(t, getApplicationType(plugins.SparkApplication_PYTHON), sj.PythonApplicationType) - assert.Equal(t, getApplicationType(plugins.SparkApplication_R), sj.RApplicationType) - assert.Equal(t, getApplicationType(plugins.SparkApplication_JAVA), sj.JavaApplicationType) - assert.Equal(t, getApplicationType(plugins.SparkApplication_SCALA), sj.ScalaApplicationType) + assert.Equal(t, getApplicationType(plugins.SparkApplication_PYTHON), sj.SparkApplicationTypePython) + assert.Equal(t, getApplicationType(plugins.SparkApplication_R), sj.SparkApplicationTypeR) + assert.Equal(t, getApplicationType(plugins.SparkApplication_JAVA), sj.SparkApplicationTypeJava) + assert.Equal(t, getApplicationType(plugins.SparkApplication_SCALA), sj.SparkApplicationTypeScala) } func TestGetEventInfo(t *testing.T) { @@ -103,7 +103,7 @@ func TestGetEventInfo(t *testing.T) { }, })) pluginContext := dummySparkPluginContext(dummySparkTaskTemplateContainer("blah-1", dummySparkConf), k8s.PluginState{}) - info, err := getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.RunningState)) + info, err := getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.ApplicationStateRunning)) assert.NoError(t, err) assert.Len(t, info.Logs, 6) assert.Equal(t, "https://spark-ui.flyte", info.CustomInfo.Fields[sparkDriverUI].GetStringValue()) @@ -123,7 +123,7 @@ func TestGetEventInfo(t *testing.T) { assert.Equal(t, expectedLinks, generatedLinks) - info, err = getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.SubmittedState)) + info, err = getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.ApplicationStateSubmitted)) generatedLinks = make([]string, 0, len(info.Logs)) for _, l := range info.Logs { generatedLinks = append(generatedLinks, l.Uri) @@ -161,7 +161,7 @@ func TestGetEventInfo(t *testing.T) { }, })) - info, err = getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.FailedState)) + info, err = getEventInfoForSpark(context.TODO(), pluginContext, dummySparkApplication(sj.ApplicationStateFailed)) assert.NoError(t, err) assert.Len(t, info.Logs, 5) assert.Equal(t, "spark-history.flyte/history/app-id", info.CustomInfo.Fields[sparkHistoryUI].GetStringValue()) @@ -211,70 +211,70 @@ func TestGetTaskPhase(t *testing.T) { ctx := context.TODO() pluginCtx := dummySparkPluginContext(dummySparkTaskTemplateContainer("", dummySparkConf), k8s.PluginState{}) - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.NewState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateNew)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseQueued) assert.NotNil(t, taskPhase.Info()) assert.NotNil(t, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.SubmittedState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateSubmitted)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseInitializing) assert.NotNil(t, taskPhase.Info()) assert.NotNil(t, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.RunningState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateRunning)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRunning) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.CompletedState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateCompleted)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseSuccess) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.InvalidatingState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateInvalidating)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRunning) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.FailingState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateFailing)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRunning) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.PendingRerunState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStatePendingRerun)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRunning) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.SucceedingState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateSucceeding)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRunning) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.FailedSubmissionState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateFailedSubmission)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRetryableFailure) assert.NotNil(t, taskPhase.Info()) assert.Equal(t, expectedLogCtx, taskPhase.Info().LogContext) assert.Nil(t, err) - taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.FailedState)) + taskPhase, err = sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateFailed)) assert.NoError(t, err) assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseRetryableFailure) assert.NotNil(t, taskPhase.Info()) @@ -293,7 +293,7 @@ func TestGetTaskPhaseIncreasePhaseVersion(t *testing.T) { } pluginCtx := dummySparkPluginContext(dummySparkTaskTemplateContainer("", dummySparkConf), pluginState) - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.SubmittedState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateSubmitted)) assert.NoError(t, err) assert.Equal(t, taskPhase.Version(), pluginsCore.DefaultPhaseVersion+1) @@ -317,8 +317,8 @@ func dummySparkApplication(state sj.ApplicationStateType) *sj.SparkApplication { }, ExecutionAttempts: 1, ExecutorState: map[string]sparkOp.ExecutorState{ - "exec-pod-1": sparkOp.ExecutorPendingState, - "exec-pod-2": sparkOp.ExecutorRunningState, + "exec-pod-1": sparkOp.ExecutorStatePending, + "exec-pod-2": sparkOp.ExecutorStateRunning, }, }, } @@ -710,6 +710,7 @@ func TestBuildResourceContainer(t *testing.T) { // Set spark custom feature config. assert.NoError(t, setSparkConfig(&Config{ + EnablePodTemplate: true, Features: []Feature{ { Name: "feature1", @@ -732,17 +733,17 @@ func TestBuildResourceContainer(t *testing.T) { assert.True(t, ok) assert.Equal(t, sparkMainClass, *sparkApp.Spec.MainClass) assert.Equal(t, sparkApplicationFile, *sparkApp.Spec.MainApplicationFile) - assert.Equal(t, sj.PythonApplicationType, sparkApp.Spec.Type) + assert.Equal(t, sj.SparkApplicationTypePython, sparkApp.Spec.Type) assert.Equal(t, testArgs, sparkApp.Spec.Arguments) assert.Equal(t, testImage, *sparkApp.Spec.Image) - assert.NotNil(t, sparkApp.Spec.Driver.SecurityContenxt) - assert.Equal(t, *sparkApp.Spec.Driver.SecurityContenxt.RunAsUser, *defaultConfig.DefaultPodSecurityContext.RunAsUser) + assert.NotNil(t, sparkApp.Spec.Driver.PodSecurityContext) + assert.Equal(t, *sparkApp.Spec.Driver.PodSecurityContext.RunAsUser, *defaultConfig.DefaultPodSecurityContext.RunAsUser) assert.NotNil(t, sparkApp.Spec.Driver.DNSConfig) assert.Equal(t, []string{"8.8.8.8", "8.8.4.4"}, sparkApp.Spec.Driver.DNSConfig.Nameservers) assert.ElementsMatch(t, defaultConfig.DefaultPodDNSConfig.Options, sparkApp.Spec.Driver.DNSConfig.Options) assert.Equal(t, []string{"ns1.svc.cluster-domain.example", "my.dns.search.suffix"}, sparkApp.Spec.Driver.DNSConfig.Searches) - assert.NotNil(t, sparkApp.Spec.Executor.SecurityContenxt) - assert.Equal(t, *sparkApp.Spec.Executor.SecurityContenxt.RunAsUser, *defaultConfig.DefaultPodSecurityContext.RunAsUser) + assert.NotNil(t, sparkApp.Spec.Executor.PodSecurityContext) + assert.Equal(t, *sparkApp.Spec.Executor.PodSecurityContext.RunAsUser, *defaultConfig.DefaultPodSecurityContext.RunAsUser) assert.NotNil(t, sparkApp.Spec.Executor.DNSConfig) assert.NotNil(t, sparkApp.Spec.Executor.DNSConfig) assert.ElementsMatch(t, defaultConfig.DefaultPodDNSConfig.Options, sparkApp.Spec.Executor.DNSConfig.Options) @@ -753,7 +754,7 @@ func TestBuildResourceContainer(t *testing.T) { execCores, _ := strconv.ParseInt(dummySparkConf["spark.executor.cores"], 10, 32) execInstances, _ := strconv.ParseInt(dummySparkConf["spark.executor.instances"], 10, 32) - assert.Equal(t, "new-val", *sparkApp.Spec.ServiceAccount) + assert.Equal(t, "new-val", *sparkApp.Spec.Executor.ServiceAccount) assert.Equal(t, int32(driverCores), *sparkApp.Spec.Driver.Cores) assert.Equal(t, int32(execCores), *sparkApp.Spec.Executor.Cores) assert.Equal(t, int32(execInstances), *sparkApp.Spec.Executor.Instances) @@ -965,12 +966,12 @@ func TestBuildResourcePodTemplate(t *testing.T) { }, sparkApp.TypeMeta) // Application spec - assert.Equal(t, flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata()), *sparkApp.Spec.ServiceAccount) - assert.Equal(t, sparkOp.PythonApplicationType, sparkApp.Spec.Type) + assert.Equal(t, flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata()), *sparkApp.Spec.Executor.ServiceAccount) + assert.Equal(t, sparkOp.SparkApplicationTypePython, sparkApp.Spec.Type) assert.Equal(t, testImage, *sparkApp.Spec.Image) assert.Equal(t, testArgs, sparkApp.Spec.Arguments) assert.Equal(t, sparkOp.RestartPolicy{ - Type: sparkOp.OnFailure, + Type: sparkOp.RestartPolicyOnFailure, OnSubmissionFailureRetries: intPtr(int32(14)), }, sparkApp.Spec.RestartPolicy) assert.Equal(t, sparkMainClass, *sparkApp.Spec.MainClass) @@ -986,7 +987,7 @@ func TestBuildResourcePodTemplate(t *testing.T) { assert.Equal(t, 10, len(sparkApp.Spec.Driver.Env)) assert.Equal(t, testImage, *sparkApp.Spec.Driver.Image) assert.Equal(t, flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata()), *sparkApp.Spec.Driver.ServiceAccount) - assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Driver.SecurityContenxt) + assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Driver.PodSecurityContext) assert.Equal(t, defaultConfig.DefaultPodDNSConfig, sparkApp.Spec.Driver.DNSConfig) assert.Equal(t, defaultConfig.EnableHostNetworkingPod, sparkApp.Spec.Driver.HostNetwork) assert.Equal(t, defaultConfig.SchedulerName, *sparkApp.Spec.Driver.SchedulerName) @@ -1022,7 +1023,7 @@ func TestBuildResourcePodTemplate(t *testing.T) { assert.Equal(t, findEnvVarByName(dummyEnvVarsWithSecretRef, "SECRET"), findEnvVarByName(sparkApp.Spec.Executor.Env, "SECRET")) assert.Equal(t, 10, len(sparkApp.Spec.Executor.Env)) assert.Equal(t, testImage, *sparkApp.Spec.Executor.Image) - assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Executor.SecurityContenxt) + assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Executor.PodSecurityContext) assert.Equal(t, defaultConfig.DefaultPodDNSConfig, sparkApp.Spec.Executor.DNSConfig) assert.Equal(t, defaultConfig.EnableHostNetworkingPod, sparkApp.Spec.Executor.HostNetwork) assert.Equal(t, defaultConfig.SchedulerName, *sparkApp.Spec.Executor.SchedulerName) @@ -1066,7 +1067,7 @@ func TestGetTaskPhaseWithNamespaceInLogContext(t *testing.T) { ctx := context.TODO() pluginCtx := dummySparkPluginContext(dummySparkTaskTemplateContainer("", dummySparkConf), k8s.PluginState{}) - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.RunningState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateRunning)) assert.NoError(t, err) assert.NotNil(t, taskPhase.Info()) assert.NotNil(t, taskPhase.Info().LogContext) @@ -1109,7 +1110,7 @@ func TestGetTaskPhaseWithFailedPod(t *testing.T) { pluginCtx := dummySparkPluginContextWithPods(dummySparkTaskTemplateContainer("", dummySparkConf), k8s.PluginState{}, pod) // Even though SparkApplication status is running, should return failure due to pod status - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.RunningState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateRunning)) assert.NoError(t, err) assert.True(t, taskPhase.Phase().IsFailure()) } @@ -1152,7 +1153,7 @@ func TestGetTaskPhaseWithPendingPodInvalidImage(t *testing.T) { pluginCtx := dummySparkPluginContextWithPods(dummySparkTaskTemplateContainer("", dummySparkConf), k8s.PluginState{}, pod) - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.SubmittedState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateSubmitted)) assert.NoError(t, err) // Should detect the InvalidImageName and return a failure phase assert.True(t, taskPhase.Phase().IsFailure()) @@ -1164,7 +1165,7 @@ func TestGetTaskPhaseContainerNameConstant(t *testing.T) { pluginCtx := dummySparkPluginContext(dummySparkTaskTemplateContainer("", dummySparkConf), k8s.PluginState{}) - taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.CompletedState)) + taskPhase, err := sparkResourceHandler.GetTaskPhase(ctx, pluginCtx, dummySparkApplication(sj.ApplicationStateCompleted)) assert.NoError(t, err) assert.NotNil(t, taskPhase.Info()) assert.NotNil(t, taskPhase.Info().LogContext) @@ -1185,16 +1186,16 @@ func TestIsTerminal(t *testing.T) { state sj.ApplicationStateType expectedResult bool }{ - {"Completed", sj.CompletedState, true}, - {"Failed", sj.FailedState, true}, - {"FailedSubmission", sj.FailedSubmissionState, true}, - {"New", sj.NewState, false}, - {"Submitted", sj.SubmittedState, false}, - {"Running", sj.RunningState, false}, - {"PendingRerun", sj.PendingRerunState, false}, - {"Invalidating", sj.InvalidatingState, false}, - {"Succeeding", sj.SucceedingState, false}, - {"Failing", sj.FailingState, false}, + {"Completed", sj.ApplicationStateCompleted, true}, + {"Failed", sj.ApplicationStateFailed, true}, + {"FailedSubmission", sj.ApplicationStateFailedSubmission, true}, + {"New", sj.ApplicationStateNew, false}, + {"Submitted", sj.ApplicationStateSubmitted, false}, + {"Running", sj.ApplicationStateRunning, false}, + {"PendingRerun", sj.ApplicationStatePendingRerun, false}, + {"Invalidating", sj.ApplicationStateInvalidating, false}, + {"Succeeding", sj.ApplicationStateSucceeding, false}, + {"Failing", sj.ApplicationStateFailing, false}, } for _, tt := range tests { @@ -1237,20 +1238,20 @@ func TestGetCompletionTime(t *testing.T) { CreationTimestamp: v1.NewTime(evenEarlier), }, Status: sj.SparkApplicationStatus{ - TerminationTime: v1.NewTime(now), - SubmissionTime: v1.NewTime(earlier), + TerminationTime: v1.NewTime(now), + LastSubmissionAttemptTime: v1.NewTime(earlier), }, }, expectedTime: now, }, { - name: "falls back to SubmissionTime", + name: "falls back to LastSubmissionAttemptTime", app: &sj.SparkApplication{ ObjectMeta: v1.ObjectMeta{ CreationTimestamp: v1.NewTime(evenEarlier), }, Status: sj.SparkApplicationStatus{ - SubmissionTime: v1.NewTime(now), + LastSubmissionAttemptTime: v1.NewTime(now), }, }, expectedTime: now, @@ -1347,3 +1348,58 @@ func TestBuildResourceSparkExecutorAffinityDilution(t *testing.T) { assert.Truef(t, found, "executor node selector term %d is missing the non-interruptible requirement", i) } } + +// setPodTemplateSupportedForTest overrides the once-per-process CRD capability probe. +func setPodTemplateSupportedForTest(supported bool) { + podTemplateCapability.once.Do(func() {}) + podTemplateCapability.supported = supported +} + +func init() { + // Unit tests run without a cluster; default to the capability-on path so the + // pod-template assertions in the tests above exercise it. Tests toggle as needed. + setPodTemplateSupportedForTest(true) +} + +func TestBuildResourcePodTemplateGating(t *testing.T) { + assert.NoError(t, setSparkConfig(&Config{EnablePodTemplate: true})) + assert.NoError(t, config.SetK8sPluginConfig(defaultPluginConfig())) + taskTemplate := dummySparkTaskTemplateContainer("gating", dummySparkConf) + taskCtx := dummySparkTaskContext(taskTemplate, true) + handler := sparkResourceHandler{} + + setPodTemplateSupportedForTest(true) + resource, err := handler.BuildResource(context.TODO(), taskCtx) + assert.NoError(t, err) + withTemplate := resource.(*sj.SparkApplication) + + setPodTemplateSupportedForTest(false) + resource, err = handler.BuildResource(context.TODO(), taskCtx) + assert.NoError(t, err) + withoutTemplate := resource.(*sj.SparkApplication) + setPodTemplateSupportedForTest(true) + + // Capability on: the full pod spec rides along as the template and the service + // account is set on both driver and executor. + assert.NotNil(t, withTemplate.Spec.Driver.Template) + assert.NotNil(t, withTemplate.Spec.Executor.Template) + assert.NotEmpty(t, withTemplate.Spec.Driver.Template.Spec.Containers) + assert.Equal(t, *withTemplate.Spec.Driver.ServiceAccount, withTemplate.Spec.Driver.Template.Spec.ServiceAccountName) + assert.NotNil(t, withTemplate.Spec.Executor.ServiceAccount) + + // Capability off: today's wire shape exactly — no template, no executor service account. + assert.Nil(t, withoutTemplate.Spec.Driver.Template) + assert.Nil(t, withoutTemplate.Spec.Executor.Template) + assert.Nil(t, withoutTemplate.Spec.Executor.ServiceAccount) + assert.NotNil(t, withoutTemplate.Spec.Driver.ServiceAccount) + + // The template is strictly additive: with it stripped, the objects are identical. + normalized := withTemplate.DeepCopy() + normalized.Spec.Driver.Template = nil + normalized.Spec.Executor.Template = nil + normalized.Spec.Executor.ServiceAccount = nil + expected := withoutTemplate.DeepCopy() + delete(normalized.Spec.SparkConf, "spark.kubernetes.driverEnv.FLYTE_START_TIME") + delete(expected.Spec.SparkConf, "spark.kubernetes.driverEnv.FLYTE_START_TIME") + assert.Equal(t, expected, normalized) +} diff --git a/flytestdlib/config/viper/viper.go b/flytestdlib/config/viper/viper.go index 98b36e4946a..c7dfdcc1c74 100644 --- a/flytestdlib/config/viper/viper.go +++ b/flytestdlib/config/viper/viper.go @@ -6,16 +6,18 @@ import ( "encoding/json" "flag" "fmt" + "os" "reflect" "strings" "sync" "github.com/fsnotify/fsnotify" - "github.com/mitchellh/mapstructure" + "github.com/go-viper/mapstructure/v2" "github.com/pkg/errors" "github.com/spf13/cobra" "github.com/spf13/pflag" viperLib "github.com/spf13/viper" + "gopkg.in/yaml.v3" "k8s.io/apimachinery/pkg/util/sets" "github.com/flyteorg/flyte/v2/flytestdlib/config" @@ -182,18 +184,22 @@ func sliceToMapHook(f reflect.Kind, t reflect.Kind, data interface{}) (interface // Only handle slice -> map conversion if f == reflect.Slice && t == reflect.Map { // this will be the target result - res := map[interface{}]interface{}{} + res := map[string]interface{}{} // It's safe to convert data into a slice since we did the type assertion above. asSlice := data.([]interface{}) for _, item := range asSlice { - asMap, casted := item.(map[interface{}]interface{}) - if !casted { + switch m := item.(type) { + case map[string]interface{}: + for key, value := range m { + res[key] = value + } + case map[interface{}]interface{}: + for key, value := range m { + res[fmt.Sprintf("%v", key)] = value + } + default: return data, nil } - - for key, value := range asMap { - res[key] = value - } } return res, nil @@ -261,7 +267,61 @@ func jsonUnmarshallerHook(_, to reflect.Type, data interface{}) (interface{}, er // Parses RootType config from parsed Viper settings. This should be called after viper has parsed config file/pflags...etc. func (v viperAccessor) parseViperConfig(root config.Section) error { // We use AllSettings instead of AllKeys to get the root level keys folded. - return v.parseViperConfigRecursive(root, v.viper.AllSettings()) + settings := v.viper.AllSettings() + + // Viper v1.21+ recursively lowercases all map keys, including those inside + // arrays. This breaks the array-based workaround for case-sensitive map keys + // (see: https://github.com/spf13/viper#does-viper-support-case-sensitive-keys). + // Re-read config files directly to restore case-sensitive keys within arrays. + for _, configFile := range v.viper.ConfigFilesUsed() { + if configFile == "" { + continue + } + + data, err := os.ReadFile(configFile) + if err != nil { + return fmt.Errorf("failed to read config file %q for case-sensitive key restoration: %w", configFile, err) + } + + var rawSettings map[string]interface{} + if err := yaml.Unmarshal(data, &rawSettings); err != nil { + return fmt.Errorf("failed to parse config file %q for case-sensitive key restoration: %w", configFile, err) + } + + restoreCaseSensitiveArrayKeys(settings, rawSettings) + } + + return v.parseViperConfigRecursive(root, settings) +} + +// restoreCaseSensitiveArrayKeys walks viperData and rawData in parallel. +// When an array value is found in viperData, it is replaced with the +// corresponding value from rawData to preserve the original key casing. +func restoreCaseSensitiveArrayKeys(viperData, rawData map[string]interface{}) { + for lowerKey, viperVal := range viperData { + // Find matching key in rawData (case-insensitive match) + var rawVal interface{} + for rawKey, rv := range rawData { + if strings.EqualFold(rawKey, lowerKey) { + rawVal = rv + break + } + } + + if rawVal == nil { + continue + } + + switch viperVal.(type) { + case []interface{}: + // Replace the lowercased array with the case-preserved original + viperData[lowerKey] = rawVal + case map[string]interface{}: + if rawMap, ok := rawVal.(map[string]interface{}); ok { + restoreCaseSensitiveArrayKeys(viperVal.(map[string]interface{}), rawMap) + } + } + } } func (v viperAccessor) parseViperConfigRecursive(root config.Section, settings interface{}) error { diff --git a/go.mod b/go.mod index 52915b9a563..2aeeabec666 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,6 @@ require ( github.com/Azure/azure-sdk-for-go/sdk/azcore v1.22.0 github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.14.0 github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azsecrets v1.5.0 - github.com/GoogleCloudPlatform/spark-on-k8s-operator v0.0.0-20200723154620-6f35a1152625 github.com/aws/aws-sdk-go v1.55.8 github.com/aws/aws-sdk-go-v2 v1.43.0 github.com/aws/aws-sdk-go-v2/config v1.32.31 @@ -40,7 +39,6 @@ require ( github.com/kubeflow/training-operator v1.9.3 github.com/lib/pq v1.12.3 github.com/magiconair/properties v1.18.11 - github.com/mitchellh/mapstructure v1.5.0 github.com/onsi/ginkgo/v2 v2.32.0 github.com/onsi/gomega v1.42.1 github.com/pkg/errors v0.9.1 @@ -55,7 +53,7 @@ require ( github.com/sirupsen/logrus v1.9.4 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 - github.com/spf13/viper v1.11.0 + github.com/spf13/viper v1.21.0 github.com/stretchr/testify v1.11.1 github.com/vmihailenco/msgpack/v5 v5.4.1 go.opentelemetry.io/otel v1.44.0 @@ -97,6 +95,9 @@ require ( github.com/Masterminds/semver/v3 v3.5.0 github.com/alicebob/miniredis/v2 v2.38.0 github.com/getsentry/sentry-go v0.48.0 + github.com/kubeflow/spark-operator/v2 v2.4.0 + gopkg.in/yaml.v3 v3.0.1 + k8s.io/apiextensions-apiserver v0.36.0 ) require ( @@ -170,7 +171,6 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.18 // indirect github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 // indirect - github.com/hashicorp/hcl v1.0.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect @@ -185,16 +185,16 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncw/swift v1.0.53 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect - github.com/pelletier/go-toml v1.9.4 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/procfs v0.21.0 // indirect + github.com/sagikazarmark/locafero v0.11.0 // indirect + github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect - github.com/spf13/cast v1.7.1 // indirect - github.com/spf13/jwalterweatherman v1.1.0 // indirect + github.com/spf13/cast v1.10.0 // indirect github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect github.com/stretchr/objx v0.5.3 // indirect github.com/subosito/gotenv v1.6.0 // indirect @@ -223,9 +223,6 @@ require ( google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect - gopkg.in/ini.v1 v1.66.4 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/apiextensions-apiserver v0.36.0 // indirect k8s.io/apiserver v0.36.0 // indirect k8s.io/component-base v0.36.0 // indirect k8s.io/kube-openapi v0.0.0-20260317180543-43fb72c5454a // indirect diff --git a/go.sum b/go.sum index 1effd1778c0..eacdda38c22 100644 --- a/go.sum +++ b/go.sum @@ -59,8 +59,6 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.53.0/go.mod h1:jUZ5LYlw40WMd07qxcQJD5M40aUxrfwqQX1g7zxYnrQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0 h1:Ron4zCA/yk6U7WOBXhTJcDpsUBG9npumK6xw2auFltQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0/go.mod h1:cSgYe11MCNYunTnRXrKiR/tHc0eoKjICUuWpNZoVCOo= -github.com/GoogleCloudPlatform/spark-on-k8s-operator v0.0.0-20200723154620-6f35a1152625 h1:cQyO5JQ2iuHnEcF3v24kdDMsgh04RjyFPDtuvD6PCE0= -github.com/GoogleCloudPlatform/spark-on-k8s-operator v0.0.0-20200723154620-6f35a1152625/go.mod h1:6PnrZv6zUDkrNMw0mIoGRmGBR7i9LulhKPmxFq4rUiM= github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAwZ/2OOE= github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= @@ -144,8 +142,8 @@ github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4= github.com/envoyproxy/protoc-gen-validate v1.3.3 h1:MVQghNeW+LZcmXe7SY1V36Z+WFMDjpqGAGacLe2T0ds= github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0= -github.com/evanphx/json-patch v5.9.0+incompatible h1:fBXyNpNMuTTDdquAq/uisOr2lShz4oaXpDTX2bLe7ls= -github.com/evanphx/json-patch v5.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= +github.com/evanphx/json-patch v5.9.11+incompatible h1:ixHHqfcGvxhWkniF1tWxBHA0yb4Z+d1UQi45df52xW8= +github.com/evanphx/json-patch v5.9.11+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjTM0wiaDU= github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= @@ -172,8 +170,8 @@ github.com/gkampitakis/go-diff v1.3.2 h1:Qyn0J9XJSDTgnsgHRdz9Zp24RaJeKMUHg2+PDZZ github.com/gkampitakis/go-diff v1.3.2/go.mod h1:LLgOrpqleQe26cte8s36HTWcTmMEur6OPYerdAAS9tk= github.com/gkampitakis/go-snaps v0.5.15 h1:amyJrvM1D33cPHwVrjo9jQxX8g/7E2wYdZ+01KS3zGE= github.com/gkampitakis/go-snaps v0.5.15/go.mod h1:HNpx/9GoKisdhw9AFOBT1N7DBs9DiHo/hGheFGBZ+mc= -github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= -github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-errors/errors v1.5.1 h1:ZwEMSLRCapFLflTpT7NKaAc7ukJ8ZPEjzlxt8rPN8bk= +github.com/go-errors/errors v1.5.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= @@ -264,8 +262,6 @@ github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iP github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= -github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= -github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4= github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= @@ -298,6 +294,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kubeflow/spark-operator/v2 v2.4.0 h1:Bo5rz5dlhqYEa7e/TLtUEVXDMNRotnVaQ4VLCyeLeOc= +github.com/kubeflow/spark-operator/v2 v2.4.0/go.mod h1:oNwv86olb4glLBGXXH8/fkT4lLUm2ZMFd0OO+LrHMbk= github.com/kubeflow/training-operator v1.9.3 h1:aaSHqskOtCY8Dn8sVd+l9p5XAczW6O0nYMPjVLAw/lc= github.com/kubeflow/training-operator v1.9.3/go.mod h1:6zI0hgeCOheiW5Z12IcVkKBuSWm714fz8mUvYu4Fid4= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= @@ -317,8 +315,6 @@ github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/mfridman/tparse v0.18.0 h1:wh6dzOKaIwkUGyKgOntDW4liXSo37qg5AXbIhkMV3vE= github.com/mfridman/tparse v0.18.0/go.mod h1:gEvqZTuCgEhPbYk/2lS3Kcxg1GmTxxU7kTC8DvP0i/A= -github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY= -github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -335,8 +331,6 @@ github.com/onsi/gomega v1.42.1 h1:iN1rCUX+44NZ1Dc97MPoeFYbFR0vh8zxoxMFwKdyZ6I= github.com/onsi/gomega v1.42.1/go.mod h1:REff/hsDsodHoKlWsP2mAPhu1+5/6hVYNf9rIEBpeSg= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= -github.com/pelletier/go-toml v1.9.4 h1:tjENF6MfZAg8e4ZmZTeWaWiT2vXtsoO6+iuOjFhECwM= -github.com/pelletier/go-toml v1.9.4/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= @@ -368,31 +362,32 @@ github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3b github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= +github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= github.com/samber/lo v1.53.0 h1:t975lj2py4kJPQ6haz1QMgtId2gtmfktACxIXArw3HM= github.com/samber/lo v1.53.0/go.mod h1:4+MXEGsJzbKGaUEQFKBq2xtfuznW9oz/WrgyzMzRoM0= github.com/shamaton/msgpack/v2 v2.4.1 h1:JtJ141QoQ3NqgPDsjq2v9VXlaON8SiQOwEaoNLEK/MQ= github.com/shamaton/msgpack/v2 v2.4.1/go.mod h1:6khjYnkx73f7VQU7wjcFS9DFjs+59naVWJv1TB7qdOI= github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= +github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= -github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= -github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= -github.com/spf13/jwalterweatherman v1.1.0 h1:ue6voC5bR5F8YxI5S67j9i582FU4Qvo2bmqnqMYADFk= -github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/spf13/viper v1.11.0 h1:7OX/1FS6n7jHD1zGrZTM7WtY13ZELRyosK4k93oPr44= -github.com/spf13/viper v1.11.0/go.mod h1:djo0X/bA5+tYVoCn+C7cAYJGcVn/qYLFTG8gdUsX7Zk= +github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU= +github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY= github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo= github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= -github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -523,8 +518,6 @@ gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnf gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= -gopkg.in/ini.v1 v1.66.4 h1:SsAcf+mM7mRZo2nJNGt8mZCjG8ZRaNGMURJw7BsIST4= -gopkg.in/ini.v1 v1.66.4/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= diff --git a/runs/config/config_flags_test.go b/runs/config/config_flags_test.go index 99dd6657d7a..1f71a1ddf41 100755 --- a/runs/config/config_flags_test.go +++ b/runs/config/config_flags_test.go @@ -10,7 +10,7 @@ import ( "strings" "testing" - "github.com/mitchellh/mapstructure" + "github.com/go-viper/mapstructure/v2" "github.com/stretchr/testify/assert" )