[executor] Send TaskMessage with PID to scheduler

miltalex · teo · commit dc05890d911e · 2020-07-23T16:33:10.000+02:00
diff --git a/executor/executable/basictask.go b/executor/executable/basictask.go
@@ -238,6 +238,13 @@ func (t *BasicTask) Launch() error {
 		Debug("basic task staged")
 
 	go t.sendStatus(mesos.TASK_RUNNING, "")
+	taskMessage := event.NewTaskMessage(t.ti.Name,t.ti.TaskID.GetValue(),int32(t.taskCmd.Process.Pid))
+	jsonEvent, err := json.Marshal(taskMessage)
+	if err != nil {
+		log.WithError(err).Warning("error marshaling message from task")
+	} else {
+		t.sendMessage(jsonEvent)
+	}
 
 	return nil
 }
diff --git a/executor/executable/controllabletask.go b/executor/executable/controllabletask.go
@@ -147,6 +147,7 @@ func (t *ControllableTask) Launch() error {
 		}
 		t.rpc.TaskCmd = taskCmd
 
+		var pid int32
 		elapsed := 0 * time.Second
 		for {
 			log.WithFields(logrus.Fields{
@@ -176,6 +177,7 @@ func (t *ControllableTask) Launch() error {
 			}
 			// NOTE: we acquire the transitioner-dependent STANDBY equivalent state
 			reachedState := t.rpc.FromDeviceState(response.GetState())
+			pid = response.GetPid()
 
 			if reachedState == "STANDBY" && err == nil {
 				log.WithField("id", t.ti.TaskID.Value).
@@ -222,6 +224,14 @@ func (t *ControllableTask) Launch() error {
 
 		// send RUNNING
 		t.sendStatus(mesos.TASK_RUNNING, "")
+		taskMessage := event.NewTaskMessage(t.ti.Name,t.ti.TaskID.GetValue(),pid)
+		jsonEvent, err := json.Marshal(taskMessage)
+		if err != nil {
+			log.WithError(err).Warning("error marshaling message from task")
+		} else {
+			t.sendMessage(jsonEvent)
+		}
+		
 
 		// Process events from task in yet another goroutine
 		go func() {
diff --git a/executor/executable/task.go b/executor/executable/task.go
@@ -54,6 +54,7 @@ var log = logger.New(logrus.StandardLogger(), "executor")
 
 type SendStatusFunc func(state mesos.TaskState, message string)
 type SendDeviceEventFunc func(event event.DeviceEvent)
+type SendMessageFunc func(message []byte)
 
 type Task interface {
 	Launch() error
@@ -68,9 +69,10 @@ type taskBase struct {
 
 	sendStatus SendStatusFunc
 	sendDeviceEvent SendDeviceEventFunc
+	sendMessage SendMessageFunc
 }
 
-func NewTask(taskInfo mesos.TaskInfo, sendStatusFunc SendStatusFunc, sendDeviceEventFunc SendDeviceEventFunc) Task {
+func NewTask(taskInfo mesos.TaskInfo, sendStatusFunc SendStatusFunc, sendDeviceEventFunc SendDeviceEventFunc, sendMessageFunc SendMessageFunc) Task {
 	var commandInfo common.TaskCommandInfo
 
 	tciData := taskInfo.GetData()
@@ -104,6 +106,7 @@ func NewTask(taskInfo mesos.TaskInfo, sendStatusFunc SendStatusFunc, sendDeviceE
 				tci: &commandInfo,
 				sendStatus: sendStatusFunc,
 				sendDeviceEvent: sendDeviceEventFunc,
+				sendMessage: sendMessageFunc,
 			},
 		}
 	case controlmode.DIRECT:
@@ -115,6 +118,7 @@ func NewTask(taskInfo mesos.TaskInfo, sendStatusFunc SendStatusFunc, sendDeviceE
 				tci: &commandInfo,
 				sendStatus: sendStatusFunc,
 				sendDeviceEvent: sendDeviceEventFunc,
+				sendMessage: sendMessageFunc,
 			},
 			rpc: nil,
 		}
diff --git a/executor/handlers.go b/executor/handlers.go
@@ -60,6 +60,13 @@ func makeSendDeviceEventFunc(state *internalState) executable.SendDeviceEventFun
 	}
 }
 
+func makeSendMessageFunc(state *internalState) executable.SendMessageFunc {
+	return func(message []byte) {
+		// to send task events using state.
+		state.messageCh <- message
+	}
+}
+
 func handleOutgoingMessage(state *internalState, message []byte) {
 	_, _ = state.cli.Send(context.TODO(), calls.NonStreaming(calls.Message(message)))
 	log.WithFields(logrus.Fields{
@@ -195,7 +202,8 @@ func handleLaunchEvent(state *internalState, taskInfo mesos.TaskInfo) {
 
 	myTask := executable.NewTask(taskInfo,
 		makeSendStatusUpdateFunc(state, taskInfo),
-		makeSendDeviceEventFunc(state))
+		makeSendDeviceEventFunc(state),
+		makeSendMessageFunc(state))
 
 	err := myTask.Launch()
 
