feat: add exponential backoff for sending data (#76)

* feat: add exponential backoff for sending data

* tests(retry): add tests for retrying mechanism

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* feat(retries): remove retry_forever

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* feat(retries): ensure backward compatibility

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* docs(retry): extend logging messages

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* docs(retries): update README

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* Update lib/fluent/plugin/out_sumologic.rb

* fix: fix setting maximum sleep_time

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* feat: rename skip_retry to use_internal_retry

Signed-off-by: Dominik Rosiek <drosiek@sumologic.com>

* Update README.md

Co-authored-by: Andrzej Stencel <astencel@sumologic.com>

* Update README.md

Co-authored-by: Andrzej Stencel <astencel@sumologic.com>

Co-authored-by: Andrzej Stencel <astencel@sumologic.com>

Author: sumo-drosiek

README.md

@@ -38,6 +38,12 @@ Configuration options for fluent.conf are:
 * `compress_encoding` - Compression encoding format, either `gzip` or `deflate` (default `gzip`)
 * `custom_fields` - Comma-separated key=value list of fields to apply to every log. [more information](https://help.sumologic.com/Manage/Fields#http-source-fields)
 * `custom_dimensions` - Comma-separated key=value list of dimensions to apply to every metric. [more information](https://help.sumologic.com/03Send-Data/Sources/02Sources-for-Hosted-Collectors/HTTP-Source/Upload-Metrics-to-an-HTTP-Source#supported-http-headers)
+* `use_internal_retry` - Enable custom retry mechanism. As this is `false` by default due to backward compatibility,
+  we recommend to enable it and configure the following parameters (`retry_min_interval`, `retry_max_interval`, `retry_timeout`, `retry_max_times`)
+* `retry_min_interval` - Minimum interval to wait between sending tries (default is `1s`)
+* `retry_max_interval` - Maximum interval to wait between sending tries (default is `5m`)
+* `retry_timeout` - Time after which the data is going to be dropped (default is `72h`) (`0s` means that there is no timeout)
+* `retry_max_times` - Maximum number of retries (default is `0`) (`0` means that there is no max retry times, retries will happen forever)
 
 __NOTE:__ <sup>*</sup> [Placeholders](https://docs.fluentd.org/v1.0/articles/buffer-section#placeholders) are supported
 
@@ -137,6 +143,14 @@ Example
 }
 ```
 
+## Retry Mechanism
+
+`retry_min_interval`, `retry_max_interval`, `retry_timeout`, `retry_max_times` are not the [buffer retries parameters][buffer_retries].
+Due to technical reason, this plugin implements it's own retrying back-off exponential mechanism.
+It is disabled by default, but we recommend to enable it by setting `use_internal_retry` to `true`.
+
+[buffer_retries]: https://docs.fluentd.org/configuration/buffer-section#retries-parameters
+
 ### TLS 1.2 Requirement
 
 Sumo Logic only accepts connections from clients using TLS version 1.2 or greater. To utilize the content of this repo, ensure that it's running in an execution environment that is configured to use TLS 1.2 or greater.

lib/fluent/plugin/out_sumologic.rb

@@ -157,6 +157,13 @@ class Fluent::Plugin::Sumologic < Fluent::Plugin::Output
   config_param :timestamp_key, :string, :default => 'timestamp'
   config_param :proxy_uri, :string, :default => nil
   config_param :disable_cookies, :bool, :default => false
+
+  config_param :use_internal_retry, :bool, :default => false
+  config_param :retry_timeout, :time, :default => 72 * 3600  # 72h
+  config_param :retry_max_times, :integer, :default => 0
+  config_param :retry_min_interval, :time, :default => 1  # 1s
+  config_param :retry_max_interval, :time, :default => 5*60  # 5m
+
   # https://help.sumologic.com/Manage/Fields
   desc 'Fields string (eg "cluster=payment, service=credit_card") which is going to be added to every log record.'
   config_param :custom_fields, :string, :default => nil
@@ -385,6 +392,7 @@ def write(chunk)
 
     end
 
+    chunk_id = "##{chunk.dump_unique_id_hex(chunk.unique_id)}"
     # Push logs to sumo
     messages_list.each do |key, messages|
       source_name, source_category, source_host, fields = key[:source_name], key[:source_category],
@@ -397,18 +405,53 @@ def write(chunk)
         fields = [fields,@custom_fields].compact.join(",")
       end
 
-      @log.debug { "Sending #{messages.count} #{@data_type} records with source category '#{source_category}', source host '#{source_host}', source name '#{source_name}'." }
-
-      @sumo_conn.publish(
-          messages.join("\n"),
-          source_host         =source_host,
-          source_category     =source_category,
-          source_name         =source_name,
-          data_type           =@data_type,
-          metric_data_format  =@metric_data_format,
-          collected_fields    =fields,
-          dimensions          =@custom_dimensions
-      )
+      retries = 0
+      start_time = Time.now
+      sleep_time = @retry_min_interval
+
+      while true
+        common_log_part = "#{@data_type} records with source category '#{source_category}', source host '#{source_host}', source name '#{source_name}', chunk #{chunk_id}, try #{retries}"
+        begin
+          @log.debug { "Sending #{messages.count}; #{common_log_part}" }
+
+          @sumo_conn.publish(
+              messages.join("\n"),
+              source_host         =source_host,
+              source_category     =source_category,
+              source_name         =source_name,
+              data_type           =@data_type,
+              metric_data_format  =@metric_data_format,
+              collected_fields    =fields,
+              dimensions          =@custom_dimensions
+          )
+          break
+        rescue => e
+          if !@use_internal_retry
+            raise e
+          end
+          # increment retries
+          retries = retries + 1
+
+          log.warn "error while sending request to sumo: #{e}; #{common_log_part}"
+          log.warn_backtrace e.backtrace
+
+          # drop data if
+          #   - we reached out the @retry_max_times retries
+          #   - or we exceeded @retry_timeout
+          if (retries >= @retry_max_times && @retry_max_times > 0) || (Time.now > start_time + @retry_timeout && @retry_timeout > 0)
+            log.warn "dropping records; #{common_log_part}"
+            break
+          end
+
+          log.info "going to retry to send data at #{Time.now + sleep_time}; #{common_log_part}"
+          sleep sleep_time
+
+          sleep_time = sleep_time * 2
+          if sleep_time > @retry_max_interval
+            sleep_time = @retry_max_interval
+          end
+        end
+      end
     end
 
   end

test/plugin/test_out_sumologic.rb

@@ -825,4 +825,91 @@ def test_warning_response_from_receiver
     end
   end
 
+  def test_resend
+    endpoint = "https://collectors.sumologic.com/v1/receivers/http/1234"
+    config = %{
+      endpoint #{endpoint}
+      retry_min_interval 0s
+      retry_max_times 3
+      use_internal_retry true
+    }
+    time = event_time
+
+    driver = create_driver(config)
+    stub_request(:post, endpoint).to_return(
+      {status: 500, headers: {content_type: 'application/json'}},
+      {status: 200, headers: {content_type: 'application/json'}}
+    )
+    driver.run do
+      driver.feed("test", time, {"message": "test"})
+    end
+    assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
+                     body: /\A{"timestamp":\d+.,"message":"test"}\z/,
+                     times:2
+  end
+
+  def test_resend_failed
+    endpoint = "https://collectors.sumologic.com/v1/receivers/http/1234"
+    config = %{
+      endpoint #{endpoint}
+      retry_min_interval 0s
+      retry_max_times 15
+      use_internal_retry true
+    }
+    time = event_time
+
+    driver = create_driver(config)
+    stub_request(:post, endpoint).to_return(
+      status: 500, headers: {content_type: 'application/json'}
+    )
+    driver.run do
+      driver.feed("test", time, {"message": "test"})
+    end
+    assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
+                     body: /\A{"timestamp":\d+.,"message":"test"}\z/,
+                     times:15
+  end
+
+  def test_resend_forever
+    endpoint = "https://collectors.sumologic.com/v1/receivers/http/1234"
+    config = %{
+      endpoint #{endpoint}
+      retry_min_interval 0s
+      retry_max_times 0
+      retry_timeout 0s
+      use_internal_retry true
+    }
+    time = event_time
+
+    driver = create_driver(config)
+    stub_request(:post, endpoint).to_return(
+      *[{status: 500, headers: {content_type: 'application/json'}}]*123,
+      {status: 200, headers: {content_type: 'application/json'}}
+    )
+    driver.run do
+      driver.feed("test", time, {"message": "test"})
+    end
+    assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
+                     body: /\A{"timestamp":\d+.,"message":"test"}\z/,
+                     times:124
+  end
+
+  def test_skip_retry
+    endpoint = "https://collectors.sumologic.com/v1/receivers/http/1234"
+    config = %{
+      endpoint #{endpoint}
+    }
+    time = event_time
+
+    driver = create_driver(config)
+    stub_request(:post, endpoint).to_return(status: 500, headers: {content_type: 'application/json'})
+
+    exception = assert_raise(RuntimeError) {
+      driver.run do
+        driver.feed("test", time, {"message": "test"})
+      end
+    }
+    assert_equal("Failed to send data to HTTP Source. 500 - ", exception.message)
+  end
+
 end