[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/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
@@ -132,3 +137,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("method", "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("method", "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("method", "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("method", "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...)
+}

core/integration/odc/client.go

@@ -29,6 +29,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common/logger"
+	"github.com/AliceO2Group/Control/common/monitoring"
 	odcpb "github.com/AliceO2Group/Control/core/integration/odc/protos"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
@@ -40,6 +41,10 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "odcclient")
 
+func newMetric() monitoring.Metric {
+	return monitoring.NewMetric("odc")
+}
+
 type RpcClient struct {
 	odcpb.ODCClient
 	conn   *grpc.ClientConn
@@ -131,3 +136,101 @@ func (m *RpcClient) Close() error {
 	m.cancel()
 	return m.conn.Close()
 }
+
+func (m *RpcClient) Initialize(ctx context.Context, in *odcpb.InitializeRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Initialize")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Initialize(ctx, in, opts...)
+}
+
+func (m *RpcClient) Submit(ctx context.Context, in *odcpb.SubmitRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Submit")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Submit(ctx, in, opts...)
+}
+
+func (m *RpcClient) Activate(ctx context.Context, in *odcpb.ActivateRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Activate")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Activate(ctx, in, opts...)
+}
+
+func (m *RpcClient) Run(ctx context.Context, in *odcpb.RunRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Run")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Run(ctx, in, opts...)
+}
+
+func (m *RpcClient) Update(ctx context.Context, in *odcpb.UpdateRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Update")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Update(ctx, in, opts...)
+}
+
+func (m *RpcClient) Configure(ctx context.Context, in *odcpb.ConfigureRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Configure")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Configure(ctx, in, opts...)
+}
+
+func (m *RpcClient) SetProperties(ctx context.Context, in *odcpb.SetPropertiesRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "SetProperties")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.SetProperties(ctx, in, opts...)
+}
+
+func (m *RpcClient) GetState(ctx context.Context, in *odcpb.StateRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "GetState")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.GetState(ctx, in, opts...)
+}
+
+func (m *RpcClient) Start(ctx context.Context, in *odcpb.StartRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Start")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Start(ctx, in, opts...)
+}
+
+func (m *RpcClient) Stop(ctx context.Context, in *odcpb.StopRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Stop")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Stop(ctx, in, opts...)
+}
+
+func (m *RpcClient) Reset(ctx context.Context, in *odcpb.ResetRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Reset")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Reset(ctx, in, opts...)
+}
+
+func (m *RpcClient) Terminate(ctx context.Context, in *odcpb.TerminateRequest, opts ...grpc.CallOption) (*odcpb.StateReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Terminate")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Terminate(ctx, in, opts...)
+}
+
+func (m *RpcClient) Shutdown(ctx context.Context, in *odcpb.ShutdownRequest, opts ...grpc.CallOption) (*odcpb.GeneralReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Shutdown")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Shutdown(ctx, in, opts...)
+}
+
+func (m *RpcClient) Status(ctx context.Context, in *odcpb.StatusRequest, opts ...grpc.CallOption) (*odcpb.StatusReply, error) {
+	metric := newMetric()
+	metric.AddTag("method", "Status")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+	return m.ODCClient.Status(ctx, in, opts...)
+}