[core] Move PID to task.Task from role

Author: miltalex

core/scheduler.go

@@ -340,11 +340,7 @@ func incomingMessageHandler(state *internalState, fidStore store.Singleton) even
 
 			t := state.taskman.GetTask(taskMessage.GetTaskId())
 			if t != nil {
-				if parentRole, ok := t.GetParentRole().(workflow.Role); ok {
-					parentRole.SetRuntimeVar("taskPID",strconv.Itoa(taskMessage.GetTaskPID()))
-				} else {
-					log.WithPrefix("scheduler").Error("TaskMessage received for task with no parent role")
-				}
+				t.SetTaskPID(taskMessage.GetTaskPID())
 			}
 		}
 		return

core/serverutil.go

@@ -103,13 +103,12 @@ func taskToShortTaskInfo(t *task.Task) (sti *pb.ShortTaskInfo) {
 			OfferId: t.GetOfferId(),
 			ExecutorId: t.GetExecutorId(),
 		},
-		Pid: "UNKNOWN",
+		Pid: t.GetTaskPID(),
 	}
 	parentRole, ok := t.GetParentRole().(workflow.Role)
 	if ok && parentRole != nil {
 		sti.Status = parentRole.GetStatus().String()
 		sti.State = parentRole.GetState().String()
-		sti.Pid = parentRole.GetTaskPID()
 	}
 	return
 }

core/task/task.go

@@ -32,6 +32,7 @@ package task
 import (
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	texttemplate "text/template"
 
@@ -112,6 +113,7 @@ type Task struct {
 	// ↑ to be filled in by NewTaskForMesosOffer in Manager
 
 	commandInfo  *common.TaskCommandInfo
+	pid          string
 }
 
 func (t *Task) IsSafeToStop() bool {
@@ -447,3 +449,17 @@ func (t *Task) GetProperties() map[string]string {
 	}
 	return propertiesMap
 }
+
+func (t *Task) SetTaskPID(pid int) {
+	if t == nil {
+		return
+	}
+	t.pid = strconv.Itoa(pid)
+}
+
+func (t *Task) GetTaskPID() string {
+	if t == nil {
+		return ""
+	}
+	return t.pid
+}

core/workflow/iteratorrole.go

@@ -271,11 +271,3 @@ func (i *iteratorRole) SetRuntimeVars(kv map[string]string) {
 	}
 	i.template.SetRuntimeVars(kv)
 }
-
-func (i *iteratorRole) GetTaskPID() string {
-	if i == nil {
-		return ""
-	}
-	pid, _ := i.template.GetUserVars().Get("taskPID")
-	return pid
-}

core/workflow/role.go

@@ -57,7 +57,6 @@ type Role interface {
 	GlobFilter(g glob.Glob) []Role
 	SetRuntimeVar(key string, value string)
 	SetRuntimeVars(kv map[string]string)
-	GetTaskPID() string
 }
 
 type Updatable interface {

core/workflow/rolebase.go

@@ -351,11 +351,3 @@ func (r *roleBase) CollectInboundChannels() (channels []channel.Inbound) {
 	}	
 	return
 }
-
-func (r *roleBase) GetTaskPID() string {
-	if r == nil {
-		return ""
-	}
-	pid, _ := r.UserVars.Get("taskPID")
-	return pid
-}