From 75519b51ea0666f0d209c3da68bb52015e4b1761 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Fri, 26 Jan 2018 18:25:24 -0500 Subject: [PATCH 1/8] batch together points waiting in queue, add async parameters, and improve testing --- lib/stathat.rb | 69 ++++++++++++++++++++++------ test/test_stathat.rb | 106 ++++++++++++++++++++++++++++++------------- 2 files changed, 130 insertions(+), 45 deletions(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index 711fb32..4c0704c 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -10,6 +10,7 @@ class Common CLASSIC_VALUE_URL = "https://api.stathat.com/v" CLASSIC_COUNT_URL = "https://api.stathat.com/c" EZ_URL = "https://api.stathat.com/ez" + EZ_URI = URI(EZ_URL) class << self def send_to_stathat(url, args) @@ -25,6 +26,15 @@ def send_to_stathat(url, args) resp = Net::HTTP.get(uri) return Response.new(resp) end + + def send_ez_batch_to_stathat(batch_args, ezkey) + resp = Net::HTTP.start(EZ_URI.host, EZ_URI.port, :use_ssl => true) do |http| + http.post EZ_URI.path, + { :ezkey => ezkey, :data => batch_args }.to_json, + "Content-Type" => "application/json" + end + return Response.new(resp.body) + end end end @@ -66,19 +76,23 @@ def post_value(stat_key, user_key, value, timestamp=nil) class API class << self - def ez_post_value(stat_name, ezkey, value, timestamp=nil, &block) + attr_accessor :max_batch + attr_accessor :pool_size + attr_accessor :max_queue_size + + def ez_post_value(stat_name, ezkey, value, timestamp=Time.now.to_i, &block) Reporter.instance.ez_post_value(stat_name, ezkey, value, timestamp, block) end - def ez_post_count(stat_name, ezkey, count, timestamp=nil, &block) + def ez_post_count(stat_name, ezkey, count, timestamp=Time.now.to_i, &block) Reporter.instance.ez_post_count(stat_name, ezkey, count, timestamp, block) end - def post_count(stat_key, user_key, count, timestamp=nil, &block) + def post_count(stat_key, user_key, count, timestamp=Time.now.to_i, &block) Reporter.instance.post_count(stat_key, user_key, count, timestamp, block) end - def post_value(stat_key, user_key, value, timestamp=nil, &block) + def post_value(stat_key, user_key, value, timestamp=Time.now.to_i, &block) Reporter.instance.post_value(stat_key, user_key, value, timestamp, block) end end @@ -134,19 +148,48 @@ def ez_post_count(stat_name, ezkey, count, timestamp, cb) def run_pool @runlock.synchronize { @running = true } @pool = [] - 5.times do |i| + (API.pool_size || 5).times do |i| @pool[i] = Thread.new do while true do - point = @que.pop - # XXX check for error? - begin - resp = Common::send_to_stathat(point[:url], point[:args]) - if point[:cb] - point[:cb].call(resp) + points = [@que.pop] + puts "Dropping StatHat queue" if API.max_queue_size && @que.length > API.max_queue_size + + while points.size < (API.max_batch || 10) && @que.length > 0 + points << @que.pop(true) rescue ThreadError + end + + groups = points.group_by{|point| point[:args][:ezkey]} + groups.each do |ezkey, batch| + if ezkey.nil? + batch.each do |point| + # XXX check for error? + begin + resp = Common::send_to_stathat(point[:url], point[:args]) + if point[:cb] + point[:cb].call(resp) + end + rescue + p $! + end + end + + else + batch_args = batch.map{|point| point[:args]} + batch_args.each{|args| args.delete(:ezkey)} + + begin + resp = Common::send_ez_batch_to_stathat(batch_args, ezkey) + batch.each do |point| + if point[:cb] + point[:cb].call(resp) + end + end + rescue + p $! end - rescue - pp $! + end end + @runlock.synchronize { break unless @running } diff --git a/test/test_stathat.rb b/test/test_stathat.rb index 76140d9..56a1552 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -2,68 +2,110 @@ class TestStathat < MiniTest::Unit::TestCase - def test_ez_value - StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92) do |resp| - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = wait_for_resp do |cb| + StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92, &cb) end - sleep(1) + assert_success(r) end def test_ez_count - StatHat::API.ez_post_value("test ez count stat", "test@stathat.com", 12) do |r| - assert(r.valid?, "response was invalid") - assert_equal(r.msg, "ok", "message should be 'ok'") - assert_equal(r.status, 200, "status should be 200") + r = wait_for_resp do |cb| + StatHat::API.ez_post_count("test ez count stat", "test@stathat.com", 12, &cb) end - sleep(1) + assert_success(r) end def test_classic_count_bad_keys - StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 12) do |r| - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + r = wait_for_resp do |cb| + StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 12, &cb) end - sleep(1) + assert_failure(r) end def test_classic_value_bad_keys - StatHat::API.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92) do |r| - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + r = wait_for_resp do |cb| + StatHat::API.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92, &cb) end - sleep(1) + assert_failure(r) end def test_ez_value_sync - resp = StatHat::SyncAPI.ez_post_value("test ez value stat", "test@stathat.com", 0.92) - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = StatHat::SyncAPI.ez_post_value("test ez value stat", "test@stathat.com", 0.92) + assert_success(r) end def test_ez_count_sync - resp = StatHat::SyncAPI.ez_post_value("test ez count stat", "test@stathat.com", 12) - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = StatHat::SyncAPI.ez_post_count("test ez count stat", "test@stathat.com", 12) + assert_success(r) end def test_classic_count_bad_keys_sync r = StatHat::SyncAPI.post_count("XXXXXXXX", "YYYYYYYY", 12) - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + assert_failure(r) end def test_classic_value_bad_keys_sync r = StatHat::SyncAPI.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92) + assert_failure(r) + end + + def test_ez_batch + StatHat::API.pool_size = 1 + StatHat::API.max_batch = 2 + + final = wait_for_resp do |cb| + StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 11) do |r| + assert_failure(r) + end + + StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92) do |r| + assert_success(r) + end + + StatHat::API.ez_post_count("test ez count stat", "test@stathat.com", 12) do |r| + assert_success(r) + end + + StatHat::API.ez_post_count("test ez count stat2", "test@stathat.com", 13, &cb) + end + + assert_success(final) + end + + private + + def assert_success(r) + assert(r.valid?, "response was invalid") + assert_equal(r.msg, "ok", "message should be 'ok'") + assert_equal(r.status, 200, "status should be 200") + end + + def assert_failure(r) assert_equal(r.valid?, false, "response was valid") assert_equal(r.msg, "invalid keys", "incorrect error message") assert_equal(r.status, 500, "incorrect status code") end + + def wait_for_resp + m = Mutex.new + cv = ConditionVariable.new + resp = nil + + cb = lambda do |r| + resp = r + m.synchronize do + cv.signal + end + end + + m.synchronize do + yield cb + start = Time.now + assert cv.wait(m, 60) + assert Time.now - start < 50 + end + + resp + end end From 346579af8b3b554a56b1d72c35c4e7820720c886 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Fri, 26 Jan 2018 19:33:35 -0500 Subject: [PATCH 2/8] actually drop queue at maximum size, add sleep option to enable waiting for events to batch --- lib/stathat.rb | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index 4c0704c..48f8f7d 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -79,6 +79,7 @@ class << self attr_accessor :max_batch attr_accessor :pool_size attr_accessor :max_queue_size + attr_accessor :batch_sleep_seconds def ez_post_value(stat_name, ezkey, value, timestamp=Time.now.to_i, &block) Reporter.instance.ez_post_value(stat_name, ezkey, value, timestamp, block) @@ -152,9 +153,7 @@ def run_pool @pool[i] = Thread.new do while true do points = [@que.pop] - puts "Dropping StatHat queue" if API.max_queue_size && @que.length > API.max_queue_size - - while points.size < (API.max_batch || 10) && @que.length > 0 + while points.size < (API.max_batch || 1) && @que.length > 0 points << @que.pop(true) rescue ThreadError end @@ -193,6 +192,8 @@ def run_pool @runlock.synchronize { break unless @running } + + sleep API.batch_sleep_seconds unless API.batch_sleep_seconds.nil? end end end @@ -209,6 +210,10 @@ def stop_pool() def enqueue(url, args, cb=nil) return false unless @running + if API.max_queue_size && @que.length > API.max_queue_size + puts "Dropping StatHat queue" + @que.clear + end point = {:url => url, :args => args, :cb => cb} @que << point true From 7d398b04e17c4d2eef260240abee9cad39eb32fa Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Mon, 29 Jan 2018 10:41:18 -0500 Subject: [PATCH 3/8] initialize before pool is created in test --- test/test_stathat.rb | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/test/test_stathat.rb b/test/test_stathat.rb index 56a1552..a117aa0 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -2,6 +2,11 @@ class TestStathat < MiniTest::Unit::TestCase + def setup + StatHat::API.pool_size = 1 + StatHat::API.max_batch = 2 + end + def test_ez_value r = wait_for_resp do |cb| StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92, &cb) @@ -51,9 +56,6 @@ def test_classic_value_bad_keys_sync end def test_ez_batch - StatHat::API.pool_size = 1 - StatHat::API.max_batch = 2 - final = wait_for_resp do |cb| StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 11) do |r| assert_failure(r) From 31c5470e5b6999161bdc64ea4fd3684d18fb2670 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Mon, 29 Jan 2018 11:19:31 -0500 Subject: [PATCH 4/8] support 204 no content response as per https://blog.stathat.com/2017/05/05/bandwidth.html --- lib/stathat.rb | 21 +++++++++++++-------- test/test_stathat.rb | 4 ++-- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index 48f8f7d..5469fa6 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -23,7 +23,7 @@ def send_to_stathat(url, args) uri.query = args.map { |arg, val| arg.to_s + "=" + CGI::escape(val.to_s) }.join('&') end - resp = Net::HTTP.get(uri) + resp = Net::HTTP.get_response(uri) return Response.new(resp) end @@ -33,7 +33,7 @@ def send_ez_batch_to_stathat(batch_args, ezkey) { :ezkey => ezkey, :data => batch_args }.to_json, "Content-Type" => "application/json" end - return Response.new(resp.body) + return Response.new(resp) end end end @@ -221,29 +221,34 @@ def enqueue(url, args, cb=nil) end class Response - def initialize(body) - @body = body + def initialize(resp) + @resp = resp @parsed = nil end + def success? + return valid? && (@resp.body.nil? || msg == "ok") + end + def valid? - return status == 200 + return @resp.body.nil? ? @resp.kind_of?(Net::HTTPSuccess) : status == 200 end def status parse - return @parsed['status'] + return @resp.body.nil? ? @resp.status : @parsed['status'] end def msg parse - return @parsed['msg'] + return @resp.body.nil? ? nil : @parsed['msg'] end private def parse return unless @parsed.nil? - @parsed = JSON.parse(@body) + return if @resp.body.nil? + @parsed = JSON.parse(@resp.body) end end end diff --git a/test/test_stathat.rb b/test/test_stathat.rb index a117aa0..65592c0 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -79,14 +79,14 @@ def test_ez_batch def assert_success(r) assert(r.valid?, "response was invalid") - assert_equal(r.msg, "ok", "message should be 'ok'") - assert_equal(r.status, 200, "status should be 200") + assert(r.success?, "response should be successful #{r.inspect}") end def assert_failure(r) assert_equal(r.valid?, false, "response was valid") assert_equal(r.msg, "invalid keys", "incorrect error message") assert_equal(r.status, 500, "incorrect status code") + refute(r.success?, "response should not be successful #{r.inspect}") end def wait_for_resp From 1e858fce4583fa5d7308a6b423ba580ae19ef2d3 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Mon, 29 Jan 2018 11:21:21 -0500 Subject: [PATCH 5/8] ruby warning for unused variable --- lib/stathat.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index 5469fa6..fd4dbd9 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -18,7 +18,7 @@ def send_to_stathat(url, args) begin uri.query = URI.encode_www_form(args) - rescue NoMethodError => e + rescue NoMethodError # backwards compatability for pre 1.9.x uri.query = args.map { |arg, val| arg.to_s + "=" + CGI::escape(val.to_s) }.join('&') end From 884f837f3740878cfb92cac8ddf546ce6297e52f Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Mon, 29 Jan 2018 11:44:48 -0500 Subject: [PATCH 6/8] let clients decide whether to prefer queue time discrepancies over local clock drift, give searchable string in puts --- lib/stathat.rb | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index fd4dbd9..56b9d17 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -81,19 +81,19 @@ class << self attr_accessor :max_queue_size attr_accessor :batch_sleep_seconds - def ez_post_value(stat_name, ezkey, value, timestamp=Time.now.to_i, &block) + def ez_post_value(stat_name, ezkey, value, timestamp=nil, &block) Reporter.instance.ez_post_value(stat_name, ezkey, value, timestamp, block) end - def ez_post_count(stat_name, ezkey, count, timestamp=Time.now.to_i, &block) + def ez_post_count(stat_name, ezkey, count, timestamp=nil, &block) Reporter.instance.ez_post_count(stat_name, ezkey, count, timestamp, block) end - def post_count(stat_key, user_key, count, timestamp=Time.now.to_i, &block) + def post_count(stat_key, user_key, count, timestamp=nil, &block) Reporter.instance.post_count(stat_key, user_key, count, timestamp, block) end - def post_value(stat_key, user_key, value, timestamp=Time.now.to_i, &block) + def post_value(stat_key, user_key, value, timestamp=nil, &block) Reporter.instance.post_value(stat_key, user_key, value, timestamp, block) end end @@ -168,7 +168,7 @@ def run_pool point[:cb].call(resp) end rescue - p $! + puts "Exception in StatHat encoding or callback #{$!.inspect}" end end @@ -184,7 +184,7 @@ def run_pool end end rescue - p $! + puts "Exception in StatHat encoding or callback #{$!.inspect}" end end end From d3b1dfede52efe52c66e886e51450626cf7cee65 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Fri, 16 Feb 2018 17:16:39 -0500 Subject: [PATCH 7/8] convert strings instead of silently failing, but raise if they arent numbers --- lib/stathat.rb | 16 ++++++++-------- test/test_stathat.rb | 11 +++++++++++ 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index 56b9d17..f1e187f 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -43,7 +43,7 @@ class << self def ez_post_value(stat_name, ezkey, value, timestamp=nil) args = { :stat => stat_name, :ezkey => ezkey, - :value => value } + :value => Float(value) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::EZ_URL, args) end @@ -51,7 +51,7 @@ def ez_post_value(stat_name, ezkey, value, timestamp=nil) def ez_post_count(stat_name, ezkey, count, timestamp=nil) args = { :stat => stat_name, :ezkey => ezkey, - :count => count } + :count => Integer(count) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::EZ_URL, args) end @@ -59,7 +59,7 @@ def ez_post_count(stat_name, ezkey, count, timestamp=nil) def post_count(stat_key, user_key, count, timestamp=nil) args = { :key => stat_key, :ukey => user_key, - :count => count } + :count => Integer(count) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::CLASSIC_COUNT_URL, args) end @@ -67,7 +67,7 @@ def post_count(stat_key, user_key, count, timestamp=nil) def post_value(stat_key, user_key, value, timestamp=nil) args = { :key => stat_key, :ukey => user_key, - :value => value } + :value => Float(value) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::CLASSIC_VALUE_URL, args) end @@ -82,19 +82,19 @@ class << self attr_accessor :batch_sleep_seconds def ez_post_value(stat_name, ezkey, value, timestamp=nil, &block) - Reporter.instance.ez_post_value(stat_name, ezkey, value, timestamp, block) + Reporter.instance.ez_post_value(stat_name, ezkey, Float(value), timestamp, block) end def ez_post_count(stat_name, ezkey, count, timestamp=nil, &block) - Reporter.instance.ez_post_count(stat_name, ezkey, count, timestamp, block) + Reporter.instance.ez_post_count(stat_name, ezkey, Integer(count), timestamp, block) end def post_count(stat_key, user_key, count, timestamp=nil, &block) - Reporter.instance.post_count(stat_key, user_key, count, timestamp, block) + Reporter.instance.post_count(stat_key, user_key, Integer(count), timestamp, block) end def post_value(stat_key, user_key, value, timestamp=nil, &block) - Reporter.instance.post_value(stat_key, user_key, value, timestamp, block) + Reporter.instance.post_value(stat_key, user_key, Float(value), timestamp, block) end end end diff --git a/test/test_stathat.rb b/test/test_stathat.rb index 65592c0..6445851 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -55,6 +55,17 @@ def test_classic_value_bad_keys_sync assert_failure(r) end + def test_string_value + r = wait_for_resp do |cb| + StatHat::API.ez_post_value("ZZZZZZZZ", "YYYYYYYYY", "0.92", &cb) + end + assert_success(r) + + assert_raises(ArgumentError) do + StatHat::API.ez_post_count("XXXXXXXX", "YYYYYYYY", "1.0") + end + end + def test_ez_batch final = wait_for_resp do |cb| StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 11) do |r| From f59578da9c5dbee397bd1176f5e87a273a146a56 Mon Sep 17 00:00:00 2001 From: Eric Jensen Date: Wed, 7 Mar 2018 11:04:48 -0500 Subject: [PATCH 8/8] add basic export support --- lib/stathat.rb | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/lib/stathat.rb b/lib/stathat.rb index f1e187f..21d29e7 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -13,7 +13,7 @@ class Common EZ_URI = URI(EZ_URL) class << self - def send_to_stathat(url, args) + def stathat_uri(url, args) uri = URI.parse(url) begin @@ -23,6 +23,11 @@ def send_to_stathat(url, args) uri.query = args.map { |arg, val| arg.to_s + "=" + CGI::escape(val.to_s) }.join('&') end + return uri + end + + def send_to_stathat(url, args) + uri = stathat_uri(url, args) resp = Net::HTTP.get_response(uri) return Response.new(resp) end @@ -38,6 +43,32 @@ def send_ez_batch_to_stathat(batch_args, ezkey) end end + class Export + class Error < StandardError + attr_reader :response + + def initialize(msg, response) + @response = response + super(msg) + end + end + + EXPORT_URL = "https://www.stathat.com/x" + + class << self + def get(access_token, paths, args) + uri = Common::stathat_uri(([EXPORT_URL, access_token] + paths).join('/'), args) + resp = Net::HTTP.get_response(uri) + + if resp.kind_of?(Net::HTTPSuccess) && resp.body + return JSON.parse(resp.body) + else + raise Error.new("export error #{resp.inspect} for #{uri.inspect}", resp) + end + end + end + end + class SyncAPI class << self def ez_post_value(stat_name, ezkey, value, timestamp=nil)