Connecting Flume Water to InfluxDB Cloud Using Telegraf

9 minutes

My house is full of smart devices: smart TVs, speakers, light switches, appliances, even my toothbrush has an app. But there is one big area that is still frustratingly “dumb”: my utility usage. In most cases, public utilities in the United States aren’t known for their cutting-edge technology, so it was time to take matters into my own hands.

Tracking my water usage

I decided to focus on my water usage. I live in California, which is a state that is constantly trying to encourage people to save water. When I look at my water bill every month, I see my usage plotted against time, but I can’t get access to that data in real time from the utility.

My real water usage for the past 6 months
My real water usage for the past 6 months

So for Christmas last year, I asked Santa for a Flume Water sensor, and he delivered! Fast forward three months, and I finally found the time to set it up and start playing with the data.

I chose the Flume Water meter for two main reasons. First, I didn’t need to cut my water pipe to install it. As it turns out, the Flume device works by sensing the spinning disk in my city water meter and estimating usage. The faster the spin, the more usage, and the device does a surprisingly good job. Second, Flume Water has an API that allows me to build applications and automation based on my water usage.

So, what does it actually look like when it’s installed? Well here’s mine attached to the water meter coming into my home.

Flume water sensor
Flume Water sensor attached to my city water meter

I admit, I felt a little strange opening up one of those panels in the sidewalk in front of my house, but their tutorials walked me through it and assured me there was nothing illegal about it.

I finished the setup and calibration through the iPhone app and my water usage is now being tracked. It turns out I use about 1.3 gallons of water whenever I flush my toilet. Neat!

Flume provides a nice web portal and iPhone app to view my usage, set alarms, etc. but working at InfluxData, I really wanted to figure out how to get that data into my InfluxDB Cloud account so I could build custom dashboards and alerting on it.

flume-graph

Building a Flume Water Golang client

I took a look at the Flume Water API documentation and it looked simple enough to interact with. I’m always looking for the easy button, so I went on the hunt for some client libraries written in a language I knew. After a bit of Googling, I found a useful project for incorporating the data into Samsung Smartthings, but nothing for languages that I knew.

I had always wanted to learn a little more about writing Go code since it’s what InfluxDB is written in, so I started building a client library to make interacting with the Flume Water API a little easier. After looking at some other Go Clients and lots of copying and pasting from other Go examples online, I had a working client that authenticated, decoded my user id from the Flume Water token, and queried my water usage. Even though I didn’t need it, I decided to add functions to interact with the entire Flume API, including subscriptions and notifications.

Use case using InfluxDB, Go and Flume water meter

I’m not a seasoned Go developer, but I was able to cobble together something that seemed to work for authenticating and querying data. Here’s an example of using the client to query data.

# These credentials can be generated in your Flume Water portal
client := NewClient(os.Getenv("FLUME_CLIENT_ID"), os.Getenv("FLUME_CLIENT_SECRET"), os.Getenv("FLUME_USERNAME"), os.Getenv("FLUME_PASSWORD"))

devices, _ := client.FetchUserDevices(FlumeWaterFetchDeviceRequest{})

query := FlumeWaterQuery{
	Bucket:        FlumeWaterBucketDay,
	SinceDatetime: "2021-03-12 00:00:00",
	RequestID:     "test",
}
results, err := client.QueryUserDevice(devices[0].ID, FlumeWaterQueryRequest{
	Queries: []FlumeWaterQuery{query},
})

If you’re interested in using this library in your project, it’s called Flume Water Go Client and is available under the MIT license for anyone to use and contribute to.

Now that I had a Go client ready to use, I needed to figure out a way to get that data into my InfluxDB Cloud account. I’m a huge fan of the Telegraf data collection agent, and the Telegraf team recently released a capability that allows users to build and host external plugins, so I figured I’d give that a try.

Writing a Telegraf external plugin

The Telegraf Execd Input Plugin is an extremely powerful way to add new plugins to Telegraf. The plugin is designed to start up and interact with a long running external application that can be written in any language and compiled into an executable. Telegraf communicates with that application over STDIN and can listen for output or signal the application at a regular interval.

This is an improvement over the existing Exec Input Plugin because the application is not restarted with each Telegraf collection interval. For larger applications, the overhead of restarting the process each time can be too much for the frequency of the data being collected.

Telegraf will also manage restarting the process if it crashes or is killed, which is also handy if your code doesn’t handle network issues very well (like mine).

I started to take a look at the existing external plugins for Telegraf to see what needed to be done to build my own. The Rand example is really great, and provides some boilerplate code to make creating your plugin super easy. I’ve contributed custom Telegraf plugins in the past, and this project is configured so that all you really need to do is implement the same Gather function as you normally would in Telegraf, so it was very intuitive.

The nice thing about running this as an external plugin is that the example provides an easy way to test the plugin as well, printing the results to the terminal so you can verify it is working correctly.

I also added a Makefile as well so that building and running these commands could be simplified.

make run
./bin/flume-water --config plugin.dev.conf --poll_interval 5s
flume_water,bridge_id=34535334534534534,device_id=34534534534534543,device_type=2,location_building_type=SINGLE_FAMILY_HOME,location_city=San\ Francisco,location_name=SF\ Bernal,location_postal_code=94110,location_state=CA,request_id=flume-water-telegraf-input,units=gallons,[email protected] value=0.10389612 1616521800000000000

What I ended up with is a custom Telegraf Flume Water Input Plugin, that can be configured as easily as a native Telegraf plugin, and run via the Telegraf Execd Input. An example configuration looks like this:

[[inputs.flume_water]]
    client_id = "clientid"
    client_secret = "secret"
    username = "username"
    password = "password"
    ## If this isn't set, we will fetch your device list and pick the first one
    #device_id = ""
    ## lookback_mins is the amount of minutes to look back when querying data. This helps catch any late arriving data
    #lookback_mins = 5
    ## units can be one of GALLONS, LITERS, CUBIC_FEET, or CUBIC_METERS
    #units = "GALLONS"

And then the corresponding execd configuration for Telegraf looks like this:

[[inputs.execd]]
  ## One program to run as daemon.
  ## NOTE: process and each argument should each be their own string
  command = ["/path/to/telegraf-flume-water-input/bin/flume-water", "--config", "/path/to/telegraf-flume-water-input/plugin.conf"]

  ## Define how the process is signaled on each collection interval.
  ## Valid values are:
  ##   "none"    : Do not signal anything. (Recommended for service inputs)
  ##               The process must output metrics by itself.
  ##   "STDIN"   : Send a newline on STDIN. (Recommended for gather inputs)
  ##   "SIGHUP"  : Send a HUP signal. Not available on Windows. (not recommended)
  ##   "SIGUSR1" : Send a USR1 signal. Not available on Windows.
  ##   "SIGUSR2" : Send a USR2 signal. Not available on Windows.
  signal = "none"

  ## Delay before the process is restarted after an unexpected termination
  restart_delay = "10s"

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"

Finally, you can start Telegraf the same way you always do, using:

telegraf --config telegraf.conf
2021-03-19T00:08:19Z I! Starting Telegraf 1.18.0
2021-03-19T00:08:19Z I! Loaded inputs: execd
2021-03-19T00:08:19Z I! Loaded aggregators:
2021-03-19T00:08:19Z I! Loaded processors:
2021-03-19T00:08:19Z I! Loaded outputs: influxdb_v2
2021-03-19T00:08:19Z I! Tags enabled: host=myhost.lan
2021-03-19T00:08:19Z I! [agent] Config: Interval:10s, Quiet:false, Hostname:"myhost.lan", Flush Interval:10s
2021-03-19T00:08:19Z I! [inputs.execd] Starting process: /Users/rsavage/workspace/telegraf-flume-water-input/bin/flume-water [--config /Users/rsavage/workspace/telegraf-flume-water-input/plugin.conf]

Alright, so I’ve got my data flowing into Telegraf via the external plugin, and now I needed to connect it into my InfluxDB Cloud account.

Connecting everything to InfluxDB Cloud

I already have a free InfluxDB Cloud account, so I opened up my browser, and jumped into the Load Data > Telegraf section of the UI. I grabbed the Output plugin configuration from the UI and added it into my Telegraf config. I was able to generate an API token via the browser as well.

I started everything up and went to explore my data. After a few seconds, I started seeing my water usage “flowing” (I mean, I had to) into my InfluxDB Cloud account.

Just the beginning

All the code for this blog post is open source, so if you’ve got a Flume Water meter, you can configure it to go to your InfluxDB Cloud account. I’m currently working on building out an InfluxDB Template that can be used to quickly get up and running once the data is flowing into your account, unless someone beats me to it.

If you’ve got a favorite IoT device, and it has an API to export that data, why not contribute your own custom Telegraf Input Plugin? We’d love to hear about what you’re working on and highlight it to the community.

And if you’d like to check out InfluxDB for yourself, sign up for a free InfluxDB Cloud account. Once you do, feel free to ask any questions in our ever-helpful InfluxDB community and community Slack channel. Enjoy!

Related Blog Posts

Leave a Comment

Your email address will not be published. Required fields are marked *

Scroll to Top