merge with master and remove launchSubTasks

.travis.yml

@@ -16,3 +16,7 @@ jobs:
     install: make install
     name: lint
     script: make lint
+  - stage: test
+    name: unit tests
+    install: make install
+    script: make test_unit_codecov

boilerplate/lyft/golang_test_targets/Makefile

@@ -39,9 +39,15 @@ test_benchmark:
 
 .PHONY: test_unit_cover
 test_unit_cover:
-	go test ./... -coverprofile /tmp/cover.out -covermode=count; go tool cover -func /tmp/cover.out
+	go test ./... -coverprofile /tmp/cover.out -covermode=count
+	go tool cover -func /tmp/cover.out
 
 .PHONY: test_unit_visual
 test_unit_visual:
-	go test ./... -coverprofile /tmp/cover.out -covermode=count; go tool cover -html=/tmp/cover.out 
-
+	go test ./... -coverprofile /tmp/cover.out -covermode=count
+	go tool cover -html=/tmp/cover.out
+
+.PHONY: test_unit_codecov
+test_unit_codecov:
+	go test ./... -race -coverprofile=coverage.txt -covermode=atomic
+	curl -s https://codecov.io/bash > codecov_bash.sh && bash codecov_bash.sh

go/tasks/pluginmachinery/flytek8s/config/config.go

@@ -49,6 +49,8 @@ type K8sPluginConfig struct {
 	// Node Selector Labels for interruptible pods: Similar to InterruptibleTolerations, these node selector labels are added for pods that can tolerate
 	// eviction.
 	InterruptibleNodeSelector map[string]string `json:"interruptible-node-selector" pflag:"-,Defines a set of node selector labels to add to the interruptible pods."`
+	// Scheduler name.
+	SchedulerName string `json:"scheduler-name" pflag:",Defines scheduler name."`
 }
 
 // Retrieves the current k8s plugin config or default.

go/tasks/pluginmachinery/flytek8s/pod_helper.go

@@ -43,6 +43,7 @@ func ToK8sPodSpec(ctx context.Context, taskExecutionMetadata pluginsCore.TaskExe
 			Tolerations:        GetPodTolerations(taskExecutionMetadata.IsInterruptible(), c.Resources),
 			ServiceAccountName: taskExecutionMetadata.GetK8sServiceAccount(),
 			NodeSelector:       config.GetK8sPluginConfig().InterruptibleNodeSelector,
+			SchedulerName:      config.GetK8sPluginConfig().SchedulerName,
 		}, nil
 	}
 	return &v1.PodSpec{
@@ -51,6 +52,7 @@ func ToK8sPodSpec(ctx context.Context, taskExecutionMetadata pluginsCore.TaskExe
 		Containers:         containers,
 		Tolerations:        GetPodTolerations(taskExecutionMetadata.IsInterruptible(), c.Resources),
 		ServiceAccountName: taskExecutionMetadata.GetK8sServiceAccount(),
+		SchedulerName:      config.GetK8sPluginConfig().SchedulerName,
 	}, nil
 
 }
@@ -138,6 +140,8 @@ func DemystifyPending(status v1.PodStatus) (pluginsCore.PhaseInfo, error) {
 							// There are a variety of reasons that can cause a pod to be in this waiting state.
 							// Waiting state may be legitimate when the container is being downloaded, started or init containers are running
 							reason := containerStatus.State.Waiting.Reason
+							finalReason := fmt.Sprintf("%s|%s", c.Reason, reason)
+							finalMessage := fmt.Sprintf("%s|%s", c.Message, containerStatus.State.Waiting.Message)
 							switch reason {
 							case "ErrImagePull", "ContainerCreating", "PodInitializing":
 								// But, there are only two "reasons" when a pod is successfully being created and hence it is in
@@ -149,12 +153,12 @@ func DemystifyPending(status v1.PodStatus) (pluginsCore.PhaseInfo, error) {
 								// ErrImagePull -> Transitionary phase to ImagePullBackOff
 								// ContainerCreating -> Image is being downloaded
 								// PodInitializing -> Init containers are running
-								return pluginsCore.PhaseInfoInitializing(c.LastTransitionTime.Time, pluginsCore.DefaultPhaseVersion, fmt.Sprintf("%s:%s", c.Reason, c.Message)), nil
+								return pluginsCore.PhaseInfoInitializing(c.LastTransitionTime.Time, pluginsCore.DefaultPhaseVersion, fmt.Sprintf("[%s]: %s", finalReason, finalMessage)), nil
 
 							case "CreateContainerError":
 								// This happens if for instance the command to the container is incorrect, ie doesn't run
 								t := c.LastTransitionTime.Time
-								return pluginsCore.PhaseInfoFailure(c.Reason, c.Message, &pluginsCore.TaskInfo{
+								return pluginsCore.PhaseInfoFailure(finalReason, finalMessage, &pluginsCore.TaskInfo{
 									OccurredAt: &t,
 								}), nil
 
@@ -168,7 +172,7 @@ func DemystifyPending(status v1.PodStatus) (pluginsCore.PhaseInfo, error) {
 								// So be default if the container is not waiting with the PodInitializing/ContainerCreating
 								// reasons, then we will assume a failure reason, and fail instantly
 								t := c.LastTransitionTime.Time
-								return pluginsCore.PhaseInfoSystemRetryableFailure(c.Reason, c.Message, &pluginsCore.TaskInfo{
+								return pluginsCore.PhaseInfoSystemRetryableFailure(finalReason, finalMessage, &pluginsCore.TaskInfo{
 									OccurredAt: &t,
 								}), nil
 							}

go/tasks/pluginmachinery/flytek8s/pod_helper_test.go

@@ -2,6 +2,9 @@ package flytek8s
 
 import (
 	"context"
+	"encoding/json"
+	"io/ioutil"
+	"path/filepath"
 	"testing"
 
 	config1 "github.com/lyft/flytestdlib/config"
@@ -443,3 +446,39 @@ func TestConvertPodFailureToError(t *testing.T) {
 		assert.Equal(t, code, "OOMKilled")
 	})
 }
+
+func TestDemystifyPending_testcases(t *testing.T) {
+
+	tests := []struct {
+		name     string
+		filename string
+		isErr    bool
+		errCode  string
+		message  string
+	}{
+		{"ImagePullBackOff", "imagepull-failurepod.json", false, "ContainersNotReady|ImagePullBackOff", "containers with unready status: [fdf98e4ed2b524dc3bf7-get-flyte-id-task-0]|Back-off pulling image \"image\""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			testFile := filepath.Join("testdata", tt.filename)
+			data, err := ioutil.ReadFile(testFile)
+			assert.NoError(t, err, "failed to read file %s", testFile)
+			pod := &v1.Pod{}
+			if assert.NoError(t, json.Unmarshal(data, pod), "failed to unmarshal json in %s. Expected of type v1.Pod", testFile) {
+				p, err := DemystifyPending(pod.Status)
+				if tt.isErr {
+					assert.Error(t, err, "Error expected from method")
+				} else {
+					assert.NoError(t, err, "Error not expected")
+					assert.NotNil(t, p)
+					assert.Equal(t, p.Phase(), pluginsCore.PhaseRetryableFailure)
+					if assert.NotNil(t, p.Err()) {
+						assert.Equal(t, p.Err().Code, tt.errCode)
+						assert.Equal(t, p.Err().Message, tt.message)
+					}
+				}
+			}
+		})
+	}
+}