Remotetap extension concept

component/go.mod

@@ -28,7 +28,10 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect

processor/internal/processor.go

@@ -3,7 +3,9 @@
 
 package internal // import "go.opentelemetry.io/collector/processor/internal"
 
-import "go.opentelemetry.io/collector/component"
+import (
+	"go.opentelemetry.io/collector/component"
+)
 
 // Settings is passed to Create* functions in Factory.
 type Settings struct {
@@ -14,4 +16,7 @@ type Settings struct {
 
 	// BuildInfo can be used by components for informational purposes
 	BuildInfo component.BuildInfo
+
+	// Publishers can be used by components to publish data after processing
+	Publishers []Publisher
 }

processor/internal/publisher.go

@@ -0,0 +1,24 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "go.opentelemetry.io/collector/processor/internal"
+
+import (
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+type ComponentID string
+
+// Publisher is used by processors to publish data as it passes through the pipeline.
+type Publisher interface {
+	// IsActive returns true when at least one connection is open for the given componentID.
+	IsActive(ComponentID) bool
+	// PublishMetrics sends metrics for a given componentID.
+	PublishMetrics(ComponentID, pmetric.Metrics)
+	// PublishTraces sends traces for a given componentID.
+	PublishTraces(ComponentID, ptrace.Traces)
+	// PublishLogs sends logs for a given componentID.
+	PublishLogs(ComponentID, plog.Logs)
+}

processor/processor.go

@@ -22,6 +22,12 @@ type Logs = internal.Logs
 // Settings is passed to Create* functions in Factory.
 type Settings = internal.Settings
 
+// Publisher is used by processors to publish data as it passes through the pipeline.
+type Publisher = internal.Publisher
+
+// ComponentID represents the ID of a component.
+type ComponentID = internal.ComponentID
+
 // Factory is Factory interface for processors.
 //
 // This interface cannot be directly implemented. Implementations must

processor/processorhelper/logs.go

@@ -46,6 +46,14 @@ func NewLogsProcessor(
 		span.AddEvent("Start processing.", eventOptions)
 		var err error
 		ld, err = logsFunc(ctx, ld)
+
+		// Publish data to active streams.
+		for _, p := range set.Publishers {
+			if p.IsActive(processor.ComponentID(set.ID.String())) {
+				p.PublishLogs(processor.ComponentID(set.ID.String()), ld)
+			}
+		}
+
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
 			if errors.Is(err, ErrSkipProcessingData) {

processor/processorhelper/metrics.go

@@ -46,6 +46,14 @@ func NewMetricsProcessor(
 		span.AddEvent("Start processing.", eventOptions)
 		var err error
 		md, err = metricsFunc(ctx, md)
+
+		// Publish data to active streams.
+		for _, p := range set.Publishers {
+			if p.IsActive(processor.ComponentID(set.ID.String())) {
+				p.PublishMetrics(processor.ComponentID(set.ID.String()), md)
+			}
+		}
+
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
 			if errors.Is(err, ErrSkipProcessingData) {

processor/processorhelper/traces.go

@@ -46,6 +46,14 @@ func NewTracesProcessor(
 		span.AddEvent("Start processing.", eventOptions)
 		var err error
 		td, err = tracesFunc(ctx, td)
+
+		// Publish data to active streams.
+		for _, p := range set.Publishers {
+			if p.IsActive(processor.ComponentID(set.ID.String())) {
+				p.PublishTraces(processor.ComponentID(set.ID.String()), td)
+			}
+		}
+
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
 			if errors.Is(err, ErrSkipProcessingData) {

service/internal/graph/graph.go

@@ -29,6 +29,7 @@ import (
 	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/internal/fanoutconsumer"
+	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/service/internal/builders"
 	"go.opentelemetry.io/collector/service/internal/capabilityconsumer"
 	"go.opentelemetry.io/collector/service/internal/status"
@@ -49,6 +50,8 @@ type Settings struct {
 	PipelineConfigs pipelines.Config
 
 	ReportStatus status.ServiceStatusFunc
+
+	Publishers []processor.Publisher
 }
 
 type Graph struct {
@@ -287,7 +290,7 @@ func (g *Graph) buildComponents(ctx context.Context, set Settings) error {
 			err = n.buildComponent(ctx, set.Telemetry, set.BuildInfo, set.ReceiverBuilder, g.nextConsumers(n.ID()))
 		case *processorNode:
 			// nextConsumers is guaranteed to be length 1.  Either it is the next processor or it is the fanout node for the exporters.
-			err = n.buildComponent(ctx, set.Telemetry, set.BuildInfo, set.ProcessorBuilder, g.nextConsumers(n.ID())[0])
+			err = n.buildComponent(ctx, set.Telemetry, set.BuildInfo, set.ProcessorBuilder, set.Publishers, g.nextConsumers(n.ID())[0])
 		case *exporterNode:
 			err = n.buildComponent(ctx, set.Telemetry, set.BuildInfo, set.ExporterBuilder)
 		case *connectorNode:

service/internal/graph/host.go

@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/featuregate"
+	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/service/extensions"
 	"go.opentelemetry.io/collector/service/internal/builders"
 	"go.opentelemetry.io/collector/service/internal/status"
@@ -64,6 +65,16 @@ func (host *Host) GetExtensions() map[component.ID]component.Component {
 	return host.ServiceExtensions.GetExtensions()
 }
 
+func (host *Host) GetPublishers() []processor.Publisher {
+	publishers := make([]processor.Publisher, 0)
+	for _, v := range host.ServiceExtensions.GetExtensions() {
+		if publisher, ok := v.(processor.Publisher); ok {
+			publishers = append(publishers, publisher)
+		}
+	}
+	return publishers
+}
+
 // Deprecated: [0.79.0] This function will be removed in the future.
 // Several components in the contrib repository use this function so it cannot be removed
 // before those cases are removed. In most cases, use of this function can be replaced by a

service/internal/graph/nodes.go

@@ -132,10 +132,11 @@ func (n *processorNode) buildComponent(ctx context.Context,
 	tel component.TelemetrySettings,
 	info component.BuildInfo,
 	builder builders.Processor,
+	publishers []processor.Publisher,
 	next baseConsumer,
 ) error {
 	tel.Logger = components.ProcessorLogger(tel.Logger, n.componentID, n.pipelineID)
-	set := processor.Settings{ID: n.componentID, TelemetrySettings: tel, BuildInfo: info}
+	set := processor.Settings{ID: n.componentID, TelemetrySettings: tel, BuildInfo: info, Publishers: publishers}
 	var err error
 	switch n.pipelineID.Type() {
 	case component.DataTypeTraces:

service/service.go

@@ -211,13 +211,13 @@ func New(ctx context.Context, set Settings, cfg Config) (*Service, error) {
 		// ignore other errors as they represent invalid state transitions and are considered benign.
 	})
 
-	if err = srv.initGraph(ctx, cfg); err != nil {
+	// process the configuration and initialize the pipeline
+	if err = srv.initExtensions(ctx, cfg.Extensions); err != nil {
 		err = multierr.Append(err, srv.shutdownTelemetry(ctx))
 		return nil, err
 	}
 
-	// process the configuration and initialize the pipeline
-	if err = srv.initExtensions(ctx, cfg.Extensions); err != nil {
+	if err = srv.initGraph(ctx, cfg); err != nil {
 		err = multierr.Append(err, srv.shutdownTelemetry(ctx))
 		return nil, err
 	}
@@ -371,6 +371,7 @@ func (srv *Service) initGraph(ctx context.Context, cfg Config) error {
 		ConnectorBuilder: srv.host.Connectors,
 		PipelineConfigs:  cfg.Pipelines,
 		ReportStatus:     srv.host.Reporter.ReportStatus,
+		Publishers:       srv.host.GetPublishers(),
 	}); err != nil {
 		return fmt.Errorf("failed to build pipelines: %w", err)
 	}