Query ElasticSearch via Ruby

es-query.rb

#!/usr/bin/ruby
require 'json'
require 'elasticsearch'
#My complex ES query, Basically detects port scanning via our network sflow data
json_search = '{
  "query": {
    "filtered": {
      "query": {
        "query_string": {
          "query": "*",
          "analyze_wildcard": true
        }
      },
      "filter": {
        "bool": {
          "must": [
            {
              "query": {
                "query_string": {
                  "query": "*",
                  "analyze_wildcard": true
                }
              }
            },
            {
              "range": {
                "@timestamp": {
                  "gt": "now-48h"
                }
              }
            }
          ],
          "must_not": []
        }
      }
    }
  },
  "size": 0,
  "aggs": {
    "3": {
      "terms": {
        "field": "sflow_tcp_dst_port",
        "size": 5,
        "order": {
          "2": "desc"
        }
      },
      "aggs": {
        "2": {
          "cardinality": {
            "field": "sflow_ipv4_dst"
          }
        },
        "4": {
          "terms": {
            "field": "sflow_ipv4_src",
            "size": 5,
            "order": {
              "2": "desc"
            }
          },
          "aggs": {
            "2": {
              "cardinality": {
                "field": "sflow_ipv4_dst"
              }
            }
          }
        }
      }
    }
  }
}'
#Connect to ES
client = Elasticsearch::Client.new host: '192.168.10.101', log: true
client.transport.reload_connections!
#Run the search off above JSON data
req = client.search body: JSON.parse(json_search)
#Test working with returned HASH
req['aggregations']['3']['buckets'].each do |level1|
      puts "\n\nStats for Port Number #{level1['key']}"
      level1['4']['buckets'].each do |level2|
        puts "Host #{level2['key_as_string']} has connected to #{level2['2']['value']} uniq destinations"
      end
end