Retry failed requests inifinitely (#9)

Wait between retries in increasing steps.
Fixes #8

Author: matrixik

README.md

@@ -224,6 +224,11 @@ input {
 
 **Important:** The user which is sending the data on behalf of other tenants requires a specific role. This role is defined in the configuration file of the monasca-log-api as ```delegate_roles```.
 
+#### Exception handling
+
+This output plugin will sleep for maximum 60 seconds after the request to
+Monasca API fails and will retry the request until successful.
+
 ## Open tasks
 * Language translations (Replace hardcoded String messages with a configuration/language file)
-* Exception handling (monasca-api requests)
+* Unit tests for failed requests handling

lib/logstash/outputs/monasca_log_api.rb

@@ -209,6 +209,7 @@ def start_time_check
 
   def stop_time_check
     #ensure that entries buffered in queue will be  handled before stop
+    @stopping = true
     @mutex.synchronize do
       send_logs
     end
@@ -226,6 +227,7 @@ def send_logs
       token = LogStash::Outputs::Keystone::Token.instance
       @logger.debug("Sending #{@logs[JSON_LOGS].size} logs")
       retry_tries = 5
+      attempt = 0
       begin
         tries ||= retry_tries
         @monasca_log_api_client.send_logs(@logs, token.id, @cross_tenant)
@@ -241,10 +243,19 @@ def send_logs
         else
           @logger.error("Unauthorized: #{e}. Requesting new token failed "\
                         "after #{retry_tries} retries.")
+
+          sleep_for_retry(60)
+          retry
         end
       rescue => e
-        @logger.error('Sending event to monasca-log-api threw exception',
+        attempt += 1
+        sleep_for = get_sleep_time(attempt)
+        @logger.error("Sending event to monasca-log-api threw exception, "\
+                      "will sleep for #{sleep_for} seconds.",
                       :exceptionew => e)
+
+        sleep_for_retry(sleep_for)
+        retry
       end
       @logs.clear
       initialize_logs_object
@@ -270,4 +281,13 @@ def check_config
       raise LogStash::ConfigurationError, err
     end
   end
+
+  def get_sleep_time(attempt)
+    sleep_for = attempt**2
+    sleep_for <= 60 ? sleep_for : 60
+  end
+
+  def sleep_for_retry(duration)
+    Stud.stoppable_sleep(duration) { @stopping }
+  end
 end

logstash-output-monasca_log_api.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-output-monasca_log_api'
-  s.version         = '1.0.3'
+  s.version         = '1.0.4'
   s.licenses = ['Apache-2.0']
   s.summary = 'This gem is a logstash output plugin to connect via http to monasca-log-api.'
   s.description = 'This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program'
@@ -22,6 +22,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'logstash-core', '~> 2.0'
   s.add_runtime_dependency 'logstash-codec-plain', '~> 2.0'
   s.add_runtime_dependency 'logstash-codec-json', '~> 2.0'
+  s.add_runtime_dependency 'stud', '~> 0.0.22'
   s.add_development_dependency 'logstash-devutils', '~> 0.0.14'
   s.add_development_dependency 'simplecov', '~> 0.10'
   s.add_development_dependency 'simplecov-rcov', '~> 0.2.0'

spec/outputs/monasca_log_api_spec.rb

@@ -422,23 +422,6 @@
     end
   end
 
-  context 'when sending logs raise exception' do
-    it 'logs a failure' do
-      expect_any_instance_of(Cabin::Channel).to receive(:error)
-
-      monasca_log_api = LogStash::Outputs::MonascaLogApi.new(complete_config)
-      expect_any_instance_of(LogStash::Outputs::Monasca::MonascaLogApiClient)
-        .to receive(:send_logs)
-        .and_raise(Errno::ETIMEDOUT)
-      expect_any_instance_of(LogStash::Outputs::Keystone::KeystoneClient)
-        .to receive(:authenticate).and_return(valid_token)
-
-      monasca_log_api.register
-      monasca_log_api.multi_receive([event, event])
-      expect { monasca_log_api.multi_receive([event]) }.to_not raise_error
-    end
-  end
-
   context 'after sending logs' do
     it 'clears collected logs' do
       expect_any_instance_of(LogStash::Outputs::Monasca::MonascaLogApiClient)
@@ -480,42 +463,6 @@
       monasca_log_api.multi_receive([event, event, event])
       expect(LogStash::Outputs::Keystone::Token.instance.id).to eq(token_id)
     end
-
-    it 'if unauthorized, renew it' do
-      stub_request(:post, monasca_log_api_url_post)
-        .with(:headers =>
-        {
-          'Accept' => '*/*',
-          'Content-Type' => 'application/json',
-          'User-Agent' => 'Ruby',
-          'X-Auth-Token' => 'f8cdafb7dce94444ad781a53ddaff693'
-        })
-        .to_return(:status => 401)
-
-      expect_any_instance_of(LogStash::Outputs::Monasca::MonascaLogApiClient)
-        .to receive(:handle_response).exactly(5).times
-        .and_raise(
-          LogStash::Outputs::Monasca::MonascaLogApiClient::InvalidTokenError
-        )
-
-      expect_any_instance_of(LogStash::Outputs::Keystone::KeystoneClient)
-        .to receive(:authenticate).exactly(6).times
-        .with(
-          complete_config['username'],
-          complete_config['user_domain_name'],
-          complete_config['password'],
-          complete_config['project_name'],
-          complete_config['project_domain_name']
-        )
-        .and_return(expired_token, valid_token)
-
-      monasca_log_api = LogStash::Outputs::MonascaLogApi.new(complete_config)
-      allow(monasca_log_api).to receive(:start_time_check)
-      monasca_log_api.register
-      expect(LogStash::Outputs::Keystone::Token.instance.id).to eq(old_token_id)
-      monasca_log_api.multi_receive([event, event, event])
-      expect(LogStash::Outputs::Keystone::Token.instance.id).to eq(token_id)
-    end
   end
 
   context 'when stopping plugin' do
@@ -528,4 +475,6 @@
     end
   end
 
+  # TODO add testing graceful recovery from exceptions
+
 end