Keen IO data collector for Elasticsearch

crontab

*/10 * * * * /usr/local/bin/keenio_collector.rb

keenio_collector.rb

#!/usr/bin/env ruby
# -*- coding: utf-8 -*-

require "net/https"
require "uri"
require "json"
require "time"

KEENIO_PROJECT_ID = "<Keen IO Project ID>"
KEENIO_READ_KEY = "<Keen IO Read Key>"
EVENT_COLLECTION = "PageView"
TIMEFRAME = "previous_10_minutes"

ELASTICSEARCH_URL = "http://localhost:9200";
TAG_KEY = "@log_name"
TAG_VALUE = "keen.io.pageview"
TYPE = "keen.io"

url = "https://api.keen.io/3.0/projects/#{KEENIO_PROJECT_ID}/queries/extraction?api_key=#{KEENIO_READ_KEY}&event_collection=#{EVENT_COLLECTION}&timeframe=#{TIMEFRAME}"
uri = URI.parse(url)

client = Net::HTTP.new(uri.host, uri.port)
client.use_ssl = true
response = nil
client.start do |http|
    response = http.get(uri.request_uri)
end

results = JSON.parse(response.body)
results[EVENT_COLLECTION] = results["result"]
results.delete("result")

results[EVENT_COLLECTION].each do |result|
    result[TAG_KEY] = TAG_VALUE
    timestamp = Time.parse(result["keen"]["timestamp"])
    result["@timestamp"] = timestamp.strftime("%Y-%m-%dT%H:%M:%S%:z")
    if !result["geoip"]["lon"].nil? and !result["geoip"]["lat"].nil? then
        result["geoip"]["coords"] = [result["geoip"]["lon"], result["geoip"]["lat"]]
        result["geoip"].delete("lon")
        result["geoip"].delete("lat")
    end
    index = "logstash-#{timestamp.strftime("%Y.%m.%d")}"
    url = "#{ELASTICSEARCH_URL}/#{index}/#{TYPE}/"
    uri = URI.parse(url)
    client = Net::HTTP.new(uri.host, uri.port)
    client.start do |http|
        http.post(uri.request_uri, JSON.dump(result))
        p "Sent event on #{timestamp}"
    end
end