CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore: improve processor ordering (#12308)

This commit is contained in:
Sven Rebhan 2022-11-30 17:28:23 +01:00 committed by GitHub
parent c4dc104bb0
commit 337e4e34bc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 60 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
config/config.go
View File

@ -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)

8
config/config_test.go
View File

@ -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 {

39
models/running_processor.go
View File

@ -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
}