ruby - 如何从logstash中的netdata聚合指标？

Question

并提前感谢您的帮助！

我正在使用 Netdata 从服务器收集指标，然后将它们发送到 Logstash 和 Elastic。

我的需要是聚合具有相同字段的指标并创建单个事件，但采用嵌套格式。

这是来自 Netdata 的输入示例：

{"host":"centosdns","@version":"1","port":52212,"@timestamp":"2019-01-19T16:16:22.117Z","message":"netdata.centosdns.disk_await.centos_swap.reads 0.0000000 1547914548"}
{"host":"centosdns","@version":"1","port":52212,"@timestamp":"2019-01-19T16:16:22.117Z","message":"netdata.centosdns.disk_await.centos_swap.writes 0.0000000 1547914548"}
{"host":"centosdns","@version":"1","port":52212,"@timestamp":"2019-01-19T16:16:22.117Z","message":"netdata.centosdns.disk_await.centos_root.reads 0.0000000 1547914548"}
{"host":"centosdns","@version":"1","port":52212,"@timestamp":"2019-01-19T16:16:22.117Z","message":"netdata.centosdns.disk_await.centos_root.writes 0.0000000 1547914548"}

我的 logstash 配置文件如下所示：

input {
    tcp {
      port => 1234
    }
}

filter {
    # I take 'message' field and separate in different fields
    grok {
      named_captures_only => "true"
      pattern_definitions => {
        "CHART" => "[a-z]\w+"
        "FAMILY" => "[_a-z0-9]+"
    }

    match => {
      "message" => "%{WORD:prefix}\.%{WORD:hostname}\.%{CHART:chart}\.%{FAMILY:family}\.%{NOTSPACE:dimension} %{NUMBER:val} %{NUMBER:timestamp}" 
    }
}

if "_grokparsefailure" not in [tags] {
    mutate {
      remove_field => [ "@version", "host", "port", "prefix" ]
    }

    # Attempt to create a nested field and then aggregate 
    mutate {
      id => "chart_field"
      add_field => { "[%{chart}][%{family}][%{dimension}][value]" => "%{val}"
      }
    }

    aggregate {
      task_id => "[%{chart}][%{family}]"
      code => "
      # I tried many codes to aggregate but without success 
      event.cancel()
      "
      push_previous_map_as_event => true
      timeout => 5
    }

    mutate {
      # Remove unnecessary fields
      id => "netdata_mutate_remove"
      remove_field => [ "timestamp", "message"]
    }
} else {
    drop{}
}

output {
# TESTING PURPOSES
if "_aggregateexception" in [tags] {

    file {
      path => "/var/log/logstash/netdata/aggregatefailures-%{+MM-dd}.log"
    }
} else {
    file {
      path => "/var/log/logstash/netdata/netdata-%{+MM-dd}-aggregate.log"
    }
}

stdout { codec => rubydebug }
}

取上面的输入：

"netdata.centosdns.disk_await.centos_swap.reads 0.0000000"
"netdata.centosdns.disk_await.centos_swap.writes 0.0000000"

我的目标是制作一个嵌套字段，例如：

disk_await: { # Chart
  centos_swap: { # Family
   [
    reads => 0.0000000, # Dimension => Value
    writes => 0.0000000 # Dimension => Value
   ]
  }
}

我假装在同一个 'Chart'\'Family' 中聚合所有 'Dimension\'Value''，这只是四行指标，但实际上我们谈论每秒 1000 个甚至更多，在某些情况下，所有指标都是动态的，几乎不可能知道所有的名字。

目前我正在使用：

Logstash v.6.5.4 on a Virtualbox CentOS 7 minimal
All plugins (inputs/filters/outputs) updated

标签： rubyelasticsearchlogstashlogstash-configuration