Add ingest count and latency metrics for ipfix collector

Also, do not expose unused metrics, such as stage duration with ipfix collector

Fixes #1032

Author: jotak

pkg/pipeline/ingest/ingest_grpc.go

@@ -45,7 +45,15 @@ func NewGRPCProtobuf(opMetrics *operational.Metrics, params config.StageParam) (
 		bufLen = defaultBufferLen
 	}
 	flowPackets := make(chan *pbflow.Records, bufLen)
-	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(flowPackets) })
+	metrics := newMetrics(
+		opMetrics,
+		params.Name,
+		params.Ingest.Type,
+		func() int { return len(flowPackets) },
+		withLatency(),
+		withBatchSizeBytes(),
+		withStageDuration(),
+	)
 	collector, err := grpc.StartCollector(netObserv.Port, flowPackets,
 		grpc.WithGRPCServerOptions(grpc2.UnaryInterceptor(instrumentGRPC(metrics))))
 	if err != nil {

pkg/pipeline/ingest/ingest_ipfix.go

@@ -176,6 +176,8 @@ func (c *ingestIPFIX) processLogLines(out chan<- config.GenericMap) {
 			ilog.Infof("Exit signal received, stop processing input")
 			return
 		case record := <-c.in:
+			c.metrics.flowsProcessed.Inc()
+			c.metrics.observeLatency(record)
 			out <- record
 		}
 	}
@@ -195,7 +197,7 @@ func NewIngestIPFIX(opMetrics *operational.Metrics, params config.StageParam) (I
 	ilog.Infof("Ingest IPFIX config: [%s]", cfg.String())
 
 	in := make(chan map[string]interface{}, channelSize)
-	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(in) })
+	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(in) }, withLatency())
 
 	return &ingestIPFIX{
 		IngestIpfix: &cfg,

pkg/pipeline/ingest/ingest_kafka.go

@@ -107,33 +107,14 @@ func (k *ingestKafka) isStopped() bool {
 	}
 }
 
-func (k *ingestKafka) processRecordDelay(record config.GenericMap) {
-	timeFlowEndInterface, ok := record["TimeFlowEndMs"]
-	if !ok {
-		// "trace" level used to minimize performance impact
-		klog.Tracef("TimeFlowEndMs missing in record %v", record)
-		k.metrics.error("TimeFlowEndMs missing")
-		return
-	}
-	timeFlowEnd, ok := timeFlowEndInterface.(int64)
-	if !ok {
-		// "trace" level used to minimize performance impact
-		klog.Tracef("Cannot parse TimeFlowEndMs of record %v", record)
-		k.metrics.error("Cannot parse TimeFlowEndMs")
-		return
-	}
-	delay := time.Since(time.UnixMilli(timeFlowEnd)).Seconds()
-	k.metrics.latency.Observe(delay)
-}
-
 func (k *ingestKafka) processRecord(record []byte, out chan<- config.GenericMap) {
 	// Decode batch
 	decoded, err := k.decoder.Decode(record)
 	if err != nil {
 		klog.WithError(err).Warnf("ignoring flow")
 		return
 	}
-	k.processRecordDelay(decoded)
+	k.metrics.observeLatency(decoded)
 
 	// Send batch
 	out <- decoded
@@ -189,7 +170,14 @@ func NewIngestKafka(opMetrics *operational.Metrics, params config.StageParam) (I
 	}
 
 	in := make(chan []byte, 2*bml)
-	metrics := newMetrics(opMetrics, params.Name, ingestType, func() int { return len(in) })
+	metrics := newMetrics(
+		opMetrics,
+		params.Name,
+		ingestType,
+		func() int { return len(in) },
+		withLatency(),
+		withBatchSizeBytes(),
+	)
 
 	return &ingestKafka{
 		kafkaReader:    kafkaReader,

pkg/pipeline/ingest/metrics.go

@@ -1,6 +1,8 @@
 package ingest
 
 import (
+	"time"
+
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
 	"github.com/prometheus/client_golang/prometheus"
@@ -44,18 +46,48 @@ type metrics struct {
 	errors         *prometheus.CounterVec
 }
 
-func newMetrics(opMetrics *operational.Metrics, stage, stageType string, inGaugeFunc func() int) *metrics {
+func newMetrics(opMetrics *operational.Metrics, stage, stageType string, inGaugeFunc func() int, opts ...metricsOption) *metrics {
 	opMetrics.CreateInQueueSizeGauge(stage, inGaugeFunc)
-	return &metrics{
+	ret := &metrics{
 		Metrics:        opMetrics,
 		stage:          stage,
 		stageType:      stageType,
-		latency:        opMetrics.NewHistogram(&latencyHistogram, []float64{.001, .01, .1, 1, 10, 100, 1000, 10000}, stage),
-		stageDuration:  opMetrics.GetOrCreateStageDurationHisto().WithLabelValues(stage),
 		flowsProcessed: opMetrics.NewCounter(&flowsProcessedCounter, stage),
-		batchSizeBytes: opMetrics.NewSummary(&batchSizeBytesSummary, stage),
 		errors:         opMetrics.NewCounterVec(&errorsCounter),
 	}
+	for _, opt := range opts {
+		ret = opt(ret)
+	}
+	return ret
+}
+
+type metricsOption func(*metrics) *metrics
+
+func withStageDuration() metricsOption {
+	return func(m *metrics) *metrics {
+		if m.stageDuration == nil {
+			m.stageDuration = m.GetOrCreateStageDurationHisto().WithLabelValues(m.stage)
+		}
+		return m
+	}
+}
+
+func withLatency() metricsOption {
+	return func(m *metrics) *metrics {
+		if m.latency == nil {
+			m.latency = m.NewHistogram(&latencyHistogram, []float64{.001, .01, .1, 1, 10, 100, 1000, 10000}, m.stage)
+		}
+		return m
+	}
+}
+
+func withBatchSizeBytes() metricsOption {
+	return func(m *metrics) *metrics {
+		if m.batchSizeBytes == nil {
+			m.batchSizeBytes = m.NewSummary(&batchSizeBytesSummary, m.stage)
+		}
+		return m
+	}
 }
 
 func (m *metrics) createOutQueueLen(out chan<- config.GenericMap) {
@@ -72,3 +104,30 @@ func (m *metrics) error(code string) {
 func (m *metrics) stageDurationTimer() *operational.Timer {
 	return operational.NewTimer(m.stageDuration)
 }
+
+func (m *metrics) observeLatency(record config.GenericMap) {
+	if m.latency == nil {
+		return
+	}
+	tfeUnknown, ok := record["TimeFlowEndMs"]
+	if !ok {
+		m.error("TimeFlowEndMs missing")
+		return
+	}
+	var tfe int64
+	switch i := tfeUnknown.(type) {
+	case int64:
+		tfe = i
+	case int:
+		tfe = int64(i)
+	case uint64:
+		tfe = int64(i)
+	case uint:
+		tfe = int64(i)
+	default:
+		m.error("Cannot parse TimeFlowEndMs")
+		return
+	}
+	delay := time.Since(time.UnixMilli(tfe)).Seconds()
+	m.latency.Observe(delay)
+}