Merge pull request #6 from usedatabrew/feat/DAT-221/processor-metrics

le-vlad · web-flow · commit f57293a29a0e · 2023-12-03T13:57:15.000+01:00
feat(DAT-221/metrics): added foundation for processor metrics
diff --git a/internal/metrics/influx/plugin.go b/internal/metrics/influx/plugin.go
@@ -15,6 +15,9 @@ type Plugin struct {
 	sinkErrorsCounter   metrics.Counter
 	sourceErrorsCounter metrics.Counter
 
+	procMetrics              map[string][]metrics.Counter
+	procExecutionTimeMetrics map[string]metrics.Gauge
+
 	client       *influxdb3.Client
 	writeOptions influxdb3.WriteOptions
 
@@ -26,14 +29,16 @@ type Plugin struct {
 
 func NewPlugin(config Config) (*Plugin, error) {
 	plugin := &Plugin{
-		groupName:           config.GroupName,
-		pipelineId:          config.PipelineId,
-		orgId:               config.Org,
-		bucket:              config.Bucket,
-		sentCounter:         metrics.NewCounter(),
-		receivedCounter:     metrics.NewCounter(),
-		sinkErrorsCounter:   metrics.NewCounter(),
-		sourceErrorsCounter: metrics.NewCounter(),
+		groupName:                config.GroupName,
+		pipelineId:               config.PipelineId,
+		orgId:                    config.Org,
+		bucket:                   config.Bucket,
+		sentCounter:              metrics.NewCounter(),
+		receivedCounter:          metrics.NewCounter(),
+		sinkErrorsCounter:        metrics.NewCounter(),
+		sourceErrorsCounter:      metrics.NewCounter(),
+		procMetrics:              map[string][]metrics.Counter{},
+		procExecutionTimeMetrics: map[string]metrics.Gauge{},
 	}
 	plugin.receivedCounter.Clear()
 	plugin.sentCounter.Clear()
@@ -79,9 +84,86 @@ func (p *Plugin) IncrementSourceErrCounter() {
 	p.sourceErrorsCounter.Inc(1)
 }
 
+func (p *Plugin) SetProcessorExecutionTime(proc string, time int64) {
+	p.procExecutionTimeMetrics[proc].Update(time)
+}
+
+func (p *Plugin) IncrementProcessorDroppedMessages(proc string) {
+	p.procMetrics[proc][0].Inc(0)
+}
+
+func (p *Plugin) IncrementProcessorReceivedMessages(proc string) {
+	p.procMetrics[proc][0].Inc(1)
+}
+
+func (p *Plugin) IncrementProcessorSentMessages(proc string) {
+	p.procMetrics[proc][0].Inc(2)
+}
+
+func (p *Plugin) RegisterProcessors(processors []string) {
+	for _, proc := range processors {
+		p.procExecutionTimeMetrics[proc] = metrics.NewGauge()
+		p.procMetrics[proc] = []metrics.Counter{
+			// dropped/filtered messages metrics
+			metrics.NewCounter(),
+			// sent messages metrics
+			metrics.NewCounter(),
+			// received metrics
+			metrics.NewCounter(),
+		}
+	}
+}
+
 func (p *Plugin) flushMetrics() {
 	t := time.Now()
 
+	for proc, counters := range p.procMetrics {
+
+		executionTimePoint := influxdb3.NewPointWithMeasurement("blink_data").
+			SetTag("group", p.groupName).
+			SetTag("pipeline", strconv.Itoa(p.pipelineId)).
+			SetTag("processor", proc).
+			SetField("execution_time", p.procExecutionTimeMetrics).
+			SetTimestamp(t)
+
+		if err := p.client.WritePointsWithOptions(context.Background(), &p.writeOptions, executionTimePoint); err != nil {
+			panic(err)
+		}
+
+		filteredMessagesPoint := influxdb3.NewPointWithMeasurement("blink_data").
+			SetTag("group", p.groupName).
+			SetTag("pipeline", strconv.Itoa(p.pipelineId)).
+			SetTag("processor", proc).
+			SetField("dropped_messages", counters[0].Count()).
+			SetTimestamp(t)
+
+		if err := p.client.WritePointsWithOptions(context.Background(), &p.writeOptions, filteredMessagesPoint); err != nil {
+			panic(err)
+		}
+
+		sentMessagesPoint := influxdb3.NewPointWithMeasurement("blink_data").
+			SetTag("group", p.groupName).
+			SetTag("pipeline", strconv.Itoa(p.pipelineId)).
+			SetTag("processor", proc).
+			SetField("sent_messages", counters[1].Count()).
+			SetTimestamp(t)
+
+		if err := p.client.WritePointsWithOptions(context.Background(), &p.writeOptions, sentMessagesPoint); err != nil {
+			panic(err)
+		}
+
+		receivedMessagesPoint := influxdb3.NewPointWithMeasurement("blink_data").
+			SetTag("group", p.groupName).
+			SetTag("pipeline", strconv.Itoa(p.pipelineId)).
+			SetTag("processor", proc).
+			SetField("sent_messages", counters[2].Count()).
+			SetTimestamp(t)
+
+		if err := p.client.WritePointsWithOptions(context.Background(), &p.writeOptions, receivedMessagesPoint); err != nil {
+			panic(err)
+		}
+	}
+
 	point := influxdb3.NewPointWithMeasurement("blink_data").
 		SetTag("group", p.groupName).
 		SetTag("pipeline", strconv.Itoa(p.pipelineId)).
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
@@ -5,4 +5,11 @@ type Metrics interface {
 	IncrementSentCounter()
 	IncrementSourceErrCounter()
 	IncrementSinkErrCounter()
+
+	RegisterProcessors(processors []string)
+
+	SetProcessorExecutionTime(proc string, time int64)
+	IncrementProcessorDroppedMessages(proc string)
+	IncrementProcessorReceivedMessages(proc string)
+	IncrementProcessorSentMessages(proc string)
 }
diff --git a/internal/metrics/prometheus/plugin.go b/internal/metrics/prometheus/plugin.go
@@ -1,6 +1,7 @@
 package prometheus
 
 import (
+	"fmt"
 	"github.com/InfluxCommunity/influxdb3-go/influxdb3"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -14,6 +15,9 @@ type Plugin struct {
 
 	writeOptions influxdb3.WriteOptions
 
+	procCounters            map[string][]prometheus.Counter
+	procExecutionTimeGauges map[string]prometheus.Gauge
+
 	groupName  string
 	pipelineId int
 }
@@ -38,6 +42,8 @@ func NewPlugin(config Config) (*Plugin, error) {
 			Name: "source_errors",
 			Help: "The total number of errors from source connector",
 		}),
+		procCounters:            map[string][]prometheus.Counter{},
+		procExecutionTimeGauges: map[string]prometheus.Gauge{},
 	}
 	return plugin, nil
 }
@@ -57,3 +63,42 @@ func (p *Plugin) IncrementSinkErrCounter() {
 func (p *Plugin) IncrementSourceErrCounter() {
 	p.sourceErrorsCounter.Inc()
 }
+
+func (p *Plugin) RegisterProcessors(processors []string) {
+	for _, proc := range processors {
+		p.procExecutionTimeGauges[proc] = promauto.NewGauge(prometheus.GaugeOpts{
+			Name: fmt.Sprintf("%s_execution_time", proc),
+			Help: "Time taken to process message",
+		})
+		p.procCounters[proc] = []prometheus.Counter{
+			promauto.NewCounter(prometheus.CounterOpts{
+				Name: fmt.Sprintf("%s_dropped_messages", proc),
+				Help: "Messages that were dropped, filtered by the processor",
+			}),
+			promauto.NewCounter(prometheus.CounterOpts{
+				Name: fmt.Sprintf("%s_sent_messages", proc),
+				Help: "The total number of messages send to the next sink/processor plugin",
+			}),
+			promauto.NewCounter(prometheus.CounterOpts{
+				Name: fmt.Sprintf("%s_received_messages", proc),
+				Help: "The total number of messages send to the sink plugin",
+			}),
+		}
+	}
+}
+
+func (p *Plugin) SetProcessorExecutionTime(proc string, time int64) {
+	p.procExecutionTimeGauges[proc].Set(float64(time))
+}
+
+func (p *Plugin) IncrementProcessorDroppedMessages(proc string) {
+	p.procCounters[proc][0].Inc()
+}
+
+func (p *Plugin) IncrementProcessorReceivedMessages(proc string) {
+	p.procCounters[proc][1].Inc()
+}
+
+func (p *Plugin) IncrementProcessorSentMessages(proc string) {
+	p.procCounters[proc][2].Inc()
+}
diff --git a/public/stream/processor_wrapper.go b/public/stream/processor_wrapper.go
@@ -2,23 +2,30 @@ package stream
 
 import (
 	"blink/internal/message"
+	"blink/internal/metrics"
 	"blink/internal/processors"
 	"blink/internal/processors/openai"
 	"blink/internal/schema"
 	"blink/internal/sources"
 	"blink/internal/stream_context"
 	"errors"
+	"time"
 )
 
 // ProcessorWrapper wraps plan sink writer plugin in order to
 // measure performance, build proper configuration and control the context
 type ProcessorWrapper struct {
 	processorDriver processors.DataProcessor
 	ctx             *stream_context.Context
+	metrics         metrics.Metrics
+	procDriver      string
 }
 
 func NewProcessorWrapper(pluginType processors.ProcessorDriver, config interface{}, appctx *stream_context.Context) ProcessorWrapper {
-	loader := ProcessorWrapper{}
+	loader := ProcessorWrapper{
+		metrics:    appctx.Metrics,
+		procDriver: string(pluginType),
+	}
 	loader.ctx = appctx
 	loadedDriver, err := loader.LoadDriver(pluginType, config)
 	if err != nil {
@@ -29,7 +36,16 @@ func NewProcessorWrapper(pluginType processors.ProcessorDriver, config interface
 }
 
 func (p *ProcessorWrapper) Process(msg sources.MessageEvent) (message.Message, error) {
-	return p.processorDriver.Process(p.ctx.GetContext(), msg.Message)
+	p.metrics.IncrementProcessorReceivedMessages(p.procDriver)
+	execStart := time.Now()
+	procMsg, err := p.processorDriver.Process(p.ctx.GetContext(), msg.Message)
+	if err == nil {
+		p.metrics.IncrementProcessorSentMessages(string(p.procDriver))
+	}
+
+	execEnd := time.Since(execStart)
+	p.metrics.SetProcessorExecutionTime(p.procDriver, execEnd.Milliseconds())
+	return procMsg, err
 }
 
 func (p *ProcessorWrapper) EvolveSchema(s *schema.StreamSchemaObj) error {
diff --git a/public/stream/stream.go b/public/stream/stream.go
@@ -28,12 +28,18 @@ type Stream struct {
 func InitFromConfig(config config.Configuration) (*Stream, error) {
 	streamContext := stream_context.CreateContext()
 	streamContext.Logger.Info("Bootstrapping blink Stream-ETL")
+
+	var processorList []string
+	for _, proc := range config.Processors {
+		processorList = append(processorList, string(proc.Driver))
+	}
 	if config.Service.InfluxEnabled {
 		metrics, err := loadInfluxMetrics(config.Service.Influx)
 		if err != nil {
 			streamContext.Logger.WithPrefix("Metrics").Error("failed to load influx metrics")
 			return nil, err
 		}
+		metrics.RegisterProcessors(processorList)
 		streamContext.SetMetrics(metrics)
 		streamContext.Logger.WithPrefix("Metrics").Info("Component has been loaded")
 	} else {
@@ -44,6 +50,7 @@ func InitFromConfig(config config.Configuration) (*Stream, error) {
 			streamContext.Logger.WithPrefix("Metrics").Error("failed to load local prometheus metrics")
 			return nil, err
 		}
+		metrics.RegisterProcessors(processorList)
 		streamContext.SetMetrics(metrics)
 		streamContext.Logger.WithPrefix("Metrics").Info("Component has been loaded")
 	}
