Logentries token Rsyslog-forwarder with auto-created hosts/logs

logentries_tokenizer

#!/usr/bin/env ruby
#encoding utf-8
require 'logger'
require 'net/http'
require 'socket'
require 'json'

ACCOUNT_KEY=''
LOG_FILE = '/var/log/logentries_tokenizer.log'
UDP_HOST = '127.0.0.1'
UDP_PORT = ARGV[0].to_i

logger = Logger.new LOG_FILE
logger.level = Logger::INFO

#in case we have an earlier ruby
unless [].respond_to? :to_h
  class Array
    def to_h
      Hash[*self.flatten]
    end
  end
end

class LogentriesAPI
  def initialize(account_key)
    @account_key = account_key

    @get_uri_base = 'http://api.logentries.com/' + @account_key
    @post_uri = URI('http://api.logentries.com')
  end

  def create_host(host)
    request = {
      'request' => 'register',
      'user_key'=> @account_key,
      'name'=> host,
      'distver' => '',
      'system' => '',
      'distname' => ''
    }

    hosts_logs[host] = {}
    @hosts[host] = post_json(request)['host_key']
  end

  def create_log(host, log)
    request = {
      'request' => 'new_log',
      'user_key'=> @account_key,
      'host_key' => hosts[host],
      'name' => log,
      'type' => '',
      'retention' => '-1',
      'source' => 'token'
    }

    hosts_logs[host][log] = post_json(request)['log']['token']
  end

  # returns a mapping (host name)
  def hosts
    @hosts ||= hosts!
  end

  # returns a mapping (host name) -> (host key)
  # bypasses memoization
  def hosts!
    @hosts =
      get_list('/hosts').map do |h|
      [h['name'], h['key']]
      end.to_h
  end

  # returns a mapping (host name) -> (log name) -> (log token)
  def hosts_logs
    @hosts_logs ||= hosts_logs!
  end

  # returns a mapping (host name) -> (log name) -> (log token)
  # bypasses log memoization (uses memoized hosts)
  def hosts_logs!
    @hosts_logs =
      hosts.map do |h,k|
      logs = get_list("/hosts/#{k}/")
      [h, logs.map {|l| [l['name'], l['token']] }.to_h]
      end.to_h
  end

  private
  def post_json(request_obj)
    res = Net::HTTP.post_form(@post_uri, request_obj)
    raise IOError, "HTTP status code: #{res.code} - #{res.message}; we sent #{request_obj.inspect}" unless res.is_a?(Net::HTTPSuccess)
    j = JSON.parse(res.body)
    raise ArgumentError, "API returned non-ok #{j.inspect}; we sent #{request_obj.inspect}" unless j['response'] == 'ok'
    j
  end

  def get_list(path)
    path[/^\/?/] = '/'
    uri = URI(@get_uri_base + path)

    res = Net::HTTP.get_response(uri)

    if res.is_a?(Net::HTTPSuccess)
      j = JSON.parse(res.body)
      raise ArgumentError, "API returned non-ok #{j.inspect} requesting #{path}" unless j['response'] == 'ok'
      j['list']
    else
      raise IOError, "HTTP status code: #{res.code} - #{res.message} requesting #{path}"
    end
  end
end

begin
  le = LogentriesAPI.new(ACCOUNT_KEY)
  socket = UDPSocket.new
  socket.connect(UDP_HOST, UDP_PORT)

  while raw = $stdin.gets
    /^.* (?<host>[a-z][a-z0-9-]+)-(?<id>[0-9a-f]+) (?<log>[\w\.]+) (?<line>.*$)/ =~ raw

    if host.empty? || id.empty? || log.empty?
      logger.warn("Failed to parse line: '#{raw}'")
      next
    end

    unless le.hosts.has_key? host
      host_key = le.create_host(host)
      logger.info("Created host #{host} with key #{host_key}")
    end

    token = le.hosts_logs[host][log]
    if token
      logger.debug("fetched token for #{host}:#{log} = #{token}")
    else
      token = le.create_log(host, log)
      logger.info("created log #{host}/#{log} = #{token}")
    end

    socket.send "#{token} #{id}: #{line}\n", 0
  end
rescue Exception => e
  logger.fatal(e)
end

rsyslog.d_1-logentries.conf

module(load="imtcp")
module(load="imudp")
module(load="omprog")

template(name="Verbatim" type="string" string="%rawmsg%")

ruleset (name="translate_to_tokens"){
  action(type="omprog" binary="/usr/local/bin/logentries_tokenizer 10001" template="RSYSLOG_TraditionalFileFormat")
}

ruleset (name="forward_to_logentries"){
  action(type="omfwd" Target="data.logentries.com" Port="10000" Protocol="tcp" Template="Verbatim")
}

input(type="imudp" address="127.0.0.1" port="10001" ruleset="forward_to_logentries")
input(type="imtcp" port="10000" ruleset="translate_to_tokens")