使用 Telegraf 连接 Flume Water 到 InfluxDB Cloud
作者：Russ Savage / 产品，用例
2021年3月29日
导航至
我喜欢我的设备与我交流并告诉我事情。实际上，我的家里充满了智能设备：智能电视、扬声器、灯开关、家电，甚至我的牙刷都有应用程序。然而，一个神秘的大问题是我的公用事业使用情况。在大多数情况下，美国的公用事业公司并不以尖端技术而闻名。是时候自己动手了。以下是我将在本文中讨论的主题
- 使用 Flume Water Monitor 跟踪我的用水量
- 在您的水表上安装 Flume Smart Water Monitor
- 设置 Flume 应用
- 构建 Flume Water Golang 客户端
- Telegraf Exec 输入插件是什么
- Telegraf 的现有外部插件
- Makefile
- 自定义 Telegraf Flume Water 输入插件
- 将所有内容连接到 InfluxDB Cloud
- 这只是开始
使用 Flume Water Monitor 跟踪我的用水量
我决定专注于我的用水量。我住在加利福尼亚州，该州一直在鼓励人们节约用水。这是因为 加利福尼亚州容易发生干旱，新2022年加利福尼亚州法律将家庭用水量限制在每人 55加仑。
当我每个月查看我的水费账单时，我看到我的用水量与时间的关系，但无法从公用事业公司实时获取这些数据。
在2021年圣诞节，我向圣诞老人要了一个 Flume Water 传感器，一个智能家居水监控器。Flume 智能水监控器设备通过 感应城市水表中的旋转盘 来估计用水量。旋转越快，我用的水越多。该设备在跟踪用水量方面做得非常出色。我选择 Flume Water 水表有两个主要原因
- 我不需要进行任何管道切割，安装只需要几分钟。我只需将它的橡胶带固定在表上，他们甚至还把一对橡胶手套随设备一起提供。
- 其次，Flume Water 有一个 API，允许我根据我的用水量构建应用程序和自动化。
安装后它实际上是什么样子？这是我将它固定在我家进来的水表上的样子。
在您的水表上安装 Flume Smart Water Monitor
您可以通过简单的自助安装将 Flume 智能家居水监控器安装在水表上。您可以根据他们网站上提供的 教程 进行操作。
我承认，当我打开我家门口人行道上的一块面板时，感觉有些奇怪。但是，他们的教程引导我完成了这个过程。它也让我确信，我在美国所做的一切都是完全合法的。
不仅仅是设置Flume应用
我在我的iPhone上下载了Flume应用，但它也适用于Android。Flume还有一个直观的Web门户。
通过iPhone应用完成设置和校准后，Flume开始跟踪我的用水量。结果发现，每当我冲马桶时，我大约使用1.3加仑的水。真不错！
Flume提供了一个不错的Web门户和iPhone Flume应用，可以查看我的用水量、设置警报等。但在InfluxData工作，我想弄清楚如何将数据导入我的InfluxDB Cloud账户，以构建自定义仪表板和警报。所以，我开始着手制作。
构建Flume Water Golang客户端
我查阅了Flume Water API文档，看起来交互起来很简单。我总是寻找简单的解决方案，所以我开始寻找一些用我所了解的语言编写的客户端库。经过搜索，我发现了一个用于将数据集成到三星Smartthings的有用项目。然而，没有我所了解的语言的库。
我一直想更多地了解Go代码的编写，因为InfluxDB就是用Go编写的。这就是我开始构建客户端库的原因，使与Flume Water API交互变得更容易。
在查看了一些其他Go客户端并从网上其他Go示例中复制粘贴后，我得到了一个工作的客户端。客户端进行了认证，从Flume Water令牌中解码我的用户ID，并查询了我的用水量。尽管我不需要它，但我决定添加与Flume API交互的功能，包括订阅和通知。
虽然我不是经验丰富的Go开发者，但我能够构建出一个看起来可以用于认证和查询数据的东西。以下是如何使用客户端查询数据的一个示例。
# 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},
})
如果你对你的项目中使用这个库感兴趣，它被称为Flume Water Go客户端。它可以在MIT许可下供任何人使用和贡献。
现在我已经有了可以使用的Go客户端，我需要想出一种方法将那些数据导入我的InfluxDB Cloud账户。我非常喜爱Telegraf数据收集代理，Telegraf团队最近发布了一个功能，允许用户构建和托管外部插件。我觉得值得一试。
什么是Telegraf外部插件？
Telegraf的Execd输入插件是一种非常强大的方式来向Telegraf添加新插件。该插件旨在启动并交互一个长时间运行的外部应用程序。该应用程序可以用任何语言编写并编译成可执行文件。Telegraf通过STDIN与应用程序通信，并可以定期监听输出或向应用程序发送信号。
这是对现有的 Exec 输入插件 的改进，因为应用程序不会在每个 Telegraf 收集间隔后重新启动。对于大型应用程序，每次重新启动进程的开销可能会因为收集数据的频率而过大。
Telegraf 还会管理在进程崩溃或被杀死时重新启动进程。如果你的代码（像我的一样）不擅长处理网络问题，这也会很有用。
Telegraf 的现有外部插件
我开始查看 Telegraf 的现有外部插件 （见此文档），以了解构建自己的插件需要做什么。Rand 示例 非常出色。它提供了一些样板代码，使创建插件变得超级简单。我之前曾贡献过自定义的 Telegraf 插件。这个项目的配置是这样的，你实际上只需要实现一个与在 Telegraf 中通常所做相同的 Gather 函数，因此它非常直观。
运行此插件作为外部插件的好处是，示例提供了一个简单的方法来测试插件并将结果打印到终端，以便你可以验证它是否正常工作。
Makefile
我还添加了一个 Makefile 来简化构建和运行这些命令。
./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结果：一个定制的 Telegraf Flume Water 输入插件
最终我得到了一个定制的 Telegraf Flume Water 输入插件，它可以像原生 Telegraf 插件一样轻松配置，并通过 Telegraf Execd 输入运行。一个示例配置看起来像这样
[[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"相应的 Telegraf execd 配置如下
[[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"最后，我以通常的方式启动 Telegraf
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]好了，我已经通过外部插件将数据流入 Telegraf。现在我需要将其连接到我的 InfluxDB Cloud 账户。
将所有内容连接到 InfluxDB Cloud
我已经有一个 免费的 InfluxDB Cloud 账户，所以我打开了浏览器，跳转到 UI 中的 加载数据 > Telegraf 部分。我从 UI 中获取了输出插件配置并将其添加到我的 Telegraf 配置中。我还可以通过浏览器生成 API 令牌。
我启动了一切并探索了我的数据。几秒钟后，我开始看到我的水使用情况“流入”（我的意思是，我必须这么做）我的 InfluxDB Cloud 账户。我现在可以确切地看到我正在使用多少水。
这只是开始
本博客文章的所有代码都是开源的。如果你有一个 Flume Water 水表，你可以配置它将数据发送到你的 InfluxDB Cloud 账户。我目前正在构建一个 InfluxDB 模板。在数据流入你的账户后，你可以快速使用它，除非有人先我一步。
如果你有一个你最喜欢的物联网设备，并且它有一个用于导出数据的 API，为什么不 贡献 你自己的自定义 Telegraf 输入插件呢？我们很乐意了解你正在做什么，并向社区展示你的工作。
如果您想亲自体验InfluxDB，请注册一个免费InfluxDB Cloud账户。注册后，您可以在我们始终乐于助人的InfluxDB社区和社区Slack频道中提出任何问题。在您将Flume水表配置到InfluxDB Cloud账户时，祝您享受过程！
常见问题解答（FAQ's）
Flume水传感器是如何工作的？
Flume智能水表由三个部分组成
- Flume水传感器
- Flume桥接器
- Flume水应用程序
要安装Flume，只需将其固定到您的表上。它测量您的表产生的磁场，并使用射频将此信息发送到Flume桥接器。桥接器连接到您家中的电源并连接到您的Wi-Fi网络。然后它将数据传输到云中。您可以通过运行在智能手机上的Flume水应用程序查看您的流量。它还可以检测泄漏，并在检测到泄漏时发出警报。
Flume水监测器准确吗？
Flume提供的详细水报告与您的供水公司报告的水使用量非常接近。Flume不玩猜测游戏。他们在Sensus Sr2、Badger 25和Hersey 420等多种水表上测试了Flume。在这些测试中，Flume关于水消耗的数据与表相比，只有±1%的偏差。
安装Flume的好处是什么？
水是我们最宝贵的资源之一，有了Flume，您可以实时查看您的用水量。此外，它还充当漏检器，告诉您家中是否有异常的水流，即使是从滴水的水龙头中。这些实时数据可以帮助您节省用水，从而节省金钱。
