Transition to failed state if the k8s resource cannot be created

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

@@ -188,19 +188,19 @@ func (e *PluginManager) LaunchResource(ctx context.Context, tCtx pluginsCore.Tas
 
 	tmpl, err := tCtx.TaskReader().Read(ctx)
 	if err != nil {
-		return pluginsCore.Transition{}, err
+		return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("Failed to read task template", err.Error(), nil)), nil
 	}
 
 	k8sTaskCtxMetadata, err := newTaskExecutionMetadata(tCtx.TaskExecutionMetadata(), tmpl)
 	if err != nil {
-		return pluginsCore.Transition{}, err
+		return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("Failed to create task context", err.Error(), nil)), nil
 	}
 
 	k8sTaskCtx := newTaskExecutionContext(tCtx, k8sTaskCtxMetadata)
 
 	o, err := e.plugin.BuildResource(ctx, k8sTaskCtx)
 	if err != nil {
-		return pluginsCore.UnknownTransition, err
+		return pluginsCore.DoTransition(pluginsCore.PhaseInfoFailure("Failed to build k8s resource", err.Error(), nil)), nil
 	}
 
 	e.AddObjectMetadata(k8sTaskCtxMetadata, o, config.GetK8sPluginConfig())

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

@@ -306,6 +306,28 @@ func TestK8sTaskExecutor_Handle_LaunchResource(t *testing.T) {
 		assert.NoError(t, fakeClient.Delete(ctx, createdPod))
 	})
 
+	t.Run("failed to build k8s resource", func(t *testing.T) {
+		tCtx := getMockTaskContext(PluginPhaseNotStarted, PluginPhaseStarted)
+		// common setup code
+		mockResourceHandler := &pluginsk8sMock.Plugin{}
+		mockResourceHandler.OnGetProperties().Return(k8s.PluginProperties{})
+		mockResourceHandler.OnBuildResourceMatch(mock.Anything, mock.Anything).Return(nil, errors.New("failed"))
+		fakeClient := fake.NewClientBuilder().WithRuntimeObjects().Build()
+		pluginManager, err := NewPluginManager(ctx, dummySetupContext(fakeClient), k8s.PluginEntry{
+			ID:              "x",
+			ResourceToWatch: &v1.Pod{},
+			Plugin:          mockResourceHandler,
+		}, NewResourceMonitorIndex())
+		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())
+	})
+
 	t.Run("jobAlreadyExists", func(t *testing.T) {
 		tctx := getMockTaskContext(PluginPhaseNotStarted, PluginPhaseStarted)
 		// common setup code