Moving away from passing around *ast.Tables.

Config in the config directory will need to be merged into the main
config, which is difficult to do using the *ast.Tables. Get the config
into structs as soon as possible and then merge the structs.
This commit is contained in:
gotyaoi 2015-10-16 21:47:13 -07:00 committed by Ellison Marks
parent 553208a960
commit 03d79996de
3 changed files with 293 additions and 173 deletions

View File

@ -52,7 +52,7 @@ type Agent struct {
Debug bool Debug bool
Hostname string Hostname string
Config *Config Tags map[string]string
outputs []*runningOutput outputs []*runningOutput
plugins []*runningPlugin plugins []*runningPlugin
@ -61,7 +61,7 @@ type Agent struct {
// NewAgent returns an Agent struct based off the given Config // NewAgent returns an Agent struct based off the given Config
func NewAgent(config *Config) (*Agent, error) { func NewAgent(config *Config) (*Agent, error) {
agent := &Agent{ agent := &Agent{
Config: config, Tags: make(map[string]string),
Interval: Duration{10 * time.Second}, Interval: Duration{10 * time.Second},
FlushInterval: Duration{10 * time.Second}, FlushInterval: Duration{10 * time.Second},
FlushRetries: 2, FlushRetries: 2,
@ -84,11 +84,7 @@ func NewAgent(config *Config) (*Agent, error) {
agent.Hostname = hostname agent.Hostname = hostname
} }
if config.Tags == nil { agent.Tags["host"] = agent.Hostname
config.Tags = map[string]string{}
}
config.Tags["host"] = agent.Hostname
return agent, nil return agent, nil
} }
@ -125,10 +121,10 @@ func (a *Agent) Close() error {
} }
// LoadOutputs loads the agent's outputs // LoadOutputs loads the agent's outputs
func (a *Agent) LoadOutputs(filters []string) ([]string, error) { func (a *Agent) LoadOutputs(filters []string, config *Config) ([]string, error) {
var names []string var names []string
for _, name := range a.Config.OutputsDeclared() { for _, name := range config.OutputsDeclared() {
creator, ok := outputs.Outputs[name] creator, ok := outputs.Outputs[name]
if !ok { if !ok {
return nil, fmt.Errorf("Undefined but requested output: %s", name) return nil, fmt.Errorf("Undefined but requested output: %s", name)
@ -140,7 +136,7 @@ func (a *Agent) LoadOutputs(filters []string) ([]string, error) {
} }
output := creator() output := creator()
err := a.Config.ApplyOutput(name, output) err := config.ApplyOutput(name, output)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -156,10 +152,10 @@ func (a *Agent) LoadOutputs(filters []string) ([]string, error) {
} }
// LoadPlugins loads the agent's plugins // LoadPlugins loads the agent's plugins
func (a *Agent) LoadPlugins(filters []string) ([]string, error) { func (a *Agent) LoadPlugins(filters []string, config *Config) ([]string, error) {
var names []string var names []string
for _, name := range a.Config.PluginsDeclared() { for _, name := range config.PluginsDeclared() {
creator, ok := plugins.Plugins[name] creator, ok := plugins.Plugins[name]
if !ok { if !ok {
return nil, fmt.Errorf("Undefined but requested plugin: %s", name) return nil, fmt.Errorf("Undefined but requested plugin: %s", name)
@ -168,7 +164,7 @@ func (a *Agent) LoadPlugins(filters []string) ([]string, error) {
if sliceContains(name, filters) || len(filters) == 0 { if sliceContains(name, filters) || len(filters) == 0 {
plugin := creator() plugin := creator()
config, err := a.Config.ApplyPlugin(name, plugin) config, err := config.ApplyPlugin(name, plugin)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -203,7 +199,7 @@ func (a *Agent) gatherParallel(pointChan chan *client.Point) error {
acc := NewAccumulator(plugin.config, pointChan) acc := NewAccumulator(plugin.config, pointChan)
acc.SetDebug(a.Debug) acc.SetDebug(a.Debug)
acc.SetPrefix(plugin.name + "_") acc.SetPrefix(plugin.name + "_")
acc.SetDefaultTags(a.Config.Tags) acc.SetDefaultTags(a.Tags)
if err := plugin.plugin.Gather(acc); err != nil { if err := plugin.plugin.Gather(acc); err != nil {
log.Printf("Error in plugin [%s]: %s", plugin.name, err) log.Printf("Error in plugin [%s]: %s", plugin.name, err)
@ -236,7 +232,7 @@ func (a *Agent) gatherSeparate(
acc := NewAccumulator(plugin.config, pointChan) acc := NewAccumulator(plugin.config, pointChan)
acc.SetDebug(a.Debug) acc.SetDebug(a.Debug)
acc.SetPrefix(plugin.name + "_") acc.SetPrefix(plugin.name + "_")
acc.SetDefaultTags(a.Config.Tags) acc.SetDefaultTags(a.Tags)
if err := plugin.plugin.Gather(acc); err != nil { if err := plugin.plugin.Gather(acc); err != nil {
log.Printf("Error in plugin [%s]: %s", plugin.name, err) log.Printf("Error in plugin [%s]: %s", plugin.name, err)

View File

@ -90,7 +90,7 @@ func main() {
ag.Debug = true ag.Debug = true
} }
outputs, err := ag.LoadOutputs(outputFilters) outputs, err := ag.LoadOutputs(outputFilters, config)
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
@ -99,7 +99,7 @@ func main() {
os.Exit(1) os.Exit(1)
} }
plugins, err := ag.LoadPlugins(pluginFilters) plugins, err := ag.LoadPlugins(pluginFilters, config)
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }

432
config.go
View File

@ -4,6 +4,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"reflect"
"sort" "sort"
"strings" "strings"
"time" "time"
@ -35,20 +36,28 @@ func (d *Duration) UnmarshalTOML(b []byte) error {
// will be logging to, as well as all the plugins that the user has // will be logging to, as well as all the plugins that the user has
// specified // specified
type Config struct { type Config struct {
// This lives outside the agent because mergeStruct doesn't need to handle maps normally.
// We just copy the elements manually in applyAgent.
Tags map[string]string Tags map[string]string
agent *ast.Table agent *Agent
plugins map[string]*ast.Table plugins map[string]plugins.Plugin
outputs map[string]*ast.Table pluginConfigurations map[string]*ConfiguredPlugin
outputs map[string]outputs.Output
agentFieldsSet []string
pluginFieldsSet map[string][]string
pluginConfigurationFieldsSet map[string][]string
outputFieldsSet map[string][]string
} }
// Plugins returns the configured plugins as a map of name -> plugin toml // Plugins returns the configured plugins as a map of name -> plugins.Plugin
func (c *Config) Plugins() map[string]*ast.Table { func (c *Config) Plugins() map[string]plugins.Plugin {
return c.plugins return c.plugins
} }
// Outputs returns the configured outputs as a map of name -> output toml // Outputs returns the configured outputs as a map of name -> outputs.Output
func (c *Config) Outputs() map[string]*ast.Table { func (c *Config) Outputs() map[string]outputs.Output {
return c.outputs return c.outputs
} }
@ -123,187 +132,65 @@ func (cp *ConfiguredPlugin) ShouldPass(measurement string, tags map[string]strin
return true return true
} }
// ApplyOutput loads the toml config into the given interface // ApplyOutput loads the Output struct built from the config into the given Output struct.
// Overrides only values in the given struct that were set in the config.
func (c *Config) ApplyOutput(name string, v interface{}) error { func (c *Config) ApplyOutput(name string, v interface{}) error {
if c.outputs[name] != nil { if c.outputs[name] != nil {
return toml.UnmarshalTable(c.outputs[name], v) return mergeStruct(v, c.outputs[name], c.outputFieldsSet[name])
} }
return nil return nil
} }
// ApplyAgent loads the toml config into the given Agent object, overriding // ApplyAgent loads the Agent struct built from the config into the given Agent struct.
// defaults (such as collection duration) with the values from the toml config. // Overrides only values in the given struct that were set in the config.
func (c *Config) ApplyAgent(a *Agent) error { func (c *Config) ApplyAgent(a *Agent) error {
if c.agent != nil { if c.agent != nil {
return toml.UnmarshalTable(c.agent, a) return mergeStruct(a, c.agent, c.agentFieldsSet)
}
for key, value := range c.Tags {
a.Tags[key] = value
} }
return nil return nil
} }
// ApplyPlugin takes defined plugin names and applies them to the given // ApplyPlugin loads the Plugin struct built from the config into the given Plugin struct.
// interface, returning a ConfiguredPlugin object in the end that can // Overrides only values in the given struct that were set in the config.
// be inserted into a runningPlugin by the agent. // Additionally return a ConfiguredPlugin, which is always generated from the config.
func (c *Config) ApplyPlugin(name string, v interface{}) (*ConfiguredPlugin, error) { func (c *Config) ApplyPlugin(name string, v interface{}) (*ConfiguredPlugin, error) {
cp := &ConfiguredPlugin{Name: name} if c.plugins[name] != nil {
err := mergeStruct(v, c.plugins[name], c.pluginFieldsSet[name])
if tbl, ok := c.plugins[name]; ok {
if node, ok := tbl.Fields["pass"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
cp.Pass = append(cp.Pass, str.Value)
}
}
}
}
}
if node, ok := tbl.Fields["drop"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
cp.Drop = append(cp.Drop, str.Value)
}
}
}
}
}
if node, ok := tbl.Fields["interval"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if str, ok := kv.Value.(*ast.String); ok {
dur, err := time.ParseDuration(str.Value)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return c.pluginConfigurations[name], nil
cp.Interval = dur
}
}
} }
if node, ok := tbl.Fields["tagpass"]; ok { return nil, nil
if subtbl, ok := node.(*ast.Table); ok {
for name, val := range subtbl.Fields {
if kv, ok := val.(*ast.KeyValue); ok {
tagfilter := &TagFilter{Name: name}
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
tagfilter.Filter = append(tagfilter.Filter, str.Value)
}
}
}
cp.TagPass = append(cp.TagPass, *tagfilter)
}
}
}
}
if node, ok := tbl.Fields["tagdrop"]; ok {
if subtbl, ok := node.(*ast.Table); ok {
for name, val := range subtbl.Fields {
if kv, ok := val.(*ast.KeyValue); ok {
tagfilter := &TagFilter{Name: name}
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
tagfilter.Filter = append(tagfilter.Filter, str.Value)
}
}
}
cp.TagDrop = append(cp.TagDrop, *tagfilter)
}
}
}
}
delete(tbl.Fields, "drop")
delete(tbl.Fields, "pass")
delete(tbl.Fields, "interval")
delete(tbl.Fields, "tagdrop")
delete(tbl.Fields, "tagpass")
return cp, toml.UnmarshalTable(tbl, v)
}
return cp, nil
} }
// Couldn't figure out how to get this to work with the declared function.
// PluginsDeclared returns the name of all plugins declared in the config. // PluginsDeclared returns the name of all plugins declared in the config.
func (c *Config) PluginsDeclared() []string { func (c *Config) PluginsDeclared() []string {
return declared(c.plugins) var names []string
for name := range c.plugins {
names = append(names, name)
}
sort.Strings(names)
return names
} }
// OutputsDeclared returns the name of all outputs declared in the config. // OutputsDeclared returns the name of all outputs declared in the config.
func (c *Config) OutputsDeclared() []string { func (c *Config) OutputsDeclared() []string {
return declared(c.outputs)
}
func declared(endpoints map[string]*ast.Table) []string {
var names []string var names []string
for name := range c.outputs {
for name := range endpoints {
names = append(names, name) names = append(names, name)
} }
sort.Strings(names) sort.Strings(names)
return names return names
} }
var errInvalidConfig = errors.New("invalid configuration")
// LoadConfig loads the given config file and returns a *Config pointer
func LoadConfig(path string) (*Config, error) {
data, err := ioutil.ReadFile(path)
if err != nil {
return nil, err
}
tbl, err := toml.Parse(data)
if err != nil {
return nil, err
}
c := &Config{
Tags: make(map[string]string),
plugins: make(map[string]*ast.Table),
outputs: make(map[string]*ast.Table),
}
for name, val := range tbl.Fields {
subtbl, ok := val.(*ast.Table)
if !ok {
return nil, errInvalidConfig
}
switch name {
case "agent":
c.agent = subtbl
case "tags":
if err := toml.UnmarshalTable(subtbl, c.Tags); err != nil {
return nil, errInvalidConfig
}
case "outputs":
for outputName, outputVal := range subtbl.Fields {
outputSubtbl, ok := outputVal.(*ast.Table)
if !ok {
return nil, errInvalidConfig
}
c.outputs[outputName] = outputSubtbl
}
default:
c.plugins[name] = subtbl
}
}
return c, nil
}
// ListTags returns a string of tags specified in the config, // ListTags returns a string of tags specified in the config,
// line-protocol style // line-protocol style
func (c *Config) ListTags() string { func (c *Config) ListTags() string {
@ -474,3 +361,240 @@ func PrintPluginConfig(name string) error {
} }
return nil return nil
} }
// Used for fuzzy matching struct field names in FieldByNameFunc calls below
func fieldMatch(field string) func(string) bool {
return func(name string) bool {
r := strings.NewReplacer("_", "")
return strings.ToLower(name) == strings.ToLower(r.Replace(field))
}
}
// A very limited merge. Merges the fields named in the fields parameter, replacing most values, but appending to arrays.
func mergeStruct(base, overlay interface{}, fields []string) error {
baseValue := reflect.ValueOf(base).Elem()
overlayValue := reflect.ValueOf(overlay).Elem()
if baseValue.Kind() != reflect.Struct {
return fmt.Errorf("Tried to merge something that wasn't a struct: type %v was %v", baseValue.Type(), baseValue.Kind())
}
if baseValue.Type() != overlayValue.Type() {
return fmt.Errorf("Tried to merge two different types: %v and %v", baseValue.Type(), overlayValue.Type())
}
for _, field := range fields {
overlayFieldValue := overlayValue.FieldByNameFunc(fieldMatch(field))
if !overlayFieldValue.IsValid() {
return fmt.Errorf("could not find field in %v matching %v", overlayValue.Type(), field)
}
if overlayFieldValue.Kind() == reflect.Slice {
baseFieldValue := baseValue.FieldByNameFunc(fieldMatch(field))
baseFieldValue.Set(reflect.AppendSlice(baseFieldValue, overlayFieldValue))
} else {
baseValue.FieldByNameFunc(fieldMatch(field)).Set(overlayFieldValue)
}
}
return nil
}
// hazmat area. Keeping the ast parsing here.
// LoadConfig loads the given config file and returns a *Config pointer
func LoadConfig(path string) (*Config, error) {
data, err := ioutil.ReadFile(path)
if err != nil {
return nil, err
}
tbl, err := toml.Parse(data)
if err != nil {
return nil, err
}
c := &Config{
Tags: make(map[string]string),
plugins: make(map[string]plugins.Plugin),
pluginConfigurations: make(map[string]*ConfiguredPlugin),
outputs: make(map[string]outputs.Output),
pluginFieldsSet: make(map[string][]string),
pluginConfigurationFieldsSet: make(map[string][]string),
outputFieldsSet: make(map[string][]string),
}
for name, val := range tbl.Fields {
subtbl, ok := val.(*ast.Table)
if !ok {
return nil, errors.New("invalid configuration")
}
switch name {
case "agent":
err := c.parseAgent(subtbl)
if err != nil {
return nil, err
}
case "tags":
if err = toml.UnmarshalTable(subtbl, c.Tags); err != nil {
return nil, err
}
case "outputs":
for outputName, outputVal := range subtbl.Fields {
outputSubtbl, ok := outputVal.(*ast.Table)
if !ok {
return nil, err
}
err = c.parseOutput(outputName, outputSubtbl)
if err != nil {
return nil, err
}
}
default:
err = c.parsePlugin(name, subtbl)
if err != nil {
return nil, err
}
}
}
return c, nil
}
// Needs to have the field names, for merging later.
func extractFieldNames(ast *ast.Table) []string {
// A reasonable capacity?
var names []string
for name := range ast.Fields {
names = append(names, name)
}
return names
}
// Parse the agent config out of the given *ast.Table.
func (c *Config) parseAgent(agentAst *ast.Table) error {
c.agentFieldsSet = extractFieldNames(agentAst)
agent := &Agent{}
err := toml.UnmarshalTable(agentAst, agent)
if err != nil {
return err
}
c.agent = agent
return nil
}
// Parse an output config out of the given *ast.Table.
func (c *Config) parseOutput(name string, outputAst *ast.Table) error {
c.outputFieldsSet[name] = extractFieldNames(outputAst)
creator, ok := outputs.Outputs[name]
if !ok {
return fmt.Errorf("Undefined but requested output: %s", name)
}
output := creator()
err := toml.UnmarshalTable(outputAst, output)
if err != nil {
return err
}
c.outputs[name] = output
return nil
}
// Parse a plugin config, plus plugin meta-config, out of the given *ast.Table.
func (c *Config) parsePlugin(name string, pluginAst *ast.Table) error {
creator, ok := plugins.Plugins[name]
if !ok {
return fmt.Errorf("Undefined but requested plugin: %s", name)
}
plugin := creator()
cp := &ConfiguredPlugin{Name: name}
cpFields := make([]string, 0, 5)
if node, ok := pluginAst.Fields["pass"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
cp.Pass = append(cp.Pass, str.Value)
}
}
cpFields = append(cpFields, "pass")
}
}
}
if node, ok := pluginAst.Fields["drop"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
cp.Drop = append(cp.Drop, str.Value)
}
}
cpFields = append(cpFields, "drop")
}
}
}
if node, ok := pluginAst.Fields["interval"]; ok {
if kv, ok := node.(*ast.KeyValue); ok {
if str, ok := kv.Value.(*ast.String); ok {
dur, err := time.ParseDuration(str.Value)
if err != nil {
return err
}
cp.Interval = dur
cpFields = append(cpFields, "interval")
}
}
}
if node, ok := pluginAst.Fields["tagpass"]; ok {
if subtbl, ok := node.(*ast.Table); ok {
for name, val := range subtbl.Fields {
if kv, ok := val.(*ast.KeyValue); ok {
tagfilter := &TagFilter{Name: name}
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
tagfilter.Filter = append(tagfilter.Filter, str.Value)
}
}
}
cp.TagPass = append(cp.TagPass, *tagfilter)
cpFields = append(cpFields, "tagpass")
}
}
}
}
if node, ok := pluginAst.Fields["tagdrop"]; ok {
if subtbl, ok := node.(*ast.Table); ok {
for name, val := range subtbl.Fields {
if kv, ok := val.(*ast.KeyValue); ok {
tagfilter := &TagFilter{Name: name}
if ary, ok := kv.Value.(*ast.Array); ok {
for _, elem := range ary.Value {
if str, ok := elem.(*ast.String); ok {
tagfilter.Filter = append(tagfilter.Filter, str.Value)
}
}
}
cp.TagDrop = append(cp.TagDrop, *tagfilter)
cpFields = append(cpFields, "tagdrop")
}
}
}
}
delete(pluginAst.Fields, "drop")
delete(pluginAst.Fields, "pass")
delete(pluginAst.Fields, "interval")
delete(pluginAst.Fields, "tagdrop")
delete(pluginAst.Fields, "tagpass")
c.pluginFieldsSet[name] = extractFieldNames(pluginAst)
c.pluginConfigurationFieldsSet[name] = cpFields
err := toml.UnmarshalTable(pluginAst, plugin)
if err != nil {
return err
}
c.plugins[name] = plugin
c.pluginConfigurations[name] = cp
return nil
}