feat: add dispatcher metrics

Author: xBlaz3kx

ocppj/dispatcher.go

@@ -3,6 +3,7 @@ package ocppj
 import (
 	"context"
 	"fmt"
+	"go.opentelemetry.io/otel/metric"
 	"sync"
 	"time"
 
@@ -373,6 +374,7 @@ type DefaultServerDispatcher struct {
 	onRequestCancel     CanceledRequestHandler
 	network             ws.Server
 	mutex               sync.RWMutex
+	metrics             *dispatcherMetrics
 }
 
 // Handler function to be invoked when a request gets canceled (either due to timeout or to other external factors).
@@ -389,14 +391,25 @@ func (c clientTimeoutContext) isActive() bool {
 }
 
 // NewDefaultServerDispatcher creates a new DefaultServerDispatcher struct.
-func NewDefaultServerDispatcher(queueMap ServerQueueMap) *DefaultServerDispatcher {
+func NewDefaultServerDispatcher(queueMap ServerQueueMap, provider metric.MeterProvider) *DefaultServerDispatcher {
+	dispatcherMetrics, err := newDispatcherMetrics(provider)
+	if err != nil {
+		log.Errorf("failed to create dispatcher metrics: %v", err)
+		return nil
+	}
+
 	d := &DefaultServerDispatcher{
 		queueMap:         queueMap,
 		requestChannel:   nil,
 		readyForDispatch: make(chan string, 1),
 		timeout:          defaultMessageTimeout,
+		metrics:          dispatcherMetrics,
 	}
 	d.pendingRequestState = NewServerState(&d.mutex)
+
+	dispatcherMetrics.ObserveQueues(queueMap)
+	dispatcherMetrics.ObserveInFlightRequests(d.pendingRequestState.(*serverState))
+
 	return d
 }
 

ocppj/dispatcher_metrics.go

@@ -0,0 +1,104 @@
+package ocppj
+
+import (
+	"context"
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const (
+	dispatcherQueueSize      = "dispatcher_queue_size"
+	dispatcherPendingRequest = "dispatcher_pending_requests"
+)
+
+type dispatcherMetrics struct {
+	meter metric.Meter
+
+	requestQueue    metric.Int64ObservableGauge
+	pendingRequests metric.Int64ObservableUpDownCounter
+}
+
+func newDispatcherMetrics(meterProvider metric.MeterProvider) (*dispatcherMetrics, error) {
+	if meterProvider == nil {
+		return nil, errors.New("meterProvider is nil")
+	}
+
+	meter := meterProvider.Meter("server_dispatcher")
+
+	clientQueue, err := meter.Int64ObservableGauge(
+		dispatcherQueueSize,
+		metric.WithDescription("Number of messages in the dispatcher's queue"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	clientPendingRequest, err := meter.Int64ObservableUpDownCounter(
+		dispatcherPendingRequest,
+		metric.WithDescription("Number of pending requests in the dispatcher"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	dispatcher := &dispatcherMetrics{
+		meter:           meter,
+		requestQueue:    clientQueue,
+		pendingRequests: clientPendingRequest,
+	}
+
+	return dispatcher, nil
+}
+
+func (d *dispatcherMetrics) ObserveInFlightRequests(state *serverState) {
+	if d.meter == nil {
+		return
+	}
+
+	_, err := d.meter.RegisterCallback(
+		func(ctx context.Context, obs metric.Observer) error {
+			state.mutex.RLock()
+			currentState := state.pendingRequestState
+			state.mutex.RUnlock()
+
+			for clientID, clientState := range currentState {
+				inFlightRequest := int64(0)
+				if clientState.HasPendingRequest() {
+					inFlightRequest = 1
+				}
+				obs.ObserveInt64(
+					d.pendingRequests,
+					inFlightRequest,
+					metric.WithAttributes(attribute.String("client_id", clientID)),
+				)
+			}
+			return nil
+		},
+		d.pendingRequests)
+	if err != nil {
+		log.Errorf("failed to register callback for inflight queue size: %v", err)
+	}
+}
+
+func (d *dispatcherMetrics) ObserveQueues(queue ServerQueueMap) {
+	if d.meter == nil {
+		return
+	}
+
+	_, err := d.meter.RegisterCallback(
+		func(ctx context.Context, obs metric.Observer) error {
+			for clientID, queueSize := range queue.SizePerClient() {
+				obs.ObserveInt64(
+					d.requestQueue,
+					int64(queueSize),
+					metric.WithAttributes(attribute.String("client_id", clientID)),
+				)
+			}
+			return nil
+		},
+		d.requestQueue)
+	if err != nil {
+		log.Errorf("failed to register callback for dispatcher queue size: %v", err)
+	}
+}

ocppj/server.go

@@ -54,13 +54,9 @@ func NewServer(wsServer ws.Server, dispatcher ServerDispatcher, stateHandler Ser
 	}
 
 	if dispatcher == nil {
-		perClientQueue := NewFIFOQueueMap(0)
-
-		// Observe queue lengths
-		metrics.ObserveRequestQueue(context.Background(), perClientQueue)
-
-		dispatcher = NewDefaultServerDispatcher(perClientQueue)
+		dispatcher = NewDefaultServerDispatcher(NewFIFOQueueMap(0), meterProvider)
 	}
+
 	if stateHandler == nil {
 		d, ok := dispatcher.(*DefaultServerDispatcher)
 		if !ok {
@@ -69,9 +65,11 @@ func NewServer(wsServer ws.Server, dispatcher ServerDispatcher, stateHandler Ser
 			stateHandler = d.pendingRequestState
 		}
 	}
+
 	if wsServer == nil {
 		wsServer = ws.NewServer()
 	}
+
 	dispatcher.SetNetworkServer(wsServer)
 	dispatcher.SetPendingRequestState(stateHandler)
 

ocppj/server_metrics.go

@@ -12,7 +12,6 @@ import (
 const (
 	requestsInboundMetric  = "ocpp_requests_inbound"
 	requestsOutboundMetric = "ocpp_requests_outbound"
-	requestQueueMetric     = "ocpp_request_queue_size"
 )
 
 const (
@@ -33,10 +32,9 @@ var (
 )
 
 type ocppMetrics struct {
-	requestsIn         metric.Int64Histogram
-	requestsOut        metric.Int64Histogram
-	requestQueueMetric metric.Int64ObservableGauge
-	meter              metric.Meter
+	requestsIn  metric.Int64Histogram
+	requestsOut metric.Int64Histogram
+	meter       metric.Meter
 }
 
 // newOcppServerMetrics Creates a new metrics instance
@@ -66,19 +64,10 @@ func newOcppServerMetrics(meterProvider metric.MeterProvider, ocppVersion string
 		return nil, errors.Wrap(err, fmt.Sprintf("failed to create %s metric", requestsOutboundMetric))
 	}
 
-	queueSize, err := meter.Int64ObservableGauge(
-		requestQueueMetric,
-		metric.WithDescription("Current queue length"),
-	)
-	if err != nil {
-		return nil, errors.Wrap(err, fmt.Sprintf("failed to create %s metric", requestQueueMetric))
-	}
-
 	metrics := &ocppMetrics{
-		requestsIn:         requestsIn,
-		requestsOut:        requestsOut,
-		requestQueueMetric: queueSize,
-		meter:              meter,
+		requestsIn:  requestsIn,
+		requestsOut: requestsOut,
+		meter:       meter,
 	}
 	return metrics, nil
 }
@@ -118,16 +107,3 @@ func (m *ocppMetrics) IncrementOutboundRequests(ctx context.Context, chargePoint
 	metricAttrs := metric.WithAttributes(attrs...)
 	m.requestsOut.Record(ctx, 1, metricAttrs)
 }
-
-func (m *ocppMetrics) ObserveRequestQueue(ctx context.Context, queueMap ServerQueueMap) {
-	_, err := m.meter.RegisterCallback(
-		func(ctx context.Context, observer metric.Observer) error {
-			observer.ObserveInt64(m.requestQueueMetric, int64(queueMap.Size()))
-			return nil
-		},
-		m.requestQueueMetric,
-	)
-	if err != nil {
-		log.Error(err)
-	}
-}