forked from influxdata/telegraf
-
Notifications
You must be signed in to change notification settings - Fork 0
/
config.go
174 lines (151 loc) · 4.23 KB
/
config.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
package shim
import (
"errors"
"fmt"
"log" //nolint:depguard // Allow exceptional but valid use of log here.
"os"
"github.com/BurntSushi/toml"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
"github.com/influxdata/telegraf/plugins/outputs"
"github.com/influxdata/telegraf/plugins/processors"
)
type config struct {
Inputs map[string][]toml.Primitive
Processors map[string][]toml.Primitive
Outputs map[string][]toml.Primitive
}
type loadedConfig struct {
Input telegraf.Input
Processor telegraf.StreamingProcessor
Output telegraf.Output
}
// LoadConfig Adds plugins to the shim
func (s *Shim) LoadConfig(filePath *string) error {
conf, err := LoadConfig(filePath)
if err != nil {
return err
}
if conf.Input != nil {
if err = s.AddInput(conf.Input); err != nil {
return fmt.Errorf("failed to add Input: %w", err)
}
} else if conf.Processor != nil {
if err = s.AddStreamingProcessor(conf.Processor); err != nil {
return fmt.Errorf("failed to add Processor: %w", err)
}
} else if conf.Output != nil {
if err = s.AddOutput(conf.Output); err != nil {
return fmt.Errorf("failed to add Output: %w", err)
}
}
return nil
}
// LoadConfig loads the config and returns inputs that later need to be loaded.
func LoadConfig(filePath *string) (loaded loadedConfig, err error) {
var data string
conf := config{}
if filePath != nil && *filePath != "" {
b, err := os.ReadFile(*filePath)
if err != nil {
return loadedConfig{}, err
}
data = expandEnvVars(b)
} else {
conf = DefaultImportedPlugins()
}
md, err := toml.Decode(data, &conf)
if err != nil {
return loadedConfig{}, err
}
return createPluginsWithTomlConfig(md, conf)
}
func expandEnvVars(contents []byte) string {
return os.Expand(string(contents), getEnv)
}
func getEnv(key string) string {
v := os.Getenv(key)
return envVarEscaper.Replace(v)
}
func createPluginsWithTomlConfig(md toml.MetaData, conf config) (loadedConfig, error) {
loadedConf := loadedConfig{}
for name, primitives := range conf.Inputs {
creator, ok := inputs.Inputs[name]
if !ok {
return loadedConf, errors.New("unknown input " + name)
}
plugin := creator()
if len(primitives) > 0 {
primitive := primitives[0]
if err := md.PrimitiveDecode(primitive, plugin); err != nil {
return loadedConf, err
}
}
loadedConf.Input = plugin
break
}
for name, primitives := range conf.Processors {
creator, ok := processors.Processors[name]
if !ok {
return loadedConf, errors.New("unknown processor " + name)
}
plugin := creator()
if len(primitives) > 0 {
primitive := primitives[0]
var p telegraf.PluginDescriber = plugin
if processor, ok := plugin.(unwrappable); ok {
p = processor.Unwrap()
}
if err := md.PrimitiveDecode(primitive, p); err != nil {
return loadedConf, err
}
}
loadedConf.Processor = plugin
break
}
for name, primitives := range conf.Outputs {
creator, ok := outputs.Outputs[name]
if !ok {
return loadedConf, errors.New("unknown output " + name)
}
plugin := creator()
if len(primitives) > 0 {
primitive := primitives[0]
if err := md.PrimitiveDecode(primitive, plugin); err != nil {
return loadedConf, err
}
}
loadedConf.Output = plugin
break
}
return loadedConf, nil
}
// DefaultImportedPlugins defaults to whatever plugins happen to be loaded and
// have registered themselves with the registry. This makes loading plugins
// without having to define a config dead easy.
func DefaultImportedPlugins() config {
conf := config{
Inputs: map[string][]toml.Primitive{},
Processors: map[string][]toml.Primitive{},
Outputs: map[string][]toml.Primitive{},
}
for name := range inputs.Inputs {
log.Println("No config found. Loading default config for plugin", name)
conf.Inputs[name] = []toml.Primitive{}
return conf
}
for name := range processors.Processors {
log.Println("No config found. Loading default config for plugin", name)
conf.Processors[name] = []toml.Primitive{}
return conf
}
for name := range outputs.Outputs {
log.Println("No config found. Loading default config for plugin", name)
conf.Outputs[name] = []toml.Primitive{}
return conf
}
return conf
}
type unwrappable interface {
Unwrap() telegraf.Processor
}