[processor/servicegraph] update own telemetry to use otel (#29917)

This updates the servicegraph processor to emit telemetry using
OpenTelemetry instead of OpenCensus.

Related #29867

---------

Signed-off-by: Alex Boten <[email protected]>

Author: Alex Boten

.chloggen/codeboten_rm-census-servicegraph.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: servicegraphprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: update own telemetry to use otel
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [29917]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

processor/servicegraphprocessor/factory.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"time"
 
-	"go.opencensus.io/stats/view"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/consumer"
@@ -51,9 +50,6 @@ func init() {
 
 // NewFactory creates a factory for the servicegraph processor.
 func NewFactory() processor.Factory {
-	// TODO: Handle this err
-	_ = view.Register(serviceGraphProcessorViews()...)
-
 	return processor.NewFactory(
 		typeStr,
 		createDefaultConfig,
@@ -63,8 +59,6 @@ func NewFactory() processor.Factory {
 
 // NewConnectorFactoryFunc creates a function that returns a factory for the servicegraph connector.
 var NewConnectorFactoryFunc = func(cfgType component.Type, tracesToMetricsStability component.StabilityLevel) connector.Factory {
-	// TODO: Handle this err
-	_ = view.Register(serviceGraphProcessorViews()...)
 	return connector.NewFactory(
 		cfgType,
 		createDefaultConfig,
@@ -84,13 +78,13 @@ func createDefaultConfig() component.Config {
 }
 
 func createTracesProcessor(_ context.Context, params processor.CreateSettings, cfg component.Config, nextConsumer consumer.Traces) (processor.Traces, error) {
-	p := newProcessor(params.Logger, cfg)
+	p := newProcessor(params.TelemetrySettings, cfg)
 	p.tracesConsumer = nextConsumer
 	return p, nil
 }
 
 func createTracesToMetricsConnector(_ context.Context, params connector.CreateSettings, cfg component.Config, nextConsumer consumer.Metrics) (connector.Traces, error) {
-	c := newProcessor(params.Logger, cfg)
+	c := newProcessor(params.TelemetrySettings, cfg)
 	c.metricsConsumer = nextConsumer
 	return c, nil
 }

processor/servicegraphprocessor/go.mod

@@ -5,9 +5,9 @@ go 1.20
 
 require (
 	github.com/stretchr/testify v1.8.4
-	go.opencensus.io v0.24.0
 	go.opentelemetry.io/collector/component v0.91.0
 	go.opentelemetry.io/collector/config/configgrpc v0.91.0
+	go.opentelemetry.io/collector/config/configtelemetry v0.91.0
 	go.opentelemetry.io/collector/connector v0.91.0
 	go.opentelemetry.io/collector/consumer v0.91.0
 	go.opentelemetry.io/collector/exporter v0.91.0
@@ -17,6 +17,8 @@ require (
 	go.opentelemetry.io/collector/pdata v1.0.0
 	go.opentelemetry.io/collector/processor v0.91.0
 	go.opentelemetry.io/collector/semconv v0.91.0
+	go.opentelemetry.io/otel/metric v1.21.0
+	go.opentelemetry.io/otel/sdk/metric v1.21.0
 	go.uber.org/zap v1.26.0
 )
 
@@ -67,12 +69,12 @@ require (
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.3 // indirect
+	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/configauth v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/configcompression v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/confignet v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/configopaque v0.91.0 // indirect
-	go.opentelemetry.io/collector/config/configtelemetry v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/configtls v0.91.0 // indirect
 	go.opentelemetry.io/collector/config/internal v0.91.0 // indirect
 	go.opentelemetry.io/collector/confmap v0.91.0 // indirect
@@ -93,9 +95,7 @@ require (
 	go.opentelemetry.io/otel/exporters/prometheus v0.44.1-0.20231201153405-6027c1ae76f2 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.44.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.21.0 // indirect
-	go.opentelemetry.io/otel/metric v1.21.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.21.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.21.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect

processor/servicegraphprocessor/metrics.go

@@ -12,17 +12,19 @@ import (
 	"sync"
 	"time"
 
-	"go.opencensus.io/stats"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
 	semconv "go.opentelemetry.io/collector/semconv/v1.13.0"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/servicegraphprocessor/internal/metadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/servicegraphprocessor/internal/store"
 )
 
@@ -76,10 +78,14 @@ type serviceGraphProcessor struct {
 	metricMutex sync.RWMutex
 	keyToMetric map[string]metricSeries
 
+	statDroppedSpans metric.Int64Counter
+	statTotalEdges   metric.Int64Counter
+	statExpiredEdges metric.Int64Counter
+
 	shutdownCh chan any
 }
 
-func newProcessor(logger *zap.Logger, config component.Config) *serviceGraphProcessor {
+func newProcessor(set component.TelemetrySettings, config component.Config) *serviceGraphProcessor {
 	pConfig := config.(*Config)
 
 	bounds := defaultLatencyHistogramBuckets
@@ -102,9 +108,28 @@ func newProcessor(logger *zap.Logger, config component.Config) *serviceGraphProc
 		pConfig.VirtualNodePeerAttributes = defaultPeerAttributes
 	}
 
+	scopeName := "processor/servicegraphprocessor"
+	meter := set.MeterProvider.Meter(scopeName)
+
+	droppedSpan, _ := meter.Int64Counter(
+		processorhelper.BuildCustomMetricName(metadata.Type, "dropped_spans"),
+		metric.WithDescription("Number of spans dropped when trying to add edges"),
+		metric.WithUnit("1"),
+	)
+	totalEdges, _ := meter.Int64Counter(
+		processorhelper.BuildCustomMetricName(metadata.Type, "total_edges"),
+		metric.WithDescription("Total number of unique edges"),
+		metric.WithUnit("1"),
+	)
+	expiredEdges, _ := meter.Int64Counter(
+		processorhelper.BuildCustomMetricName(metadata.Type, "expired_edges"),
+		metric.WithDescription("Number of edges that expired before finding its matching span"),
+		metric.WithUnit("1"),
+	)
+
 	return &serviceGraphProcessor{
 		config:                               pConfig,
-		logger:                               logger,
+		logger:                               set.Logger,
 		startTime:                            time.Now(),
 		reqTotal:                             make(map[string]int64),
 		reqFailedTotal:                       make(map[string]int64),
@@ -117,6 +142,9 @@ func newProcessor(logger *zap.Logger, config component.Config) *serviceGraphProc
 		reqDurationBounds:                    bounds,
 		keyToMetric:                          make(map[string]metricSeries),
 		shutdownCh:                           make(chan any),
+		statDroppedSpans:                     droppedSpan,
+		statTotalEdges:                       totalEdges,
+		statExpiredEdges:                     expiredEdges,
 	}
 }
 
@@ -299,7 +327,7 @@ func (p *serviceGraphProcessor) aggregateMetrics(ctx context.Context, td ptrace.
 
 				if errors.Is(err, store.ErrTooManyItems) {
 					totalDroppedSpans++
-					stats.Record(ctx, statDroppedSpans.M(1))
+					p.statDroppedSpans.Add(ctx, 1)
 					continue
 				}
 
@@ -309,7 +337,7 @@ func (p *serviceGraphProcessor) aggregateMetrics(ctx context.Context, td ptrace.
 				}
 
 				if isNew {
-					stats.Record(ctx, statTotalEdges.M(1))
+					p.statTotalEdges.Add(ctx, 1)
 				}
 			}
 		}
@@ -354,7 +382,7 @@ func (p *serviceGraphProcessor) onExpire(e *store.Edge) {
 		zap.Stringer("trace_id", e.TraceID),
 	)
 
-	stats.Record(context.Background(), statExpiredEdges.M(1))
+	p.statExpiredEdges.Add(context.Background(), 1)
 
 	if virtualNodeFeatureGate.IsEnabled() {
 		e.ConnectionType = store.VirtualNode