canary.io -> Heka lua decoder

canary-heka.lua

require "cjson"

-- Generic decoder for Canary JSON data. This will extract all JSON
-- keys and add them to the `Fields` variable of the created
-- Heka message.
--
-- Example use:
--
-- [CanaryDecoder]
-- type = "SandboxDecoder"
-- script_type = "lua"
-- filename = "/usr/share/heka/lua_decoders/canary.lua"
-- module_directory = "/usr/share/lua"
--
-- [HttpInput]
-- url = "https://api.canary.io/checks/https-github.com/measurements?range=10"
-- ticker_interval = 1
-- decoder = "CanaryDecoder"
--
-- [fileoutput]
-- type = "FileOutput"
-- message_matcher = "Fields[location] == 'do-ny2'"
-- path = "/var/log/heka/canary-output.log"
-- prefix_ts = true
-- perm = "666"

function process_message()
   local raw_message = read_message("Payload")

   local json = cjson.decode(raw_message)
   if not json then
     -- When plain text is found, ship it in it's raw form.
     inject_message(raw_message)
     return 0
   end

   for element = 1, #json do
     local cj2 = cjson.new()

     local fields = json[element]
     if fields.check then
       fields.check_id = fields.check.id
       fields.check_url = fields.check.url
       fields.check = nil
     end

     local message = {
       Type = "Canary",
       Payload = cj2.encode(json[element]),
       Fields = fields
     }

     inject_message(message)
   end
   return 0
end

Resulting log from the example above:

[2014/May/08:19:34:19 -0400] {"primary_ip":"192.30.252.129","starttransfer_time":0.081333,"id":"db871b93-b4b7-4d03-6e94-e1f137b59540","t":1399592053,"namelookup_time":0.041654,"local_ip":"107.170.68.66","location":"do-ny2","exit_status":0,"http_status":200,"check_id":"https-github.com","check_url":"https:\/\/github.com","connect_time":0.048234,"total_time":0.088734}
[2014/May/08:19:34:19 -0400] {"primary_ip":"192.30.252.129","starttransfer_time":0.081591,"id":"325a339c-8340-4019-43a6-b5e9b9d5dceb","t":1399592052,"namelookup_time":0.040701,"local_ip":"107.170.68.66","location":"do-ny2","exit_status":0,"http_status":200,"check_id":"https-github.com","check_url":"https:\/\/github.com","connect_time":0.047392,"total_time":0.089207}
[2014/May/08:19:34:19 -0400] {"primary_ip":"192.30.252.129","starttransfer_time":0.082788,"id":"fc44e487-7361-446f-5ae4-0bc3f89e3617","t":1399592051,"namelookup_time":0.041927,"local_ip":"107.170.68.66","location":"do-ny2","exit_status":0,"http_status":200,"check_id":"https-github.com","check_url":"https:\/\/github.com","connect_time":0.0486,"total_time":0.090782}
[2014/May/08:19:34:19 -0400] {"primary_ip":"192.30.252.129","starttransfer_time":0.082124,"id":"5022851f-6183-4c95-77f9-53c54ba48ac2","t":1399592050,"namelookup_time":0.041687,"local_ip":"107.170.68.66","location":"do-ny2","exit_status":0,"http_status":200,"check_id":"https-github.com","check_url":"https:\/\/github.com","connect_time":0.048542,"total_time":0.089086}