[chore] Remove telemetry from graph initialization (#10446)

djaglowski · web-flow · commit b3c781b90e39 · 2024-06-21T12:32:21.000+02:00
This PR refactors the service graph initialization so that the graph can
be assembled without the intention to start it.
1. Do not save telemetry on the graph. The only use of this was for
reporting component status. Instead, pass in a reporter when starting or
stopping. Correspondingly, add an internal `statustest` package to make
it easy to pass in a status reporter in tests.
2. Decouple graph building from extension building. There isn't any
direct relationship so these things should be separated.
diff --git a/service/internal/graph/graph.go b/service/internal/graph/graph.go
@@ -32,6 +32,7 @@ import (
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/service/internal/capabilityconsumer"
 	"go.opentelemetry.io/collector/service/internal/servicetelemetry"
+	"go.opentelemetry.io/collector/service/internal/status"
 	"go.opentelemetry.io/collector/service/pipelines"
 )
 
@@ -69,7 +70,6 @@ func Build(ctx context.Context, set Settings) (*Graph, error) {
 		componentGraph: simple.NewDirectedGraph(),
 		pipelines:      make(map[component.ID]*pipelineNodes, len(set.PipelineConfigs)),
 		instanceIDs:    make(map[int64]*component.InstanceID),
-		telemetry:      set.Telemetry,
 	}
 	for pipelineID := range set.PipelineConfigs {
 		pipelines.pipelines[pipelineID] = &pipelineNodes{
@@ -394,7 +394,7 @@ type pipelineNodes struct {
 	exporters map[int64]graph.Node
 }
 
-func (g *Graph) StartAll(ctx context.Context, host component.Host) error {
+func (g *Graph) StartAll(ctx context.Context, host component.Host, reporter status.Reporter) error {
 	nodes, err := topo.Sort(g.componentGraph)
 	if err != nil {
 		return err
@@ -413,25 +413,25 @@ func (g *Graph) StartAll(ctx context.Context, host component.Host) error {
 		}
 
 		instanceID := g.instanceIDs[node.ID()]
-		g.telemetry.Status.ReportStatus(
+		reporter.ReportStatus(
 			instanceID,
 			component.NewStatusEvent(component.StatusStarting),
 		)
 
 		if compErr := comp.Start(ctx, host); compErr != nil {
-			g.telemetry.Status.ReportStatus(
+			reporter.ReportStatus(
 				instanceID,
 				component.NewPermanentErrorEvent(compErr),
 			)
 			return compErr
 		}
 
-		g.telemetry.Status.ReportOKIfStarting(instanceID)
+		reporter.ReportOKIfStarting(instanceID)
 	}
 	return nil
 }
 
-func (g *Graph) ShutdownAll(ctx context.Context) error {
+func (g *Graph) ShutdownAll(ctx context.Context, reporter status.Reporter) error {
 	nodes, err := topo.Sort(g.componentGraph)
 	if err != nil {
 		return err
@@ -452,21 +452,21 @@ func (g *Graph) ShutdownAll(ctx context.Context) error {
 		}
 
 		instanceID := g.instanceIDs[node.ID()]
-		g.telemetry.Status.ReportStatus(
+		reporter.ReportStatus(
 			instanceID,
 			component.NewStatusEvent(component.StatusStopping),
 		)
 
 		if compErr := comp.Shutdown(ctx); compErr != nil {
 			errs = multierr.Append(errs, compErr)
-			g.telemetry.Status.ReportStatus(
+			reporter.ReportStatus(
 				instanceID,
 				component.NewPermanentErrorEvent(compErr),
 			)
 			continue
 		}
 
-		g.telemetry.Status.ReportStatus(
+		reporter.ReportStatus(
 			instanceID,
 			component.NewStatusEvent(component.StatusStopped),
 		)
diff --git a/service/internal/graph/graph_test.go b/service/internal/graph/graph_test.go
@@ -30,6 +30,7 @@ import (
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"go.opentelemetry.io/collector/service/internal/servicetelemetry"
 	"go.opentelemetry.io/collector/service/internal/status"
+	"go.opentelemetry.io/collector/service/internal/status/statustest"
 	"go.opentelemetry.io/collector/service/internal/testcomponents"
 	"go.opentelemetry.io/collector/service/pipelines"
 )
@@ -154,13 +155,13 @@ func TestGraphStartStop(t *testing.T) {
 				pg.componentGraph.SetEdge(simple.Edge{F: f, T: t})
 			}
 
-			require.NoError(t, pg.StartAll(ctx, componenttest.NewNopHost()))
+			require.NoError(t, pg.StartAll(ctx, componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
 			for _, edge := range tt.edges {
 				assert.Greater(t, ctx.order[edge[0]], ctx.order[edge[1]])
 			}
 
 			ctx.order = map[component.ID]int{}
-			require.NoError(t, pg.ShutdownAll(ctx))
+			require.NoError(t, pg.ShutdownAll(ctx, statustest.NewNopStatusReporter()))
 			for _, edge := range tt.edges {
 				assert.Less(t, ctx.order[edge[0]], ctx.order[edge[1]])
 			}
@@ -188,11 +189,11 @@ func TestGraphStartStopCycle(t *testing.T) {
 	pg.componentGraph.SetEdge(simple.Edge{F: c1, T: e1})
 	pg.componentGraph.SetEdge(simple.Edge{F: c1, T: p1}) // loop back
 
-	err := pg.StartAll(context.Background(), componenttest.NewNopHost())
+	err := pg.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter())
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `topo: no topological ordering: cyclic components`)
 
-	err = pg.ShutdownAll(context.Background())
+	err = pg.ShutdownAll(context.Background(), statustest.NewNopStatusReporter())
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `topo: no topological ordering: cyclic components`)
 }
@@ -216,8 +217,8 @@ func TestGraphStartStopComponentError(t *testing.T) {
 		F: r1,
 		T: e1,
 	})
-	assert.EqualError(t, pg.StartAll(context.Background(), componenttest.NewNopHost()), "foo")
-	assert.EqualError(t, pg.ShutdownAll(context.Background()), "bar")
+	assert.EqualError(t, pg.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()), "foo")
+	assert.EqualError(t, pg.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()), "bar")
 }
 
 func TestConnectorPipelinesGraph(t *testing.T) {
@@ -768,7 +769,7 @@ func TestConnectorPipelinesGraph(t *testing.T) {
 
 			assert.Equal(t, len(test.pipelineConfigs), len(pg.pipelines))
 
-			assert.NoError(t, pg.StartAll(context.Background(), componenttest.NewNopHost()))
+			assert.NoError(t, pg.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
 
 			mutatingPipelines := make(map[component.ID]bool, len(test.pipelineConfigs))
 
@@ -892,7 +893,7 @@ func TestConnectorPipelinesGraph(t *testing.T) {
 			}
 
 			// Shut down the entire component graph
-			assert.NoError(t, pg.ShutdownAll(context.Background()))
+			assert.NoError(t, pg.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()))
 
 			// Check each pipeline individually, ensuring that all components are stopped.
 			for pipelineID := range test.pipelineConfigs {
@@ -2148,8 +2149,8 @@ func TestGraphFailToStartAndShutdown(t *testing.T) {
 			}
 			pipelines, err := Build(context.Background(), set)
 			assert.NoError(t, err)
-			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost()))
-			assert.Error(t, pipelines.ShutdownAll(context.Background()))
+			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
+			assert.Error(t, pipelines.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()))
 		})
 
 		t.Run(dt.String()+"/processor", func(t *testing.T) {
@@ -2162,8 +2163,8 @@ func TestGraphFailToStartAndShutdown(t *testing.T) {
 			}
 			pipelines, err := Build(context.Background(), set)
 			assert.NoError(t, err)
-			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost()))
-			assert.Error(t, pipelines.ShutdownAll(context.Background()))
+			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
+			assert.Error(t, pipelines.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()))
 		})
 
 		t.Run(dt.String()+"/exporter", func(t *testing.T) {
@@ -2176,8 +2177,8 @@ func TestGraphFailToStartAndShutdown(t *testing.T) {
 			}
 			pipelines, err := Build(context.Background(), set)
 			assert.NoError(t, err)
-			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost()))
-			assert.Error(t, pipelines.ShutdownAll(context.Background()))
+			assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
+			assert.Error(t, pipelines.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()))
 		})
 
 		for _, dt2 := range dataTypes {
@@ -2196,8 +2197,8 @@ func TestGraphFailToStartAndShutdown(t *testing.T) {
 				}
 				pipelines, err := Build(context.Background(), set)
 				assert.NoError(t, err)
-				assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost()))
-				assert.Error(t, pipelines.ShutdownAll(context.Background()))
+				assert.Error(t, pipelines.StartAll(context.Background(), componenttest.NewNopHost(), statustest.NewNopStatusReporter()))
+				assert.Error(t, pipelines.ShutdownAll(context.Background(), statustest.NewNopStatusReporter()))
 			})
 		}
 	}
@@ -2350,8 +2351,8 @@ func TestStatusReportedOnStartupShutdown(t *testing.T) {
 			}
 			pg.componentGraph.SetEdge(simple.Edge{F: e0, T: e1})
 
-			assert.Equal(t, tc.startupErr, pg.StartAll(context.Background(), componenttest.NewNopHost()))
-			assert.Equal(t, tc.shutdownErr, pg.ShutdownAll(context.Background()))
+			assert.Equal(t, tc.startupErr, pg.StartAll(context.Background(), componenttest.NewNopHost(), rep))
+			assert.Equal(t, tc.shutdownErr, pg.ShutdownAll(context.Background(), rep))
 			assertEqualStatuses(t, tc.expectedStatuses, actualStatuses)
 		})
 	}
diff --git a/service/internal/servicetelemetry/telemetry_settings.go b/service/internal/servicetelemetry/telemetry_settings.go
@@ -37,7 +37,7 @@ type TelemetrySettings struct {
 
 	// Status contains a Reporter that allows the service to report status on behalf of a
 	// component.
-	Status *status.Reporter
+	Status status.Reporter
 }
 
 // ToComponentTelemetrySettings returns a TelemetrySettings for a specific component derived from
diff --git a/service/internal/status/status.go b/service/internal/status/status.go
@@ -96,7 +96,13 @@ type ServiceStatusFunc func(*component.InstanceID, *component.StatusEvent)
 var ErrStatusNotReady = errors.New("report component status is not ready until service start")
 
 // Reporter handles component status reporting
-type Reporter struct {
+type Reporter interface {
+	Ready()
+	ReportStatus(id *component.InstanceID, ev *component.StatusEvent)
+	ReportOKIfStarting(id *component.InstanceID)
+}
+
+type reporter struct {
 	mu                  sync.Mutex
 	ready               bool
 	fsmMap              map[*component.InstanceID]*fsm
@@ -106,23 +112,23 @@ type Reporter struct {
 
 // NewReporter returns a reporter that will invoke the NotifyStatusFunc when a component's status
 // has changed.
-func NewReporter(onStatusChange NotifyStatusFunc, onInvalidTransition InvalidTransitionFunc) *Reporter {
-	return &Reporter{
+func NewReporter(onStatusChange NotifyStatusFunc, onInvalidTransition InvalidTransitionFunc) Reporter {
+	return &reporter{
 		fsmMap:              make(map[*component.InstanceID]*fsm),
 		onStatusChange:      onStatusChange,
 		onInvalidTransition: onInvalidTransition,
 	}
 }
 
 // Ready enables status reporting
-func (r *Reporter) Ready() {
+func (r *reporter) Ready() {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.ready = true
 }
 
 // ReportStatus reports status for the given InstanceID
-func (r *Reporter) ReportStatus(
+func (r *reporter) ReportStatus(
 	id *component.InstanceID,
 	ev *component.StatusEvent,
 ) {
@@ -137,7 +143,7 @@ func (r *Reporter) ReportStatus(
 	}
 }
 
-func (r *Reporter) ReportOKIfStarting(id *component.InstanceID) {
+func (r *reporter) ReportOKIfStarting(id *component.InstanceID) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	if !r.ready {
@@ -152,7 +158,7 @@ func (r *Reporter) ReportOKIfStarting(id *component.InstanceID) {
 }
 
 // Note: a lock must be acquired before calling this method.
-func (r *Reporter) componentFSM(id *component.InstanceID) *fsm {
+func (r *reporter) componentFSM(id *component.InstanceID) *fsm {
 	fsm, ok := r.fsmMap[id]
 	if !ok {
 		fsm = newFSM(func(ev *component.StatusEvent) { r.onStatusChange(id, ev) })
diff --git a/service/internal/status/statustest/statustest.go b/service/internal/status/statustest/statustest.go
@@ -0,0 +1,21 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package statustest // import "go.opentelemetry.io/collector/service/internal/status/statustest"
+
+import (
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/service/internal/status"
+)
+
+func NewNopStatusReporter() status.Reporter {
+	return &nopStatusReporter{}
+}
+
+type nopStatusReporter struct{}
+
+func (r *nopStatusReporter) Ready() {}
+
+func (r *nopStatusReporter) ReportStatus(*component.InstanceID, *component.StatusEvent) {}
+
+func (r *nopStatusReporter) ReportOKIfStarting(*component.InstanceID) {}
diff --git a/service/internal/status/statustest/statustest_test.go b/service/internal/status/statustest/statustest_test.go
@@ -0,0 +1,13 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package statustest // import "go.opentelemetry.io/collector/service/internal/status/statustest"
+
+import "testing"
+
+func TestNopStatusReporter(*testing.T) {
+	nop := NewNopStatusReporter()
+	nop.Ready()
+	nop.ReportOKIfStarting(nil)
+	nop.ReportStatus(nil, nil)
+}
diff --git a/service/service.go b/service/service.go

Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,7 @@ import (`
`32`	`32`	`"go.opentelemetry.io/collector/receiver"`
`33`	`33`	`"go.opentelemetry.io/collector/service/internal/capabilityconsumer"`
`34`	`34`	`"go.opentelemetry.io/collector/service/internal/servicetelemetry"`
	`35`	`+ "go.opentelemetry.io/collector/service/internal/status"`
`35`	`36`	`"go.opentelemetry.io/collector/service/pipelines"`
`36`	`37`	`)`
`37`	`38`
`@@ -69,7 +70,6 @@ func Build(ctx context.Context, set Settings) (*Graph, error) {`
`69`	`70`	`componentGraph: simple.NewDirectedGraph(),`
`70`	`71`	`pipelines: make(map[component.ID]*pipelineNodes, len(set.PipelineConfigs)),`
`71`	`72`	`instanceIDs: make(map[int64]*component.InstanceID),`
`72`		`- telemetry: set.Telemetry,`
`73`	`73`	`}`
`74`	`74`	`for pipelineID := range set.PipelineConfigs {`
`75`	`75`	`pipelines.pipelines[pipelineID] = &pipelineNodes{`
`@@ -394,7 +394,7 @@ type pipelineNodes struct {`
`394`	`394`	`exporters map[int64]graph.Node`
`395`	`395`	`}`
`396`	`396`
`397`		`-func (g *Graph) StartAll(ctx context.Context, host component.Host) error {`
	`397`	`+func (g *Graph) StartAll(ctx context.Context, host component.Host, reporter status.Reporter) error {`
`398`	`398`	`nodes, err := topo.Sort(g.componentGraph)`
`399`	`399`	`if err != nil {`
`400`	`400`	`return err`
`@@ -413,25 +413,25 @@ func (g *Graph) StartAll(ctx context.Context, host component.Host) error {`
`413`	`413`	`}`
`414`	`414`
`415`	`415`	`instanceID := g.instanceIDs[node.ID()]`
`416`		`- g.telemetry.Status.ReportStatus(`
	`416`	`+ reporter.ReportStatus(`
`417`	`417`	`instanceID,`
`418`	`418`	`component.NewStatusEvent(component.StatusStarting),`
`419`	`419`	`)`
`420`	`420`
`421`	`421`	`if compErr := comp.Start(ctx, host); compErr != nil {`
`422`		`- g.telemetry.Status.ReportStatus(`
	`422`	`+ reporter.ReportStatus(`
`423`	`423`	`instanceID,`
`424`	`424`	`component.NewPermanentErrorEvent(compErr),`
`425`	`425`	`)`
`426`	`426`	`return compErr`
`427`	`427`	`}`
`428`	`428`
`429`		`- g.telemetry.Status.ReportOKIfStarting(instanceID)`
	`429`	`+ reporter.ReportOKIfStarting(instanceID)`
`430`	`430`	`}`
`431`	`431`	`return nil`
`432`	`432`	`}`
`433`	`433`
`434`		`-func (g *Graph) ShutdownAll(ctx context.Context) error {`
	`434`	`+func (g *Graph) ShutdownAll(ctx context.Context, reporter status.Reporter) error {`
`435`	`435`	`nodes, err := topo.Sort(g.componentGraph)`
`436`	`436`	`if err != nil {`
`437`	`437`	`return err`
`@@ -452,21 +452,21 @@ func (g *Graph) ShutdownAll(ctx context.Context) error {`
`452`	`452`	`}`
`453`	`453`
`454`	`454`	`instanceID := g.instanceIDs[node.ID()]`
`455`		`- g.telemetry.Status.ReportStatus(`
	`455`	`+ reporter.ReportStatus(`
`456`	`456`	`instanceID,`
`457`	`457`	`component.NewStatusEvent(component.StatusStopping),`
`458`	`458`	`)`
`459`	`459`
`460`	`460`	`if compErr := comp.Shutdown(ctx); compErr != nil {`
`461`	`461`	`errs = multierr.Append(errs, compErr)`
`462`		`- g.telemetry.Status.ReportStatus(`
	`462`	`+ reporter.ReportStatus(`
`463`	`463`	`instanceID,`
`464`	`464`	`component.NewPermanentErrorEvent(compErr),`
`465`	`465`	`)`
`466`	`466`	`continue`
`467`	`467`	`}`
`468`	`468`
`469`		`- g.telemetry.Status.ReportStatus(`
	`469`	`+ reporter.ReportStatus(`
`470`	`470`	`instanceID,`
`471`	`471`	`component.NewStatusEvent(component.StatusStopped),`
`472`	`472`	`)`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ type TelemetrySettings struct {`
`37`	`37`
`38`	`38`	`// Status contains a Reporter that allows the service to report status on behalf of a`
`39`	`39`	`// component.`
`40`		`- Status *status.Reporter`
	`40`	`+ Status status.Reporter`
`41`	`41`	`}`
`42`	`42`
`43`	`43`	`// ToComponentTelemetrySettings returns a TelemetrySettings for a specific component derived from`