From 3a15b857a1c1f51ce3d02e9c90d0eb85c4d1447e Mon Sep 17 00:00:00 2001 From: Wesley Dawson Date: Mon, 22 Feb 2016 14:50:53 -0800 Subject: [PATCH 1/2] Add systemd input to the data-pipeline build --- bin/build_pipeline_heka.sh | 20 ++- heka/sandbox/inputs/systemd.lua | 259 ++++++++++++++++++++++++++++++++ heka/sandbox/modules/debug.lua | 56 +++++++ 3 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 heka/sandbox/inputs/systemd.lua create mode 100644 heka/sandbox/modules/debug.lua diff --git a/bin/build_pipeline_heka.sh b/bin/build_pipeline_heka.sh index 47faf21..6dd1d49 100755 --- a/bin/build_pipeline_heka.sh +++ b/bin/build_pipeline_heka.sh @@ -19,7 +19,7 @@ set -o errtrace pushd . # Machine config: -# sudo yum install -y git hg golang cmake rpmdevtools GeoIP-devel rpmrebuild +# sudo yum install -y git hg golang cmake rpmdevtools GeoIP-devel rpmrebuild systemd-devel BUILD_BRANCH=$1 if [ -z "$BUILD_BRANCH" ]; then @@ -87,6 +87,7 @@ if [ ! -f "patches_applied" ]; then echo "Patching to build 'heka-s3list' and 'heka-s3cat'" patch CMakeLists.txt < $BASE/heka/patches/0003-Add-more-cmds.patch + echo 'install(DIRECTORY "${CMAKE_SOURCE_DIR}/sandbox/lua/inputs/" DESTINATION "share/${CMAKE_PROJECT_NAME}/lua_inputs")' >> CMakeLists.txt echo "Adding external plugin for s3splitfile output" echo "add_external_plugin(git https://github.com/mozilla-services/data-pipeline/s3splitfile :local)" >> cmake/plugin_loader.cmake @@ -218,6 +219,23 @@ mkdir -p $HEKA_MODS/fx cd $BASE gcc $CFLAGS -I${LUA_INCLUDE_PATH} $SO_FLAGS --std=c99 heka/plugins/fx/executive_report.c heka/plugins/fx/xxhash.c heka/plugins/fx/common.c -o $HEKA_MODS/fx/executive_report.so +if [[ "$UNAME" == "Linux" ]] ; then + # build systemd input + echo 'Installing lua-systemd lib' + cd $BASE/build + if [ ! -d lua-systemd ]; then + git clone https://github.com/whd/lua-systemd + fi + cd lua-systemd + + # Use a known revision (current "master" 2015-04-02) + git checkout 48c8db982dd3af531c9033eca0d9853f8b56c92a + mkdir -p release + cd release + cmake -DLUA_INCLUDE_DIR="${LUA_INCLUDE_PATH}" -DINSTALL_CMOD="" -DINSTALL_LMOD="" -DCMAKE_INSTALL_PREFIX="" -DCMAKE_BUILD_TYPE=release .. + make install DESTDIR=$HEKA_MODS +fi + cd $BASE/build/heka/build case $UNAME in diff --git a/heka/sandbox/inputs/systemd.lua b/heka/sandbox/inputs/systemd.lua new file mode 100644 index 0000000..beb7c89 --- /dev/null +++ b/heka/sandbox/inputs/systemd.lua @@ -0,0 +1,259 @@ +-- 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/. + +--[[ +Input to read entries from the systemd journal + +Config: + matches = "[]" + embedded JSON array of matches to apply. See + http://www.freedesktop.org/software/systemd/man/journalctl.html for a + description of matches. By default, all journal entries that the user + running Heka has access to are matched. + + use_fields = true + whether to map systemd fields to top-level Heka fields + e.g. _HOSTNAME -> Fields[Hostname] etc. This is superseded by + process_module_entry_point. + + process_module = nil + module to load (if any) to further transform messages before injecting + (similar to a decoder) + + process_module_entry_point = nil + method to call handing off the current table for further decoding. It is + expected that this method call inject_message(). + + offset_method = "manual_oldest" + The method used to determine at which offset to begin consuming messages. + The valid values are: + + - *manual_oldest* + Heka will track the offset and resume from where it last left off, or + else from the beginning of the journal if no checkpoint file exists + (default). + - *manual_newest* + Heka will track the offset and resume from where it last left off, or + else from the end of the journal if no checkpoint file exists. + - *newest* + Heka will start reading from the most recent available offset. + - *oldest* + Heka will start reading from the oldest available offset. + + offset_file = nil (required if offset_method is "manual_oldest" or "manual_newest") + File to store the checkpoint in. Must be unique. Currently the sandbox API + does not provide access to Logger information so this file must be + specified explicitly. + +*Example Heka Configuration* + +.. code-block:: ini + + [SystemdInput] + type = "SandboxInput" + filename = "lua_inputs/systemd.lua" + + [SystemdInput.config] + matches = '["_SYSTEMD_UNIT=fxa-auth.service"]' + offset_method = "newest" +--]] + +require "io" +require "os" +require "cjson" +local s = require "string" +local format = s.format +local dbg = require "debug" +local function debug (...) + dbg.debug("SystemdInput: " .. format(...)) +end + +local sj = require "systemd.journal" + +-- TODO support disjunction +local matches = cjson.decode(read_config("matches") or "[]") + +local offset_method = read_config("offset_method") or "manual_oldest" +local cursor, checkpoint_file +if offset_method == "manual_newest" or offset_method == "manual_oldest" then + checkpoint_file = read_config("offset_file") or error("must specify offset_file") +elseif offset_method and not ( + offset_method == "newest" or offset_method == "oldest") then + error(format("offset_method must be one of '%s', '%s', or '%s'", + "manual_newest", "manual_oldest", "newest", "oldest")) +end + +-- Try to format systemd's default MESSAGE_ID field into a format heka +-- supports. *Note* this value is potentially user-supplied and might be +-- garbage. +local function format_uuid(str) + local s = "" + local clean = str:gsub("-", "") + if not clean:match("^%x+$") then return str end + for i in string.gmatch(clean, "(..)") do + s = s .. string.char(tonumber(i, 16)) + end + return s +end + +-- This is a cursory attempt at making a more "heka-ish" +-- message. process_module_entry_point can be used to for arbitrary mappings +-- and transformations. +local fields_map = { + MESSAGE = "Payload" + , MESSAGE_ID = { + "Uuid", + function(x) format_uuid(x) end + } + , _HOSTNAME = "Hostname" + , _SOURCE_REALTIME_TIMESTAMP = { + "Timestamp", + function (i) return i * 1e3 end + } + , __REALTIME_TIMESTAMP = { + "Timestamp", + function (i) return i * 1e3 end + } + , _PID = "PID" + , PRIORITY = "Severity" + , _SYSTEMD_UNIT = "Logger" + , _SYSTEMD_USER_UNIT = "Logger" + , SYSLOG_IDENTIFIER = "Logger" +} + +-- see +-- http://www.freedesktop.org/software/systemd/man/systemd.journal-fields.html +-- for a list of fields. Currently "any-wins" semantics when multiple journal +-- fields map to a single heka one since tables are traversed in arbitrary +-- order. +local function map_fields(tbl) + for key, value in pairs(fields_map) do + local newkey, method + if type(value) == "table" then + newkey, method = unpack(value) + else + newkey = value + end + + local v = tbl.Fields[key] + if v then + if method then + tbl[newkey] = method(v) + else + tbl[newkey] = v + end + tbl.Fields[key] = nil + end + end + inject_message(tbl) + return 0 +end + +local mod = read_config("process_module") +local decoder_name = read_config("process_module_entry_point") +if mod and not decoder_name then + error(format( + "must provide process_module_entry_point for module %s", mod)) +elseif mod then + mod = require(mod) + decoder = mod[decoder_name] or error( + format("can't find function %s in module %s", + decoder_name, mod)) +elseif read_config("use_fields") then + decoder = map_fields + decoder_name = "map_fields" +end + +function process_message() + local j = assert(sj.open()) + j:set_data_threshold(0) + local checkpoint = offset_method:match("manual") + + local fh, cursor + if checkpoint then + fh = io.open(checkpoint_file, "r") + if fh then + -- the systemd cursor is variable length, so we write out the cursor + -- with a newline to avoid having to truncate the checkpoint file + cursor = fh:read("*line") + if cursor then + debug("cursor: %s", cursor) + else + debug("empty checkpoint file") + end + fh:close() + fh = assert(io.open(checkpoint_file, "r+")) + else + debug("checkpoint file %s doesn't exist, creating", + checkpoint_file) + fh = assert(io.open(checkpoint_file, "w+")) + end + fh:setvbuf("no") + end + + for i, match in ipairs(matches) do + debug("adding match %s", match) + assert(j:add_match(match)) + end + + debug("using offset method '%s'", offset_method) + if cursor then + if not j:seek_cursor(cursor) then + fh:close() + os.remove(checkpoint_file) + error(format("failed to seek to cursor %s, removing checkpoint and stopping", cursor)) + end + -- Note that [seek_cursor] does not actually make any entry the new + -- current entry, this needs to be done in a separate step with a + -- subsequent sd_journal_next(3) invocation (or a similar call) + j:next() + + -- maybe issue a warning instead if this fails + assert(j:test_cursor(cursor)) + elseif offset_method:match("newest") then + assert(j:seek_tail()) + j:previous() + elseif offset_method:match("oldest") then + assert(j:seek_head()) + j:next() + end + + local ready = j:next() + + while true do + while not ready do + if j:wait(1) ~= sj.WAKEUP.NOP then + ready = j:next() + end + end + + local cursor = j:get_cursor() + + local msg = { + Type = "heka.systemd", + Timestamp = os.time() * 1e9, + Fields = j:to_table() + } + if decoder then + local r, err = decoder(msg) + if r ~= 0 then + debug("%s failed: %s", decoder_name, err or "") + end + else + inject_message(msg) + end + + if checkpoint then + local offset, err = fh:seek("set") + if err then error(err) end + assert(offset == 0) + fh:write(cursor .. "\n") + fh:flush() + end + + ready = j:next() + end + fh:close() + return 0 +end diff --git a/heka/sandbox/modules/debug.lua b/heka/sandbox/modules/debug.lua new file mode 100644 index 0000000..d84451a --- /dev/null +++ b/heka/sandbox/modules/debug.lua @@ -0,0 +1,56 @@ +-- 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/. + +-- Simple debug module for lua input and output modules to provide comparable +-- functionality to go plugin logging facilities. + +-- To see debug output, add configuration like: +-- [PayloadEncoder] +-- [LogOutput] +-- message_matcher = "Type == 'heka.debug'" +-- encoder = "PayloadEncoder" + +--[[ +API +^^^ +**debug(payload, type, ns)** + + Emit a debug message. + + *Arguments* + - payload (any type that supports tostring) + Value to use in Payload field, converted with tostring. + - type (string or nil) + String to use in the `Type` field. Defaults to "heka.debug". + - ns (number or nil) + Nanosecond timestamp to use for any strftime field interpolation + into the above fields. Current system time will be used if nil. + +--]] + +local os = require "os" +local tostring = tostring +local inject_message = inject_message + +-- *NOTE* this may conflict with filter-specific configuration, and overrides +-- the lua debug library (not available from the sandbox anyway). +local dbg = read_config("debug") + +local M = {} +setfenv(1, M) -- Remove external access to contain everything in the module. + +--[[ Public Interface --]] + +function debug(msg, typ, ns) + if dbg then + local msg = { + Timestamp = ns or os.time(), + Type = typ or "heka.debug", + Payload = tostring(msg) + } + inject_message(msg) + end +end + +return M From 78711091fd9439cc613f962341349cc5b115a3df Mon Sep 17 00:00:00 2001 From: Wesley Dawson Date: Mon, 29 Feb 2016 14:34:23 -0800 Subject: [PATCH 2/2] Try to fix travis for systemd input --- .travis.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.travis.yml b/.travis.yml index bab1e4e..1adc065 100644 --- a/.travis.yml +++ b/.travis.yml @@ -9,7 +9,7 @@ before_install: - sudo add-apt-repository ppa:andykimpe/cmake3 -y - sudo add-apt-repository ppa:maxmind/ppa -y - sudo apt-get update -qq - - sudo apt-get install -y protobuf-compiler cmake libgeoip-dev libpq-dev + - sudo apt-get install -y protobuf-compiler cmake libgeoip-dev libpq-dev lua5.1-dev install: - bash bin/build_pipeline_heka.sh