diff --git a/config/config.go b/config/config.go index ba870ae90..52c91fec0 100644 --- a/config/config.go +++ b/config/config.go @@ -21,7 +21,6 @@ import ( "time" "github.com/coreos/go-semver/semver" - "github.com/google/uuid" "github.com/influxdata/toml" "github.com/influxdata/toml/ast" @@ -69,8 +68,11 @@ type Config struct { Outputs []*models.RunningOutput Aggregators []*models.RunningAggregator // Processors have a slice wrapper type because they need to be sorted - Processors models.RunningProcessors - AggProcessors models.RunningProcessors + Processors models.RunningProcessors + AggProcessors models.RunningProcessors + fileProcessors OrderedPlugins + fileAggProcessors OrderedPlugins + // Parsers are created by their inputs during gather. Config doesn't keep track of them // like the other plugins because they need to be garbage collected (See issue #11809) @@ -78,6 +80,17 @@ type Config struct { version *semver.Version } +// Ordered plugins used to keep the order in which they appear in a file +type OrderedPlugin struct { + Line int + plugin any +} +type OrderedPlugins []*OrderedPlugin + +func (op OrderedPlugins) Len() int { return len(op) } +func (op OrderedPlugins) Swap(i, j int) { op[i], op[j] = op[j], op[i] } +func (op OrderedPlugins) Less(i, j int) bool { return op[i].Line < op[j].Line } + // NewConfig creates a new struct to hold the Telegraf config. // For historical reasons, It holds the actual instances of the running plugins // once the configuration is parsed. @@ -95,14 +108,16 @@ func NewConfig() *Config { LogfileRotationMaxArchives: 5, }, - Tags: make(map[string]string), - Inputs: make([]*models.RunningInput, 0), - Outputs: make([]*models.RunningOutput, 0), - Processors: make([]*models.RunningProcessor, 0), - AggProcessors: make([]*models.RunningProcessor, 0), - InputFilters: make([]string, 0), - OutputFilters: make([]string, 0), - Deprecations: make(map[string][]int64), + Tags: make(map[string]string), + Inputs: make([]*models.RunningInput, 0), + Outputs: make([]*models.RunningOutput, 0), + Processors: make([]*models.RunningProcessor, 0), + AggProcessors: make([]*models.RunningProcessor, 0), + fileProcessors: make([]*OrderedPlugin, 0), + fileAggProcessors: make([]*OrderedPlugin, 0), + InputFilters: make([]string, 0), + OutputFilters: make([]string, 0), + Deprecations: make(map[string][]int64), } // Handle unknown version @@ -391,14 +406,16 @@ func (c *Config) LoadAll(configFiles ...string) error { } } + // Sort the processors according to their `order` setting while + // using a stable sort to keep the file loading / file position order. + sort.Stable(c.Processors) + sort.Stable(c.AggProcessors) + return nil } // LoadConfigData loads TOML-formatted config data func (c *Config) LoadConfigData(data []byte) error { - // Create unique identifier for plugins to identify when using multiple configurations - id := uuid.New() - tbl, err := parseConfig(data) if err != nil { return fmt.Errorf("error parsing data: %s", err) @@ -450,6 +467,10 @@ func (c *Config) LoadConfigData(data []byte) error { return fmt.Errorf("line %d: configuration specified the fields %q, but they weren't used", tbl.Line, keys(c.UnusedFields)) } + // Initialize the file-sorting slices + c.fileProcessors = make(OrderedPlugins, 0) + c.fileAggProcessors = make(OrderedPlugins, 0) + // Parse all the rest of the plugins: for name, val := range tbl.Fields { subTable, ok := val.(*ast.Table) @@ -510,7 +531,7 @@ func (c *Config) LoadConfigData(data []byte) error { switch pluginSubTable := pluginVal.(type) { case []*ast.Table: for _, t := range pluginSubTable { - if err = c.addProcessor(id.String(), pluginName, t); err != nil { + if err = c.addProcessor(pluginName, t); err != nil { return fmt.Errorf("error parsing %s, %w", pluginName, err) } } @@ -555,8 +576,16 @@ func (c *Config) LoadConfigData(data []byte) error { } } - if len(c.Processors) > 1 { - sort.Sort(c.Processors) + // Sort the processor according to the order they appeared in this file + // In a later stage, we sort them using the `order` option. + sort.Sort(c.fileProcessors) + for _, op := range c.fileProcessors { + c.Processors = append(c.Processors, op.plugin.(*models.RunningProcessor)) + } + + sort.Sort(c.fileAggProcessors) + for _, op := range c.fileAggProcessors { + c.AggProcessors = append(c.AggProcessors, op.plugin.(*models.RunningProcessor)) } return nil @@ -758,7 +787,7 @@ func (c *Config) addParser(parentcategory, parentname string, table *ast.Table) return running, err } -func (c *Config) addProcessor(id string, name string, table *ast.Table) error { +func (c *Config) addProcessor(name string, table *ast.Table) error { creator, ok := processors.Processors[name] if !ok { // Handle removed, deprecated plugins @@ -780,7 +809,7 @@ func (c *Config) addProcessor(id string, name string, table *ast.Table) error { c.setLocalMissingTomlFieldTracker(missCount) defer c.resetMissingTomlFieldTracker() - processorConfig, err := c.buildProcessor(id, name, table) + processorConfig, err := c.buildProcessor(name, table) if err != nil { return err } @@ -791,7 +820,7 @@ func (c *Config) addProcessor(id string, name string, table *ast.Table) error { return err } rf := models.NewRunningProcessor(processorBefore, processorConfig) - c.Processors = append(c.Processors, rf) + c.fileProcessors = append(c.fileProcessors, &OrderedPlugin{table.Line, rf}) // Setup another (new) processor instance running after the aggregator processorAfter, _, err := c.setupProcessor(processorConfig.Name, creator, table) @@ -799,7 +828,7 @@ func (c *Config) addProcessor(id string, name string, table *ast.Table) error { return err } rf = models.NewRunningProcessor(processorAfter, processorConfig) - c.AggProcessors = append(c.AggProcessors, rf) + c.fileAggProcessors = append(c.fileAggProcessors, &OrderedPlugin{table.Line, rf}) // Check the number of misses against the threshold if hasParser { @@ -1074,12 +1103,8 @@ func (c *Config) buildParser(name string, tbl *ast.Table) *models.ParserConfig { // buildProcessor parses Processor specific items from the ast.Table, // builds the filter and returns a // models.ProcessorConfig to be inserted into models.RunningProcessor -func (c *Config) buildProcessor(id string, name string, tbl *ast.Table) (*models.ProcessorConfig, error) { - conf := &models.ProcessorConfig{ - ID: id, - Name: name, - Line: tbl.Line, - } +func (c *Config) buildProcessor(name string, tbl *ast.Table) (*models.ProcessorConfig, error) { + conf := &models.ProcessorConfig{Name: name} c.getFieldInt64(tbl, "order", &conf.Order) c.getFieldString(tbl, "alias", &conf.Alias) diff --git a/config/config_test.go b/config/config_test.go index 9bc5be432..64fdc5fb5 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -825,9 +825,11 @@ func TestConfig_MultipleProcessorsOrder(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { c := NewConfig() - for _, f := range test.filename { - require.NoError(t, c.LoadConfig(filepath.Join("./testdata/processor_order", f))) + filenames := make([]string, 0, len(test.filename)) + for _, fn := range test.filename { + filenames = append(filenames, filepath.Join("./testdata/processor_order", fn)) } + require.NoError(t, c.LoadAll(filenames...)) require.Equal(t, len(test.expectedOrder), len(c.Processors)) @@ -862,7 +864,7 @@ func TestConfig_ProcessorsWithParsers(t *testing.T) { } c := NewConfig() - require.NoError(t, c.LoadConfig("./testdata/processors_with_parsers.toml")) + require.NoError(t, c.LoadAll("./testdata/processors_with_parsers.toml")) require.Len(t, c.Processors, len(formats)) override := map[string]struct { diff --git a/models/running_processor.go b/models/running_processor.go index 52fc40f60..0e4385741 100644 --- a/models/running_processor.go +++ b/models/running_processor.go @@ -16,48 +16,15 @@ type RunningProcessor struct { type RunningProcessors []*RunningProcessor -func (rp RunningProcessors) Len() int { - return len(rp) -} -func (rp RunningProcessors) Swap(i, j int) { - rp[i], rp[j] = rp[j], rp[i] -} -func (rp RunningProcessors) Less(i, j int) bool { - // If the processors are defined in separate files only sort based on order - if rp[i].Config.ID != rp[j].Config.ID { - return rp[i].Config.Order < rp[j].Config.Order - } - - // If Order is defined for both processors, sort according to the number set - if rp[i].Config.Order != 0 && rp[j].Config.Order != 0 { - // If both orders are equal, ensure config order is maintained - if rp[i].Config.Order == rp[j].Config.Order { - return rp[i].Config.Line < rp[j].Config.Line - } - - return rp[i].Config.Order < rp[j].Config.Order - } - - // If "Order" is defined for one processor but not another, - // the processor without an "Order" will always take precedence. - // This adheres to the original implementation. - if rp[i].Config.Order != 0 { - return false - } - if rp[j].Config.Order != 0 { - return true - } - - return rp[i].Config.Line < rp[j].Config.Line -} +func (rp RunningProcessors) Len() int { return len(rp) } +func (rp RunningProcessors) Swap(i, j int) { rp[i], rp[j] = rp[j], rp[i] } +func (rp RunningProcessors) Less(i, j int) bool { return rp[i].Config.Order < rp[j].Config.Order } // ProcessorConfig containing a name and filter type ProcessorConfig struct { - ID string Name string Alias string Order int64 - Line int Filter Filter }