[core] Added monitoring to hooks and communication with outside services

- we monitor every trigger
- we monitor every process run by a trigger (more granularity)
- we monitor outside communication with services defined in integration
- DCS: we monitor duration of it's stream Recv() function
- DCS: we monitor duration of whole gRPC stream communication for PFR,
SOR, EOR

Author: Michal Tichák

common/event/writer.go

@@ -110,8 +110,7 @@ func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
 		metric.SetFieldUInt64("messages_failed", 0)
 
 		metricDuration := writer.newMetric(KAFKAWRITER)
-		defer monitoring.SendHistogrammable(&metricDuration)
-		defer monitoring.TimerNS(&metricDuration)()
+		defer monitoring.TimerSendHist(&metricDuration, monitoring.Nanosecond)()
 
 		if err := writer.WriteMessages(context.Background(), messages...); err != nil {
 			metric.SetFieldUInt64("messages_failed", uint64(len(messages)))
@@ -250,8 +249,7 @@ func (w *KafkaWriter) WriteEventWithTimestamp(e interface{}, timestamp time.Time
 
 	metric := w.newMetric(KAFKAPREPARE)
 
-	defer monitoring.SendHistogrammable(&metric)
-	defer monitoring.TimerNS(&metric)()
+	defer monitoring.TimerSendHist(&metric, monitoring.Nanosecond)()
 
 	wrappedEvent, key, err := internalEventToKafkaEvent(e, timestamp)
 	if err != nil {

common/monitoring/grpcinterceptor.go

@@ -0,0 +1,79 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2025 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package monitoring
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+)
+
+type measuredClientStream struct {
+	grpc.ClientStream
+	method     string
+	metricName string
+}
+
+func (t *measuredClientStream) RecvMsg(m interface{}) error {
+	metric := NewMetric(t.metricName)
+	metric.AddTag("method", t.method)
+	defer TimerSendSingle(&metric, Millisecond)()
+
+	err := t.ClientStream.RecvMsg(m)
+	return err
+}
+
+type NameConvertType func(string) string
+
+func SetupStreamClientInterceptor(metricName string, convert NameConvertType) grpc.StreamClientInterceptor {
+	return func(
+		ctx context.Context,
+		desc *grpc.StreamDesc,
+		cc *grpc.ClientConn,
+		method string,
+		streamer grpc.Streamer,
+		opts ...grpc.CallOption,
+	) (grpc.ClientStream, error) {
+		clientStream, err := streamer(ctx, desc, cc, method, opts...)
+		if err != nil {
+			return nil, err
+		}
+
+		return &measuredClientStream{
+			ClientStream: clientStream,
+			method:       convert(method),
+			metricName:   metricName,
+		}, nil
+	}
+}
+
+func SetupUnaryClientInterceptor(name string, convert NameConvertType) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		metric := NewMetric(name)
+		metric.AddTag("method", convert(method))
+		defer TimerSendSingle(&metric, Millisecond)()
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}

common/monitoring/metric.go

@@ -27,6 +27,7 @@ package monitoring
 import (
 	"fmt"
 	"io"
+	"sort"
 	"time"
 
 	lp "github.com/influxdata/line-protocol/v2/lineprotocol"
@@ -110,6 +111,10 @@ func Format(writer io.Writer, metrics []Metric) error {
 	var enc lp.Encoder
 
 	for _, metric := range metrics {
+		// AddTag requires tags sorted lexicografically
+		sort.Slice(metric.tags, func(i int, j int) bool {
+			return metric.tags[i].name < metric.tags[j].name
+		})
 		enc.StartLine(metric.name)
 		for _, tag := range metric.tags {
 			enc.AddTag(tag.name, tag.value)

common/monitoring/monitoring_test.go

@@ -546,8 +546,8 @@ func TestMetricsHistogramObject(t *testing.T) {
 }
 
 func measureFunc(metric *Metric) {
-	defer TimerMS(metric)()
-	defer TimerNS(metric)()
+	defer Timer(metric, Millisecond)()
+	defer Timer(metric, Nanosecond)()
 	time.Sleep(100 * time.Millisecond)
 }
 

common/monitoring/timer.go

@@ -1,19 +1,59 @@
 package monitoring
 
-import "time"
+import (
+	"time"
 
-// Timer* functions are meant to be used with defer statement to measure runtime of given function:
-// defer TimerNS(&metric)()
-func TimerMS(metric *Metric) func() {
-	start := time.Now()
-	return func() {
-		metric.SetFieldInt64("execution_time_ms", time.Since(start).Milliseconds())
-	}
+	"github.com/AliceO2Group/Control/common/logger/infologger"
+)
+
+type TimeResolution int
+
+const (
+	Millisecond TimeResolution = iota
+	Nanosecond
+)
+
+// Timer function is meant to be used with defer statement to measure runtime of given function:
+// defer Timer(&metric, Milliseconds)()
+func Timer(metric *Metric, unit TimeResolution) func() {
+	return timer(metric, unit, false, false)
 }
 
-func TimerNS(metric *Metric) func() {
+// Timer function is meant to be used with defer statement to measure runtime of given function:
+// defer Timer(&metric, Milliseconds)()
+// sends measured value as Send(metric)
+func TimerSendSingle(metric *Metric, unit TimeResolution) func() {
+	return timer(metric, unit, true, false)
+}
+
+// Timer function is meant to be used with defer statement to measure runtime of given function:
+// defer Timer(&metric, Milliseconds)()
+// sends measured value as SendHistogrammable(metric)
+func TimerSendHist(metric *Metric, unit TimeResolution) func() {
+	return timer(metric, unit, true, true)
+}
+
+func timer(metric *Metric, unit TimeResolution, send bool, sendHistogrammable bool) func() {
 	start := time.Now()
+
 	return func() {
-		metric.SetFieldInt64("execution_time_ns", time.Since(start).Nanoseconds())
+		dur := time.Since(start)
+		// we are setting default value as Nanoseconds
+		switch unit {
+		case Millisecond:
+			metric.SetFieldInt64("execution_time_ms", dur.Milliseconds())
+		case Nanosecond:
+			metric.SetFieldInt64("execution_time_ns", dur.Nanoseconds())
+		default:
+			log.WithField("level", infologger.IL_Devel).Warnf("trying to use unknown time resolution in monitoring.timer function [%d], skipping", unit)
+		}
+
+		if send {
+			if sendHistogrammable {
+				SendHistogrammable(metric)
+			} else {
+				Send(metric)
+			}
+		}
 	}
 }

core/environment/environment.go

@@ -42,6 +42,7 @@ import (
 	"github.com/AliceO2Group/Control/common/gera"
 	"github.com/AliceO2Group/Control/common/logger"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	pb "github.com/AliceO2Group/Control/common/protos"
 	"github.com/AliceO2Group/Control/common/runtype"
 	"github.com/AliceO2Group/Control/common/system"
@@ -647,6 +648,11 @@ func (env *Environment) handleHooks(workflow workflow.Role, trigger string, weig
 	hooksMapForTrigger := workflow.GetHooksMapForTrigger(trigger)
 	callsMapForAwait := env.callsPendingAwait[trigger]
 
+	metric := monitoring.NewMetric("hooks")
+	metric.AddTag("trigger", trigger)
+	metric.AddTag("envId", env.id.String())
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	allWeightsSet := make(callable.HooksMap)
 	for k := range hooksMapForTrigger {
 		allWeightsSet[k] = callable.Hooks{}

core/integration/ccdb/plugin.go

@@ -37,6 +37,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 
 	"github.com/AliceO2Group/Control/core/environment"
 
@@ -109,7 +110,6 @@ func getFlpIdList(envId string) (flps []string, err error) {
 }
 
 func NewGRPObject(varStack map[string]string) *GeneralRunParameters {
-
 	envId, ok := varStack["environment_id"]
 	if !ok {
 		log.WithField("level", infologger.IL_Support).
@@ -366,8 +366,8 @@ func (p *Plugin) ObjectStack(_ map[string]string, _ map[string]string) (stack ma
 
 func (p *Plugin) NewCcdbGrpWriteCommand(grp *GeneralRunParameters, ccdbUrl string, refresh bool) (cmd string, err error) {
 	// o2-ecs-grp-create -h
-	//Create GRP-ECS object and upload to the CCDB
-	//Usage:
+	// Create GRP-ECS object and upload to the CCDB
+	// Usage:
 	//  o2-ecs-grp-create:
 	//  -h [ --help ]                         Print this help message
 	//  -p [ --period ] arg                   data taking period
@@ -524,7 +524,6 @@ func (p *Plugin) CallStack(data interface{}) (stack map[string]interface{}) {
 }
 
 func (p *Plugin) uploadCurrentGRP(grp *GeneralRunParameters, envId string, refresh bool) error {
-
 	if grp == nil {
 		return errors.New(fmt.Sprintf("Failed to create a GRP object"))
 	}
@@ -547,6 +546,10 @@ func (p *Plugin) uploadCurrentGRP(grp *GeneralRunParameters, envId string, refre
 	ctx, cancel := context.WithTimeout(context.Background(), timeoutSeconds*time.Second)
 	defer cancel()
 
+	metric := monitoring.NewMetric("ccdb")
+	metric.AddTag("envId", envId)
+	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+
 	cmd := exec.CommandContext(ctx, "bash", "-c", cmdStr)
 	// execute the DPL command in the repo of the workflow used
 	cmd.Dir = "/tmp"

core/integration/dcs/client.go

@@ -29,6 +29,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common/logger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	dcspb "github.com/AliceO2Group/Control/core/integration/dcs/protos"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
@@ -40,6 +41,22 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "dcsclient")
 
+func convertMethodName(method string) (converted string) {
+	switch method {
+	case dcspb.Configurator_Subscribe_FullMethodName:
+		converted = "Subscribe"
+	case dcspb.Configurator_PrepareForRun_FullMethodName:
+		converted = "PFR"
+	case dcspb.Configurator_StartOfRun_FullMethodName:
+		converted = "SOR"
+	case dcspb.Configurator_EndOfRun_FullMethodName:
+		converted = "EOR"
+	default:
+		converted = "Unknown"
+	}
+	return
+}
+
 type RpcClient struct {
 	dcspb.ConfiguratorClient
 	conn   *grpc.ClientConn
@@ -67,6 +84,7 @@ func NewClient(cxt context.Context, cancel context.CancelFunc, endpoint string)
 			Timeout:             time.Second,
 			PermitWithoutStream: true,
 		}),
+		grpc.WithStreamInterceptor(monitoring.SetupStreamClientInterceptor("dcsgrpcstream", convertMethodName)),
 	}
 	if !viper.GetBool("dcsServiceUseSystemProxy") {
 		dialOptions = append(dialOptions, grpc.WithNoProxy())