From 9b6f5aecb407c81af78ec5e67ef72aca3e8ed6ac Mon Sep 17 00:00:00 2001 From: andig Date: Wed, 18 Mar 2020 16:56:15 +0100 Subject: [PATCH] Add support for writing to InfluxDB (#8) --- cmd/config.go | 9 +++ cmd/root.go | 48 +++++++++++-- go.mod | 11 +-- go.sum | 13 ++++ server/influxdb.go | 169 +++++++++++++++++++++++++++++++++++++++++++++ server/socket.go | 4 +- 6 files changed, 242 insertions(+), 12 deletions(-) create mode 100644 server/influxdb.go diff --git a/cmd/config.go b/cmd/config.go index f8166919b..062c9d5ae 100644 --- a/cmd/config.go +++ b/cmd/config.go @@ -12,6 +12,7 @@ type config struct { Log string Interval time.Duration Mqtt mqttConfig + Influx influxConfig Menu []server.MenuConfig Pushover messagingConfig Meters []meterConfig @@ -31,6 +32,14 @@ type mqttConfig struct { Password string } +type influxConfig struct { + URL string + Database string + User string + Password string + Interval time.Duration +} + type providerConfig struct { Type string Topic string diff --git a/cmd/root.go b/cmd/root.go index 8f88ff5b4..dae5e3d03 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -136,6 +136,25 @@ func checkVersion() { } } +var teeIsChained bool // controles piping of first channel in teed chain + +func tee(in chan core.Param) (chan core.Param, <-chan core.Param) { + gen := make(chan core.Param) + tee := make(chan core.Param) + + go func(teeIsChained bool) { + for i := range gen { + if teeIsChained { + in <- i + } + tee <- i + } + }(teeIsChained) + + teeIsChained = true + return gen, tee +} + func run(cmd *cobra.Command, args []string) { level, _ := cmd.PersistentFlags().GetString("log") configureLogging(level) @@ -170,19 +189,38 @@ func run(cmd *cobra.Command, args []string) { loadPoints := loadConfig(conf, notificationChan) go notificationHub.Run(notificationChan) + // start broadcasting values + valueChan := make(chan core.Param) + triggerChan := make(chan struct{}) + + // setup influx + if viper.Get("influx") != nil { + influx := server.NewInfluxClient( + conf.Influx.URL, + conf.Influx.Database, + conf.Influx.Interval, + conf.Influx.User, + conf.Influx.Password, + ) + + var teeChan <-chan core.Param + valueChan, teeChan = tee(valueChan) + + go influx.Run(teeChan) + } + // create webserver socketHub := server.NewSocketHub() httpd := server.NewHttpd(uri, conf.Menu, loadPoints[0], socketHub) - // start broadcasting values - uiChan := make(chan core.Param) - triggerChan := make(chan struct{}) - go socketHub.Run(uiChan, triggerChan) + var teeChan <-chan core.Param + valueChan, teeChan = tee(valueChan) + go socketHub.Run(teeChan, triggerChan) // start all loadpoints for _, lp := range loadPoints { lp.Dump() - lp.Prepare(uiChan, notificationChan) + lp.Prepare(valueChan, notificationChan) go lp.Run(conf.Interval) } diff --git a/go.mod b/go.mod index 925be1cf6..63ca2e5f5 100644 --- a/go.mod +++ b/go.mod @@ -6,8 +6,8 @@ require ( github.com/asaskevich/EventBus v0.0.0-20180315140547-d46933a94f05 github.com/benbjohnson/clock v1.0.0 github.com/eclipse/paho.mqtt.golang v1.2.0 - github.com/fsnotify/fsnotify v1.4.8 // indirect - github.com/golang/mock v1.4.1 + github.com/fsnotify/fsnotify v1.4.9 // indirect + github.com/golang/mock v1.4.3 github.com/google/go-github v17.0.0+incompatible // indirect github.com/google/go-querystring v1.0.0 // indirect github.com/gorilla/handlers v1.4.2 @@ -17,6 +17,7 @@ require ( github.com/grid-x/modbus v0.0.0-20200108122021-57d05a9f1e1a github.com/grid-x/serial v0.0.0-20191104121038-e24bc9bf6f08 // indirect github.com/hashicorp/go-version v1.2.0 // indirect + github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 github.com/mjibson/esc v0.2.0 github.com/pelletier/go-toml v1.6.0 // indirect @@ -29,9 +30,9 @@ require ( github.com/spf13/viper v1.6.2 github.com/tcnksm/go-latest v0.0.0-20170313132115-e3007ae9052e golang.org/x/net v0.0.0-20200301022130-244492dfa37a // indirect - golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 // indirect - golang.org/x/tools v0.0.0-20200311090712-aafaee8bce8c - gopkg.in/ini.v1 v1.54.0 // indirect + golang.org/x/sys v0.0.0-20200317113312-5766fd39f98d // indirect + golang.org/x/tools v0.0.0-20200318150045-ba25ddc85566 + gopkg.in/ini.v1 v1.55.0 // indirect gopkg.in/yaml.v2 v2.2.8 // indirect ) diff --git a/go.sum b/go.sum index 64e3d2956..a86fbbb33 100644 --- a/go.sum +++ b/go.sum @@ -50,6 +50,8 @@ github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.8 h1:vmjGRBE6skJRlqYIcSsh9B823pS/MqsvxevqbMnBPbI= github.com/fsnotify/fsnotify v1.4.8/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= +github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4= +github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= @@ -65,6 +67,8 @@ github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y= github.com/golang/mock v1.4.1 h1:ocYkMQY5RrXTYgXl7ICpV0IXwlEQGwKIsery4gyXa1U= github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw= +github.com/golang/mock v1.4.3 h1:GV+pQPG/EUUbkh47niozDcADz6go/dUwhVzdUQHIVRw= +github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= @@ -126,6 +130,8 @@ github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2p github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc= github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= +github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d h1:/WZQPMZNsjZ7IlCpsLGdQBINg5bxKQ1K1sh6awxLtkA= +github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo= github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= @@ -229,6 +235,7 @@ github.com/tcnksm/go-latest v0.0.0-20170313132115-e3007ae9052e h1:IWllFTiDjjLIf2 github.com/tcnksm/go-latest v0.0.0-20170313132115-e3007ae9052e/go.mod h1:d7u6HkTYKSv5m6MCKkOQlHwaShTMl3HjqSGW3XtVhXM= github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U= github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= +github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= @@ -305,6 +312,8 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 h1:uYVVQ9WP/Ds2ROhcaGPeIdVq0RIXVLwsHlnvJ+cT1So= golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200317113312-5766fd39f98d h1:62ap6LNOjDU6uGmKXHJbSfciMoV+FeI1sRXx/pLDL44= +golang.org/x/sys v0.0.0-20200317113312-5766fd39f98d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -334,6 +343,8 @@ golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200311090712-aafaee8bce8c h1:9WR4YuzLDuQMqEmLQrG0DiMmE2/HvX1dlrujzjmNVFg= golang.org/x/tools v0.0.0-20200311090712-aafaee8bce8c/go.mod h1:o4KQGtdN14AW+yjsvvwRTJJuXz8XRtIHtEnmAXLyFUw= +golang.org/x/tools v0.0.0-20200318150045-ba25ddc85566 h1:OXjomkWHhzUx4+HldlJ2TsMxJdWgEo5CTtspD1wdhdk= +golang.org/x/tools v0.0.0-20200318150045-ba25ddc85566/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= @@ -368,6 +379,8 @@ gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno= gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.54.0 h1:oM5ElzbIi7gwLnNbPX2M25ED1vSAK3B6dex50eS/6Fs= gopkg.in/ini.v1 v1.54.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= +gopkg.in/ini.v1 v1.55.0 h1:E8yzL5unfpW3M6fz/eB7Cb5MQAYSZ7GKo4Qth+N2sgQ= +gopkg.in/ini.v1 v1.55.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo= gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/server/influxdb.go b/server/influxdb.go new file mode 100644 index 000000000..6903b4eb0 --- /dev/null +++ b/server/influxdb.go @@ -0,0 +1,169 @@ +package server + +import ( + "sync" + "time" + + "github.com/andig/evcc/api" + "github.com/andig/evcc/core" + influxdb "github.com/influxdata/influxdb1-client/v2" +) + +const ( + influxWriteTimeout = 10 * time.Second + influxWriteInterval = 30 * time.Second + precision = "s" +) + +// Influx is a influx publisher +type Influx struct { + sync.Mutex + log *api.Logger + client influxdb.Client + points []*influxdb.Point + pointsConf influxdb.BatchPointsConfig + interval time.Duration + measurement string +} + +// NewInfluxClient creates new publisher for influx +func NewInfluxClient( + url string, + database string, + interval time.Duration, + user string, + password string, +) *Influx { + log := api.NewLogger("iflx") + + if database == "" { + log.FATAL.Fatal("missing database") + } + if interval == 0 { + interval = influxWriteInterval + } + + client, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{ + Addr: url, + Username: user, + Password: password, + Timeout: influxWriteTimeout, + }) + if err != nil { + log.FATAL.Fatalf("error creating client: %v", err) + } + + // check connection + go func(client influxdb.Client) { + if _, _, err := client.Ping(influxWriteTimeout); err != nil { + log.FATAL.Fatalf("%v", err) + } + }(client) + + return &Influx{ + log: log, + client: client, + interval: interval, + pointsConf: influxdb.BatchPointsConfig{ + Database: database, + Precision: precision, + }, + } +} + +// writeBatchPoints asynchronously writes the collected points +func (m *Influx) writeBatchPoints() { + m.Lock() + + // get current batch + if len(m.points) == 0 { + m.Unlock() + return + } + + // create new batch + batch, err := influxdb.NewBatchPoints(m.pointsConf) + if err != nil { + m.log.ERROR.Print(err) + m.Unlock() + return + } + + // replace current batch + points := m.points + m.points = nil + m.Unlock() + + // write batch + batch.AddPoints(points) + m.log.TRACE.Printf("writing %d point(s)", len(points)) + + if err := m.client.Write(batch); err != nil { + m.log.ERROR.Print(err) + + // put points back at beginning of next batch + m.Lock() + m.points = append(points, m.points...) + m.Unlock() + } +} + +// asyncWriter periodically calls writeBatchPoints +func (m *Influx) asyncWriter(exit <-chan struct{}) <-chan struct{} { + done := make(chan struct{}) // signal writer stopped + + // async batch writer + go func() { + ticker := time.NewTicker(m.interval) + for { + select { + case <-ticker.C: + m.writeBatchPoints() + case <-exit: + ticker.Stop() + m.writeBatchPoints() + close(done) + return + } + } + }() + + return done +} + +// Run Influx publisher +func (m *Influx) Run(in <-chan core.Param) { + // run async writer + exit := make(chan struct{}) // exit signals to stop writer + done := m.asyncWriter(exit) // done signals writer stopped + + // add points to batch for async writing + for param := range in { + if _, ok := param.Val.(float64); !ok { + continue + } + + p, err := influxdb.NewPoint( + param.Key, + map[string]string{}, + map[string]interface{}{ + "value": param.Val, + }, + time.Now(), + ) + if err != nil { + m.log.ERROR.Printf("failed creating point: %v", err) + continue + } + + m.Lock() + m.points = append(m.points, p) + m.Unlock() + } + + // close write loop + exit <- struct{}{} + <-done + + m.client.Close() +} diff --git a/server/socket.go b/server/socket.go index b9bfa1207..8c416b8fa 100644 --- a/server/socket.go +++ b/server/socket.go @@ -11,7 +11,7 @@ import ( const ( // Time allowed to write a message to the peer - writeTimeout = 10 * time.Second + socketWriteTimeout = 10 * time.Second ) var upgrader = websocket.Upgrader{ @@ -39,7 +39,7 @@ func (c *SocketClient) writePump() { for { msg := <-c.send - if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + if err := c.conn.SetWriteDeadline(time.Now().Add(socketWriteTimeout)); err != nil { return } if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {