[OCTRL-1092] Improve metrics error reporting

common/monitoring/metric.go

@@ -71,6 +71,17 @@ func (metric *Metric) AddTag(tagName string, value string) {
 	metric.tags = append(metric.tags, Tag{name: tagName, value: value})
 }
 
+const (
+	ERROR     = "error"
+	SUCCESS   = "success"
+	CANCELLED = "cancelled"
+	TIMEOUT   = "timeout"
+)
+
+func (metric *Metric) AddResult(result string) {
+	metric.AddTag("result", result)
+}
+
 func (metric *Metric) setField(fieldName string, field any) {
 	if metric.fields == nil {
 		metric.fields = make(FieldsType)

core/environment/transition.go

@@ -27,7 +27,8 @@ package environment
 import (
 	"errors"
 
-	"github.com/AliceO2Group/Control/core/protos"
+	"github.com/AliceO2Group/Control/common/monitoring"
+	pb "github.com/AliceO2Group/Control/core/protos"
 	"github.com/AliceO2Group/Control/core/task"
 )
 
@@ -74,3 +75,10 @@ func (t baseTransition) check() (err error) {
 func (t baseTransition) eventName() string {
 	return t.name
 }
+
+func (t baseTransition) transitionDoMetric(env *Environment) monitoring.Metric {
+	metric := monitoring.NewMetric("transition_do")
+	metric.AddTag("transition", t.name)
+	metric.AddTag("envId", env.Id().String())
+	return metric
+}

core/environment/transition_configure.go

@@ -30,6 +30,7 @@ import (
 	"github.com/AliceO2Group/Control/core/workflow"
 
 	"github.com/AliceO2Group/Control/common/event"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/taskop"
 )
@@ -52,6 +53,9 @@ func (t ConfigureTransition) do(env *Environment) (err error) {
 		return errors.New("cannot transition in NIL environment")
 	}
 
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	wf := env.Workflow()
 
 	activeTasks := workflow.GetActiveTasks(wf)
@@ -64,9 +68,11 @@ func (t ConfigureTransition) do(env *Environment) (err error) {
 	incomingEv := <-env.stateChangedCh
 	// If some tasks failed to transition
 	if tasksStateErrors := incomingEv.GetTasksStateChangedError(); tasksStateErrors != nil {
+		metric.AddResult(monitoring.ERROR)
 		return tasksStateErrors
 	}
 
 	env.sendEnvironmentEvent(&event.EnvironmentEvent{EnvironmentID: env.Id().String(), State: "CONFIGURED"})
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }

core/environment/transition_deploy.go

@@ -36,6 +36,7 @@ import (
 
 	"github.com/AliceO2Group/Control/common/event"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/sm"
 	"github.com/AliceO2Group/Control/core/task/taskop"
@@ -66,6 +67,9 @@ func (t DeployTransition) do(env *Environment) (err error) {
 		return errors.New("cannot transition in NIL environment")
 	}
 
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	wf := env.Workflow()
 
 	// Skip cleanup for anything other than readout-dataflow
@@ -347,10 +351,12 @@ func (t DeployTransition) do(env *Environment) (err error) {
 		log.WithField("level", infologger.IL_Ops).
 			WithField("partition", env.Id().String()).
 			Error(err)
+		metric.AddResult(monitoring.ERROR)
 		return
 	}
 
 	env.sendEnvironmentEvent(&event.EnvironmentEvent{EnvironmentID: env.Id().String(), State: "DEPLOYED"})
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }
 

core/environment/transition_goerror.go

@@ -25,6 +25,7 @@
 package environment
 
 import (
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/controlcommands"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/sm"
@@ -44,6 +45,8 @@ type GoErrorTransition struct {
 }
 
 func (t GoErrorTransition) do(env *Environment) (err error) {
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
 
 	// we stop all tasks which are in RUNNING
 	toStop := env.Workflow().GetTasks().Filtered(func(t *task.Task) bool {
@@ -72,5 +75,6 @@ func (t GoErrorTransition) do(env *Environment) (err error) {
 		<-env.stateChangedCh
 	}
 
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }

core/environment/transition_reset.go

@@ -28,6 +28,7 @@ import (
 	"errors"
 
 	"github.com/AliceO2Group/Control/common/event"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/sm"
 	"github.com/AliceO2Group/Control/core/workflow"
@@ -51,6 +52,9 @@ func (t ResetTransition) do(env *Environment) (err error) {
 		return errors.New("cannot transition in NIL environment")
 	}
 
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	taskmanMessage := task.NewTransitionTaskMessage(
 		workflow.GetActiveTasks(env.Workflow()),
 		sm.CONFIGURED.String(),
@@ -64,9 +68,11 @@ func (t ResetTransition) do(env *Environment) (err error) {
 	incomingEv := <-env.stateChangedCh
 	// If some tasks failed to transition
 	if tasksStateErrors := incomingEv.GetTasksStateChangedError(); tasksStateErrors != nil {
+		metric.AddResult(monitoring.ERROR)
 		return tasksStateErrors
 	}
 
 	env.sendEnvironmentEvent(&event.EnvironmentEvent{EnvironmentID: env.Id().String(), State: "RESET"})
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }

core/environment/transition_startactivity.go

@@ -33,6 +33,7 @@ import (
 
 	"github.com/AliceO2Group/Control/common/event"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/controlcommands"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/iancoleman/strcase"
@@ -72,6 +73,9 @@ func (t StartActivityTransition) do(env *Environment) (err error) {
 		return errors.New("cannot transition in NIL environment")
 	}
 
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	runNumber := env.currentRunNumber
 
 	log.WithField(infologger.Run, runNumber).
@@ -120,6 +124,7 @@ func (t StartActivityTransition) do(env *Environment) (err error) {
 	incomingEv := <-env.stateChangedCh
 	// If some tasks failed to transition
 	if tasksStateErrors := incomingEv.GetTasksStateChangedError(); tasksStateErrors != nil {
+		metric.AddResult(monitoring.ERROR)
 		return tasksStateErrors
 	}
 
@@ -133,5 +138,6 @@ func (t StartActivityTransition) do(env *Environment) (err error) {
 		Run:           env.currentRunNumber,
 	})
 
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }

core/environment/transition_stopactivity.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/AliceO2Group/Control/common/event"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	"github.com/AliceO2Group/Control/core/controlcommands"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/sm"
@@ -63,6 +64,9 @@ func (t StopActivityTransition) do(env *Environment) (err error) {
 		return errors.New("cannot transition in NIL environment")
 	}
 
+	metric := t.transitionDoMetric(env)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	log.WithField(infologger.Run, env.currentRunNumber).
 		WithField("partition", env.Id().String()).
 		WithField(infologger.Level, infologger.IL_Support).
@@ -98,6 +102,7 @@ func (t StopActivityTransition) do(env *Environment) (err error) {
 	incomingEv := <-env.stateChangedCh
 	// If some tasks failed to transition
 	if tasksStateErrors := incomingEv.GetTasksStateChangedError(); tasksStateErrors != nil {
+		metric.AddResult(monitoring.ERROR)
 		return tasksStateErrors
 	}
 	env.sendEnvironmentEvent(&event.EnvironmentEvent{EnvironmentID: env.Id().String(), State: "CONFIGURED"})
@@ -107,5 +112,6 @@ func (t StopActivityTransition) do(env *Environment) (err error) {
 		WithField(infologger.Level, infologger.IL_Support).
 		Info("run stopped")
 
+	metric.AddResult(monitoring.SUCCESS)
 	return
 }

core/environment/utils.go

@@ -52,7 +52,7 @@ type WorkflowPublicInfo struct {
 func parseWorkflowPublicInfo(workflowExpr string) (WorkflowPublicInfo, error) {
 	repoManager := the.RepoManager()
 
-	resolvedWorkflowPath, _, err := repoManager.GetWorkflow(workflowExpr) //Will fail if repo unknown
+	resolvedWorkflowPath, _, err := repoManager.GetWorkflow(workflowExpr) // Will fail if repo unknown
 	if err != nil {
 		return WorkflowPublicInfo{}, err
 	}

core/workflow/callable/call.go

@@ -110,14 +110,23 @@ func (s Calls) AwaitAll() map[*Call]error {
 	return errs
 }
 
+func (c *Call) callableMetric(name string) monitoring.Metric {
+	metric := monitoring.NewMetric(name)
+	metric.AddTag("runtype", c.getRunTypeTag())
+	metric.AddTag("name", c.GetName())
+	metric.AddTag("trigger", c.GetTraits().Trigger)
+	metric.AddTag("envId", c.parentRole.GetEnvironmentId().String())
+	return metric
+}
+
 func (c *Call) Call() error {
 	log.WithField("trigger", c.Traits.Trigger).
 		WithField("await", c.Traits.Await).
 		WithField("partition", c.parentRole.GetEnvironmentId().String()).
 		WithField("level", infologger.IL_Devel).
 		Debugf("calling hook function %s", c.Func)
 
-	metric := c.newMetric("callablecall")
+	metric := c.callableMetric("callablecall")
 	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
 
 	the.EventWriterWithTopic(topic.Call).WriteEvent(&evpb.Ev_CallEvent{
@@ -178,6 +187,7 @@ func (c *Call) Call() error {
 			EnvironmentId: c.parentRole.GetEnvironmentId().String(),
 		})
 
+		metric.AddResult(monitoring.ERROR)
 		return err
 	}
 	if len(returnVar) > 0 {
@@ -206,6 +216,7 @@ func (c *Call) Call() error {
 			EnvironmentId: c.parentRole.GetEnvironmentId().String(),
 		})
 
+		metric.AddResult(monitoring.ERROR)
 		return errors.New(errMsg)
 	}
 
@@ -224,32 +235,31 @@ func (c *Call) Call() error {
 		EnvironmentId: c.parentRole.GetEnvironmentId().String(),
 	})
 
+	metric.AddResult(monitoring.SUCCESS)
 	return nil
 }
 
-func (c *Call) newMetric(name string) monitoring.Metric {
-	metric := monitoring.NewMetric(name)
-	metric.AddTag("runtype", c.getRunTypeTag())
-	metric.AddTag("name", c.GetName())
-	metric.AddTag("trigger", c.GetTraits().Trigger)
-	metric.AddTag("envId", c.parentRole.GetEnvironmentId().String())
-	return metric
-}
-
 func (c *Call) Start() {
 	c.await = make(chan error)
 	ctx, cancel := context.WithCancel(context.Background())
 	c.awaitCancel = cancel
 	go func() {
-		metric := c.newMetric("callablewrapped")
+		metric := c.callableMetric("callablewrapped")
 		defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
 
 		callId := fmt.Sprintf("hook:%s:%s", c.GetTraits().Trigger, c.GetName())
 		log.Debugf("%s started", callId)
 		defer utils.TimeTrack(time.Now(), callId, log.WithPrefix("callable"))
+		err := c.Call()
 		select {
-		case c.await <- c.Call():
+		case c.await <- err:
+			if err == nil {
+				metric.AddResult(monitoring.SUCCESS)
+			} else {
+				metric.AddResult(monitoring.ERROR)
+			}
 		case <-ctx.Done():
+			metric.AddResult(monitoring.CANCELLED)
 			log.Debugf("%s cancelled", callId)
 		}
 		close(c.await)