feat: upgrade v0.98

Makefile

@@ -33,7 +33,7 @@ build:
 
 .PHONY: run
 run:
-	go run cmd/otel-collector/*.go --config ${CONFIG_FILE} --feature-gates=telemetry.useOtelForInternalMetrics
+	go run cmd/otel-collector/*.go --config ${CONFIG_FILE}
 
 .PHONY: fmt
 fmt:

README.md

@@ -106,7 +106,28 @@ receivers:
             - targets: ['exporter:8080']
   influxdb:
     endpoint: 0.0.0.0:8086
-
+connectors:
+  servicegraph:
+    latency_histogram_buckets: [ 100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms ]
+    dimensions: [ cluster, namespace ]
+    store:
+      ttl: 2s
+      max_items: 1000
+    cache_loop: 2m
+    store_expiration_loop: 2s
+    virtual_node_peer_attributes:
+      - db.name
+      - rpc.service
+  spanmetrics:
+    namespace: span.metrics
+    exemplars:
+      enabled: false
+    dimensions_cache_size: 1000
+    aggregation_temporality: 'AGGREGATION_TEMPORALITY_CUMULATIVE'
+    metrics_flush_interval: 30s
+    metrics_expiration: 5m
+    events:
+      enabled: false
 processors:
   batch:
     send_batch_size: 10000
@@ -124,17 +145,6 @@ processors:
       - key: service.name
         value: "serviceName"
         action: upsert
-  spanmetrics:
-    metrics_exporter: otlp/spanmetrics
-    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
-    dimensions_cache_size: 1500
-  servicegraph:
-    metrics_exporter: otlp/spanmetrics
-    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
-    dimensions: [cluster, namespace]
-    store:
-      ttl: 2s
-      max_items: 200
   metricstransform:
     transforms:
       - include: calls_total
@@ -174,14 +184,10 @@ service:
       exporters: [qryn]
     traces:
       receivers: [otlp, jaeger, zipkin, skywalking]
-      processors: [memory_limiter, resourcedetection/system, resource, spanmetrics, servicegraph, batch]
-      exporters: [qryn]
-    metrics/spanmetrics:
-      receivers: [otlp]
-      processors: [metricstransform]
-      exporters: [qryn]
+      processors: [memory_limiter, resourcedetection/system, resource, batch]
+      exporters: [qryn, spanmetrics, servicegraph]
     metrics:
-      receivers: [prometheus, influxdb]
+      receivers: [prometheus, influxdb, spanmetrics, servicegraph]
       processors: [memory_limiter, resourcedetection/system, resource, batch]
       exporters: [qryn]
 ```

cmd/otel-collector/components.go

@@ -26,7 +26,6 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/zipkinencodingextension"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/headerssetterextension"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckextension"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/httpforwarderextension"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/jaegerremotesampling"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/oauth2clientauthextension"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/dockerobserver"
@@ -182,7 +181,6 @@ func components() (otelcol.Factories, error) {
 		bearertokenauthextension.NewFactory(),
 		headerssetterextension.NewFactory(),
 		healthcheckextension.NewFactory(),
-		httpforwarderextension.NewFactory(),
 		jaegerremotesampling.NewFactory(),
 		oauth2clientauthextension.NewFactory(),
 		ecsobserver.NewFactory(),

config/example-config.yaml

@@ -23,34 +23,44 @@ processors:
     detectors: ['system']
     system:
       hostname_sources: ['os']
-  spanmetrics:
-    metrics_exporter: 'otlp'
-    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
-    dimensions_cache_size: 1500
-    dimensions:
-      - name: http.method
-        default: GET
-      - name: http.status_code
-      - name: span.name
+
+connectors:
   servicegraph:
-    metrics_exporter: 'otlp'
     latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
     dimensions: [cluster, namespace]
     store:
       ttl: 2s
-      max_items: 200
-  metricstransform:
-    transforms:
-      - include: calls_total
-        action: update
-        new_name: traces_spanmetrics_calls_total
-        operations:
-          - action: update_label
-            label: operation
-            new_label: span.name
-      - include: latency
-        action: update
-        new_name: traces_spanmetrics_latency
+      max_items: 1000
+    cache_loop: 2m
+    store_expiration_loop: 2s
+    virtual_node_peer_attributes:
+     - db.name
+     - rpc.service
+  spanmetrics:
+    namespace: span.metrics
+    dimensions:
+      - name: exchange
+      - name: k8s.namespace.name
+      - name: queue
+      - name: k8s.cluster.name
+      - name: host.name
+      - name: trace.root
+      - name: db.system
+      - name: db.name
+      - name: db.type
+      - name: db.addr
+      - name: db.method
+      - name: net.peer.ip
+      - name: k8s.pod.name
+      - name: http.response.status_code
+    exemplars:
+      enabled: false
+    dimensions_cache_size: 1000
+    aggregation_temporality: 'AGGREGATION_TEMPORALITY_CUMULATIVE'
+    metrics_flush_interval: 30s
+    metrics_expiration: 5m
+    events:
+      enabled: false
 
 exporters:
   qryn:
@@ -84,8 +94,6 @@ extensions:
     mutex_profile_fraction: 100
     block_profile_fraction: 100
   zpages:
-  memory_ballast:
-    size_mib: 1000
 service:
   telemetry:
     metrics:
@@ -104,17 +112,10 @@ service:
       exporters: [clickhouseprofileexporter]
     traces:
       receivers: [otlp]
-      processors: [resourcedetection/system, spanmetrics, batch]
-      exporters: [qryn]
-    traces/servicegraph:
-      receivers: [otlp]
-      processors: [servicegraph]
-      exporters: [otlp]
-    metrics/spanmetrics:
-      receivers: [otlp/dummy]
-      exporters: [otlp]
+      processors: [resourcedetection/system, batch]
+      exporters: [qryn, spanmetrics, servicegraph]
     metrics:
-      receivers: [otlp]
+      receivers: [otlp, spanmetrics, servicegraph]
       processors:
-        [memory_limiter, resourcedetection/system, metricstransform, batch]
+        [memory_limiter, resourcedetection/system, batch]
       exporters: [qryn]

exporter/qrynexporter/config_test.go

@@ -27,11 +27,11 @@ func TestLoadConfig(t *testing.T) {
 		expected component.Config
 	}{
 		{
-			id:       component.NewIDWithName(typeStr, ""),
+			id:       component.NewIDWithName(component.MustNewType(typeStr), ""),
 			expected: defaultCfg,
 		},
 		{
-			id: component.NewIDWithName(typeStr, "full"),
+			id: component.NewIDWithName(component.MustNewType(typeStr), "full"),
 			expected: &Config{
 				DSN: defaultDSN,
 				TimeoutSettings: exporterhelper.TimeoutSettings{

exporter/qrynexporter/example/otel-collector-config.yaml

@@ -11,6 +11,23 @@ receivers:
       grpc:
         endpoint: 'localhost:12345'
 
+connectors:
+  servicegraph:
+    metrics_exporter: 'otlp'
+    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
+    dimensions: [cluster, namespace]
+    store:
+      ttl: 2s
+      max_items: 200
+  spanmetrics:
+    namespace: span.metrics
+    metrics_exporter: 'otlp'
+    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
+    dimensions_cache_size: 1500
+    dimensions:
+      - name: http.method
+        default: GET
+      - name: http.status_code
 processors:
   batch:
     send_batch_size: 1000
@@ -23,21 +40,6 @@ processors:
     detectors: ['system']
     system:
       hostname_sources: ['os']
-  spanmetrics:
-    metrics_exporter: 'otlp'
-    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
-    dimensions_cache_size: 1500
-    dimensions:
-      - name: http.method
-        default: GET
-      - name: http.status_code
-  servicegraph:
-    metrics_exporter: 'otlp'
-    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms]
-    dimensions: [cluster, namespace]
-    store:
-      ttl: 2s
-      max_items: 200
   metricstransform:
     transforms:
       - include: calls_total
@@ -73,8 +75,6 @@ extensions:
   health_check:
   pprof:
   zpages:
-  memory_ballast:
-    size_mib: 1000
 
 service:
   extensions: [pprof, zpages, health_check]
@@ -85,17 +85,10 @@ service:
       exporters: [qryn]
     traces:
       receivers: [otlp]
-      processors: [resourcedetection/system, spanmetrics, batch]
-      exporters: [qryn]
-    traces/servicegraph:
-      receivers: [otlp]
-      processors: [servicegraph]
-      exporters: [otlp]
-    metrics/spanmetrics:
-      receivers: [otlp/dummy]
-      exporters: [otlp]
+      processors: [resourcedetection/system, batch]
+      exporters: [qryn, spanmetrics, servicegraph]
     metrics:
-      receivers: [otlp]
+      receivers: [otlp, servicegraph, spanmetrics]
       processors:
         [memory_limiter, resourcedetection/system, metricstransform, batch]
       exporters: [qryn]

exporter/qrynexporter/traces.go

@@ -243,11 +243,17 @@ func extractServiceName(tags map[string]string) string {
 	return serviceName
 }
 
-func mergeAttributes(span ptrace.Span, resource pcommon.Resource) {
+func mergeAttributes(span ptrace.Span, resource pcommon.Resource) pcommon.Map {
+	newAttributes := pcommon.NewMap()
 	resource.Attributes().Range(func(k string, v pcommon.Value) bool {
-		span.Attributes().PutStr(k, v.AsString())
+		newAttributes.PutStr(k, v.AsString())
 		return true
 	})
+	span.Attributes().Range(func(k string, v pcommon.Value) bool {
+		newAttributes.PutStr(k, v.AsString())
+		return true
+	})
+	return newAttributes
 }
 
 func sliceToArray(vs pcommon.Slice) []*commonv1.AnyValue {
@@ -329,7 +335,7 @@ func spanEventSliceToSpanEvents(spanEvents ptrace.SpanEventSlice) []*tracev1.Spa
 	return events
 }
 
-func marshalSpanToJSON(span ptrace.Span) ([]byte, error) {
+func marshalSpanToJSON(span ptrace.Span, mergedAttributes pcommon.Map) ([]byte, error) {
 	traceID := span.TraceID()
 	spanID := span.SpanID()
 	parentSpanID := span.ParentSpanID()
@@ -342,7 +348,7 @@ func marshalSpanToJSON(span ptrace.Span) ([]byte, error) {
 		Kind:                   tracev1.Span_SpanKind(span.Kind()),
 		StartTimeUnixNano:      uint64(span.StartTimestamp()),
 		EndTimeUnixNano:        uint64(span.EndTimestamp()),
-		Attributes:             mapToKeyValueList(span.Attributes()),
+		Attributes:             mapToKeyValueList(mergedAttributes),
 		DroppedAttributesCount: span.DroppedAttributesCount(),
 		Events:                 spanEventSliceToSpanEvents(span.Events()),
 		DroppedEventsCount:     span.DroppedEventsCount(),
@@ -357,7 +363,6 @@ func marshalSpanToJSON(span ptrace.Span) ([]byte, error) {
 }
 
 func convertTracesInput(span ptrace.Span, resource pcommon.Resource, serviceName string, tags map[string]string) (*Trace, error) {
-	mergeAttributes(span, resource)
 	durationNano := uint64(span.EndTimestamp() - span.StartTimestamp())
 	tags = aggregateSpanTags(span, tags)
 	tags["service.name"] = serviceName
@@ -369,7 +374,7 @@ func convertTracesInput(span ptrace.Span, resource pcommon.Resource, serviceName
 	for k, v := range tags {
 		mTags = append(mTags, []string{k, v})
 	}
-	payload, err := marshalSpanToJSON(span)
+	payload, err := marshalSpanToJSON(span, mergeAttributes(span, resource))
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal span: %w", err)
 	}