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

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.TimerSend(&metricDuration, monitoring.Nanoseconds)()
 
 		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.TimerSend(&metric, monitoring.Nanoseconds)()
 
 	wrappedEvent, key, err := internalEventToKafkaEvent(e, timestamp)
 	if err != nil {

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, Milliseconds)()
+	defer Timer(metric, Nanoseconds)()
 	time.Sleep(100 * time.Millisecond)
 }
 

common/monitoring/timer.go

@@ -1,19 +1,42 @@
 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())
-	}
+type TimeUnit int
+
+const (
+	Milliseconds TimeUnit = iota
+	Nanoseconds
+)
+
+// 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 TimeUnit) func() {
+	return timer(metric, unit, false)
 }
 
-func TimerNS(metric *Metric) func() {
+// TimerSend function is meant to be used with defer statement to measure runtime of given function:
+// defer TimerSend(&metric, Milliseconds)()
+func TimerSend(metric *Metric, unit TimeUnit) func() {
+	return timer(metric, unit, true)
+}
+
+func timer(metric *Metric, unit TimeUnit, send 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
+		if unit == Milliseconds {
+			metric.SetFieldInt64("execution_time_ms", dur.Milliseconds())
+		} else {
+			metric.SetFieldInt64("execution_time_ns", dur.Nanoseconds())
+		}
+
+		if send {
+			SendHistogrammable(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.TimerSend(&metric, monitoring.Milliseconds)()
+
 	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.TimerSend(&metric, monitoring.Milliseconds)()
+
 	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,10 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "dcsclient")
 
+func newMetric() monitoring.Metric {
+	return monitoring.NewMetric("dcs")
+}
+
 type RpcClient struct {
 	dcspb.ConfiguratorClient
 	conn   *grpc.ClientConn
@@ -67,6 +72,7 @@ func NewClient(cxt context.Context, cancel context.CancelFunc, endpoint string)
 			Timeout:             time.Second,
 			PermitWithoutStream: true,
 		}),
+		grpc.WithStreamInterceptor(setupStreamClientInterceptor()),
 	}
 	if !viper.GetBool("dcsServiceUseSystemProxy") {
 		dialOptions = append(dialOptions, grpc.WithNoProxy())
@@ -121,6 +127,57 @@ func NewClient(cxt context.Context, cancel context.CancelFunc, endpoint string)
 	return client
 }
 
+type measuredClientStream struct {
+	grpc.ClientStream
+	method string
+}
+
+func (t *measuredClientStream) RecvMsg(m interface{}) error {
+	metric := monitoring.NewMetric("dcs_stream")
+	metric.AddTag("method", t.method)
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	err := t.ClientStream.RecvMsg(m)
+
+	return err
+}
+
+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
+}
+
+func setupStreamClientInterceptor() 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:       convertMethodName(method),
+		}, nil
+	}
+}
+
 func (m *RpcClient) GetConnState() connectivity.State {
 	if m.conn == nil {
 		return connectivity.Idle
@@ -132,3 +189,38 @@ func (m *RpcClient) Close() error {
 	m.cancel()
 	return m.conn.Close()
 }
+
+func (m *RpcClient) Subscribe(ctx context.Context, in *dcspb.SubscriptionRequest, opts ...grpc.CallOption) (dcspb.Configurator_SubscribeClient, error) {
+	metric := newMetric()
+	metric.AddTag("stream_setup", "Subscribe")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ConfiguratorClient.Subscribe(ctx, in, opts...)
+}
+
+func (m *RpcClient) PrepareForRun(ctx context.Context, in *dcspb.PfrRequest, opts ...grpc.CallOption) (dcspb.Configurator_PrepareForRunClient, error) {
+	metric := newMetric()
+	metric.AddTag("stream_setup", "PFR")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ConfiguratorClient.PrepareForRun(ctx, in, opts...)
+}
+
+func (m *RpcClient) StartOfRun(ctx context.Context, in *dcspb.SorRequest, opts ...grpc.CallOption) (dcspb.Configurator_StartOfRunClient, error) {
+	metric := newMetric()
+	metric.AddTag("stream_setup", "SOR")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ConfiguratorClient.StartOfRun(ctx, in, opts...)
+}
+
+func (m *RpcClient) EndOfRun(ctx context.Context, in *dcspb.EorRequest, opts ...grpc.CallOption) (dcspb.Configurator_EndOfRunClient, error) {
+	metric := newMetric()
+	metric.AddTag("stream_setup", "EOR")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ConfiguratorClient.EndOfRun(ctx, in, opts...)
+}
+
+func (m *RpcClient) GetStatus(ctx context.Context, in *dcspb.StatusRequest, opts ...grpc.CallOption) (*dcspb.StatusReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "GetStatus")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ConfiguratorClient.GetStatus(ctx, in, opts...)
+}

core/integration/ddsched/client.go

@@ -29,6 +29,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common/logger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	ddpb "github.com/AliceO2Group/Control/core/integration/ddsched/protos"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
@@ -40,6 +41,10 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "ddschedclient")
 
+func newMetric() monitoring.Metric {
+	return monitoring.NewMetric("ddsched")
+}
+
 type RpcClient struct {
 	ddpb.DataDistributionControlClient
 	conn   *grpc.ClientConn
@@ -131,3 +136,24 @@ func (m *RpcClient) Close() error {
 	m.cancel()
 	return m.conn.Close()
 }
+
+func (m *RpcClient) PartitionInitialize(ctx context.Context, in *ddpb.PartitionInitRequest, opts ...grpc.CallOption) (*ddpb.PartitionResponse, error) {
+	metric := newMetric()
+	metric.AddTag("method", "PartitionInitialize")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.DataDistributionControlClient.PartitionInitialize(ctx, in, opts...)
+}
+
+func (m *RpcClient) PartitionTerminate(ctx context.Context, in *ddpb.PartitionTermRequest, opts ...grpc.CallOption) (*ddpb.PartitionResponse, error) {
+	metric := newMetric()
+	metric.AddTag("method", "PartitionTerminate")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.DataDistributionControlClient.PartitionTerminate(ctx, in, opts...)
+}
+
+func (m *RpcClient) PartitionStatus(ctx context.Context, in *ddpb.PartitionInfo, opts ...grpc.CallOption) (*ddpb.PartitionResponse, error) {
+	metric := newMetric()
+	metric.AddTag("method", "PartitionStatus")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.DataDistributionControlClient.PartitionStatus(ctx, in, opts...)
+}