[dcs] sending SOR,PFR,EOR metrics per detector

Author: Michal Tichák

core/integration/dcs/plugin.go

@@ -289,18 +289,18 @@ func (p *Plugin) Init(instanceId string) error {
 		in := &dcspb.SubscriptionRequest{
 			InstanceId: instanceId,
 		}
-		
+
 		// Always start the goroutine, even if initial subscription fails
 		go func() {
 			var evStream dcspb.Configurator_SubscribeClient
 			var err error
-			
+
 			for {
 				// Try to establish subscription if we don't have one
 				if evStream == nil {
 					log.WithField("endpoint", viper.GetString("dcsServiceEndpoint")).
 						Debug("attempting to subscribe to DCS service")
-					
+
 					evStream, err = p.dcsClient.Subscribe(context.Background(), in, grpc.EmptyCallOption{})
 					if err != nil {
 						log.WithField("endpoint", viper.GetString("dcsServiceEndpoint")).
@@ -1462,8 +1462,13 @@ func EORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 	payloadJsonForKafka []byte, stream dcspb.Configurator_EndOfRunClient, detectorStatusMap map[dcspb.Detector]dcspb.DetectorState,
 	callFailedStr string, payload map[string]interface{},
 ) (error, []byte) {
-	metric := newMetric("EOR")
-	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+	eor := "EOR"
+	detectorDurations := map[dcspb.Detector]time.Duration{}
+	start := time.Now()
+
+	wholeMetric := newMetric(eor)
+	wholeMetric.AddTag("detector", "All")
+	defer monitoring.TimerSendSingle(&wholeMetric, monitoring.Millisecond)()
 
 	var dcsEvent *dcspb.RunEvent
 	var err error
@@ -1576,6 +1581,7 @@ func EORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 		}
 
 		detectorStatusMap[dcsEvent.GetDetector()] = dcsEvent.GetState()
+		detectorDurations[dcsEvent.GetDetector()] = time.Since(start)
 		ecsDet := dcsToEcsDetector(dcsEvent.GetDetector())
 
 		if dcsEvent.GetState() == dcspb.DetectorState_EOR_FAILURE {
@@ -1687,15 +1693,23 @@ func EORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 			})
 		}
 	}
+
+	convertAndSendDetectorDurationsAndStates(eor, detectorStatusMap, detectorDurations, &wholeMetric)
+
 	return err, payloadJsonForKafka
 }
 
 func SORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *callable.Call, envId string,
 	payloadJsonForKafka []byte, stream dcspb.Configurator_StartOfRunClient, detectorStatusMap map[dcspb.Detector]dcspb.DetectorState,
 	callFailedStr string, payload map[string]interface{},
 ) (error, []byte) {
-	metric := newMetric("SOR")
-	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+	sor := "SOR"
+	detectorDurations := map[dcspb.Detector]time.Duration{}
+	start := time.Now()
+
+	wholeMetric := newMetric(sor)
+	wholeMetric.AddTag("detector", "All")
+	defer monitoring.TimerSendSingle(&wholeMetric, monitoring.Millisecond)()
 
 	var dcsEvent *dcspb.RunEvent
 	var err error
@@ -1808,6 +1822,7 @@ func SORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 		}
 
 		detectorStatusMap[dcsEvent.GetDetector()] = dcsEvent.GetState()
+		detectorDurations[dcsEvent.GetDetector()] = time.Since(start)
 		ecsDet := dcsToEcsDetector(dcsEvent.GetDetector())
 
 		if dcsEvent.GetState() == dcspb.DetectorState_SOR_FAILURE {
@@ -1956,15 +1971,41 @@ func SORgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 			})
 		}
 	}
+
+	convertAndSendDetectorDurationsAndStates(sor, detectorStatusMap, detectorDurations, &wholeMetric)
+
 	return err, payloadJsonForKafka
 }
 
+func convertAndSendDetectorDurationsAndStates(method string, detectorStatusMap map[dcspb.Detector]dcspb.DetectorState, detectorDurations map[dcspb.Detector]time.Duration, wholeMetric *monitoring.Metric) {
+	resultsMap := make(map[dcspb.DetectorState]int)
+	for dcsDet, state := range detectorStatusMap {
+		metric := newMetric(method)
+		det := dcsToEcsDetector(dcsDet)
+		metric.AddTag("detector", det)
+		metric.AddTag("state", dcspb.DetectorState_name[int32(state)])
+		resultsMap[state] += 1
+		if duration, ok := detectorDurations[dcsDet]; ok {
+			metric.SetFieldInt64("execution_time_ms", duration.Milliseconds())
+			monitoring.Send(&metric)
+		}
+	}
+	for detectorState, detectorCount := range resultsMap {
+		wholeMetric.SetFieldInt64(dcspb.DetectorState_name[int32(detectorState)], int64(detectorCount))
+	}
+}
+
 func PFRgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *callable.Call, envId string,
 	payloadJsonForKafka []byte, stream dcspb.Configurator_StartOfRunClient, detectorStatusMap map[dcspb.Detector]dcspb.DetectorState,
 	callFailedStr string, payload map[string]interface{},
 ) (error, []byte) {
-	metric := newMetric("PFR")
-	defer monitoring.TimerSendSingle(&metric, monitoring.Millisecond)()
+	pfr := "PFR"
+	detectorDurations := map[dcspb.Detector]time.Duration{}
+	start := time.Now()
+
+	wholeMetric := newMetric(pfr)
+	wholeMetric.AddTag("detector", "All")
+	defer monitoring.TimerSendSingle(&wholeMetric, monitoring.Millisecond)()
 
 	var err error
 	var dcsEvent *dcspb.RunEvent
@@ -2078,6 +2119,7 @@ func PFRgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 		}
 
 		detectorStatusMap[dcsEvent.GetDetector()] = dcsEvent.GetState()
+		detectorDurations[dcsEvent.GetDetector()] = time.Since(start)
 		ecsDet := dcsToEcsDetector(dcsEvent.GetDetector())
 
 		if dcsEvent.GetState() == dcspb.DetectorState_SOR_FAILURE {
@@ -2225,6 +2267,9 @@ func PFRgRPCCommunicationLoop(ctx context.Context, timeout time.Duration, call *
 			})
 		}
 	}
+
+	convertAndSendDetectorDurationsAndStates(pfr, detectorStatusMap, detectorDurations, &wholeMetric)
+
 	return err, payloadJsonForKafka
 }