STAC-0: send (limited) topology when receiving traces as well

exporter/ststopologyexporter/exporter.go

@@ -15,6 +15,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap"
 )
 
@@ -59,7 +60,6 @@ func (t *topologyExporter) logAttrs(msg string, attrs *pcommon.Map) {
 }
 
 func (t *topologyExporter) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
-	log := t.logger
 
 	componentsByApiKey := make(map[string]*internal.ComponentsCollection, 0)
 	rms := md.ResourceMetrics()
@@ -132,6 +132,37 @@ func (t *topologyExporter) ConsumeMetrics(ctx context.Context, md pmetric.Metric
 		}
 	}
 
+	t.sendCollection(componentsByApiKey)
+
+	return nil
+}
+
+func (t *topologyExporter) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
+	componentsByApiKey := make(map[string]*internal.ComponentsCollection, 0)
+	rms := td.ResourceSpans()
+	for i := 0; i < rms.Len(); i++ {
+		rs := rms.At(i)
+		resource := rs.Resource()
+		attrs := resource.Attributes()
+		sts_api_key_value, key_exists := attrs.Get("sts_api_key")
+		if key_exists {
+			sts_api_key := sts_api_key_value.AsString()
+			attrs.Remove("sts_api_key")
+			collection := getOrDefault(componentsByApiKey, sts_api_key)
+			if !collection.AddResource(&attrs) {
+				t.logAttrs("Skipping resource without necessary attributes", &attrs)
+			}
+		}
+	}
+
+	t.sendCollection(componentsByApiKey)
+
+	return nil
+}
+
+func (t *topologyExporter) sendCollection(componentsByApiKey map[string]*internal.ComponentsCollection) error {
+	log := t.logger
+
 	for apiKey, collection := range componentsByApiKey {
 		components := collection.GetComponents()
 		relations := collection.GetRelations()
@@ -190,6 +221,5 @@ func (t *topologyExporter) ConsumeMetrics(ctx context.Context, md pmetric.Metric
 			)
 		}
 	}
-
 	return nil
 }

exporter/ststopologyexporter/exporter_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -58,6 +59,42 @@ func TestExporter_skipVirtualNodes(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestExporter_simpleTrace(t *testing.T) {
+	testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
+		require.Equal(t, "APIKEY", req.Header[http.CanonicalHeaderKey("sts-api-key")][0])
+		require.Equal(t, "ms", req.Header[http.CanonicalHeaderKey("sts-time-format")][0])
+
+		var payload internal.IntakeTopology
+		err := json.NewDecoder(req.Body).Decode(&payload)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(payload.Topologies))
+
+		require.Equal(t, internal.Instance{
+			Type: "opentelemetry",
+			URL:  "collector",
+		}, payload.Topologies[0].Instance)
+
+		require.Equal(t, 3, len(payload.Topologies[0].Components))
+		for _, component := range payload.Topologies[0].Components {
+			tags := component.Data.Tags
+			_, ok := tags["sts_api_key"]
+			require.False(t, ok)
+		}
+		// just service <-> instance
+		require.Equal(t, 1, len(payload.Topologies[0].Relations))
+		for _, relation := range payload.Topologies[0].Relations {
+			tags := relation.Data.Tags
+			_, ok := tags["sts_api_key"]
+			require.False(t, ok)
+		}
+		res.WriteHeader(200)
+	}))
+	exporter := newTestExporter(t, testServer.URL)
+	err := exporter.ConsumeTraces(context.TODO(), simpleTrace())
+	require.NoError(t, err)
+
+}
+
 // simpleMetrics there will be added two ResourceMetrics and each of them have count data point
 func simpleMetrics() pmetric.Metrics {
 	metrics := pmetric.NewMetrics()
@@ -107,6 +144,14 @@ func virtualNodeMetrics() pmetric.Metrics {
 	return metrics
 }
 
+func simpleTrace() ptrace.Traces {
+	traces := ptrace.NewTraces()
+	rt := traces.ResourceSpans().AppendEmpty()
+	rt.Resource().Attributes().PutStr("service.name", "demo 1")
+	rt.Resource().Attributes().PutStr("sts_api_key", "APIKEY")
+	return traces
+}
+
 func newTestExporter(t *testing.T, url string) *topologyExporter {
 	exporter, err := newTopologyExporter(zaptest.NewLogger(t), &Config{
 		TimeoutSettings: exporterhelper.TimeoutSettings{

exporter/ststopologyexporter/factory.go

@@ -4,6 +4,7 @@ package ststopologyexporter
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/stackvista/sts-opentelemetry-collector/exporter/ststopologyexporter/internal/metadata"
@@ -17,6 +18,7 @@ func NewFactory() exporter.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		exporter.WithMetrics(createMetricsExporter, metadata.MetricsStability),
+		exporter.WithTraces(createTracesExporter, metadata.TracesStability),
 	)
 }
 
@@ -47,3 +49,26 @@ func createMetricsExporter(ctx context.Context, settings exporter.CreateSettings
 		exporterhelper.WithQueue(c.QueueSettings),
 	)
 }
+
+// createTracesExporter creates a new exporter for traces.
+// Traces are directly insert into clickhouse.
+func createTracesExporter(
+	ctx context.Context,
+	set exporter.CreateSettings,
+	cfg component.Config,
+) (exporter.Traces, error) {
+	c := cfg.(*Config)
+	exporter, err := newTopologyExporter(set.Logger, c)
+	if err != nil {
+		return nil, fmt.Errorf("cannot configure clickhouse traces exporter: %w", err)
+	}
+
+	return exporterhelper.NewTracesExporter(
+		ctx,
+		set,
+		cfg,
+		exporter.ConsumeTraces,
+		exporterhelper.WithTimeout(c.TimeoutSettings),
+		exporterhelper.WithQueue(c.QueueSettings),
+	)
+}

exporter/ststopologyexporter/metadata.yaml

@@ -3,6 +3,6 @@ type: ststopology
 status:
   class: exporter
   stability:
-    development: [metrics]
+    development: [traces, metrics]
   codeowners:
     active: StackState B.V.