-
Notifications
You must be signed in to change notification settings - Fork 5.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[WIP] Etcd integration for configuration
- Loading branch information
1 parent
c13927f
commit 121fd61
Showing
3 changed files
with
224 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,174 @@ | ||
| package etcd | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "golang.org/x/net/context" | ||
| "io/ioutil" | ||
| "log" | ||
| "os" | ||
| "strings" | ||
|
|
||
| "github.com/BurntSushi/toml" | ||
| "github.com/coreos/etcd/client" | ||
|
|
||
| "github.com/influxdb/telegraf/internal/config" | ||
| ) | ||
|
|
||
| type EtcdClient struct { | ||
| Kapi client.KeysAPI | ||
| Folder string | ||
| } | ||
|
|
||
| func (e *EtcdClient) LaunchWatcher(shutdown chan struct{}, reload chan bool) { | ||
| // TODO: All telegraf client will reload for each changes... | ||
| // Maybe we want to reload on those we need to ??? | ||
| // So we need to create a watcher by labels ?? | ||
| watcherOpts := client.WatcherOptions{AfterIndex: 0, Recursive: true} | ||
| w := e.Kapi.Watcher(e.Folder, &watcherOpts) | ||
| r, err := w.Next(context.Background()) | ||
| if err != nil { | ||
| // TODO What we have to do here ???? | ||
| log.Fatal("Error occurred", err) | ||
| } | ||
| if r.Action == "set" || r.Action == "update" { | ||
| // do something with Response r here | ||
| log.Printf("Changes detected in etcd (%s action detected)\n", r.Action) | ||
| log.Print("Reloading Telegraf config\n") | ||
| <-reload | ||
| reload <- true | ||
| close(shutdown) | ||
| } | ||
| } | ||
|
|
||
| func NewEtcdClient(urls string, folder string) *EtcdClient { | ||
|
|
||
| cfg := client.Config{ | ||
| Endpoints: []string{"http://127.0.0.1:2379"}, | ||
| Transport: client.DefaultTransport, | ||
| } | ||
|
|
||
| e := &EtcdClient{} | ||
| c, err := client.New(cfg) | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| kapi := client.NewKeysAPI(c) | ||
|
|
||
| e.Kapi = kapi | ||
| e.Folder = folder | ||
|
|
||
| return e | ||
| } | ||
|
|
||
| func (e *EtcdClient) Connect() error { | ||
| //c, err := eclient.New(cfg) | ||
| return nil | ||
| } | ||
|
|
||
| func (e *EtcdClient) WriteConfig(label string, path string) error { | ||
| // Read config file, get conf in tomlformat, convert to json | ||
| // Then write it to etcd | ||
| // TODO: Maybe we just want to store toml in etcd ? Is json really needed ???? | ||
| // Read file | ||
| raw_data, err := ioutil.ReadFile(path) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| // Get toml | ||
| var data interface{} | ||
| _, err = toml.Decode(string(raw_data), &data) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| // Get json | ||
| json_data, _ := json.Marshal(&data) | ||
| // Write it | ||
| key := e.Folder + "/labels/" + label | ||
| resp, _ := e.Kapi.Get(context.Background(), key, nil) | ||
| if resp == nil { | ||
| _, err = e.Kapi.Set(context.Background(), key, string(json_data), nil) | ||
| } else { | ||
| _, err = e.Kapi.Update(context.Background(), key, string(json_data)) | ||
| } | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| return err | ||
| } else { | ||
| log.Printf("Config written with label %s\n", label) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| //func (e *EtcdClient) ReadConfig(labels []string) (*config.Config, error) { | ||
| func (e *EtcdClient) ReadConfig(labels string) (*config.Config, error) { | ||
| c := config.NewConfig() | ||
|
|
||
| // Get default config in etcd | ||
| // key = /telegraf/default | ||
| key := e.Folder + "/default" | ||
| resp, err := e.Kapi.Get(context.Background(), key, nil) | ||
| if err != nil { | ||
| log.Printf("WARNING: [etcd] %s", err) | ||
| } else { | ||
| // Put it in toml | ||
| data, err := json2toml(resp) | ||
| if err != nil { | ||
| log.Printf("WARNING: [etcd] %s", err) | ||
| } | ||
| c.LoadConfigFromText(data) | ||
| } | ||
|
|
||
| // Get specific host config | ||
| // key = /telegraf/hosts/HOSTNAME | ||
| hostname, err := os.Hostname() | ||
| if err != nil { | ||
| log.Printf("WARNING: [etcd] %s", err) | ||
| } else if hostname != "" { | ||
| key = e.Folder + "/hosts/" + hostname | ||
| resp, err := e.Kapi.Get(context.Background(), key, nil) | ||
| if err != nil { | ||
| log.Printf("WARNING: [etcd] %s", err) | ||
| } else { | ||
| // Put it in toml | ||
| data, err := json2toml(resp) | ||
| if err != nil { | ||
| log.Print(err) | ||
| } | ||
| c.LoadConfigFromText(data) | ||
| } | ||
| } | ||
|
|
||
| // Iterate on all labels | ||
| // TODO check label order of importance ? | ||
| for _, label := range strings.Split(labels, ",") { | ||
| // Read from etcd | ||
| // key = /telegraf/labels/LABEL | ||
| key := e.Folder + "/labels/" + label | ||
| resp, err := e.Kapi.Get(context.Background(), key, nil) | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| continue | ||
| } | ||
| // Put it in toml | ||
| data, err := json2toml(resp) | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| continue | ||
| } | ||
| // Load config | ||
| c.LoadConfigFromText(data) | ||
| } | ||
|
|
||
| return c, nil | ||
| } | ||
|
|
||
| func json2toml(resp *client.Response) ([]byte, error) { | ||
| var json_data interface{} | ||
| var data []byte | ||
| json.Unmarshal([]byte(resp.Node.Value), &json_data) | ||
| buf := new(bytes.Buffer) | ||
| err := toml.NewEncoder(buf).Encode(json_data) | ||
| data = buf.Bytes() | ||
| return data, err | ||
| } |