网站首页 文章专栏 telegraf+influxdb+grafana监控(二)
telegraf+influxdb+grafana监控(二)
编辑时间:2022-04-02 10:47:51 作者:linxi 浏览量:55

在前面的文章已经详细讲解了MQTT、telegraf、influxDB等组件,这里要将三者之间的的关系理顺,并详细介绍配置及使用方法MQTT(mosquitto)->telegraf->influxDB

一、启动 mosquitto 服务

image.png

二、配置 telegraf 连接 influxDB、mosquitto

1、telegraf 采集MQTT数据,写入influxDB

# Configuration for telegraf agent
[agent]
 ## Default data collection interval for all inputs
 ##interval = "10s"
 interval = "5s"   #采集间隔5秒
 ## Rounds collection interval to 'interval'
 ## ie, if interval="10s" then always collect on :00, :10, :20, etc.
 round_interval = true

 ## Telegraf will send metrics to outputs in batches of at most
 ## metric_batch_size metrics.
 ## This controls the size of writes that Telegraf sends to output plugins.
 metric_batch_size = 1000

 ## Maximum number of unwritten metrics per output.  Increasing this value
 ## allows for longer periods of output downtime without dropping metrics at the
 ## cost of higher maximum memory usage.
 metric_buffer_limit = 10000

 ## Collection jitter is used to jitter the collection by a random amount.
 ## Each plugin will sleep for a random time within jitter before collecting.
 ## This can be used to avoid many plugins querying things like sysfs at the
 ## same time, which can have a measurable effect on the system.
 collection_jitter = "0s"

 ## Collection offset is used to shift the collection by the given amount.
 ## This can be be used to avoid many plugins querying constraint devices
 ## at the same time by manually scheduling them in time.
 # collection_offset = "0s"

 ## Default flushing interval for all outputs. Maximum flush_interval will be
 ## flush_interval + flush_jitter
 ##flush_interval = "10s"
 flush_interval = "5s"   #写入influxDB间隔5秒
 flush_jitter = "0s"
 precision = "0s"

# Configuration for sending metrics to InfluxDB
[[outputs.influxdb]]
 ## The full HTTP or UDP URL for your InfluxDB instance.
 ##
 ## Multiple URLs can be specified for a single cluster, only ONE of the
 ## urls will be written to each interval.
 # urls = ["unix:///var/run/influxdb.sock"]
 # urls = ["udp://127.0.0.1:8089"]
 urls = ["http://127.0.0.1:8086"]

 ## The target database for metrics; will be created as needed.
 ## For UDP url endpoint database needs to be configured on server side.
 database = "telegraf"
 ## Timeout for HTTP messages.
 timeout = "5s"

 ## HTTP Basic Auth
 username = "xxxx"    #连接influxDB数据库用户名
 password = "xxxx"    #连接influxDB数据库密码

# # Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
#   ## Broker URLs for the MQTT server or cluster.  To connect to multiple
#   ## clusters or standalone servers, use a separate plugin instance.
#   ##   example: servers = ["tcp://localhost:1883"]
#   ##            servers = ["ssl://localhost:1883"]
#   ##            servers = ["ws://localhost:1883"]
servers = ["tcp://127.0.0.1:1883"]
#   ## Topics that will be subscribed to.
#   topics = [
#     "telegraf/host01/cpu",
#     "telegraf/+/mem",
#     "sensors/#",
#   ]
topics = [
     "dev",
     "test",
   ]

#   # topic_fields = "_/_/_/temperature"
#   ## The message topic will be stored in a tag specified by this value.  If set
#   ## to the empty string no topic tag will be created.
#   # topic_tag = "topic"
#   ## QoS policy for messages
#   ##   0 = at most once
#   ##   1 = at least once
#   ##   2 = exactly once
#   ##
qos = 2
#   ## When using a QoS of 1 or 2, you should enable persistent_session to allow
#   ## resuming unacknowledged messages.
#   # qos = 0
#   ## Connection timeout for initial connection in seconds
#   # connection_timeout = "30s"
connection_timeout = "30s"
persistent_session = false
#   ## If unset, a random client ID will be generated.
#   # client_id = ""
#   ## Username and password to connect MQTT server.
#   # username = "telegraf"
username = "xxxx"   #连接MQTT的用户名
#   # password = "metricsmetricsmetricsmetrics"
password = "xxxx"   #连接MQTT的密码
#   ## Optional TLS Config
#   # tls_ca = "/etc/telegraf/ca.pem"
#   # tls_cert = "/etc/telegraf/cert.pem"
#   # tls_key = "/etc/telegraf/key.pem"
#   ## Use TLS but skip chain & host verification
#   # insecure_skip_verify = false
#   ## 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"
data_format = "value"
data_type = "string"

上例中,数据格式设置为 value,数据类型为 string ,模拟测试结果:

image.png

2、json 格式数据解析

可见我们将 json 字符串存入了 influxDB,如何完成解析呢?

https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md

https://docs.influxdata.com/telegraf/v1.21/data_formats/input/json/

测试后发现,修改为以下参数

# # Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
......
data_format = "value"
data_type = "json"

tag_keys = [
 "id",
 "name",
 "collect_time",
]

上例中,数据格式设置为 json,标签 tags 为 id、name、collect_time ,模拟测试结果:

# select * from mqtt_consumer;
# show series from mqtt_consumer;

image.png

可见,该表中有 equipment-1、equipment-2 两个系列,也就是成图时有两条曲线。测试数据结构如下:

image.png

3、修改默认表 mqtt_consumer

测试后发现,修改为以下参数

# # Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
......
data_format = "value"
data_type = "json"
tag_keys = [
 "id",
 "name",
 "collect_time",
]
json_name_key = "measurement"

上例中,数据格式设置为 json_name_key 为表名取自哪个 json 字段,模拟测试结果:

image.png

可见,我们在 influxDB 中成功创建了 mqtt_test 表,并且插入了 5 条设备数据。测试数据结构如下:

image.png

其中,measurement 与  json_name_key 的设置值一一对应,当我们实例化赋值为 mqtt_test 时,该条数据插入时,就会创建 mqtt_test 表。

来说两句吧
最新评论