[pkg/stanza] Fix operator sorting internal to stanza now that its int…

…egrated (open-telemetry#10288)

Fix operator sorting internal to stanza now that its integrated

CHANGELOG.md

@@ -31,6 +31,7 @@
 - `mongodbreceiver`: Fix issue where receiver startup could hang (#10111)
 - `transformprocessor`: Fix issue where metric.aggregation_temporality and metric.is_monotic were not actually gettable or settable (#10197)
 - `podmanreceiver`: Container Stats Error structure (#9397)
+- `pkg/stanza`: pipeline.Operators() will return a consistently ordered list of operators whenever possible (#9761)
 
 ## v0.51.0
 

pkg/stanza/pipeline/directed.go

@@ -98,6 +98,15 @@ func (p *DirectedPipeline) Render() ([]byte, error) {
 // Operators returns a slice of operators that make up the pipeline graph
 func (p *DirectedPipeline) Operators() []operator.Operator {
  operators := make([]operator.Operator, 0)
+ if nodes, err := topo.Sort(p.Graph); err == nil {
+ for _, node := range nodes {
+ operators = append(operators, node.(OperatorNode).Operator())
+ }
+ return operators
+ }
+
+ // If for some unexpected reason an Unorderable error is returned,
+ // when using topo.Sort, return the list without ordering
  nodes := p.Graph.Nodes()
  for nodes.Next() {
  operators = append(operators, nodes.Node().(OperatorNode).Operator())

pkg/stanza/pipeline/directed_test.go

@@ -309,5 +309,5 @@ func TestPipelineOperators(t *testing.T) {
  require.NoError(t, err)
 
  operators := pipeline.Operators()
- require.ElementsMatch(t, []operator.Operator{mockOperator1, mockOperator2, mockOperator3}, operators)
+ require.Equal(t, []operator.Operator{mockOperator1, mockOperator2, mockOperator3}, operators)
 }

processor/logstransformprocessor/go.mod

@@ -10,7 +10,6 @@ require (
  go.opentelemetry.io/collector v0.51.1-0.20220519211145-c56d20e9e0af
  go.opentelemetry.io/collector/pdata v0.51.1-0.20220519211145-c56d20e9e0af
  go.uber.org/zap v1.21.0
- gonum.org/v1/gonum v0.11.0
 )
 
 require (
@@ -36,6 +35,7 @@ require (
  golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
  golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
  golang.org/x/text v0.3.7 // indirect
+ gonum.org/v1/gonum v0.11.0 // indirect
  google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
  google.golang.org/grpc v1.46.2 // indirect
  google.golang.org/protobuf v1.28.0 // indirect

processor/logstransformprocessor/processor.go

@@ -26,7 +26,6 @@ import (
  "go.opentelemetry.io/collector/config"
  "go.opentelemetry.io/collector/pdata/plog"
  "go.uber.org/zap"
- "gonum.org/v1/gonum/graph/topo"
 
  "github.com/open-telemetry/opentelemetry-collector-contrib/internal/stanza"
  "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
@@ -98,15 +97,11 @@ func (ltp *logsTransformProcessor) Start(ctx context.Context, host component.Hos
  }
 
  ltp.pipe = pipe
-
- orderedNodes, err := topo.Sort(pipe.Graph)
- if err != nil {
- return err
- }
- if len(orderedNodes) == 0 {
+ pipelineOperators := pipe.Operators()
+ if len(pipelineOperators) == 0 {
  return errors.New("processor requires at least one operator to be configured")
  }
- ltp.firstOperator = orderedNodes[0].(pipeline.OperatorNode).Operator()
+ ltp.firstOperator = pipelineOperators[0]
 
  wkrCount := int(math.Max(1, float64(runtime.NumCPU())))
  if baseCfg.Converter.WorkerCount > 0 {