telegraf设置多输入源多输出源

0、导

很多场景下，需要使用telegraf采集不同来源的数据，并输出到不同的数据源。打个比方，从emqx server采集很多不同系列的订阅信息，然后输出到influxdb不同的db中。

而telegraf支持将所采集到的不同数据源的数据输出到不同的位置，并指定输入与输出的对应关系。多数据源这个功能基于输出插件配置中的Metric Filtering参数。

The metric filtering parameters can be used to limit what metrics are emitted from the input/output plugin.

The metric filtering parameters can be used to limit what metrics are handled by the processor. Excluded metrics are passed downstream to the next processor

The metric filtering parameters can be used to limit what metrics are handled by the aggregator. Excluded metrics are passed downstream to the next aggregator.

1、Metric Filtering相关参数

关于Metric Filtering的完整介绍可以去往参考2查看详细介绍。

这里实现多数据输入源，多输出源所用到的是Metric Filtering 的 Selectors 功能。Selectors用来包含、排除指标(相当于白名单、黑名单的作用)。当输入/输出插件指定了白名单时，那仅会采集与输入/输出所匹配的指标。如果指定了黑名单，那些匹配的指标将会被丢弃掉。具体参数包括：

namepass：它相当于一个白名单，输入/输出插件都可以使用。只有measurement name与namepass相匹配了，该输入/输出插件指定的指标才会被采集/发送。namepass是一个数组，元素都是string，多个元素之间的关系是或，即只要匹配到任何一个元素即可。

namedrop：仅与namepass作用相反，与namedrop所匹配的会被丢弃。

tagpass：标签白名单，只有与tagpass所匹配的tag的会被发出。
tagdrop：标签黑名单，与tagpass相反。

nameXXXX匹配的是measurement name,tagXXXX匹配的则是标签

2、部分官方示例

# Using tagpass and tagdrop

[[inputs.cpu]]
  percpu = true
  totalcpu = false
  fielddrop = ["cpu_time"]
  # Don't collect CPU data for cpu6 & cpu7
  [inputs.cpu.tagdrop]
    cpu = [ "cpu6", "cpu7" ]

[[inputs.disk]]
  [inputs.disk.tagpass]
    # tagpass conditions are OR, not AND.
    # If the (filesystem is ext4 or xfs) OR (the path is /opt or /home)
    # then the metric passes
    fstype = [ "ext4", "xfs" ]
    # Globs can also be used on the tag values
    path = [ "/opt", "/home*" ]

[[inputs.win_perf_counters]]
  [[inputs.win_perf_counters.object]]
    ObjectName = "Network Interface"
    Instances = ["*"]
    Counters = [
      "Bytes Received/sec",
      "Bytes Sent/sec"
    ]
    Measurement = "win_net"
  # Don't send metrics where the Windows interface name (instance) begins with isatap or Local
  [inputs.win_perf_counters.tagdrop]
    instance = ["isatap*", "Local*"]

# Using namepass and namedrop
# Drop all metrics about containers for kubelet
[[inputs.prometheus]]
  urls = ["http://kube-node-1:4194/metrics"]
  namedrop = ["container_*"]

# Only store rest client related metrics for kubelet
[[inputs.prometheus]]
  urls = ["http://kube-node-1:4194/metrics"]
  namepass = ["rest_client_*"]

# Metrics can be routed to different outputs using the metric name and tags
[[outputs.influxdb]]
  urls = [ "http://localhost:8086" ]
  database = "telegraf"
  # Drop all measurements that start with "aerospike"
  namedrop = ["aerospike*"]

[[outputs.influxdb]]
  urls = [ "http://localhost:8086" ]
  database = "telegraf-aerospike-data"
  # Only accept aerospike data:
  namepass = ["aerospike*"]

[[outputs.influxdb]]
  urls = [ "http://localhost:8086" ]
  database = "telegraf-cpu0-data"
  # Only store measurements where the tag "cpu" matches the value "cpu0"
  [outputs.influxdb.tagpass]
    cpu = ["cpu0"]

# Routing metrics to different outputs based on the input
## Metrics are tagged with influxdb_database in the input, which is then used to select the output. The tag is removed in the outputs before writing.

[[outputs.influxdb]]
  urls = ["http://influxdb.example.com"]
  database = "db_default"
  [outputs.influxdb.tagdrop]
    influxdb_database = ["*"]

[[outputs.influxdb]]
  urls = ["http://influxdb.example.com"]
  database = "db_other"
  tagexclude = ["influxdb_database"]
  [outputs.influxdb.tagpass]
    influxdb_database = ["other"]

[[inputs.disk]]
  [inputs.disk.tags]
    influxdb_database = "other"

3、多输入源多输出源具体实现

上面第二个章节提到了多输出源涉及到的几个参数，这里为了实现多输入源与多输出源，可以结合输入插件的name_suffix、name_prefix参数搭配使用

    [[outputs.influxdb]]
      urls = ["http://xxxxx:xxxxx"]
      namepass = ["telegraf_*"]
      database = "telegraf"
      timeout = "5s"
      username = "admin"
      password = "xxxxx"
    [[outputs.influxdb]]
      urls = ["http://xxxxx:xxxxx"]
      namepass = ["internal_mqtt_*"]
      database = "internal_mqtt"
      timeout = "5s"
      username = "admin"
      password = "xxxxx"
    [[inputs.mqtt_consumer]]
      name_prefix = "internal_mqtt_"
      servers = ["tcp://xxxxxx:xxxxx"]
      topics = [
        "SS/BATTERY/#"
      ]
      qos = 0
      connection_timeout = "30s"
      data_format = "json"
      json_string_fields = ["cmd","lat","lng","timestamp","temperature","accelerate","power","voltage","current","electricity","speed","pullout","status"]
    [[inputs.mqtt_consumer]]
      servers = ["tcp://xxxxx:xxxxx"]
      name_prefix = "telegraf_"
      topics = [
        "DS/BATTERY/#",
        "DS/RED/#",
        "SD/BATTERY/#"
      ]
      qos = 0
      connection_timeout = "30s"
      # json_string_fields = ["cmd","lng","lat","datatime","Time"]
      data_format = "json"

在上边的例子中，两个input都是从mqtt拿数据，两个input都是将数据存到influxdb中。input不同的是订阅，output不同的是influxdb不同的库。

name_prefix = "telegraf_" 的input将会被存放到influxdb的telegraf库的telegraf_mqtt_consumer下;

而name_prefix = "internal_mqtt_"的input将会被存放到influxdb的internal_mqtt库的internal_mqtt_mqtt_consumer下

参考：

1、telegraf官方文档

2、Metric Filtering plugin github introduce

3、telegraf多输入输出