Fix the bug that TCP metrics are not aggregated correctly

CHANGELOG.md

@@ -14,6 +14,7 @@
 - Support to identify the MySQL protocol with statements `commit` and `set`. ([#417](https://github.com/KindlingProject/kindling/pull/417))
 
 ### Bug fixes
+- Fix the bug that TCP metrics are not aggregated correctly. ([#444](https://github.com/KindlingProject/kindling/pull/444))
 - Fix the bug that cpuanalyzer missed some trigger events due to the incorrect variable reference. This may cause some traces can't correlate with on/off CPU data. ([#424](https://github.com/KindlingProject/kindling/pull/424))
 
 ## v0.6.0 - 2022-12-21

collector/pkg/aggregator/defaultaggregator/default_aggregator_test.go

@@ -34,7 +34,7 @@ func TestConcurrentAggregator(t *testing.T) {
 	go func() {
 		for i := 0; i < runLoop; i++ {
 			metricValues := []*model.Metric{
-				{Name: "duration", Data: &model.Metric_Int{Int: &model.Int{Value: duration}}},
+				{Name: "duration", Data: &model.Int{Value: duration}},
 			}
 			dataGroup := model.NewDataGroup("testMetric", labels, 0, metricValues...)
 			aggregatorInstance.Aggregate(dataGroup, labelSelectors)

collector/pkg/aggregator/defaultaggregator/value_recorder.go

@@ -22,7 +22,9 @@ func newValueRecorder(recorderName string, aggKindMap map[string][]KindConfig) *
 	}
 }
 
-// Record is thread-safe, and return the result value
+// Record is thread-safe, and return the result value.
+// A recorder can record only the metrics that are the same as the initial ones when using the same key.
+// But it can record different metrics with different keys.
 func (r *valueRecorder) Record(key *aggregator.LabelKeys, metricValues []*model.Metric, timestamp uint64) {
 	if key == nil {
 		return

collector/pkg/aggregator/defaultaggregator/value_recorder_test.go

@@ -4,6 +4,8 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/Kindling-project/kindling/collector/pkg/aggregator"
 	"github.com/Kindling-project/kindling/collector/pkg/model"
 )
@@ -78,3 +80,38 @@ func TestRecord(t *testing.T) {
 		t.Errorf("expected %+v, got %+v", expectedValue, histogramValue.GetHistogram())
 	}
 }
+
+// TestRecordDiffMetricsWithSameKey validates that a recorder can record only the metrics that are same as the initial ones.
+// But it allows to record different metrics with different keys.
+func TestRecordDiffMetricsWithSameKey(t *testing.T) {
+	aggKindMap := AggregatedConfig{KindMap: map[string][]KindConfig{
+		"duration": {
+			{Kind: SumKind, OutputName: "duration_sum"},
+		},
+		"last": {{Kind: LastKind, OutputName: "last"}},
+	}}
+	recorder := newValueRecorder("testRecorder", aggKindMap.KindMap)
+	keys := aggregator.NewLabelKeys([]aggregator.LabelKey{
+		{
+			Name:  "stringKey",
+			Value: "stringValue",
+			VType: aggregator.StringType,
+		},
+	}...)
+	for i := 0; i < 100; i++ {
+		metricValues := []*model.Metric{
+			model.NewIntMetric("duration", int64(100)),
+		}
+		recorder.Record(keys, metricValues, 0)
+		metricValues = []*model.Metric{
+			model.NewIntMetric("last", int64(i)),
+		}
+		recorder.Record(keys, metricValues, 0)
+	}
+	retMetricGroup := recorder.dump()
+	sumValue, _ := retMetricGroup[0].GetMetric("duration_sum")
+	assert.Equal(t, int64(10000), sumValue.GetInt().Value)
+	// last is not aggregated because it is not the initial one using the current key
+	_, ok := retMetricGroup[0].GetMetric("last")
+	assert.Equal(t, false, ok)
+}

collector/pkg/component/analyzer/tcpmetricanalyzer/tcp_analyzer.go

@@ -3,16 +3,17 @@ package tcpmetricanalyzer
 import (
 	"fmt"
 
+	"github.com/hashicorp/go-multierror"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+
 	"github.com/Kindling-project/kindling/collector/pkg/component"
 	"github.com/Kindling-project/kindling/collector/pkg/component/analyzer"
 	"github.com/Kindling-project/kindling/collector/pkg/component/consumer"
 	conntrackerpackge "github.com/Kindling-project/kindling/collector/pkg/metadata/conntracker"
 	"github.com/Kindling-project/kindling/collector/pkg/model"
 	"github.com/Kindling-project/kindling/collector/pkg/model/constlabels"
 	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
-	"github.com/hashicorp/go-multierror"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )
 
 const (
@@ -25,7 +26,7 @@ type TcpMetricAnalyzer struct {
 	telemetry   *component.TelemetryTools
 }
 
-func NewTcpMetricAnalyzer(cfg interface{}, telemetry *component.TelemetryTools, nextConsumers []consumer.Consumer) analyzer.Analyzer {
+func NewTcpMetricAnalyzer(_ interface{}, telemetry *component.TelemetryTools, nextConsumers []consumer.Consumer) analyzer.Analyzer {
 	retAnalyzer := &TcpMetricAnalyzer{
 		consumers: nextConsumers,
 		telemetry: telemetry,
@@ -101,7 +102,7 @@ func (a *TcpMetricAnalyzer) generateRtt(event *model.KindlingEvent) (*model.Data
 		return nil, nil
 	}
 	metric := model.NewIntMetric(constnames.TcpRttMetricName, int64(rtt))
-	return model.NewDataGroup(constnames.TcpMetricGroupName, labels, event.Timestamp, metric), nil
+	return model.NewDataGroup(constnames.TcpRttMetricGroupName, labels, event.Timestamp, metric), nil
 }
 
 func (a *TcpMetricAnalyzer) generateRetransmit(event *model.KindlingEvent) (*model.DataGroup, error) {
@@ -110,7 +111,7 @@ func (a *TcpMetricAnalyzer) generateRetransmit(event *model.KindlingEvent) (*mod
 		return nil, err
 	}
 	metric := model.NewIntMetric(constnames.TcpRetransmitMetricName, 1)
-	return model.NewDataGroup(constnames.TcpMetricGroupName, labels, event.Timestamp, metric), nil
+	return model.NewDataGroup(constnames.TcpRetransmitMetricGroupName, labels, event.Timestamp, metric), nil
 }
 
 func (a *TcpMetricAnalyzer) generateDrop(event *model.KindlingEvent) (*model.DataGroup, error) {
@@ -119,7 +120,7 @@ func (a *TcpMetricAnalyzer) generateDrop(event *model.KindlingEvent) (*model.Dat
 		return nil, err
 	}
 	metric := model.NewIntMetric(constnames.TcpDropMetricName, 1)
-	return model.NewDataGroup(constnames.TcpMetricGroupName, labels, event.Timestamp, metric), nil
+	return model.NewDataGroup(constnames.TcpDropMetricGroupName, labels, event.Timestamp, metric), nil
 }
 
 func (a *TcpMetricAnalyzer) getTupleLabels(event *model.KindlingEvent) (*model.AttributeMap, error) {

collector/pkg/component/consumer/exporter/otelexporter/instrument_test.go

@@ -6,16 +6,17 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Kindling-project/kindling/collector/pkg/component"
-	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter/tools/adapter"
-	"github.com/Kindling-project/kindling/collector/pkg/model"
-	"github.com/Kindling-project/kindling/collector/pkg/model/constlabels"
-	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
 	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
 	otelprocessor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
+
+	"github.com/Kindling-project/kindling/collector/pkg/component"
+	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter/tools/adapter"
+	"github.com/Kindling-project/kindling/collector/pkg/model"
+	"github.com/Kindling-project/kindling/collector/pkg/model/constlabels"
+	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
 )
 
 func Test_instrumentFactory_recordLastValue(t *testing.T) {
@@ -61,7 +62,7 @@ func Test_instrumentFactory_recordLastValue(t *testing.T) {
 		controller.WithResource(nil),
 	)
 
-	cont.Start(context.Background())
+	_ = cont.Start(context.Background())
 
 	ins := newInstrumentFactory(cont.Meter("test"), component.NewDefaultTelemetryTools(), nil)
 
@@ -77,7 +78,7 @@ func Test_instrumentFactory_recordLastValue(t *testing.T) {
 
 func makeTcpGroup(rttLatency int64) *model.DataGroup {
 	return model.NewDataGroup(
-		constnames.TcpMetricGroupName,
+		constnames.TcpRttMetricGroupName,
 		model.NewAttributeMapWithValues(
 			map[string]model.AttributeValue{
 				constlabels.SrcIp:           model.NewStringValue("src-ip"),
@@ -170,8 +171,8 @@ func Test_instrumentFactory_recordTraceAsMetric(t *testing.T) {
 					t1 = lastTraceAsMetric[i]
 
 					// value check
-					if metric, ok := t1.GetMetric(constnames.TraceAsMetric); ok {
-						if metric.GetInt().Value != randTime {
+					if m, ok := t1.GetMetric(constnames.TraceAsMetric); ok {
+						if m.GetInt().Value != randTime {
 							t.Errorf("Value check failed")
 						}
 					} else {

collector/pkg/component/consumer/exporter/otelexporter/otelexporter.go

@@ -7,10 +7,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/Kindling-project/kindling/collector/pkg/component"
-	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter"
-	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter/tools/adapter"
-	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
@@ -29,6 +25,11 @@ import (
 	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
+
+	"github.com/Kindling-project/kindling/collector/pkg/component"
+	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter"
+	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter/tools/adapter"
+	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
 )
 
 const (
@@ -44,17 +45,6 @@ const (
 
 var serviceName string
 
-type labelKey struct {
-	metric          string
-	srcIp           string
-	dstIp           string
-	dstPort         int64
-	requestContent  string
-	responseContent string
-	statusCode      string
-	protocol        string
-}
-
 type OtelOutputExporters struct {
 	metricExporter exportmetric.Exporter
 	traceExporter  sdktrace.SpanExporter
@@ -151,7 +141,8 @@ func NewExporter(config interface{}, telemetry *component.TelemetryTools) export
 					StorePodDetail:     cfg.AdapterConfig.NeedPodDetail,
 					StoreExternalSrcIP: cfg.AdapterConfig.StoreExternalSrcIP,
 				}),
-				adapter.NewSimpleAdapter([]string{constnames.TcpMetricGroupName, constnames.TcpConnectMetricGroupName}, customLabels),
+				adapter.NewSimpleAdapter([]string{constnames.TcpRttMetricGroupName, constnames.TcpRetransmitMetricGroupName,
+					constnames.TcpDropMetricGroupName, constnames.TcpConnectMetricGroupName}, customLabels),
 			},
 		}
 		go func() {
@@ -218,7 +209,8 @@ func NewExporter(config interface{}, telemetry *component.TelemetryTools) export
 					StorePodDetail:     cfg.AdapterConfig.NeedPodDetail,
 					StoreExternalSrcIP: cfg.AdapterConfig.StoreExternalSrcIP,
 				}),
-				adapter.NewSimpleAdapter([]string{constnames.TcpMetricGroupName, constnames.TcpConnectMetricGroupName}, customLabels),
+				adapter.NewSimpleAdapter([]string{constnames.TcpRttMetricGroupName, constnames.TcpRetransmitMetricGroupName,
+					constnames.TcpDropMetricGroupName, constnames.TcpConnectMetricGroupName}, customLabels),
 			},
 		}
 

collector/pkg/component/consumer/exporter/otelexporter/otelexporter_test.go

@@ -7,19 +7,20 @@ import (
 	"testing"
 	"time"
 
+	"github.com/spf13/viper"
+	"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
+	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
+	otelprocessor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
+	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
+	"go.uber.org/zap"
+
 	"github.com/Kindling-project/kindling/collector/pkg/component"
 	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter"
 	"github.com/Kindling-project/kindling/collector/pkg/component/consumer/exporter/tools/adapter"
 	"github.com/Kindling-project/kindling/collector/pkg/model"
 	"github.com/Kindling-project/kindling/collector/pkg/model/constlabels"
 	"github.com/Kindling-project/kindling/collector/pkg/model/constnames"
 	"github.com/Kindling-project/kindling/collector/pkg/model/constvalues"
-	"github.com/spf13/viper"
-	"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
-	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
-	otelprocessor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
-	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
-	"go.uber.org/zap"
 )
 
 func InitOtelExporter(t *testing.T) exporter.Exporter {
@@ -171,13 +172,13 @@ func BenchmarkOtelExporter_Consume(b *testing.B) {
 	}
 
 	telemetry := component.NewDefaultTelemetryTools()
-	exporter, _ := newExporters(context.Background(), cfg, telemetry)
+	myExporter, _ := newExporters(context.Background(), cfg, telemetry)
 
 	cont := controller.New(
 		otelprocessor.NewFactory(simple.NewWithHistogramDistribution(
 			histogram.WithExplicitBoundaries(exponentialInt64NanosecondsBoundaries),
-		), exporter.metricExporter),
-		controller.WithExporter(exporter.metricExporter),
+		), myExporter.metricExporter),
+		controller.WithExporter(myExporter.metricExporter),
 		controller.WithCollectPeriod(cfg.StdoutCfg.CollectPeriod),
 		controller.WithResource(nil),
 	)
@@ -198,7 +199,8 @@ func BenchmarkOtelExporter_Consume(b *testing.B) {
 				StorePodDetail:     cfg.AdapterConfig.NeedPodDetail,
 				StoreExternalSrcIP: cfg.AdapterConfig.StoreExternalSrcIP,
 			}),
-			adapter.NewSimpleAdapter([]string{constnames.TcpMetricGroupName}, nil),
+			adapter.NewSimpleAdapter([]string{constnames.TcpRttMetricGroupName, constnames.TcpRetransmitMetricGroupName,
+				constnames.TcpDropMetricGroupName}, nil),
 		},
 	}
 
@@ -254,7 +256,7 @@ func BenchmarkOtelExporter_Consume(b *testing.B) {
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		otelexporter.Consume(metricsGroupsSlice[recordCounter%dimension])
+		_ = otelexporter.Consume(metricsGroupsSlice[recordCounter%dimension])
 		recordCounter++
 	}