Add support for using task execution ID fields in log URI templates

go/tasks/logs/logging_utils.go

@@ -18,7 +18,7 @@ type logPlugin struct {
 }
 
 // Internal
-func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, pod *v1.Pod, index uint32, nameSuffix string) ([]*core.TaskLog, error) {
+func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, taskExecID *core.TaskExecutionIdentifier, pod *v1.Pod, index uint32, nameSuffix string, extraLogTemplateVars ...tasklog.TemplateVars) ([]*core.TaskLog, error) {
 	if logPlugin == nil {
 		return nil, nil
 	}
@@ -43,17 +43,19 @@ func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, pod
 
 	logs, err := logPlugin.GetTaskLogs(
 		tasklog.Input{
-			PodName:              pod.Name,
-			PodUID:               string(pod.GetUID()),
-			Namespace:            pod.Namespace,
-			ContainerName:        pod.Spec.Containers[index].Name,
-			ContainerID:          pod.Status.ContainerStatuses[index].ContainerID,
-			LogName:              nameSuffix,
-			PodRFC3339StartTime:  time.Unix(startTime, 0).Format(time.RFC3339),
-			PodRFC3339FinishTime: time.Unix(finishTime, 0).Format(time.RFC3339),
-			PodUnixStartTime:     startTime,
-			PodUnixFinishTime:    finishTime,
+			PodName:                 pod.Name,
+			PodUID:                  string(pod.GetUID()),
+			Namespace:               pod.Namespace,
+			ContainerName:           pod.Spec.Containers[index].Name,
+			ContainerID:             pod.Status.ContainerStatuses[index].ContainerID,
+			LogName:                 nameSuffix,
+			PodRFC3339StartTime:     time.Unix(startTime, 0).Format(time.RFC3339),
+			PodRFC3339FinishTime:    time.Unix(finishTime, 0).Format(time.RFC3339),
+			PodUnixStartTime:        startTime,
+			PodUnixFinishTime:       finishTime,
+			TaskExecutionIdentifier: taskExecID,
 		},
+		extraLogTemplateVars...,
 	)
 
 	if err != nil {
@@ -67,12 +69,18 @@ type taskLogPluginWrapper struct {
 	logPlugins []logPlugin
 }
 
-func (t taskLogPluginWrapper) GetTaskLogs(input tasklog.Input) (logOutput tasklog.Output, err error) {
+func (t taskLogPluginWrapper) GetTaskLogs(input tasklog.Input, extraTemplateVars ...tasklog.TemplateVars) (logOutput tasklog.Output, err error) {
 	logs := make([]*core.TaskLog, 0, len(t.logPlugins))
 	suffix := input.LogName
+
+	mergedTemplateVars := make(tasklog.TemplateVars)
+	if err := mergedTemplateVars.Merge(extraTemplateVars...); err != nil {
+		return tasklog.Output{}, err
+	}
+
 	for _, plugin := range t.logPlugins {
 		input.LogName = plugin.Name + suffix
-		o, err := plugin.Plugin.GetTaskLogs(input)
+		o, err := plugin.Plugin.GetTaskLogs(input, mergedTemplateVars)
 		if err != nil {
 			return tasklog.Output{}, err
 		}

go/tasks/logs/logging_utils_test.go

@@ -14,10 +14,29 @@ import (
 
 const podName = "PodName"
 
+var dummyTaskExecID = &core.TaskExecutionIdentifier{
+	TaskId: &core.Identifier{
+		ResourceType: core.ResourceType_TASK,
+		Name:         "my-task-name",
+		Project:      "my-project",
+		Domain:       "my-domain",
+		Version:      "1",
+	},
+	NodeExecutionId: &core.NodeExecutionIdentifier{
+		NodeId: "n0",
+		ExecutionId: &core.WorkflowExecutionIdentifier{
+			Name:    "my-execution-name",
+			Project: "my-project",
+			Domain:  "my-domain",
+		},
+	},
+	RetryAttempt: 1,
+}
+
 func TestGetLogsForContainerInPod_NoPlugins(t *testing.T) {
 	logPlugin, err := InitializeLogPlugins(&LogConfig{})
 	assert.NoError(t, err)
-	l, err := GetLogsForContainerInPod(context.TODO(), logPlugin, nil, 0, " Suffix")
+	l, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, nil, 0, " Suffix")
 	assert.NoError(t, err)
 	assert.Nil(t, l)
 }
@@ -29,7 +48,7 @@ func TestGetLogsForContainerInPod_NoLogs(t *testing.T) {
 		CloudwatchLogGroup:  "/kubernetes/flyte-production",
 	})
 	assert.NoError(t, err)
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, nil, 0, " Suffix")
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, nil, 0, " Suffix")
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -60,7 +79,7 @@ func TestGetLogsForContainerInPod_BadIndex(t *testing.T) {
 	}
 	pod.Name = podName
 
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 1, " Suffix")
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 1, " Suffix")
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -85,7 +104,7 @@ func TestGetLogsForContainerInPod_MissingStatus(t *testing.T) {
 	}
 	pod.Name = podName
 
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 1, " Suffix")
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 1, " Suffix")
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -115,7 +134,7 @@ func TestGetLogsForContainerInPod_Cloudwatch(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 0, " Suffix")
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 0, " Suffix")
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -145,7 +164,7 @@ func TestGetLogsForContainerInPod_K8s(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 0, " Suffix")
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 0, " Suffix")
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -178,7 +197,7 @@ func TestGetLogsForContainerInPod_All(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 0, " Suffix")
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 0, " Suffix")
 	assert.Nil(t, err)
 	assert.Len(t, logs, 2)
 }
@@ -209,7 +228,7 @@ func TestGetLogsForContainerInPod_Stackdriver(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 0, " Suffix")
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 0, " Suffix")
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -283,7 +302,7 @@ func assertTestSucceeded(tb testing.TB, config *LogConfig, expectedTaskLogs []*c
 		},
 	}
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, pod, 0, " my-Suffix")
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID, pod, 0, " my-Suffix")
 	assert.Nil(tb, err)
 	assert.Len(tb, logs, len(expectedTaskLogs))
 	if diff := deep.Equal(logs, expectedTaskLogs); len(diff) > 0 {
@@ -301,12 +320,24 @@ func TestGetLogsForContainerInPod_Templates(t *testing.T) {
 				},
 				MessageFormat: core.TaskLog_JSON,
 			},
+			{
+				DisplayName: "Internal",
+				TemplateURIs: []string{
+					"https://flyte.corp.net/console/projects/{{ .taskExecution.node_execution_id.execution_id.project }}/domains/{{ .taskExecution.node_execution_id.execution_id.domain }}/executions/{{ .taskExecution.node_execution_id.execution_id.name }}/nodeId/{{ .taskExecution.node_execution_id.node_id }}/taskId/{{ .taskExecution.task_id.name }}/attempt/{{ .taskExecution.retry_attempt }}/view/logs",
+				},
+				MessageFormat: core.TaskLog_JSON,
+			},
 		},
 	}, []*core.TaskLog{
 		{
 			Uri:           "https://my-log-server/my-namespace/my-pod/ContainerName/ContainerID",
 			MessageFormat: core.TaskLog_JSON,
 			Name:          "StackDriver my-Suffix",
 		},
+		{
+			Uri:           "https://flyte.corp.net/console/projects/my-project/domains/my-domain/executions/my-execution-name/nodeId/n0/taskId/my-task-name/attempt/1/view/logs",
+			MessageFormat: core.TaskLog_JSON,
+			Name:          "Internal my-Suffix",
+		},
 	})
 }

go/tasks/pluginmachinery/tasklog/plugin.go

@@ -2,29 +2,32 @@ package tasklog
 
 import "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 
+type TemplateVars map[string]interface{}
+
 // Input contains all available information about task's execution that a log plugin can use to construct task's
 // log links.
 type Input struct {
-	HostName             string `json:"hostname"`
-	PodName              string `json:"podName"`
-	Namespace            string `json:"namespace"`
-	ContainerName        string `json:"containerName"`
-	ContainerID          string `json:"containerId"`
-	LogName              string `json:"logName"`
-	PodRFC3339StartTime  string `json:"podRFC3339StartTime"`
-	PodRFC3339FinishTime string `json:"podRFC3339FinishTime"`
-	PodUnixStartTime     int64  `json:"podUnixStartTime"`
-	PodUnixFinishTime    int64  `json:"podUnixFinishTime"`
-	PodUID               string `json:"podUID"`
+	HostName                string
+	PodName                 string
+	Namespace               string
+	ContainerName           string
+	ContainerID             string
+	LogName                 string
+	PodRFC3339StartTime     string
+	PodRFC3339FinishTime    string
+	PodUnixStartTime        int64
+	PodUnixFinishTime       int64
+	PodUID                  string
+	TaskExecutionIdentifier *core.TaskExecutionIdentifier
 }
 
 // Output contains all task logs a plugin generates for a given Input.
 type Output struct {
-	TaskLogs []*core.TaskLog `json:"taskLogs"`
+	TaskLogs []*core.TaskLog
 }
 
 // Plugin represents an interface for task log plugins to implement to plug generated task log links into task events.
 type Plugin interface {
 	// Generates a TaskLog object given necessary computation information
-	GetTaskLogs(input Input) (logs Output, err error)
+	GetTaskLogs(i Input, v ...TemplateVars) (logs Output, err error)
 }