Improve error message when cycle is detected in component graph (open-telemetry#7206)

See open-telemetry#7045 (comment)

Author: djaglowski

service/graph.go

@@ -16,7 +16,10 @@ package service // import "go.opentelemetry.io/collector/service"
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"net/http"
+	"strings"
 
 	"go.uber.org/multierr"
 	"gonum.org/v1/gonum/graph"
@@ -159,10 +162,7 @@ func (g *pipelinesGraph) createEdges() {
 func (g *pipelinesGraph) buildComponents(ctx context.Context, set pipelinesSettings) error {
 	nodes, err := topo.Sort(g.componentGraph)
 	if err != nil {
-		// TODO When there is a cycle in the graph, there is enough information
-		// within the error to construct a better error message that indicates
-		// exactly the components that are in a cycle.
-		return err
+		return cycleErr(err, topo.DirectedCyclesIn(g.componentGraph))
 	}
 
 	for i := len(nodes) - 1; i >= 0; i-- {
@@ -356,3 +356,43 @@ func (p *pipelineNodes) exporterIDs() []string {
 func (p *pipelineNodes) mutatesData() bool {
 	return p.capabilitiesNode.getConsumer().Capabilities().MutatesData
 }
+
+func cycleErr(err error, cycles [][]graph.Node) error {
+	var topoErr topo.Unorderable
+	if !errors.As(err, &topoErr) || len(cycles) == 0 || len(cycles[0]) == 0 {
+		return err
+	}
+
+	// There may be multiple cycles, but report only the first one.
+	cycle := cycles[0]
+
+	// The last node is a duplicate of the first node.
+	// Remove it because we may start from a different node.
+	cycle = cycle[:len(cycle)-1]
+
+	// A cycle always contains a connector. For the sake of consistent
+	// error messages report the cycle starting from a connector.
+	for i := 0; i < len(cycle); i++ {
+		if _, ok := cycle[i].(*connectorNode); ok {
+			cycle = append(cycle[i:], cycle[:i]...)
+			break
+		}
+	}
+
+	// Repeat the first node at the end to clarify the cycle
+	cycle = append(cycle, cycle[0])
+
+	// Build the error message
+	componentDetails := make([]string, 0, len(cycle))
+	for _, node := range cycle {
+		switch n := node.(type) {
+		case *processorNode:
+			componentDetails = append(componentDetails, fmt.Sprintf("processor %q in pipeline %q", n.componentID, n.pipelineID))
+		case *connectorNode:
+			componentDetails = append(componentDetails, fmt.Sprintf("connector %q (%s to %s)", n.componentID, n.exprPipelineType, n.rcvrPipelineType))
+		default:
+			continue // skip capabilities/fanout nodes
+		}
+	}
+	return fmt.Errorf("cycle detected: %s", strings.Join(componentDetails, " -> "))
+}

service/graph_test.go

@@ -824,14 +824,13 @@ func TestGraphBuildErrors(t *testing.T) {
 	badConnectorFactory := newBadConnectorFactory()
 
 	tests := []struct {
-		name               string
-		receiverCfgs       map[component.ID]component.Config
-		processorCfgs      map[component.ID]component.Config
-		exporterCfgs       map[component.ID]component.Config
-		connectorCfgs      map[component.ID]component.Config
-		pipelineCfgs       map[component.ID]*PipelineConfig
-		expected           string
-		expectedStartsWith string
+		name          string
+		receiverCfgs  map[component.ID]component.Config
+		processorCfgs map[component.ID]component.Config
+		exporterCfgs  map[component.ID]component.Config
+		connectorCfgs map[component.ID]component.Config
+		pipelineCfgs  map[component.ID]*PipelineConfig
+		expected      string
 	}{
 		{
 			name: "not_supported_exporter_logs",
@@ -1217,7 +1216,10 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewIDWithName("nop", "conn")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
+			expected: `cycle detected: ` +
+				`connector "nop/conn" (traces to traces) -> ` +
+				`processor "nop" in pipeline "traces" -> ` +
+				`connector "nop/conn" (traces to traces)`,
 		},
 		{
 			name: "not_allowed_simple_cycle_metrics.yaml",
@@ -1240,7 +1242,10 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewIDWithName("nop", "conn")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
+			expected: `cycle detected: ` +
+				`connector "nop/conn" (metrics to metrics) -> ` +
+				`processor "nop" in pipeline "metrics" -> ` +
+				`connector "nop/conn" (metrics to metrics)`,
 		},
 		{
 			name: "not_allowed_simple_cycle_logs.yaml",
@@ -1263,7 +1268,10 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewIDWithName("nop", "conn")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
+			expected: `cycle detected: ` +
+				`connector "nop/conn" (logs to logs) -> ` +
+				`processor "nop" in pipeline "logs" -> ` +
+				`connector "nop/conn" (logs to logs)`,
 		},
 		{
 			name: "not_allowed_deep_cycle_traces.yaml",
@@ -1303,8 +1311,12 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewID("nop")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
-			// TODO rebuild cycle in order
+			expected: `cycle detected: ` +
+				`connector "nop/conn1" (traces to traces) -> ` +
+				`processor "nop" in pipeline "traces/2" -> ` +
+				`connector "nop/conn" (traces to traces) -> ` +
+				`processor "nop" in pipeline "traces/1" -> ` +
+				`connector "nop/conn1" (traces to traces)`,
 		},
 		{
 			name: "not_allowed_deep_cycle_metrics.yaml",
@@ -1344,8 +1356,12 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewID("nop")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
-			// TODO rebuild cycle in order
+			expected: `cycle detected: ` +
+				`connector "nop/conn1" (metrics to metrics) -> ` +
+				`processor "nop" in pipeline "metrics/2" -> ` +
+				`connector "nop/conn" (metrics to metrics) -> ` +
+				`processor "nop" in pipeline "metrics/1" -> ` +
+				`connector "nop/conn1" (metrics to metrics)`,
 		},
 		{
 			name: "not_allowed_deep_cycle_logs.yaml",
@@ -1385,8 +1401,12 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewID("nop")},
 				},
 			},
-			expectedStartsWith: "topo: no topological ordering: cyclic components:",
-			// TODO rebuild cycle in order
+			expected: `cycle detected: ` +
+				`connector "nop/conn1" (logs to logs) -> ` +
+				`processor "nop" in pipeline "logs/2" -> ` +
+				`connector "nop/conn" (logs to logs) -> ` +
+				`processor "nop" in pipeline "logs/1" -> ` +
+				`connector "nop/conn1" (logs to logs)`,
 		},
 		{
 			name: "not_allowed_deep_cycle_multi_signal.yaml",
@@ -1442,7 +1462,14 @@ func TestGraphBuildErrors(t *testing.T) {
 					Exporters:  []component.ID{component.NewIDWithName("nop", "fork")}, // cannot loop back to "nop/fork"
 				},
 			},
-			expected: "topo: no topological ordering: 12 nodes in 1 cyclic components",
+			expected: `cycle detected: ` +
+				`connector "nop/rawlog" (traces to logs) -> ` +
+				`processor "nop" in pipeline "logs/raw" -> ` +
+				`connector "nop/fork" (logs to traces) -> ` +
+				`processor "nop" in pipeline "traces/copy2" -> ` +
+				`connector "nop/forkagain" (traces to traces) -> ` +
+				`processor "nop" in pipeline "traces/copy2b" -> ` +
+				`connector "nop/rawlog" (traces to logs)`,
 		},
 		{
 			name: "unknown_exporter_config",
@@ -1605,12 +1632,7 @@ func TestGraphBuildErrors(t *testing.T) {
 				PipelineConfigs: test.pipelineCfgs,
 			}
 			_, err := buildPipelinesGraph(context.Background(), set)
-			if test.expected != "" {
-				assert.EqualError(t, err, test.expected)
-			} else {
-				assert.Error(t, err)
-				assert.Contains(t, err.Error(), test.expectedStartsWith)
-			}
+			assert.EqualError(t, err, test.expected)
 		})
 	}
 }