Render task template in the agent client

go/tasks/plugins/webapi/agent/integration_test.go

@@ -8,6 +8,8 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/utils/strings/slices"
+
 	flyteIdlCore "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 	pluginCore "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core"
 	pluginCoreMocks "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core/mocks"
@@ -39,7 +41,11 @@ type MockPlugin struct {
 type MockClient struct {
 }
 
-func (m *MockClient) CreateTask(_ context.Context, _ *admin.CreateTaskRequest, _ ...grpc.CallOption) (*admin.CreateTaskResponse, error) {
+func (m *MockClient) CreateTask(_ context.Context, createTaskRequest *admin.CreateTaskRequest, _ ...grpc.CallOption) (*admin.CreateTaskResponse, error) {
+	expectedArgs := []string{"pyflyte-fast-execute", "--output-prefix", "fake://bucket/prefix/nhv"}
+	if slices.Equal(createTaskRequest.Template.GetContainer().Args, expectedArgs) {
+		return nil, fmt.Errorf("args not as expected")
+	}
 	return &admin.CreateTaskResponse{ResourceMeta: []byte{1, 2, 3, 4}}, nil
 }
 
@@ -95,6 +101,9 @@ func TestEndToEnd(t *testing.T) {
 	template := flyteIdlCore.TaskTemplate{
 		Type:   "bigquery_query_job_task",
 		Custom: st,
+		Target: &flyteIdlCore.TaskTemplate_Container{
+			Container: &flyteIdlCore.Container{Args: []string{"pyflyte-fast-execute", "--output-prefix", "{{.outputPrefix}}"}},
+		},
 	}
 	basePrefix := storage.DataReference("fake://bucket/prefix/")
 

go/tasks/plugins/webapi/agent/plugin.go

@@ -6,6 +6,8 @@
 	"encoding/gob"
 	"fmt"
 
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core/template"
+
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
 	"github.com/flyteorg/flytestdlib/config"
 	"google.golang.org/grpc/credentials"
@@ -67,7 +69,19 @@
 	if err != nil {
 		return nil, nil, err
 	}
-
+	templateParameters := template.Parameters{
+		TaskExecMetadata: taskCtx.TaskExecutionMetadata(),
+		Inputs:           taskCtx.InputReader(),
+		OutputPath:       taskCtx.OutputWriter(),
+		Task:             taskCtx.TaskReader(),
+	}
+	if taskTemplate.GetContainer() != nil {
+		modifiedArgs, err := template.Render(ctx, taskTemplate.GetContainer().Args, templateParameters)
+		if err != nil {
+			return nil, nil, err
+		}
+		taskTemplate.GetContainer().Args = modifiedArgs
+	}
 	outputPrefix := taskCtx.OutputWriter().GetOutputPrefixPath().String()
 
 	agent, err := getFinalAgent(taskTemplate.Type, p.cfg)