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

Author: Michal Tichák

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)

core/integration/dcs/client.go

@@ -72,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())
@@ -126,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
@@ -140,28 +192,28 @@ func (m *RpcClient) Close() error {
 
 func (m *RpcClient) Subscribe(ctx context.Context, in *dcspb.SubscriptionRequest, opts ...grpc.CallOption) (dcspb.Configurator_SubscribeClient, error) {
 	metric := newMetric()
-	metric.AddTag("method", "Subscribe")
+	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("method", "PFR")
+	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("method", "SOR")
+	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("method", "EOR")
+	metric.AddTag("stream_setup", "EOR")
 	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
 	return m.ConfiguratorClient.EndOfRun(ctx, in, opts...)
 }

core/workflow/callable/call.go

@@ -114,6 +114,9 @@ func (c *Call) Call() error {
 		WithField("level", infologger.IL_Devel).
 		Debugf("calling hook function %s", c.Func)
 
+	metric := c.newMetric("callable_call")
+	defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
+
 	the.EventWriterWithTopic(topic.Call).WriteEvent(&evpb.Ev_CallEvent{
 		Path:       c.GetParentRolePath(),
 		Func:       c.Func,
@@ -221,17 +224,23 @@ func (c *Call) Call() error {
 	return nil
 }
 
+func (c *Call) newMetric(name string) monitoring.Metric {
+	metric := monitoring.NewMetric(name)
+	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() {
-		callId := fmt.Sprintf("hook:%s:%s", c.GetTraits().Trigger, c.GetName())
-		metric := monitoring.NewMetric("callable")
-		metric.AddTag("id", callId)
-		metric.AddTag("envId", c.parentRole.GetEnvironmentId().String())
+		metric := c.newMetric("callable_wrapped")
 		defer monitoring.TimerSend(&metric, monitoring.Milliseconds)()
 
+		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"))
 		select {