Revert "Add Flyin propeller config (#4610)"

This reverts commit 3501675.

Signed-off-by: Eduardo Apolinario <[email protected]>

flyteplugins/go/tasks/logs/config.go

@@ -28,9 +28,6 @@ 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"`
-
 	Templates []tasklog.TemplateLogPlugin `json:"templates" pflag:"-,"`
 }
 

flyteplugins/go/tasks/logs/logging_utils.go

@@ -14,7 +14,7 @@ import (
 )
 
 // Internal
-func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, taskExecID pluginsCore.TaskExecutionID, pod *v1.Pod, index uint32, nameSuffix string, extraLogTemplateVarsByScheme *tasklog.TemplateVarsByScheme, taskTemplate *core.TaskTemplate) ([]*core.TaskLog, error) {
+func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, taskExecID pluginsCore.TaskExecutionID, pod *v1.Pod, index uint32, nameSuffix string, extraLogTemplateVarsByScheme *tasklog.TemplateVarsByScheme) ([]*core.TaskLog, error) {
 	if logPlugin == nil {
 		return nil, nil
 	}
@@ -51,7 +51,6 @@ func GetLogsForContainerInPod(ctx context.Context, logPlugin tasklog.Plugin, tas
 			PodUnixFinishTime:         finishTime,
 			TaskExecutionID:           taskExecID,
 			ExtraTemplateVarsByScheme: extraLogTemplateVarsByScheme,
-			TaskTemplate:              taskTemplate,
 		},
 	)
 
@@ -109,14 +108,6 @@ 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})
-		}
-	}
-
 	plugins = append(plugins, cfg.Templates...)
 	return templateLogPluginCollection{plugins: plugins}, nil
 }

flyteplugins/go/tasks/logs/logging_utils_test.go

@@ -44,7 +44,7 @@ func dummyTaskExecID() pluginCore.TaskExecutionID {
 func TestGetLogsForContainerInPod_NoPlugins(t *testing.T) {
 	logPlugin, err := InitializeLogPlugins(&LogConfig{})
 	assert.NoError(t, err)
-	l, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), nil, 0, " Suffix", nil, nil)
+	l, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), nil, 0, " Suffix", nil)
 	assert.NoError(t, err)
 	assert.Nil(t, l)
 }
@@ -56,7 +56,7 @@ func TestGetLogsForContainerInPod_NoLogs(t *testing.T) {
 		CloudwatchLogGroup:  "/kubernetes/flyte-production",
 	})
 	assert.NoError(t, err)
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), nil, 0, " Suffix", nil, nil)
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), nil, 0, " Suffix", nil)
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -87,7 +87,7 @@ func TestGetLogsForContainerInPod_BadIndex(t *testing.T) {
 	}
 	pod.Name = podName
 
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 1, " Suffix", nil, nil)
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 1, " Suffix", nil)
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -112,7 +112,7 @@ func TestGetLogsForContainerInPod_MissingStatus(t *testing.T) {
 	}
 	pod.Name = podName
 
-	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 1, " Suffix", nil, nil)
+	p, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 1, " Suffix", nil)
 	assert.NoError(t, err)
 	assert.Nil(t, p)
 }
@@ -142,7 +142,7 @@ func TestGetLogsForContainerInPod_Cloudwatch(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil, nil)
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil)
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -172,7 +172,7 @@ func TestGetLogsForContainerInPod_K8s(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil, nil)
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil)
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -205,7 +205,7 @@ func TestGetLogsForContainerInPod_All(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil, nil)
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil)
 	assert.Nil(t, err)
 	assert.Len(t, logs, 2)
 }
@@ -236,7 +236,7 @@ func TestGetLogsForContainerInPod_Stackdriver(t *testing.T) {
 	}
 	pod.Name = podName
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil, nil)
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " Suffix", nil)
 	assert.Nil(t, err)
 	assert.Len(t, logs, 1)
 }
@@ -252,7 +252,7 @@ func TestGetLogsForContainerInPod_LegacyTemplate(t *testing.T) {
 
 			IsStackDriverEnabled:   true,
 			StackDriverTemplateURI: "https://sd-my-log-server/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
-		}, nil, []*core.TaskLog{
+		}, []*core.TaskLog{
 			{
 				Uri:           "https://k8s-my-log-server/my-namespace/my-pod/ContainerName/ContainerID",
 				MessageFormat: core.TaskLog_JSON,
@@ -275,7 +275,7 @@ func TestGetLogsForContainerInPod_LegacyTemplate(t *testing.T) {
 		assertTestSucceeded(t, &LogConfig{
 			IsStackDriverEnabled:   true,
 			StackDriverTemplateURI: "https://sd-my-log-server/{{ .namespace }}/{{ .podName }}/{{ .containerName }}/{{ .containerId }}",
-		}, nil, []*core.TaskLog{
+		}, []*core.TaskLog{
 			{
 				Uri:           "https://sd-my-log-server/my-namespace/my-pod/ContainerName/ContainerID",
 				MessageFormat: core.TaskLog_JSON,
@@ -285,7 +285,7 @@ func TestGetLogsForContainerInPod_LegacyTemplate(t *testing.T) {
 	})
 }
 
-func assertTestSucceeded(tb testing.TB, config *LogConfig, taskTemplate *core.TaskTemplate, expectedTaskLogs []*core.TaskLog) {
+func assertTestSucceeded(tb testing.TB, config *LogConfig, expectedTaskLogs []*core.TaskLog) {
 	logPlugin, err := InitializeLogPlugins(config)
 	assert.NoError(tb, err)
 
@@ -310,7 +310,7 @@ func assertTestSucceeded(tb testing.TB, config *LogConfig, taskTemplate *core.Ta
 		},
 	}
 
-	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " my-Suffix", nil, taskTemplate)
+	logs, err := GetLogsForContainerInPod(context.TODO(), logPlugin, dummyTaskExecID(), pod, 0, " my-Suffix", nil)
 	assert.Nil(tb, err)
 	assert.Len(tb, logs, len(expectedTaskLogs))
 	if diff := deep.Equal(logs, expectedTaskLogs); len(diff) > 0 {
@@ -337,7 +337,7 @@ func TestGetLogsForContainerInPod_Templates(t *testing.T) {
 				Scheme:        tasklog.TemplateSchemeTaskExecution,
 			},
 		},
-	}, nil, []*core.TaskLog{
+	}, []*core.TaskLog{
 		{
 			Uri:           "https://my-log-server/my-namespace/my-pod/ContainerName/ContainerID",
 			MessageFormat: core.TaskLog_JSON,
@@ -350,31 +350,3 @@ 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 }}",
-		},
-		&core.TaskTemplate{
-			Config: map[string]string{
-				"link_type": "vscode",
-				"port":      "65535",
-			},
-		},
-		[]*core.TaskLog{
-			{
-				Uri:           "https://k8s.com",
-				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",
-			},
-		})
-}

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

@@ -14,7 +14,6 @@ type TemplateScheme int
 const (
 	TemplateSchemePod TemplateScheme = iota
 	TemplateSchemeTaskExecution
-	TemplateSchemeFlyin
 )
 
 // TemplateURI is a URI that accepts templates. See: go/tasks/pluginmachinery/tasklog/template.go for available templates.
@@ -31,7 +30,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
@@ -50,7 +48,6 @@ type Input struct {
 	PodUID                    string
 	TaskExecutionID           pluginsCore.TaskExecutionID
 	ExtraTemplateVarsByScheme *TemplateVarsByScheme
-	TaskTemplate              *core.TaskTemplate
 }
 
 // Output contains all task logs a plugin generates for a given Input.

flyteplugins/go/tasks/pluginmachinery/tasklog/template.go

@@ -35,7 +35,6 @@ type templateRegexes struct {
 	ExecutionProject     *regexp.Regexp
 	ExecutionDomain      *regexp.Regexp
 	GeneratedName        *regexp.Regexp
-	Port                 *regexp.Regexp
 }
 
 func initDefaultRegexes() templateRegexes {
@@ -61,7 +60,6 @@ func initDefaultRegexes() templateRegexes {
 		MustCreateRegex("executionProject"),
 		MustCreateRegex("executionDomain"),
 		MustCreateRegex("generatedName"),
-		MustCreateRegex("port"),
 	}
 }
 
@@ -87,16 +85,6 @@ func (input Input) templateVarsForScheme(scheme TemplateScheme) TemplateVars {
 	}
 
 	switch scheme {
-	case TemplateSchemeFlyin:
-		port := input.TaskTemplate.GetConfig()["port"]
-		if port == "" {
-			port = "8080"
-		}
-		vars = append(
-			vars,
-			TemplateVar{defaultRegexes.Port, port},
-		)
-		fallthrough
 	case TemplateSchemePod:
 		// Container IDs are prefixed with docker://, cri-o://, etc. which is stripped by fluentd before pushing to a log
 		// stream. Therefore, we must also strip the prefix.
@@ -193,22 +181,7 @@ func (input Input) templateVarsForScheme(scheme TemplateScheme) TemplateVars {
 func (p TemplateLogPlugin) GetTaskLogs(input Input) (Output, error) {
 	templateVars := input.templateVarsForScheme(p.Scheme)
 	taskLogs := make([]*core.TaskLog, 0, len(p.TemplateURIs))
-
-	// Grab metadata from task template and check if key "link_type" is set to "vscode".
-	// If so, add a vscode link to the task logs.
-	isFlyin := false
-	if input.TaskTemplate != nil && input.TaskTemplate.GetConfig() != nil {
-		config := input.TaskTemplate.GetConfig()
-		if config != nil && config["link_type"] == "vscode" {
-			isFlyin = true
-		}
-	}
 	for _, templateURI := range p.TemplateURIs {
-		// Skip Flyin logs if plugin is enabled but no metadata is defined in input's task template.
-		// This is to prevent Flyin logs from being generated for tasks that don't have a Flyin metadata section.
-		if p.DisplayName == "Flyin Logs" && !isFlyin {
-			continue
-		}
 		taskLogs = append(taskLogs, &core.TaskLog{
 			Uri:           replaceAll(templateURI, templateVars),
 			Name:          p.DisplayName + input.LogName,