package nomad import ( "fmt" "strconv" "strings" "time" nomad "github.com/hashicorp/nomad/api" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // createNomadTasks takes the containers in a kubernetes pod and creates // a list of Nomad tasks from them. func (p *Provider) createNomadTasks(pod *v1.Pod) []*nomad.Task { nomadTasks := make([]*nomad.Task, 0, len(pod.Spec.Containers)) for _, ctr := range pod.Spec.Containers { portMap, networkResourcess := createPortMap(ctr.Ports) image := ctr.Image labels := pod.Labels command := ctr.Command args := ctr.Args resources := createResources(ctr.Resources.Limits, networkResourcess) envVars := createEnvVars(ctr.Env) task := nomad.Task{ Name: ctr.Name, Driver: "docker", Config: map[string]interface{}{ "image": image, "port_map": portMap, "labels": labels, // TODO: Add volumes support "command": strings.Join(command, ""), "args": args, }, Resources: resources, Env: envVars, } nomadTasks = append(nomadTasks, &task) } return nomadTasks } func createPortMap(ports []v1.ContainerPort) ([]map[string]interface{}, []*nomad.NetworkResource) { var portMap []map[string]interface{} var dynamicPorts []nomad.Port var networkResources []*nomad.NetworkResource for i, port := range ports { portName := fmt.Sprintf("port_%s", strconv.Itoa(i+1)) if port.Name != "" { portName = port.Name } portMap = append(portMap, map[string]interface{}{portName: port.ContainerPort}) dynamicPorts = append(dynamicPorts, nomad.Port{Label: portName}) } return portMap, append(networkResources, &nomad.NetworkResource{DynamicPorts: dynamicPorts}) } func createResources(limits v1.ResourceList, networkResources []*nomad.NetworkResource) *nomad.Resources { taskMemory := int(limits.Memory().Value()) taskCPU := int(limits.Cpu().Value()) if taskMemory == 0 { taskMemory = 128 } if taskCPU == 0 { taskCPU = 100 } return &nomad.Resources{ Networks: networkResources, MemoryMB: &taskMemory, CPU: &taskCPU, } } func createEnvVars(podEnvVars []v1.EnvVar) map[string]string { envVars := map[string]string{} for _, v := range podEnvVars { envVars[v.Name] = v.Value } return envVars } func (p *Provider) createTaskGroups(name string, tasks []*nomad.Task) []*nomad.TaskGroup { count := 1 restartDelay := 1 * time.Second restartMode := "delay" restartAttempts := 25 return []*nomad.TaskGroup{ &nomad.TaskGroup{ Name: &name, Count: &count, RestartPolicy: &nomad.RestartPolicy{ Delay: &restartDelay, Mode: &restartMode, Attempts: &restartAttempts, }, Tasks: tasks, }, } } func (p *Provider) createJob(name string, datacenters []string, taskGroups []*nomad.TaskGroup) *nomad.Job { jobName := fmt.Sprintf("%s-%s", jobNamePrefix, name) // Create a new nomad job job := nomad.NewServiceJob(jobName, jobName, p.nomadRegion, 100) job.Datacenters = datacenters job.TaskGroups = taskGroups return job } func (p *Provider) jobToPod(job *nomad.Job, allocs []*nomad.AllocationListStub) (*v1.Pod, error) { containers := []v1.Container{} containerStatues := []v1.ContainerStatus{} jobStatus := *job.Status jobCreatedAt := *job.SubmitTime podCondition := convertJobStatusToPodCondition(jobStatus) containerStatusesMap := allocToContainerStatuses(allocs) // containerPorts are specified for task in a task // group var containerPorts []v1.ContainerPort for _, tg := range job.TaskGroups { for _, task := range tg.Tasks { for _, taskNetwork := range task.Resources.Networks { for _, dynamicPort := range taskNetwork.DynamicPorts { // TODO: Dynamic ports aren't being reported via the // Nomad `/jobs` endpoint. containerPorts = append(containerPorts, v1.ContainerPort{ Name: dynamicPort.Label, HostPort: int32(dynamicPort.Value), HostIP: taskNetwork.IP, }) } } containers = append(containers, v1.Container{ Name: task.Name, Image: fmt.Sprintf("%s", task.Config["image"]), Command: strings.Split(fmt.Sprintf("%s", task.Config["command"]), ""), Args: strings.Split(fmt.Sprintf("%s", task.Config["args"]), " "), Ports: containerPorts, }) containerStatus := containerStatusesMap[task.Name] containerStatus.Image = fmt.Sprintf("%s", task.Config["image"]) containerStatus.ImageID = fmt.Sprintf("%s", task.Config["image"]) containerStatues = append(containerStatues, containerStatus) } } pod := v1.Pod{ TypeMeta: metav1.TypeMeta{ Kind: "Pod", APIVersion: "v1", }, ObjectMeta: metav1.ObjectMeta{ Name: *job.Name, Namespace: "default", CreationTimestamp: metav1.NewTime(time.Unix(jobCreatedAt, 0)), }, Spec: v1.PodSpec{ NodeName: p.nodeName, Volumes: []v1.Volume{}, Containers: containers, }, Status: v1.PodStatus{ Phase: jobStatusToPodPhase(jobStatus), Conditions: []v1.PodCondition{podCondition}, Message: "", Reason: "", HostIP: "", // TODO: find out the HostIP PodIP: "", // TODO: find out the equalent for PodIP ContainerStatuses: containerStatues, }, } return &pod, nil } func allocToContainerStatuses(allocs []*nomad.AllocationListStub) map[string]v1.ContainerStatus { containerStatusesMap := map[string]v1.ContainerStatus{} for _, alloc := range allocs { for name, taskState := range alloc.TaskStates { containerState, readyFlag := convertTaskStateToContainerState(taskState.State, taskState.StartedAt, taskState.FinishedAt, ) containerStatusesMap[name] = v1.ContainerStatus{ Name: name, RestartCount: int32(taskState.Restarts), Ready: readyFlag, State: containerState, } } } return containerStatusesMap } func jobStatusToPodPhase(status string) v1.PodPhase { switch status { case "pending": return v1.PodPending case "running": return v1.PodRunning // TODO: Make sure we take PodFailed into account. case "dead": return v1.PodFailed } return v1.PodUnknown } func convertJobStatusToPodCondition(jobStatus string) v1.PodCondition { podCondition := v1.PodCondition{} switch jobStatus { case "pending": podCondition = v1.PodCondition{ Type: v1.PodInitialized, Status: v1.ConditionFalse, } case "running": podCondition = v1.PodCondition{ Type: v1.PodReady, Status: v1.ConditionTrue, } case "dead": podCondition = v1.PodCondition{ Type: v1.PodReasonUnschedulable, Status: v1.ConditionFalse, } default: podCondition = v1.PodCondition{ Type: v1.PodReasonUnschedulable, Status: v1.ConditionUnknown, } } return podCondition } func convertTaskStateToContainerState(taskState string, startedAt time.Time, finishedAt time.Time) (v1.ContainerState, bool) { containerState := v1.ContainerState{} readyFlag := false switch taskState { case "pending": containerState = v1.ContainerState{ Waiting: &v1.ContainerStateWaiting{}, } case "running": containerState = v1.ContainerState{ Running: &v1.ContainerStateRunning{ StartedAt: metav1.NewTime(startedAt), }, } readyFlag = true // TODO: Make sure containers that are exiting with non-zero status codes // are accounted for using events or something similar? //case v1.PodSucceeded: // podCondition = v1.PodCondition{ // Type: v1.PodReasonUnschedulable, // Status: v1.ConditionFalse, // } // containerState = v1.ContainerState{ // Terminated: &v1.ContainerStateTerminated{ // ExitCode: int32(container.State.ExitCode), // FinishedAt: metav1.NewTime(finishedAt), // }, // } case "dead": containerState = v1.ContainerState{ Terminated: &v1.ContainerStateTerminated{ ExitCode: 0, FinishedAt: metav1.NewTime(finishedAt), }, } default: containerState = v1.ContainerState{} } return containerState, readyFlag }