Merge branch 'master' into peeter/fix-contributing-page

Author: ppiegaze

flytectl/cmd/config/subcommand/matchable_attr_file_config_utils.go

@@ -52,7 +52,7 @@ func DumpTaskResourceAttr(matchableAttrConfig interface{}, fileName string) erro
 		if err := WriteConfigToFile(matchableAttrConfig, fileName); err != nil {
 			return fmt.Errorf("error dumping in file due to %v", err)
 		}
-		fmt.Printf("wrote the config to file %v", fileName)
+		fmt.Printf("wrote the config to file %v\n", fileName)
 	} else {
 		fmt.Printf("%v", String(matchableAttrConfig))
 	}

flytectl/cmd/get/matchable_workflow_execution_config.go

@@ -8,10 +8,9 @@ import (
 	sconfig "github.com/flyteorg/flyte/flytectl/cmd/config/subcommand"
 	"github.com/flyteorg/flyte/flytectl/cmd/config/subcommand/workflowexecutionconfig"
 	cmdCore "github.com/flyteorg/flyte/flytectl/cmd/core"
+	"github.com/flyteorg/flyte/flytectl/pkg/ext"
 	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
 	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 const (
@@ -141,7 +140,7 @@ func getWorkflowExecutionConfigFunc(ctx context.Context, args []string, cmdCtx c
 	// Updates the workflowExecutionConfigFileConfig with the fetched matchable attribute
 	if err := FetchAndUnDecorateMatchableAttr(ctx, project, domain, workflowName, cmdCtx.AdminFetcherExt(),
 		&workflowExecutionConfigFileConfig, admin.MatchableResource_WORKFLOW_EXECUTION_CONFIG); err != nil {
-		if grpcError := status.Code(err); grpcError == codes.NotFound && workflowexecutionconfig.DefaultFetchConfig.Gen {
+		if ext.IsNotFoundError(err) && workflowexecutionconfig.DefaultFetchConfig.Gen {
 			fmt.Println("Generating a sample workflow execution config file")
 			workflowExecutionConfigFileConfig = getSampleWorkflowExecutionFileConfig(project, domain, workflowName)
 		} else {

flytepropeller/pkg/controller/nodes/task/k8s/plugin_manager.go

@@ -262,8 +262,14 @@ func (e *PluginManager) launchResource(ctx context.Context, tCtx pluginsCore.Tas
 		} else if k8serrors.IsForbidden(err) {
 			return pluginsCore.DoTransition(pluginsCore.PhaseInfoRetryableFailure("RuntimeFailure", err.Error(), nil)), nil
 		} else if k8serrors.IsBadRequest(err) || k8serrors.IsInvalid(err) {
+			// BadRequest (HTTP 400) and Invalid (HTTP 422) errors are intrinsic
+			// to the request payload and not transient. The most common source
+			// is a validating admission webhook rejecting the pod spec; retrying
+			// with the same input will produce the same rejection. Treat as a
+			// permanent failure so the validation error surfaces to the user
+			// instead of exhausting the workflow's retry budget.
 			logger.Errorf(ctx, "Badly formatted resource for plugin [%s], err %s", e.id, err)
-			// return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("BadTaskFormat", err.Error(), nil)), nil
+			return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("BadTaskFormat", err.Error(), nil)), nil
 		} else if k8serrors.IsRequestEntityTooLargeError(err) {
 			logger.Errorf(ctx, "Badly formatted resource for plugin [%s], err %s", e.id, err)
 			return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("EntityTooLarge", err.Error(), nil)), nil

flytepropeller/pkg/controller/nodes/task/k8s/plugin_manager_test.go

@@ -408,6 +408,75 @@ func TestK8sTaskExecutor_Handle_LaunchResource(t *testing.T) {
 		assert.True(t, k8serrors.IsNotFound(err))
 	})
 
+	t.Run("jobBadRequest", func(t *testing.T) {
+		// BadRequest (HTTP 400) errors — typically from a validating admission
+		// webhook — are intrinsic to the request payload and not transient.
+		// Retrying with the same input will produce the same rejection.
+		// They should be treated as a permanent failure (PhasePermanentFailure)
+		// rather than the default retryable system error, so workflows surface
+		// the validation error to the user instead of exhausting their retry
+		// budget. See https://github.com/flyteorg/flyte/issues/6531.
+		tctx := getMockTaskContext(PluginPhaseNotStarted, PluginPhaseNotStarted)
+		mockResourceHandler := &pluginsk8sMock.Plugin{}
+		mockResourceHandler.EXPECT().GetProperties().Return(k8s.PluginProperties{})
+		mockResourceHandler.EXPECT().BuildResource(mock.Anything, mock.Anything).Return(&v1.Pod{}, nil)
+		fakeClient := extendedFakeClient{
+			Client:      fake.NewClientBuilder().WithRuntimeObjects().Build(),
+			CreateError: k8serrors.NewBadRequest("admission webhook \"deny.example.com\" denied the request: invalid pod spec"),
+		}
+		mockClientset := k8sfake.NewSimpleClientset()
+
+		pluginManager, err := NewPluginManager(ctx, dummySetupContext(fakeClient), k8s.PluginEntry{
+			ID:              "x",
+			ResourceToWatch: &v1.Pod{},
+			Plugin:          mockResourceHandler,
+		}, NewResourceMonitorIndex(), mockClientset)
+		assert.NoError(t, err)
+
+		transition, err := pluginManager.Handle(ctx, tctx)
+		assert.NoError(t, err)
+		assert.NotNil(t, transition)
+		transitionInfo := transition.Info()
+		assert.NotNil(t, transitionInfo)
+		assert.Equal(t, pluginsCore.PhasePermanentFailure, transitionInfo.Phase())
+		assert.Equal(t, "BadTaskFormat", transitionInfo.Err().GetCode())
+	})
+
+	t.Run("jobInvalid", func(t *testing.T) {
+		// Invalid (HTTP 422) errors indicate the request was well-formed but
+		// the object failed validation (e.g. an invalid field value). Like
+		// BadRequest, this is intrinsic to the payload and not transient, so
+		// it should be a permanent failure.
+		tctx := getMockTaskContext(PluginPhaseNotStarted, PluginPhaseNotStarted)
+		mockResourceHandler := &pluginsk8sMock.Plugin{}
+		mockResourceHandler.EXPECT().GetProperties().Return(k8s.PluginProperties{})
+		mockResourceHandler.EXPECT().BuildResource(mock.Anything, mock.Anything).Return(&v1.Pod{}, nil)
+		fakeClient := extendedFakeClient{
+			Client: fake.NewClientBuilder().WithRuntimeObjects().Build(),
+			CreateError: k8serrors.NewInvalid(
+				schema.GroupKind{Group: "", Kind: "Pod"},
+				"test-pod",
+				nil,
+			),
+		}
+		mockClientset := k8sfake.NewSimpleClientset()
+
+		pluginManager, err := NewPluginManager(ctx, dummySetupContext(fakeClient), k8s.PluginEntry{
+			ID:              "x",
+			ResourceToWatch: &v1.Pod{},
+			Plugin:          mockResourceHandler,
+		}, NewResourceMonitorIndex(), mockClientset)
+		assert.NoError(t, err)
+
+		transition, err := pluginManager.Handle(ctx, tctx)
+		assert.NoError(t, err)
+		assert.NotNil(t, transition)
+		transitionInfo := transition.Info()
+		assert.NotNil(t, transitionInfo)
+		assert.Equal(t, pluginsCore.PhasePermanentFailure, transitionInfo.Phase())
+		assert.Equal(t, "BadTaskFormat", transitionInfo.Err().GetCode())
+	})
+
 	t.Run("Insufficient resource blocking pod creation for the first time", func(t *testing.T) {
 		tctx := getMockTaskContext(PluginPhaseNotStarted, PluginPhaseNotStarted)
 		var tmpl *core.TaskTemplate

flytestdlib/config/tests/accessor_test.go

@@ -89,6 +89,16 @@ type ComplexType struct {
 
 type ComplexTypeArray []ComplexType
 
+type Plugin struct {
+	Name        string            `json:"name"`
+	Annotations map[string]string `json:"annotations"`
+}
+
+type DottedKeysConfig struct {
+	Annotations map[string]string `json:"annotations"`
+	Plugins     []Plugin          `json:"plugins"`
+}
+
 type ConfigWithLists struct {
 	ListOfStuff []ComplexType `json:"list"`
 	StringValue string        `json:"string-val"`
@@ -403,6 +413,27 @@ func TestAccessor_UpdateConfig(t *testing.T) {
 				assert.Equal(t, "xyz1", r.ItemsMap["itemA"]["itemAb"].ID)
 				assert.Equal(t, "xyz2", r.ItemsMap["itemB"]["itemBb"].ID)
 			})
+
+			t.Run("DottedKeysEndToEnd", func(t *testing.T) {
+				root := config.NewRootSection()
+				_, err := root.RegisterSection("dotted-keys", &DottedKeysConfig{})
+				assert.NoError(t, err)
+
+				v := provider(config.Options{
+					SearchPaths: []string{filepath.Join("testdata", "dotted_keys_config.yaml")},
+					RootSection: root,
+				})
+
+				assert.NoError(t, v.UpdateConfig(context.TODO()))
+				r := root.GetSection("dotted-keys").GetConfig().(*DottedKeysConfig)
+
+				assert.Equal(t, "false", r.Annotations["cluster-autoscaler.kubernetes.io/safe-to-evict"])
+				assert.Equal(t, "true", r.Annotations["sidecar.istio.io/inject"])
+
+				assert.Len(t, r.Plugins, 1)
+				assert.Equal(t, "KeepMyCase", r.Plugins[0].Name)
+				assert.Equal(t, "/etc/plugin", r.Plugins[0].Annotations["config.path"])
+			})
 		})
 
 		t.Run(fmt.Sprintf("[%v] Override in Env Var", provider(config.Options{}).ID()), func(t *testing.T) {

flytestdlib/config/tests/testdata/dotted_keys_config.yaml

@@ -0,0 +1,8 @@
+dotted-keys:
+  annotations:
+    cluster-autoscaler.kubernetes.io/safe-to-evict: "false"
+    sidecar.istio.io/inject: "true"
+  plugins:
+    - name: KeepMyCase
+      annotations:
+        config.path: /etc/plugin

flytestdlib/config/viper/viper.go

@@ -289,6 +289,7 @@ func (v viperAccessor) parseViperConfig(root config.Section) error {
 		}
 
 		restoreCaseSensitiveArrayKeys(settings, rawSettings)
+		restoreDottedMapKeys(settings, rawSettings)
 	}
 
 	return v.parseViperConfigRecursive(root, settings)
@@ -324,6 +325,84 @@ func restoreCaseSensitiveArrayKeys(viperData, rawData map[string]interface{}) {
 	}
 }
 
+// restoreDottedMapKeys undoes viper's splitting of dotted YAML keys.
+//
+//   - Viper hard-codes "." as its key delimiter and splits every key on it,
+//     so a YAML leaf key like "test.annotation" arrives in viperData as the
+//     nested map {"test": {"annotation": ...}}.
+//   - rawData is the authoritative source for what the user wrote.
+//   - We walk rawData, detect dotted keys, drop the nested skeleton from
+//     viperData, and reinsert the value under its original dotted key.
+func restoreDottedMapKeys(viperData, rawData map[string]interface{}) {
+	for rawKey, rawVal := range rawData {
+		if strings.Contains(rawKey, keyDelim) {
+			// Drop the nested skeleton viper built from this dotted key, then
+			// reinsert the raw value under the original key. Lowercase the
+			// path because viper lowercases all keys.
+			pruneSplitPath(viperData, strings.Split(strings.ToLower(rawKey), keyDelim))
+			viperData[rawKey] = rawVal
+			continue
+		}
+
+		// rawVal may itself be a map containing dotted keys deeper down
+		// (e.g. annotations: {test.annotation: ...}); recurse to fix them.
+		viperKey, ok := findViperKey(viperData, rawKey)
+		if !ok {
+			continue
+		}
+		if rawMap, ok := rawVal.(map[string]interface{}); ok {
+			if viperMap, ok := viperData[viperKey].(map[string]interface{}); ok {
+				restoreDottedMapKeys(viperMap, rawMap)
+			}
+		}
+	}
+}
+
+// findViperKey returns the key in viperData that case-insensitively matches
+// rawKey. Viper lowercases all keys internally, but raw YAML preserves the
+// original casing, so we must scan.
+func findViperKey(viperData map[string]interface{}, rawKey string) (string, bool) {
+	for k := range viperData {
+		if strings.EqualFold(k, rawKey) {
+			return k, true
+		}
+	}
+	return "", false
+}
+
+// pruneSplitPath removes the nested-map skeleton viper created when it split a
+// dotted key on ".". It walks the path top-down, deleting only nodes that
+// have no remaining children, so sibling keys that share a prefix are left
+// intact (e.g. removing "a.b.c" must not drop "a.b.d").
+func pruneSplitPath(m map[string]interface{}, parts []string) {
+	if len(parts) == 0 {
+		return
+	}
+	head := parts[0]
+	child, ok := m[head]
+	if !ok {
+		// head not in the map m, directly return.
+		return
+	}
+	if len(parts) == 1 {
+		// Leaf of the split path — this is the value viper produced from the
+		// dotted key. Drop it; the caller will reinsert the raw value.
+		delete(m, head)
+		return
+	}
+	childMap, ok := child.(map[string]interface{})
+	if !ok {
+		// Path diverges from what viper would have produced; leave alone.
+		return
+	}
+	pruneSplitPath(childMap, parts[1:])
+	// Drop the parent if recursion emptied it — keeps the cleanup tight so we
+	// don't leave behind empty intermediate maps from the split.
+	if len(childMap) == 0 {
+		delete(m, head)
+	}
+}
+
 func (v viperAccessor) parseViperConfigRecursive(root config.Section, settings interface{}) error {
 	errs := stdLibErrs.ErrorCollection{}
 	var mine interface{}