[core] Emit run SOSOR/EOSOR/SOEOR/EOEOR events to aliecs.run topic

Author: teo

coconut/protos/o2control.pb.go

@@ -132,6 +132,11 @@ func (w *KafkaWriter) WriteEventWithTimestamp(e interface{}, timestamp time.Time
 				Timestamp: timestamp.UnixMilli(),
 				Payload:   &pb.Event_IntegratedServiceEvent{IntegratedServiceEvent: e},
 			}
+		case *pb.Ev_RunEvent:
+			wrappedEvent = &pb.Event{
+				Timestamp: timestamp.UnixMilli(),
+				Payload:   &pb.Event_RunEvent{RunEvent: e},
+			}
 		}
 
 		if wrappedEvent == nil {

coconut/protos/o2control_grpc.pb.go

@@ -112,17 +112,28 @@ message Ev_IntegratedServiceEvent {
   string payload = 8;                 // any additional payload, depending on the integrated service; there is no schema, it can even be the raw return structure of a remote API call
 }
 
+message Ev_RunEvent {
+  string environmentId = 1;
+  uint32 runNumber = 2;
+  string state = 3;
+  string error = 4;
+  string transition = 5;
+  OpStatus transitionStatus = 6;
+  map<string, string> vars = 7;
+}
+
 message Event {
   int64 timestamp = 1;
   reserved 2 to 10;
-  reserved 16 to 100;
+  reserved 17 to 100;
 
   oneof Payload {
     Ev_EnvironmentEvent environmentEvent                 = 11;
     Ev_TaskEvent taskEvent                               = 12;
     Ev_RoleEvent roleEvent                               = 13;
     Ev_CallEvent callEvent                               = 14;
     Ev_IntegratedServiceEvent integratedServiceEvent     = 15;
+    Ev_RunEvent runEvent                                 = 16;
 
     Ev_MetaEvent_FrameworkEvent frameworkEvent           = 101;
     Ev_MetaEvent_MesosHeartbeat mesosHeartbeatEvent      = 102;

common/event/writer.go

@@ -186,9 +186,20 @@ func newEnvironment(userVars map[string]string, newId uid.ID) (env *Environment,
 					env.workflow.GetVars().Set("run_number", rnString)
 					env.workflow.GetVars().Set("runNumber", rnString)
 
-					runStartTime := strconv.FormatInt(time.Now().UnixMilli(), 10)
-					env.workflow.SetRuntimeVar("run_start_time_ms", runStartTime)
-					env.workflow.SetRuntimeVar("run_end_time_ms", "") // we delete previous EOR
+					runStartTime := time.Now()
+					runStartTimeS := strconv.FormatInt(runStartTime.UnixMilli(), 10)
+					env.workflow.SetRuntimeVar("run_start_time_ms", runStartTimeS)
+					env.workflow.SetRuntimeVar("run_end_time_ms", "") // we delete previous SOEOR
+
+					the.EventWriterWithTopic(topic.Run).WriteEventWithTimestamp(&pb.Ev_RunEvent{
+						EnvironmentId:    envId.String(),
+						RunNumber:        runNumber,
+						State:            env.Sm.Current(),
+						Error:            "",
+						Transition:       e.Event,
+						TransitionStatus: pb.OpStatus_STARTED,
+						Vars:             nil,
+					}, runStartTime)
 
 					cleanupCount := 0
 					cleanupCountS, ok := env.GlobalVars.Get("__fmq_cleanup_count")
@@ -216,17 +227,43 @@ func newEnvironment(userVars map[string]string, newId uid.ID) (env *Environment,
 				} else if e.Event == "STOP_ACTIVITY" {
 					endTime, ok := env.workflow.GetUserVars().Get("run_end_time_ms")
 					if ok && endTime == "" {
-						runEndTime := strconv.FormatInt(time.Now().UnixMilli(), 10)
-						env.workflow.SetRuntimeVar("run_end_time_ms", runEndTime)
+						runEndTime := time.Now()
+						runEndTimeS := strconv.FormatInt(runEndTime.UnixMilli(), 10)
+						env.workflow.SetRuntimeVar("run_end_time_ms", runEndTimeS)
+						env.workflow.SetRuntimeVar("run_start_time_ms", "") // we delete previous SOSOR
+
+						the.EventWriterWithTopic(topic.Run).WriteEventWithTimestamp(&pb.Ev_RunEvent{
+							EnvironmentId:    envId.String(),
+							RunNumber:        env.GetCurrentRunNumber(),
+							State:            env.Sm.Current(),
+							Error:            "",
+							Transition:       e.Event,
+							TransitionStatus: pb.OpStatus_STARTED,
+							Vars:             nil,
+						}, runEndTime)
+
 					} else {
 						log.WithField("partition", envId.String()).
 							Debug("O2 End time already set before before_STOP_ACTIVITY")
 					}
 				} else if e.Event == "GO_ERROR" {
 					endTime, ok := env.workflow.GetUserVars().Get("run_end_time_ms")
 					if ok && endTime == "" {
-						runEndTime := strconv.FormatInt(time.Now().UnixMilli(), 10)
-						env.workflow.SetRuntimeVar("run_end_time_ms", runEndTime)
+						runEndTime := time.Now()
+						runEndTimeS := strconv.FormatInt(runEndTime.UnixMilli(), 10)
+						env.workflow.SetRuntimeVar("run_end_time_ms", runEndTimeS)
+						env.workflow.SetRuntimeVar("run_start_time_ms", "") // we delete previous SOSOR
+
+						the.EventWriterWithTopic(topic.Run).WriteEventWithTimestamp(&pb.Ev_RunEvent{
+							EnvironmentId:    envId.String(),
+							RunNumber:        env.GetCurrentRunNumber(),
+							State:            env.Sm.Current(),
+							Error:            "",
+							Transition:       e.Event,
+							TransitionStatus: pb.OpStatus_STARTED,
+							Vars:             nil,
+						}, runEndTime)
+
 					} else {
 						log.WithField("partition", envId.String()).
 							Debug("O2 End time already set before before_GO_ERROR")
@@ -439,7 +476,49 @@ func newEnvironment(userVars map[string]string, newId uid.ID) (env *Environment,
 							Infof("auto stop transition scheduled, expected execution at %s", expected)
 					}
 
+					runStartCompletionTime := time.Now()
+					runStartCompletionTimeS := strconv.FormatInt(runStartCompletionTime.UnixMilli(), 10)
+					env.workflow.SetRuntimeVar("run_start_completion_time_ms", runStartCompletionTimeS)
+					env.workflow.SetRuntimeVar("run_end_completion_time_ms", "") // we delete previous EOEOR
+
+					runEvent := &pb.Ev_RunEvent{
+						EnvironmentId:    envId.String(),
+						RunNumber:        env.GetCurrentRunNumber(),
+						State:            env.Sm.Current(),
+						Error:            "",
+						Transition:       e.Event,
+						TransitionStatus: pb.OpStatus_DONE_OK,
+						Vars:             nil,
+					}
+					if e.Err != nil {
+						runEvent.Error = e.Err.Error()
+						runEvent.TransitionStatus = pb.OpStatus_DONE_ERROR
+					}
+
+					the.EventWriterWithTopic(topic.Run).WriteEventWithTimestamp(runEvent, runStartCompletionTime)
+
 				} else if e.Event == "STOP_ACTIVITY" {
+					runEndCompletionTime := time.Now()
+					runEndCompletionTimeS := strconv.FormatInt(runEndCompletionTime.UnixMilli(), 10)
+					env.workflow.SetRuntimeVar("run_end_completion_time_ms", runEndCompletionTimeS)
+					env.workflow.SetRuntimeVar("run_start_completion_time_ms", "") // we delete previous EOSOR
+
+					runEvent := &pb.Ev_RunEvent{
+						EnvironmentId:    envId.String(),
+						RunNumber:        env.GetCurrentRunNumber(),
+						State:            env.Sm.Current(),
+						Error:            "",
+						Transition:       e.Event,
+						TransitionStatus: pb.OpStatus_DONE_OK,
+						Vars:             nil,
+					}
+					if e.Err != nil {
+						runEvent.Error = e.Err.Error()
+						runEvent.TransitionStatus = pb.OpStatus_DONE_ERROR
+					}
+
+					the.EventWriterWithTopic(topic.Run).WriteEventWithTimestamp(runEvent, runEndCompletionTime)
+
 					// If the event is STOP_ACTIVITY, we remove the active run number after all hooks are done.
 					env.workflow.GetVars().Set("last_run_number", strconv.Itoa(int(env.currentRunNumber)))
 					env.currentRunNumber = 0

common/protos/events.pb.go

@@ -26,6 +26,7 @@ package environment
 
 import (
 	"errors"
+
 	"github.com/AliceO2Group/Control/core/workflow"
 
 	"github.com/AliceO2Group/Control/common/event"
@@ -36,7 +37,7 @@ import (
 func NewConfigureTransition(taskman *task.Manager) Transition {
 	return &ConfigureTransition{
 		baseTransition: baseTransition{
-			name: "CONFIGURE",
+			name:    "CONFIGURE",
 			taskman: taskman,
 		},
 	}
@@ -62,7 +63,7 @@ func (t ConfigureTransition) do(env *Environment) (err error) {
 	}
 	incomingEv := <-env.stateChangedCh
 	// If some tasks failed to transition
-	if tasksStateErrors := incomingEv.GetTasksStateChangedError();  tasksStateErrors != nil {
+	if tasksStateErrors := incomingEv.GetTasksStateChangedError(); tasksStateErrors != nil {
 		return tasksStateErrors
 	}