使用 Telegraf 发送数据到阿里云 IoT 平台

阿里云的IoT平台本质上是一个以MQTT服务为核心的物联网平台。国内的一些商业网关设备，提供了上报数据阿里云IoT平台的支持。而使用开源方案的话，Telegraf 提供了 MQTT 输出插件，配合其他的配置，也可以数据上云。

问题定义

工业设备或者 IoT 设备的数据已经被采集到本地服务器（边缘服务器），比如在 OPCUA 服务器、Kakfa 甚至本地的 MQTT 服务器中。为了数据监控、分析和应用，需要将数据传送到阿里云 IoT 平台中。待传到 IoT 平台之后，还可以利用它的数据流转服务，编写类 Javascript 代码，将数据流转到 Lindorm 时序数据库中。本文只关注上报数据到阿里云 IoT 平台。

技术调研

Telegraf

Telegraf 提供了发送数据到 MQTT 服务器的集成，具体见其代码库中的文档《MQTT Producer Output Plugin》。

阿里云 IoT 平台

在最核心 MQTT 服务器之外，阿里云IoT平台还包括了管理元数据的物模型、鉴权认证等。一个产品对应一个物模型，而一个产品可以实例化为多个设备。物模型的介绍见阿里云的文档《配置物模型》，其中介绍了《什么是物模型》。在此不做赘述。

鉴权包括了TLS证书、MQTT用户名密码认证。其中，MQTT用户名是由设备名字和产品键组成的，即deviceName+"&"+productKey，而每个设备由自己的密码。TLS证书见《MQTT-TLS连接通信》中的文档。本文作者验证了使用“Global Sign R1根证书”的方法。

Telegraf 与阿里云 IoT 集成的挑战

在真实的场景下，当使用 Telegraf 的 MQTT 输出插件与阿里云 IoT 集成时，会遇到一个问题：假如有数十个设备，那么意味着就有数十个用户名密码对。但是每一个 Telegraf 的集成，只支持填写一对 MQTT 输出插件的用户名密码。这也就意味着，针对每一个设备，都需要启动一个 Telegraf 进程。这显然带来了管理和运维上的复杂度。目前我还没有找到简化的方法。

当发送数据到阿里云时，需要构建特定格式的 JSON 格式的消息体，而不是 Influxdb 的格式。好消息是，可以使用 json_transformation 配置发送的消息体。

为了安全起见，一些企业只允许本地网络和阿里云之间通过VPN网关联通，此时阿里云IoT位于 VPC 之中。截止当前，本地的客户端无法直接访问阿里云IoT的VPC地址，还需要在 VPC 中搭建一个三层代理进行数据转发。

在早期的 Telegraf 有一个 Bug，它会在构建 MQTT 主题的时候，去掉 / 前缀，而阿里云的 MQTT 主题正好以 / 开头。有很多文章不建议 MQTT 的主题以 / 开头，因为这会产生歧义，但阿里云似乎总是不按常理出牌。在最新的版本中这个问题已经修复。

【疯狂吐槽】阿里云为什么不按常理出牌？可能是独树一帜吧！也许这样就有自己的标准了。

配置举例

使用 Nginx 做三层代理

通常 Nginx 被用于做四层代理，其实也可以作为三层代理，配置示例如下：

stream {
    upstream aliyun_iot_platform_1883 {
        server iot-xxxx-vpc.mqtt.iothub.aliyuncs.com:1883; # 服务器地址位于“查看开发设置”
    }

    server {
        listen 31883; # 本地端口，自行选择
        proxy_pass aliyun_iot_platform_1883;
    }
}

Telegraf 配置

Telegraf MQTT 输出插件的配置示例如下，注意修改deviceName、productKey、client_id、password（client_id 和 password 的计算方式见文档《MQTT-TLS连接通信》）：

[[outputs.mqtt]]
servers = ["tls://x.x.x.x:xxxx"] # 指向 VPC 中的代理
username = "deviceName&productKey"
client_id = "client_id"
password = "password"
topic = "/sys/productKey/deviceName/thing/event/property/post"
protocol = "3.1.1"
qos = 1
tls_ca = "/etc/telegraf/aliyun/root.crt" # 指向本地的 “Global Sign R1根证书”
keep_alive = 3000
connection_timeout = "5s"
insecure_skip_verify = true
timeout = "2s"

data_format = "json"
json_timestamp_units = "1ms"
# 下面的 tags.Property 是一个自定义的标签，对应的是设备的属性
json_transformation = '''
  {"id": $join([tags.Property, "-", $string(timestamp)]),
  "version":"1.0",
  "sys": {
    "ack": 0
  },
  "params": {
    tags.Property: {
      "value": fields.Value,
      "time": timestamp
    }},
    "method": "thing.event.property.post"
  }
'''

结尾

使用 Telegraf 发送数据到阿里云 IoT 平台是可行的，但是在集成的过程中，并不是很顺利。尤其在遇到很多设备的时候，如果每个设备都运行一个 Telegraf 进程，那管理起来就是噩梦。如果不需要在阿里云上构建物模型，而只是为了转存 IoT 数据到阿里云上的 Lindorm 时序数据库作为分析，那么可以直接将数据发送到 Lindorm 时序数据。