ls-pipeline-json.morden.conf

input {
    beats {
        port => "5043"
    }
}
filter {
    grok {
      match => { "to.udp_address" => "%{IP:to.host}:%{NUMBER:port}" }
      match => { "from.udp_address" => "%{IP:from.host}:%{NUMBER:port}" }
    }
    date {
        match => [ "block.time","UNIX" ]
        target => "block.TimeFromUnix"
    }
    geoip {
        source => "to.host"
        target => "geoip"
        add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
        add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
    }
    geoip {
        source => "from.host"
        target => "geoip"
        add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
        add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
    }
    mutate {
        convert => [
            "[geoip][coordinates]", "float",
            "object.balance", "string",
            "sub_balance.amount", "string",
            "add_balance.amount", "string"
        ]
    }
    ruby {
        code => "
            begin
                if event.get('event') != 'blockchain.write.block' then
                    return
                end
                block_time = DateTime.parse(Time.at(event.get('block.time')).to_s).to_time
                block_received_at = DateTime.parse(event.get('block.received_at')).to_time

                event.set('block.time_vs_received_elapsed', block_received_at - block_time)
            rescue Exception => e
                event.set('logstash_ruby_exception', 'elapsed: ' + e.message)
            end
        "
    }
    ruby {
        code => "
            require 'bigdecimal'
            begin
                eth_wei = BigDecimal.new('1e+18')
                ['object.balance', 'sub_balance.amount', 'add_balance.amount'].each do |f|
                    e = event.get(f)
                    if e.nil?
                        next
                    end
                    big_val = BigDecimal.new(e)
                    eth =  big_val.div(eth_wei,8)
                    event.set(f + '_eth', eth.to_s('F').to_f)
                end
            rescue Exception => e
                event.set('logstash_ruby_exception', '_eth: ' + e.message)
            end
        "
    }

}
output {
    stdout { }
    elasticsearch {
        hosts => "localhost:9200"
        index => "morden-%{+YYYY.MM.dd}"
    }
}