[executor] Forward DeviceEvents from OCC to core

Author: teo

executor/executor.go

@@ -44,6 +44,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common"
+	"github.com/AliceO2Group/Control/common/event"
 	"github.com/AliceO2Group/Control/common/logger"
 	"github.com/AliceO2Group/Control/executor/executorcmd"
 	"github.com/AliceO2Group/Control/executor/protos"
@@ -491,10 +492,10 @@ func launch(state *internalState, task mesos.TaskInfo) {
 			} else {
 				log.WithField("state", response.GetState()).WithField("task", task.Name).Debug("task status queried")
 			}
+			// NOTE: we acquire the transitioner-dependent STANDBY equivalent state
 			reachedState := state.rpcClients[task.TaskID].FromDeviceState(response.GetState())
 			state.mu.RUnlock()
 
-			// FIXME: we should compare the reached state against transitioner-dependent IDLE instead of plain IDLE
 			if reachedState == "STANDBY" && err == nil {
 				log.WithField("id", task.TaskID.Value).
 					WithField("task", task.Name).
@@ -519,6 +520,23 @@ func launch(state *internalState, task mesos.TaskInfo) {
 			}
 		}
 
+		// Set up event stream from task
+		state.mu.RLock()
+		esc, err := state.rpcClients[task.TaskID].EventStream(context.TODO(), &pb.EventStreamRequest{}, grpc.EmptyCallOption{})
+		state.mu.RUnlock()
+		if err != nil {
+			log.WithField("task", task.Name).WithError(err).Error("cannot set up event stream from task")
+			status.State = mesos.TASK_FAILED.Enum()
+			status.Message = protoString(err.Error())
+
+			state.mu.Lock()
+			state.failedTasks[task.TaskID] = status
+			state.rpcClients[task.TaskID].Close()
+			delete(state.rpcClients, task.TaskID)
+			state.mu.Unlock()
+			return
+		}
+
 		log.WithField("task", task.Name).Debug("notifying of task running state")
 		status.State = mesos.TASK_RUNNING.Enum()
 
@@ -561,6 +579,44 @@ func launch(state *internalState, task mesos.TaskInfo) {
 		}
 		state.mu.Unlock()
 
+		// Process events from task
+		go func() {
+			deo := event.DeviceEventOrigin{
+				AgentId: task.AgentID,
+				ExecutorId: task.GetExecutor().ExecutorID,
+				TaskId: task.TaskID,
+			}
+			for {
+				esr, err := esc.Recv()
+				if err == io.EOF {
+					log.WithError(err).Warning("event stream EOF")
+					break
+				}
+				if err != nil {
+					log.WithError(err).Warning("error receiving event from task")
+					continue
+				}
+				ev := esr.GetEvent()
+
+				deviceEvent := event.NewDeviceEvent(deo, ev.GetType())
+
+				jsonEvent, err := json.Marshal(deviceEvent)
+				if err != nil {
+					log.WithError(err).Warning("error marshaling event from task")
+					continue
+				}
+
+				state.mu.RLock()
+				state.cli.Send(context.TODO(), calls.NonStreaming(calls.Message(jsonEvent)))
+				log.WithFields(logrus.Fields{
+						"task": task.TaskID.Value,
+						"event": string(jsonEvent),
+					}).
+					Debug("event sent")
+				state.mu.RUnlock()
+			}
+		}()
+
 		err = taskCmd.Wait()
 
 		state.mu.Lock()

executor/executorcmd/client.go

@@ -98,7 +98,7 @@ func (r *RpcClient) doTransition(ei transitioner.EventInfo) (newState string, er
 	log.WithField("event", ei.Evt).Debug("executor<->occplugin interface requesting transition")
 	var response *pb.TransitionReply
 	response, err = r.Transition(context.TODO(), &pb.TransitionRequest{
-		Event: ei.Evt,
+		TransitionEvent: ei.Evt,
 		Arguments: func() (cfg []*pb.ConfigEntry) {
 			cfg = make([]*pb.ConfigEntry, 0)
 			if len(ei.Args) == 0 {
@@ -140,7 +140,7 @@ func (r *RpcClient) doTransition(ei transitioner.EventInfo) (newState string, er
 	if response != nil &&
 		response.GetOk() &&
 		response.GetTrigger() == pb.StateChangeTrigger_EXECUTOR &&
-		response.GetEvent() == ei.Evt &&
+		response.GetTransitionEvent() == ei.Evt &&
 		response.GetState() == ei.Dst {
 		newState = response.GetState()
 		err = nil
@@ -149,7 +149,7 @@ func (r *RpcClient) doTransition(ei transitioner.EventInfo) (newState string, er
 		err = errors.New(fmt.Sprintf("transition unsuccessful: ok: %s, trigger: %s, event: %s, state: %s",
 			strconv.FormatBool(response.GetOk()),
 			response.GetTrigger().String(),
-			response.GetEvent(),
+			response.GetTransitionEvent(),
 			response.GetState()))
 	} else {
 		newState = ""