From 7c334afc9dd2768ec7025d06ae1c533ee41f61b8 Mon Sep 17 00:00:00 2001 From: Denis Zhuk Date: Tue, 9 May 2017 17:32:41 +0200 Subject: [PATCH 1/2] Fix connection timeout --- README.md | 40 +++++++++++++++++++++++++++++++++------- lib/druid/client.rb | 3 ++- lib/druid/data_source.rb | 23 ++++++++++++----------- 3 files changed, 47 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 8534b45..aabf363 100644 --- a/README.md +++ b/README.md @@ -29,22 +29,48 @@ gem install ruby-druid ## Usage -A query can be constructed and sent like so: +1. Connect: ```ruby -data_source = Druid::Client.new('zk1:2181,zk2:2181/druid').data_source('service/source') -query = Druid::Query::Builder.new.long_sum(:aggregate1).last(1.day).granularity(:all) -result = data_source.post(query) +client = Druid::Client.new('zk1:2181,zk2:2181/druid', opts) +datasource = client.data_source('druid:broker/datasource_name') ``` -The `post` method on the `DataSource` returns the parsed response from the Druid server as an array. +if broker is behind of load balancer you can connect to static host without service discovery: -If you don't want to use ZooKeeper for broker discovery, you can explicitly construct a `DataSource`: ```ruby -data_source = Druid::DataSource.new('service/source', 'http://localhost:8080/druid/v2') +datasource = Druid::DataSource.new('datasource_name', 'http://broker-host:8080/druid/v2/', opts) ``` +`opts` is an optional hash of connection options: + +| key | description | type | default | +| ------------------- | -------------------------------------------------- | ------ | ------------ | +| :connection_timeout | connection timeout for druid services (in seconds) | int | 60 | +| :read_timeout | read timeout for druid services (in seconds) | int | nil | +| :discovery_path | druid service discovery path in zookeeper | string | '/discovery' | + +3. Create query: + +```ruby +query = Druid::Query::Builder.new +``` +4. Build query, e.g.: +```ruby +query.granularity(:all) +query.long_sum(:aggregate1) +# .... +``` + +5. Send request: + +```ruby +result = datasource.post(query) +``` + +The `post` method returns the parsed response from the druid server as an array. If the response is not empty it contains one `ResponseRow` object for each row. The timestamp by can be received by a method with the same name (i.e. `row.timestamp`), all row values by hashlike syntax (i.e. `row['dimension']) + ### GroupBy A [GroupByQuery](http://druid.io/docs/latest/querying/groupbyquery.html) sets the diff --git a/lib/druid/client.rb b/lib/druid/client.rb index 1462cff..c050c6d 100644 --- a/lib/druid/client.rb +++ b/lib/druid/client.rb @@ -7,12 +7,13 @@ class Client attr_reader :zk def initialize(zookeeper, opts = {}) + @opts = opts @zk = ZK.new(zookeeper, opts) end def data_source(source) uri = @zk.data_sources[source] - Druid::DataSource.new(source, uri) + Druid::DataSource.new(source, uri, @opts) end def data_sources diff --git a/lib/druid/data_source.rb b/lib/druid/data_source.rb index dbfe543..0a2c04d 100644 --- a/lib/druid/data_source.rb +++ b/lib/druid/data_source.rb @@ -6,8 +6,10 @@ class DataSource attr_reader :name, :uri, :metrics, :dimensions - def initialize(name, uri) + def initialize(name, uri, opt = {}) @name = name.split('/').last + @connection_timeout = opt[:connection_timeout] || 10 # if druid is down fail fast + @read_timeout = opt[:read_timeout] # we wait until druid is finished uri = uri.sample if uri.is_a?(Array) if uri.is_a?(String) @uri = URI(uri) @@ -32,11 +34,11 @@ def metadata!(opts = {}) end req = Net::HTTP::Get.new(meta_path) - response = Net::HTTP.new(uri.host, uri.port).start do |http| - http.open_timeout = 10 # if druid is down fail fast - http.read_timeout = nil # we wait until druid is finished - http.request(req) - end + + http = Net::HTTP.new(uri.host, uri.port) + http.open_timeout = @connection_timeout + http.read_timeout = @read_timeout + response = http.start { |h| h.request(req) } if response.code != '200' raise "Request failed: #{response.code}: #{response.body}" @@ -61,11 +63,10 @@ def post(query) req = Net::HTTP::Post.new(uri.path, { 'Content-Type' => 'application/json' }) req.body = query.to_json - response = Net::HTTP.new(uri.host, uri.port).start do |http| - http.open_timeout = 10 # if druid is down fail fast - http.read_timeout = nil # we wait until druid is finished - http.request(req) - end + http = Net::HTTP.new(uri.host, uri.port) + http.open_timeout = @connection_timeout + http.read_timeout = @read_timeout + response = http.start { |h| h.request(req) } if response.code != '200' # ignore GroupBy cache issues and try again without cached results From 1b07fdc89dd8259c164933c74a26481e678f0a90 Mon Sep 17 00:00:00 2001 From: Denis Zhuk Date: Tue, 9 May 2017 18:07:48 +0200 Subject: [PATCH 2/2] Rename connection option to --- README.md | 2 +- lib/druid/data_source.rb | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index aabf363..217312c 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ datasource = Druid::DataSource.new('datasource_name', 'http://broker-host:8080/d | key | description | type | default | | ------------------- | -------------------------------------------------- | ------ | ------------ | -| :connection_timeout | connection timeout for druid services (in seconds) | int | 60 | +| :open_timeout | open timeout for druid services (in seconds) | int | 60 | | :read_timeout | read timeout for druid services (in seconds) | int | nil | | :discovery_path | druid service discovery path in zookeeper | string | '/discovery' | diff --git a/lib/druid/data_source.rb b/lib/druid/data_source.rb index 0a2c04d..fdfe3ae 100644 --- a/lib/druid/data_source.rb +++ b/lib/druid/data_source.rb @@ -8,7 +8,7 @@ class DataSource def initialize(name, uri, opt = {}) @name = name.split('/').last - @connection_timeout = opt[:connection_timeout] || 10 # if druid is down fail fast + @open_timeout = opt[:open_timeout] || 10 # if druid is down fail fast @read_timeout = opt[:read_timeout] # we wait until druid is finished uri = uri.sample if uri.is_a?(Array) if uri.is_a?(String) @@ -36,7 +36,7 @@ def metadata!(opts = {}) req = Net::HTTP::Get.new(meta_path) http = Net::HTTP.new(uri.host, uri.port) - http.open_timeout = @connection_timeout + http.open_timeout = @open_timeout http.read_timeout = @read_timeout response = http.start { |h| h.request(req) } @@ -64,7 +64,7 @@ def post(query) req.body = query.to_json http = Net::HTTP.new(uri.host, uri.port) - http.open_timeout = @connection_timeout + http.open_timeout = @open_timeout http.read_timeout = @read_timeout response = http.start { |h| h.request(req) }