From 482375f2e7998dadad6c5d3fda213a4e7a5ed320 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Tue, 29 Sep 2015 17:16:27 +0300 Subject: [PATCH 1/6] added encoder for OpenTSDB --- sandbox/lua/encoders/opentsdb_http.lua | 190 +++++++++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 sandbox/lua/encoders/opentsdb_http.lua diff --git a/sandbox/lua/encoders/opentsdb_http.lua b/sandbox/lua/encoders/opentsdb_http.lua new file mode 100644 index 000000000..f36427568 --- /dev/null +++ b/sandbox/lua/encoders/opentsdb_http.lua @@ -0,0 +1,190 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- License, v. 2.0. If a copy of the MPL was not distributed with this +-- file, You can obtain one at http://mozilla.org/MPL/2.0/. + +--[=[ +Extracts Field data from messages and generates some OpenTSDB-compliant JSON +(http://opentsdb.net/docs/build/html/api_http/put.html). + +Can optionally buffer and flush events after a configurable number of messages +(flush_count) or after a configurable number of seconds have passed between the +last Timestamp and the current one (flush_delta) (encoders currently have no +support for timer_event()s) + +Config: + +- flush_count (number, default 1) + Flush the buffer every 'flush_count' messages. + +- flush_delta (number, default 0) + Flush the buffer if 'flush_delta' seconds have elapsed since the last + Timestamp and the current one. + +- tag_prefix (string, optional, default "") + Convert Fields to OpenTSDB tags if they match the tag_prefix. + (tag_prefix is stripped from the tag name) + +- tag_list (string, optional, default "") + A list of fields (separated by spaces) which will be converted into tags. + You can use either "tag_list", "tag_prefix" or both. + +- skip_fields (string, optional, default "") + A list of fields (separated by spaces) which will be dropped from processing. + +- type_as_prefix (boolean, optional, default true) + Prepend metrics when encoding them to OpenTSDB format by Type field. + +- ts_from_message (boolean, optional, default false) + Use the Timestamp field (otherwise "now") + +- add_hostname_if_missing (boolean, optional, default false) + If no 'host' tag has been seen, append one with the value of the + Hostname field. Deprecated in favour of using the 'fieldfix' filter. + + +*Example Heka Configuration* +.. code-block:: ini + + [OpentsdbEncoder] + type = "SandboxEncoder" + filename = "lua_encoders/opentsdb_http.lua" + + [OpentsdbEncoder.config] + ts_from_message = true + add_hostname_if_missing = true + type_as_prefix = true + tag_prefix = "tag_" + tag_list = "env dc" + skip_fields = "Pid" + + [opentsdb] + type = "HttpOutput" + message_matcher = "Fields[MetricOpentsdb] != NIL" + address = "http://opentsdb.example.com:4242/api/put" + encoder = "OpentsdbEncoder" + +*Example Output* +.. code-block:: json + [{"value":0.05,"timestamp":1443535912,"metric":"stats.loadavg.15MinAvg","tags":{"env":"test","host":"example.com"}},{"value":2,"timestamp":1443535912,"metric":"stats.loadavg.NumProcesses","tags":{"env":"test","host":"example.com"}},{"value":0.01,"timestamp":1443535912,"metric":"stats.loadavg.5MinAvg","tags":{"env":"test","host":"example.com"}},{"value":0,"timestamp":1443535912,"metric":"stats.loadavg.1MinAvg","tags":{"env":"test","host":"example.com"}}] + +--]=] + +require "cjson" +require "table" +require "string" +require "os" + +local l = require 'lpeg' +l.locale(l) + +local flush_count = read_config("flush_count") or 1 +local flush_delta = read_config("flush_delta") or 0 +local tag_prefix = read_config("tag_prefix") +local tag_list_str = read_config("tag_list") or "" +local skip_fields_str = read_config("skip_fields") or "" +local type_as_prefix = read_config("type_as_prefix") +local ts_from_message = read_config("ts_from_message") +local add_hostname = read_config("add_hostname_if_missing") +_PRESERVATION_VERSION = read_config("preservation_version") or 1 + +if not tag_prefix then + return -1 +end + +if type_as_prefix == nil then + type_as_prefix = true +end + +if add_hostname == nil then + add_hostname = true +end + +if ts_from_message == nil then + ts_from_message = true +end + +tag_list = {} +for field in tag_list_str:gmatch("[%S]+") do + tag_list[field] = true +end + +skip_fields = {} +for field in skip_fields_str:gmatch("[%S]+") do + skip_fields[field] = true +end + +prefix = l.P(tag_prefix) +chars = l.alnum + l.S("_-./") +grammar = prefix * l.C(l.alpha * chars^0) + +buffer = {} +last_flush = 0 + +function flush() + add_to_payload(cjson.encode(buffer)) + inject_payload() + last_flush = os.time() + buffer = {} +end + + +function process_message() + + local ts + if ts_from_message then + ts = read_message("Timestamp") / 1e9 + else + ts = os.time() + end + + if not ts then return -1 end + + local typ=read_message("Type") + local tags = {} + local metrics = {} + + local msg = decode_message(read_message("raw")) + if not msg.Fields then return -1 end + + for _, field in ipairs(msg.Fields) do + local name = field["name"] -- these are related to way Heka works + local value = field["value"][1] -- + + local tag = grammar:match(name) or (tag_list[name] ~= nil and name) + if tag then + tags[tag] = value + else + if type(value) == "number" and skip_fields[name] == nil then + if type_as_prefix then + name = typ .. "." .. name + end + metrics[name] = value + end + end + end + + if not tags["host"] and add_hostname then + tags["host"] = msg.Hostname + end + + opentsdb_message = {} + + for name, value in pairs(metrics) do + local message = {} + message["metric"] = name + message["value"] = value + message["timestamp"] = ts + message["tags"] = tags + buffer[#buffer+1] = message + end + + -- flush the buffer + if (flush_count > 0 and #buffer >= flush_count) or + (flush_delta > 0 and os.time() - last_flush > flush_delta) then + flush() + else + return -2 + end + + return 0 +end From 2d4aa4d8c8bd25ac568e80ae270124066190ef72 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Mon, 12 Oct 2015 13:28:24 +0300 Subject: [PATCH 2/6] updated OpenTSDB encoder based on feedback from pull request --- sandbox/lua/encoders/opentsdb_http.lua | 66 +++++++++++--------------- 1 file changed, 28 insertions(+), 38 deletions(-) diff --git a/sandbox/lua/encoders/opentsdb_http.lua b/sandbox/lua/encoders/opentsdb_http.lua index f36427568..893b24508 100644 --- a/sandbox/lua/encoders/opentsdb_http.lua +++ b/sandbox/lua/encoders/opentsdb_http.lua @@ -31,15 +31,15 @@ Config: - skip_fields (string, optional, default "") A list of fields (separated by spaces) which will be dropped from processing. -- type_as_prefix (boolean, optional, default true) +- type_as_prefix (boolean, optional, default false) Prepend metrics when encoding them to OpenTSDB format by Type field. -- ts_from_message (boolean, optional, default false) +- ts_from_message (boolean, optional, default true) Use the Timestamp field (otherwise "now") - add_hostname_if_missing (boolean, optional, default false) If no 'host' tag has been seen, append one with the value of the - Hostname field. Deprecated in favour of using the 'fieldfix' filter. + Hostname field. *Example Heka Configuration* @@ -68,6 +68,8 @@ Config: [{"value":0.05,"timestamp":1443535912,"metric":"stats.loadavg.15MinAvg","tags":{"env":"test","host":"example.com"}},{"value":2,"timestamp":1443535912,"metric":"stats.loadavg.NumProcesses","tags":{"env":"test","host":"example.com"}},{"value":0.01,"timestamp":1443535912,"metric":"stats.loadavg.5MinAvg","tags":{"env":"test","host":"example.com"}},{"value":0,"timestamp":1443535912,"metric":"stats.loadavg.1MinAvg","tags":{"env":"test","host":"example.com"}}] --]=] +_PRESERVATION_VERSION = read_config("preservation_version") or 0 +_PRESERVATION_VERSION = _PRESERVATION_VERSION + 1 -- force a revision update due to internal changes require "cjson" require "table" @@ -79,25 +81,12 @@ l.locale(l) local flush_count = read_config("flush_count") or 1 local flush_delta = read_config("flush_delta") or 0 -local tag_prefix = read_config("tag_prefix") +local tag_prefix = read_config("tag_prefix") or error("`tag_prefix` setting required") local tag_list_str = read_config("tag_list") or "" local skip_fields_str = read_config("skip_fields") or "" local type_as_prefix = read_config("type_as_prefix") local ts_from_message = read_config("ts_from_message") local add_hostname = read_config("add_hostname_if_missing") -_PRESERVATION_VERSION = read_config("preservation_version") or 1 - -if not tag_prefix then - return -1 -end - -if type_as_prefix == nil then - type_as_prefix = true -end - -if add_hostname == nil then - add_hostname = true -end if ts_from_message == nil then ts_from_message = true @@ -113,11 +102,12 @@ for field in skip_fields_str:gmatch("[%S]+") do skip_fields[field] = true end -prefix = l.P(tag_prefix) -chars = l.alnum + l.S("_-./") -grammar = prefix * l.C(l.alpha * chars^0) +local prefix = l.P(tag_prefix) +local chars = l.alnum + l.S("_-./") +local grammar = prefix * l.C(l.alpha * chars^0) buffer = {} +buffered_messages = 0 last_flush = 0 function flush() @@ -125,6 +115,7 @@ function flush() inject_payload() last_flush = os.time() buffer = {} + buffered_messages = 0 end @@ -137,49 +128,48 @@ function process_message() ts = os.time() end - if not ts then return -1 end - local typ=read_message("Type") local tags = {} local metrics = {} local msg = decode_message(read_message("raw")) - if not msg.Fields then return -1 end + if not msg.Fields then + return -1, "Malformed message (no fields)" + end for _, field in ipairs(msg.Fields) do - local name = field["name"] -- these are related to way Heka works - local value = field["value"][1] -- + local name = field.name -- these are related to way Heka works + local value = field.value[1] -- - local tag = grammar:match(name) or (tag_list[name] ~= nil and name) + local tag = grammar:match(name) or (tag_list[name] and name) if tag then tags[tag] = value else - if type(value) == "number" and skip_fields[name] == nil then + if type(value) == "number" and not skip_fields[name] then if type_as_prefix then - name = typ .. "." .. name + name = string.format("%s.%s", typ, name) end metrics[name] = value end end end - if not tags["host"] and add_hostname then - tags["host"] = msg.Hostname + if not tags.host and add_hostname then + tags.host = msg.Hostname end - opentsdb_message = {} - for name, value in pairs(metrics) do local message = {} - message["metric"] = name - message["value"] = value - message["timestamp"] = ts - message["tags"] = tags - buffer[#buffer+1] = message + message.metric = name + message.value = value + message.timestamp = ts + message.tags = tags + buffered_messages = buffered_messages + 1 + buffer[buffered_messages] = message end -- flush the buffer - if (flush_count > 0 and #buffer >= flush_count) or + if (flush_count > 0 and buffered_messages >= flush_count) or (flush_delta > 0 and os.time() - last_flush > flush_delta) then flush() else From 0338b2a26a4c3ee561f582ca31d100f6f77621c3 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Mon, 12 Oct 2015 13:45:09 +0300 Subject: [PATCH 3/6] renamed opentsdb_http into opentsdb_batch to avoid name clash with @hynd's encoder --- sandbox/lua/encoders/{opentsdb_http.lua => opentsdb_batch.lua} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename sandbox/lua/encoders/{opentsdb_http.lua => opentsdb_batch.lua} (99%) diff --git a/sandbox/lua/encoders/opentsdb_http.lua b/sandbox/lua/encoders/opentsdb_batch.lua similarity index 99% rename from sandbox/lua/encoders/opentsdb_http.lua rename to sandbox/lua/encoders/opentsdb_batch.lua index 893b24508..bfafb4040 100644 --- a/sandbox/lua/encoders/opentsdb_http.lua +++ b/sandbox/lua/encoders/opentsdb_batch.lua @@ -47,7 +47,7 @@ Config: [OpentsdbEncoder] type = "SandboxEncoder" - filename = "lua_encoders/opentsdb_http.lua" + filename = "lua_encoders/opentsdb_batch.lua" [OpentsdbEncoder.config] ts_from_message = true From 43b548a77e736013ea2b279b076d867539c764e7 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Mon, 12 Oct 2015 13:46:33 +0300 Subject: [PATCH 4/6] add docs for OpenTSDB encoder --- CHANGES.txt | 2 ++ docs/source/config/encoders/index.rst | 1 + docs/source/config/encoders/index_noref.rst | 3 +++ docs/source/config/encoders/opentsdb_http.rst | 13 +++++++++++++ docs/source/sandbox/encoder.rst | 6 ++++++ 5 files changed, 25 insertions(+) create mode 100644 docs/source/config/encoders/opentsdb_http.rst diff --git a/CHANGES.txt b/CHANGES.txt index 1a33cb8ce..2f2a41e90 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -22,6 +22,8 @@ Features into alternate locations; useful for relocating forks of Go packages into their original import paths. +* Adde encoder for OpenTSDB + 0.10.0 (2015-??-??) ===================== diff --git a/docs/source/config/encoders/index.rst b/docs/source/config/encoders/index.rst index dcac6cda6..b4ee2600c 100644 --- a/docs/source/config/encoders/index.rst +++ b/docs/source/config/encoders/index.rst @@ -17,6 +17,7 @@ Available Encoder Plugins esjson eslogstashv0 espayload + opentsdb_batch payload protobuf rst diff --git a/docs/source/config/encoders/index_noref.rst b/docs/source/config/encoders/index_noref.rst index 80414a96e..4c504dce8 100644 --- a/docs/source/config/encoders/index_noref.rst +++ b/docs/source/config/encoders/index_noref.rst @@ -18,6 +18,9 @@ Encoders .. include:: /config/encoders/espayload.rst :start-line: 1 +.. include:: /config/encoders/opentsdb_batch.rst + :start-line: 1 + .. include:: /config/encoders/payload.rst :start-line: 1 diff --git a/docs/source/config/encoders/opentsdb_http.rst b/docs/source/config/encoders/opentsdb_http.rst new file mode 100644 index 000000000..bb10ad4b7 --- /dev/null +++ b/docs/source/config/encoders/opentsdb_http.rst @@ -0,0 +1,13 @@ +.. _config_opentsdb_http_encoder: + +OpenTSDB Encoder +============= + +.. versionadded:: 0.10 + +| Plugin Name: **SandboxEncoder** +| File Name: **lua_encoders/opentsdb_http.lua** + +.. include:: /../../sandbox/lua/encoders/opentsdb_http.lua + :start-after: --[[ + :end-before: --]] diff --git a/docs/source/sandbox/encoder.rst b/docs/source/sandbox/encoder.rst index 617019324..c2868fc79 100644 --- a/docs/source/sandbox/encoder.rst +++ b/docs/source/sandbox/encoder.rst @@ -26,6 +26,12 @@ ESPayloadEncoder :start-after: --[[ :end-before: --]] +OpenTSDBEncoder +^^^^^^^^^^^^^^^^ +.. include:: /../../sandbox/lua/encoders/opentsdb_batch.lua + :start-after: --[[ + :end-before: --]] + Schema Carbon Line Encoder ^^^^^^^^^^^^^^^^^^^^^^^^^^ .. include:: /../../sandbox/lua/encoders/schema_carbon_line.lua From 7e9c7dd98b7d72400ef025a3be96e271540f8e85 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Tue, 13 Oct 2015 13:26:48 +0300 Subject: [PATCH 5/6] opentsdb_batch: try to convert all strings to numbers instead of checking for a type --- sandbox/lua/encoders/opentsdb_batch.lua | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sandbox/lua/encoders/opentsdb_batch.lua b/sandbox/lua/encoders/opentsdb_batch.lua index bfafb4040..3b8d8c216 100644 --- a/sandbox/lua/encoders/opentsdb_batch.lua +++ b/sandbox/lua/encoders/opentsdb_batch.lua @@ -145,7 +145,8 @@ function process_message() if tag then tags[tag] = value else - if type(value) == "number" and not skip_fields[name] then + value = tonumber(value) + if value and not skip_fields[name] then if type_as_prefix then name = string.format("%s.%s", typ, name) end From a3c1204acea13fa3f62897f66eca195608159428 Mon Sep 17 00:00:00 2001 From: Timur Batyrshin Date: Thu, 15 Oct 2015 14:22:54 +0300 Subject: [PATCH 6/6] make sure we are sending only integers to OpenTSDB --- sandbox/lua/encoders/opentsdb_batch.lua | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sandbox/lua/encoders/opentsdb_batch.lua b/sandbox/lua/encoders/opentsdb_batch.lua index 3b8d8c216..9aca573a0 100644 --- a/sandbox/lua/encoders/opentsdb_batch.lua +++ b/sandbox/lua/encoders/opentsdb_batch.lua @@ -76,7 +76,8 @@ require "table" require "string" require "os" -local l = require 'lpeg' +local math = require "math" +local l = require "lpeg" l.locale(l) local flush_count = read_config("flush_count") or 1 @@ -123,7 +124,7 @@ function process_message() local ts if ts_from_message then - ts = read_message("Timestamp") / 1e9 + ts = math.floor(read_message("Timestamp") / 1e9) else ts = os.time() end