Dynamic log links

flyteplugins/go/tasks/logs/config.go

@@ -28,8 +28,8 @@ type LogConfig struct {
 	StackdriverLogResourceName string              `json:"stackdriver-logresourcename" pflag:",Name of the logresource in stackdriver"`
 	StackDriverTemplateURI     tasklog.TemplateURI `json:"stackdriver-template-uri" pflag:",Template Uri to use when building stackdriver log links"`
 
-	IsFlyinEnabled   bool                `json:"flyin-enabled" pflag:",Enable Log-links to flyin logs"`
-	FlyinTemplateURI tasklog.TemplateURI `json:"flyin-template-uri" pflag:",Template Uri to use when building flyin log links"`
+	IsDynamicLogLinksEnabled bool                           `json:"dynamic-log-links-enabled" pflag:",Enable dynamic log links"`
+	DynamicLogLinks          map[string]tasklog.TemplateURI `json:"dynamic-log-links" pflag:",Map of dynamic log links"`
 
 	Templates []tasklog.TemplateLogPlugin `json:"templates" pflag:"-,"`
 }

flyteplugins/go/tasks/logs/logging_utils.go

@@ -63,13 +63,14 @@ func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, tas
 }
 
 type templateLogPluginCollection struct {
-	plugins []tasklog.TemplateLogPlugin
+	plugins        []tasklog.TemplateLogPlugin
+	dynamicPlugins []tasklog.TemplateLogPlugin
 }
 
 func (t templateLogPluginCollection) GetTaskLogs(input tasklog.Input) (tasklog.Output, error) {
 	var taskLogs []*core.TaskLog
 
-	for _, plugin := range t.plugins {
+	for _, plugin := range append(t.plugins, t.dynamicPlugins...) {
 		o, err := plugin.GetTaskLogs(input)
 		if err != nil {
 			return tasklog.Output{}, err
@@ -84,6 +85,7 @@ func (t templateLogPluginCollection) GetTaskLogs(input tasklog.Input) (tasklog.O
 func InitializeLogPlugins(cfg *LogConfig) (tasklog.Plugin, error) {
 	// Use a list to maintain order.
 	var plugins []tasklog.TemplateLogPlugin
+	var dynamicPlugins []tasklog.TemplateLogPlugin
 
 	if cfg.IsKubernetesEnabled {
 		if len(cfg.KubernetesTemplateURI) > 0 {
@@ -109,14 +111,20 @@ func InitializeLogPlugins(cfg *LogConfig) (tasklog.Plugin, error) {
 		}
 	}
 
-	if cfg.IsFlyinEnabled {
-		if len(cfg.FlyinTemplateURI) > 0 {
-			plugins = append(plugins, tasklog.TemplateLogPlugin{DisplayName: "Flyin Logs", Scheme: tasklog.TemplateSchemeFlyin, TemplateURIs: []tasklog.TemplateURI{cfg.FlyinTemplateURI}, MessageFormat: core.TaskLog_JSON})
-		} else {
-			plugins = append(plugins, tasklog.TemplateLogPlugin{DisplayName: "Flyin Logs", Scheme: tasklog.TemplateSchemeFlyin, TemplateURIs: []tasklog.TemplateURI{fmt.Sprintf("https://flyin.%s/logs/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}", cfg.GCPProjectName)}, MessageFormat: core.TaskLog_JSON})
+	if cfg.IsDynamicLogLinksEnabled {
+		for logLinkType, dynamicLogLink := range cfg.DynamicLogLinks {
+			dynamicPlugins = append(
+				dynamicPlugins,
+				tasklog.TemplateLogPlugin{
+					DisplayName: fmt.Sprintf("%s logs", logLinkType),
+					Scheme:      tasklog.TemplateSchemeDynamic,
+					DynamicTemplateURIs: []tasklog.DynamicTemplateURI{
+						{TemplateURI: dynamicLogLink, Kind: logLinkType}},
+					MessageFormat: core.TaskLog_JSON,
+				})
 		}
 	}
 
 	plugins = append(plugins, cfg.Templates...)
-	return templateLogPluginCollection{plugins: plugins}, nil
+	return templateLogPluginCollection{plugins: plugins, dynamicPlugins: dynamicPlugins}, nil
 }

flyteplugins/go/tasks/logs/logging_utils_test.go

@@ -352,29 +352,122 @@ func TestGetLogsForContainerInPod_Templates(t *testing.T) {
 }
 
 func TestGetLogsForContainerInPod_Flyin(t *testing.T) {
-	assertTestSucceeded(t,
-		&LogConfig{
-			IsKubernetesEnabled:   true,
-			KubernetesTemplateURI: "https://k8s.com",
-			IsFlyinEnabled:        true,
-			FlyinTemplateURI:      "https://flyin.mydomain.com:{{ .port }}/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+	tests := []struct {
+		name             string
+		config           *LogConfig
+		template         *core.TaskTemplate
+		expectedTaskLogs []*core.TaskLog
+	}{
+		{
+			"Flyin enabled but no task template",
+			&LogConfig{
+				IsDynamicLogLinksEnabled: true,
+				DynamicLogLinks: map[string]tasklog.TemplateURI{
+					"flyin": "https://flyin.mydomain.com:{{ .taskConfig.port }}/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				},
+			},
+			nil,
+			nil,
 		},
-		&core.TaskTemplate{
-			Config: map[string]string{
-				"link_type": "vscode",
-				"port":      "65535",
+		{
+			"Flyin enabled but config not found in task template",
+			&LogConfig{
+				IsDynamicLogLinksEnabled: true,
+				DynamicLogLinks: map[string]tasklog.TemplateURI{
+					"flyin": "https://flyin.mydomain.com:{{ .taskConfig.port }}/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				},
 			},
+			&core.TaskTemplate{},
+			nil,
 		},
-		[]*core.TaskLog{
-			{
-				Uri:           "https://k8s.com",
-				MessageFormat: core.TaskLog_JSON,
-				Name:          "Kubernetes Logs my-Suffix",
+		{
+			"Flyin enabled but no port in task template",
+			&LogConfig{
+				IsDynamicLogLinksEnabled: true,
+				DynamicLogLinks: map[string]tasklog.TemplateURI{
+					"flyin": "https://flyin.mydomain.com:{{ .taskConfig.port }}/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				},
 			},
-			{
-				Uri:           "https://flyin.mydomain.com:65535/my-namespace/my-pod/ContainerName/ContainerID",
-				MessageFormat: core.TaskLog_JSON,
-				Name:          "Flyin Logs my-Suffix",
+			&core.TaskTemplate{
+				Config: map[string]string{
+					"link_type": "vscode",
+				},
+			},
+			[]*core.TaskLog{
+				{
+					Uri:           "https://flyin.mydomain.com:8080/my-namespace/my-pod/ContainerName/ContainerID",
+					MessageFormat: core.TaskLog_JSON,
+					Name:          "flyin logs my-Suffix",
+				},
+			},
+		},
+		{
+			"Flyin disabled but config present in TaskTemplate",
+			&LogConfig{
+				IsDynamicLogLinksEnabled: false,
+			},
+			&core.TaskTemplate{
+				Config: map[string]string{
+					"link_type": "vscode",
+					"port":      "65535",
+				},
 			},
+			nil,
+		},
+		{
+			"Flyin disabled and K8s enabled and flyin config present in TaskTemplate",
+			&LogConfig{
+				IsKubernetesEnabled:      true,
+				KubernetesTemplateURI:    "https://k8s.com/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				IsDynamicLogLinksEnabled: false,
+			},
+			&core.TaskTemplate{
+				Config: map[string]string{
+					"link_type": "vscode",
+					"port":      "65535",
+				},
+			},
+			[]*core.TaskLog{
+				{
+					Uri:           "https://k8s.com/my-namespace/my-pod/ContainerName/ContainerID",
+					MessageFormat: core.TaskLog_JSON,
+					Name:          "Kubernetes Logs my-Suffix",
+				},
+			},
+		},
+		{
+			"Flyin and K8s enabled",
+			&LogConfig{
+				IsKubernetesEnabled:      true,
+				KubernetesTemplateURI:    "https://k8s.com/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				IsDynamicLogLinksEnabled: true,
+				DynamicLogLinks: map[string]tasklog.TemplateURI{
+					"flyin": "https://flyin.mydomain.com:{{ .taskConfig.port }}/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
+				},
+			},
+			&core.TaskTemplate{
+				Config: map[string]string{
+					"link_type": "vscode",
+					"port":      "65535",
+				},
+			},
+			[]*core.TaskLog{
+				{
+					Uri:           "https://k8s.com/my-namespace/my-pod/ContainerName/ContainerID",
+					MessageFormat: core.TaskLog_JSON,
+					Name:          "Kubernetes Logs my-Suffix",
+				},
+				{
+					Uri:           "https://flyin.mydomain.com:65535/my-namespace/my-pod/ContainerName/ContainerID",
+					MessageFormat: core.TaskLog_JSON,
+					Name:          "flyin logs my-Suffix",
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assertTestSucceeded(t, tt.config, tt.template, tt.expectedTaskLogs)
 		})
+	}
 }

flyteplugins/go/tasks/pluginmachinery/tasklog/plugin.go

@@ -14,12 +14,17 @@ type TemplateScheme int
 const (
 	TemplateSchemePod TemplateScheme = iota
 	TemplateSchemeTaskExecution
-	TemplateSchemeFlyin
+	TemplateSchemeDynamic
 )
 
 // TemplateURI is a URI that accepts templates. See: go/tasks/pluginmachinery/tasklog/template.go for available templates.
 type TemplateURI = string
 
+type DynamicTemplateURI struct {
+	TemplateURI TemplateURI
+	Kind        string // Kind describes the kind of the dynamic template. Currently only "flyin" is supported.
+}
+
 type TemplateVar struct {
 	Regex *regexp.Regexp
 	Value string
@@ -31,7 +36,6 @@ type TemplateVarsByScheme struct {
 	Common        TemplateVars
 	Pod           TemplateVars
 	TaskExecution TemplateVars
-	Flyin         TemplateVars
 }
 
 // Input contains all available information about task's execution that a log plugin can use to construct task's
@@ -65,8 +69,9 @@ type Plugin interface {
 }
 
 type TemplateLogPlugin struct {
-	DisplayName   string                     `json:"displayName" pflag:",Display name for the generated log when displayed in the console."`
-	TemplateURIs  []TemplateURI              `json:"templateUris" pflag:",URI Templates for generating task log links."`
-	MessageFormat core.TaskLog_MessageFormat `json:"messageFormat" pflag:",Log Message Format."`
-	Scheme        TemplateScheme             `json:"scheme" pflag:",Templating scheme to use. Supported values are Pod and TaskExecution."`
+	DisplayName         string                     `json:"displayName" pflag:",Display name for the generated log when displayed in the console."`
+	TemplateURIs        []TemplateURI              `json:"templateUris" pflag:",URI Templates for generating task log links."`
+	DynamicTemplateURIs []DynamicTemplateURI       `json:"dynamicTemplateUris" pflag:",Dynamic URI Templates for generating task log links."`
+	MessageFormat       core.TaskLog_MessageFormat `json:"messageFormat" pflag:",Log Message Format."`
+	Scheme              TemplateScheme             `json:"scheme" pflag:",Templating scheme to use. Supported values are Pod and TaskExecution."`
 }