feat: support addTask and startTask for k8 RP [DET-3416, DET-3419] #798

aaron276h · 2020-06-30T17:54:37Z

Description

This PR supports addTask and startTask for Kubernetes. It includes the initial implementation of the Kubernetes RP actor, the pods actor, and the pod actor. So far these actors take task spec and translate it to a pod spec amd submit the pod spec to k8.

Test Plan

Tested manually by running on a k8 cluster. Adding CI will be part of M4.

Commentary

One thing that I wish we could do better is not using scheduler/agentState to represent the Kubernetes resources in the scheduler. However, refactoring this will be a significant amount of work that we will plan to complete as future tech debt work.

dzhu · 2020-07-07T19:46:23Z

master/internal/core.go

@@ -267,6 +267,39 @@ func (m *Master) rwCoordinatorWebSocket(socket *websocket.Conn, c echo.Context)
 	return actorRef.AwaitTermination()
 }

+func (m *Master) initializeResourceProviders(proxyRef *actor.Ref, provisionerSlotsPerInstance int) {
+	var resourceProvider *actor.Ref
+	if m.config.Scheduler.ResourceProvider.DefaultRPConfig != nil {


blocking: Be more explicit about selecting which case to use:

switch { case ...DefaultRPConfig != nil: ... case ...KubernetesRPConfig != nil: ... default: panic("...") }

dzhu · 2020-07-07T20:20:39Z

master/pkg/etc/etc.go

@@ -28,6 +28,8 @@ const (
 	TrialEntrypointScriptResource = "entrypoint.sh"
 	// AgentSetupScriptTemplateResource is the template for the script to run a dynamic agent.
 	AgentSetupScriptTemplateResource = "agent_setup_script.sh.template"
+	// InitContainerEntryScriptResource is the script to run the init container on k8s.
+	InitContainerEntryScriptResource = "init_container_entrypoint.sh"


non-blocking: Put something about Kubernetes into the name?

dzhu · 2020-07-07T20:32:14Z

master/internal/kubernetes/pods.go

+	)
+
+	if !ok {
+		return errors.Errorf(fmt.Sprintf("pod actor %s already exits", ref.Address().String()))


Suggested change

return errors.Errorf(fmt.Sprintf("pod actor %s already exits", ref.Address().String()))

return errors.Errorf("pod actor %s already exists", ref.Address().String())

dzhu · 2020-07-07T22:11:04Z

master/internal/scheduler/assignment.go

-	return newTaskSummary(a.task)
+}
+
+type podAssignment struct {


non-blocking: Probably these (both this and containerAssignment) should just be moved into the respective *_resource_provider.go files. That feels like a more natural grouping to me.

moved them over, although it doesn't seem that great adding this to the default_resource_provider as that has bloated to 600+ lines.

dzhu · 2020-07-08T02:10:51Z

master/internal/scheduler/kubernetes_resource_provider.go

+		k.receiveStartTask(ctx, msg)
+
+	default:
+		ctx.Log().Error("Unexpected message", reflect.TypeOf(msg))


It looks like we don't normally bother logging these, so I'd probably just drop this (and the other instance in the PR). But if you really want to have it here, it should use %T rather than reflect.

Yeah I find these logs useful, especially during development. I updated to not use reflect.

master/internal/kubernetes/pod.go

As part of this change added a pods actor that is part of the k8 RP, and added several additional configurations to the k8 RP config.

Translate task spec to pod spec and submit to k8.

aaron276h

@dzhu I made the changes as a separate new commit.

aaron276h · 2020-07-09T11:20:36Z

master/internal/core.go

@@ -267,6 +267,39 @@ func (m *Master) rwCoordinatorWebSocket(socket *websocket.Conn, c echo.Context)
 	return actorRef.AwaitTermination()
 }

+func (m *Master) initializeResourceProviders(proxyRef *actor.Ref, provisionerSlotsPerInstance int) {
+	var resourceProvider *actor.Ref
+	if m.config.Scheduler.ResourceProvider.DefaultRPConfig != nil {


master/internal/kubernetes/pod.go

aaron276h · 2020-07-09T11:30:45Z

master/internal/kubernetes/pods.go

+	)
+
+	if !ok {
+		return errors.Errorf(fmt.Sprintf("pod actor %s already exits", ref.Address().String()))


aaron276h · 2020-07-09T11:35:33Z

master/internal/scheduler/assignment.go

-	return newTaskSummary(a.task)
+}
+
+type podAssignment struct {


moved them over, although it doesn't seem that great adding this to the default_resource_provider as that has bloated to 600+ lines.

aaron276h · 2020-07-09T11:46:45Z

master/internal/scheduler/kubernetes_resource_provider.go

+		k.receiveStartTask(ctx, msg)
+
+	default:
+		ctx.Log().Error("Unexpected message", reflect.TypeOf(msg))


Yeah I find these logs useful, especially during development. I updated to not use reflect.

aaron276h · 2020-07-09T11:48:00Z

master/pkg/etc/etc.go

@@ -28,6 +28,8 @@ const (
 	TrialEntrypointScriptResource = "entrypoint.sh"
 	// AgentSetupScriptTemplateResource is the template for the script to run a dynamic agent.
 	AgentSetupScriptTemplateResource = "agent_setup_script.sh.template"
+	// InitContainerEntryScriptResource is the script to run the init container on k8s.
+	InitContainerEntryScriptResource = "init_container_entrypoint.sh"


master/internal/kubernetes/pod.go

dzhu

Mostly nitpicks; see kubernetes_resource_provider.go and volumes.go for some that may actually matter.

dzhu · 2020-07-10T19:37:15Z

master/internal/scheduler/kubernetes_resource_provider.go

+func (k *kubernetesResourceProvider) receiveAddTask(ctx *actor.Context, msg AddTask) {
+	actors.NotifyOnStop(ctx, msg.TaskHandler, taskStopped{Ref: msg.TaskHandler})
+
+	if task, ok := k.tasksByHandler[ctx.Sender()]; ok {


blocking: Should this be msg.TaskHandler? Isn't the sender always the resource providers actor due to forwarding?

Oh really good catch. Had the same mistake in the default RP also.

dzhu · 2020-07-10T19:45:40Z

master/internal/scheduler/kubernetes_resource_provider.go

+
+		if task.SlotsNeeded()%k.slotsPerNode != 0 {
+			ctx.Log().WithField("task ID", task.ID).Error(
+				"task number of slots is not schedulable on the configured slots_per_node")


non-blocking: This could bear to be more explicit:

Suggested change

"task number of slots is not schedulable on the configured slots_per_node")

"task is not schedulable: number of slots (%d) is not a multiple of slots_per_node (%d)", task.SlotsNeeded(), k.slotsPerNode)

(adjusted for line length)

dzhu · 2020-07-10T21:01:30Z

master/internal/kubernetes/pod.go

+		},
+		Spec: v1.PodSpec{
+			Volumes:        volumes,
+			HostNetwork:    p.taskSpec.TaskContainerDefaults.NetworkMode == hostNetwork,


blocking:

Suggested change

HostNetwork: p.taskSpec.TaskContainerDefaults.NetworkMode == hostNetwork,

HostNetwork: p.taskSpec.TaskContainerDefaults.NetworkMode.IsHost(),

oh yeah good idea

dzhu · 2020-07-11T00:12:21Z

master/internal/kubernetes/pod.go

+	"github.com/docker/docker/api/types/mount"
+
+	"github.com/determined-ai/determined/master/pkg/container"
+


non-blocking: Put all the internal imports into one block.

dzhu · 2020-07-11T00:12:56Z

master/internal/kubernetes/pod.go

+	"github.com/determined-ai/determined/master/pkg/model"
+	"github.com/determined-ai/determined/master/pkg/tasks"
+
+	v1 "k8s.io/api/core/v1"


non-blocking: corev1 or k8sv1, maybe?

dzhu · 2020-07-11T01:22:43Z

master/pkg/tasks/ports.go

+// LocalRendezvousPort is the start of the range of ports used for rendezvous by tasks.
+const LocalRendezvousPort = 1734
+
+// LocalRendezvousPortOffset is the offset for rendezvous ports.


non-blocking: Just "the offset" isn't super clear. Maybe "...the difference between the two rendezvous ports for each container"?

dzhu · 2020-07-11T01:23:13Z

master/pkg/tasks/task.go

@@ -18,16 +18,17 @@ import (
 )

 const (
-	containerWorkDir  = "/run/determined/workdir"
+	// ContainerWorkDir is working directory for containers.


Suggested change

// ContainerWorkDir is working directory for containers.

// ContainerWorkDir is the working directory for tasks.

dzhu · 2020-07-11T01:26:37Z

master/pkg/tasks/task.go

+	return user
+}
+
+// ConfigureCommandEnvVars configures environment variables for cmd tasks.


non-blocking: ConfigureXXX for all of the new functions here feels a bit off, since that sounds like something that actually goes out and tweaks settings somewhere as part of the function. I think just the XXX part usually sounds good (CommandEnvVars, CommandArchives, etc.), or maybe XXXConfiguration if that's too bare.

yep, updated these

dzhu · 2020-07-11T01:28:11Z

master/pkg/tasks/task.go

+	return envVarsMap
+}
+
+// ConfigureCommandArchives returns the additional files for a c as an archive.


Suggested change

// ConfigureCommandArchives returns the additional files for a c as an archive.

// ConfigureCommandArchives returns the additional files for a command as an archive.

dzhu · 2020-07-11T01:31:14Z

master/pkg/tasks/task.go

-	envVars := defaultEnvVars()
-	envVars = append(envVars, gcc.ExperimentConfig.Environment.EnvironmentVariables.For(deviceType)...)
-	mounts := toDockerMounts(gcc.ExperimentConfig.BindMounts)
+// ConfigureGCEnvVars returns environment variables for gc.


non-blocking:

Suggested change

// ConfigureGCEnvVars returns environment variables for gc.

// ConfigureGCEnvVars returns environment variables for checkpoint GC.

(Also several spots below.)

dzhu · 2020-07-12T04:23:48Z

master/internal/scheduler/kubernetes_resource_provider.go

+}
+
+func (k *kubernetesResourceProvider) getOrCreateGroup(
+	handler *actor.Ref,


non-blocking: Let's keep the context first everywhere it's passed as an argument.

aaron276h requested a review from dzhu June 30, 2020 17:54

aaron276h assigned dzhu Jun 30, 2020

cla-bot bot added the cla-signed label Jun 30, 2020

aaron276h force-pushed the DET-3419 branch from 7c6b58a to b4ed6e5 Compare June 30, 2020 18:43

aaron276h mentioned this pull request Jul 7, 2020

feat: support logs and pod lifecycles for kubernetes [DET-3420, DET-3417] #839

Closed

dzhu reviewed Jul 9, 2020

View reviewed changes

aaron276h assigned aaron276h and unassigned dzhu Jul 9, 2020

aaron276h added 6 commits July 9, 2020 07:59

feat: add kubernetes resource provider

b8fd5e7

feat: add k8 clienset initilization

38f9d1b

As part of this change added a pods actor that is part of the k8 RP, and added several additional configurations to the k8 RP config.

feat: support addTask in k8s RP

0033545

feat: support startTask for k8 RP

7eb1661

Translate task spec to pod spec and submit to k8.

chore: remove out of cluster config for k8

601dc0c

fix: use backets for slot ids env variable

3ce08b2

aaron276h force-pushed the DET-3419 branch from 97091a1 to 4687c9e Compare July 9, 2020 14:31

aaron276h commented Jul 9, 2020

View reviewed changes

aaron276h assigned dzhu and unassigned aaron276h Jul 9, 2020

aaron276h added 2 commits July 10, 2020 10:37

chore: addressing review comments

cc1e008

chore: add comment for kubernetes package

7fcbcc7

aaron276h force-pushed the DET-3419 branch from 5cb850d to 7fcbcc7 Compare July 10, 2020 14:38

aaron276h mentioned this pull request Jul 10, 2020

feat: support terminating tasks in k8 RP [DET-3418] #865

Merged

dzhu approved these changes Jul 11, 2020

View reviewed changes

dzhu assigned aaron276h and unassigned dzhu Jul 11, 2020

dzhu reviewed Jul 12, 2020

View reviewed changes

chore: addressing review feedback

589d7dc

aaron276h merged commit 9e1664c into determined-ai:master Jul 13, 2020

dannysauer added this to the 0.12.12 milestone Feb 6, 2024

	return errors.Errorf(fmt.Sprintf("pod actor %s already exits", ref.Address().String()))
	return errors.Errorf("pod actor %s already exists", ref.Address().String())

	"task number of slots is not schedulable on the configured slots_per_node")
	"task is not schedulable: number of slots (%d) is not a multiple of slots_per_node (%d)", task.SlotsNeeded(), k.slotsPerNode)

	HostNetwork: p.taskSpec.TaskContainerDefaults.NetworkMode == hostNetwork,
	HostNetwork: p.taskSpec.TaskContainerDefaults.NetworkMode.IsHost(),

		"github.com/docker/docker/api/types/mount"

		"github.com/determined-ai/determined/master/pkg/container"

	// ContainerWorkDir is working directory for containers.
	// ContainerWorkDir is the working directory for tasks.

	// ConfigureCommandArchives returns the additional files for a c as an archive.
	// ConfigureCommandArchives returns the additional files for a command as an archive.

	// ConfigureGCEnvVars returns environment variables for gc.
	// ConfigureGCEnvVars returns environment variables for checkpoint GC.

feat: support addTask and startTask for k8 RP [DET-3416, DET-3419] #798

feat: support addTask and startTask for k8 RP [DET-3416, DET-3419] #798

Conversation

aaron276h commented Jun 30, 2020

Description

Test Plan

Commentary

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aaron276h left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dzhu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment