Add retry logic in pod and namespace watches when Kubernetes API connection gets closed. (#204)

* Add retry logic in pod and namespace watches when Kubernetes API connection gets closed.

* Refactor and add backup.

Change-Id: I6771e12b4fa359b92f7de757b9cf56704693833e

* Fix tests.

Change-Id: I688a867f6b0a7b88e198b5d5ac5463f02c4022b5

* remove unused variables.

Change-Id: I36f2dd8ef34d9bdeb890b0b073a802e6401ace32

* add stats.

Change-Id: I5a116f5a91a9251de849a2d679ec277af6272a52

* Add tests for retry.

Change-Id: I7fb939bc5dbb3851ed07c674ec22416a43c92432

Author: qingling128

README.md

@@ -61,6 +61,7 @@ when true (default: `true`)
 * `skip_container_metadata` - Skip some of the container data of the metadata. The metadata will not contain the container_image and container_image_id fields.
 * `skip_master_url` - Skip the master_url field from the metadata.
 * `skip_namespace_metadata` - Skip the namespace_id field from the metadata. The fetch_namespace_metadata function will be skipped. The plugin will be faster and cpu consumption will be less.
+* `watch_retry_interval` - The time interval in seconds for retry backoffs when watch connections fail. (default: `10`)
 
 **NOTE:** As of the release 2.1.x of this plugin, it no longer supports parsing the source message into JSON and attaching it to the
 payload.  The following configuration options are removed:

lib/fluent/plugin/filter_kubernetes_metadata.rb

@@ -22,6 +22,8 @@
 require_relative 'kubernetes_metadata_stats'
 require_relative 'kubernetes_metadata_watch_namespaces'
 require_relative 'kubernetes_metadata_watch_pods'
+
+require 'fluent/plugin_helper/thread'
 require 'fluent/plugin/filter'
 require 'resolv'
 
@@ -37,6 +39,8 @@ class KubernetesMetadataFilter < Fluent::Plugin::Filter
 
     Fluent::Plugin.register_filter('kubernetes_metadata', self)
 
+    helpers :thread
+
     config_param :kubernetes_url, :string, default: nil
     config_param :cache_size, :integer, default: 1000
     config_param :cache_ttl, :integer, default: 60 * 60
@@ -80,6 +84,12 @@ class KubernetesMetadataFilter < Fluent::Plugin::Filter
     config_param :skip_container_metadata, :bool, default: false
     config_param :skip_master_url, :bool, default: false
     config_param :skip_namespace_metadata, :bool, default: false
+    # The time interval in seconds for retry backoffs when watch connections fail.
+    config_param :watch_retry_interval, :bool, default: 1
+    # The base number of exponential backoff for retries.
+    config_param :watch_retry_exponential_backoff_base, :bool, default: 2
+    # The maximum number of times to retry pod and namespace watches.
+    config_param :watch_retry_max_times, :bool, default: 10
 
     def fetch_pod_metadata(namespace_name, pod_name)
       log.trace("fetching pod metadata: #{namespace_name}/#{pod_name}") if log.trace?
@@ -264,9 +274,14 @@ def log.trace?
         end
 
         if @watch
-          thread = Thread.new(self) { |this| this.start_pod_watch }
-          thread.abort_on_exception = true
-          namespace_thread = Thread.new(self) { |this| this.start_namespace_watch }
+          pod_thread = thread_create :"pod_watch_thread" do
+            set_up_pod_thread
+          end
+          pod_thread.abort_on_exception = true
+
+          namespace_thread = thread_create :"namespace_watch_thread" do
+            set_up_namespace_thread
+          end
           namespace_thread.abort_on_exception = true
         end
       end

lib/fluent/plugin/kubernetes_metadata_watch_namespaces.rb

@@ -23,26 +23,81 @@ module WatchNamespaces
 
     include ::KubernetesMetadata::Common
 
-    def start_namespace_watch
-      begin
-        options = {
-          resource_version: '0'  # Fetch from API server.
-        }
-        namespaces = @client.get_namespaces(options)
-        namespaces.each do |namespace|
-          cache_key = namespace.metadata['uid']
-          @namespace_cache[cache_key] = parse_namespace_metadata(namespace)
-          @stats.bump(:namespace_cache_host_updates)
+    def set_up_namespace_thread
+      # Any failures / exceptions in the initial setup should raise
+      # Fluent:ConfigError, so that users can inspect potential errors in
+      # the configuration.
+      namespace_watcher = start_namespace_watch
+      Thread.current[:namespace_watch_retry_backoff_interval] = @watch_retry_interval
+      Thread.current[:namespace_watch_retry_count] = 0
+
+      # Any failures / exceptions in the followup watcher notice
+      # processing will be swallowed and retried. These failures /
+      # exceptions could be caused by Kubernetes API being temporarily
+      # down. We assume the configuration is correct at this point.
+      while thread_current_running?
+        begin
+          namespace_watcher ||= get_namespaces_and_start_watcher
+          process_namespace_watcher_notices(namespace_watcher)
+        rescue Exception => e
+          @stats.bump(:namespace_watch_failures)
+          if Thread.current[:namespace_watch_retry_count] < @watch_retry_max_times
+            # Instead of raising exceptions and crashing Fluentd, swallow
+            # the exception and reset the watcher.
+            log.info(
+              "Exception encountered parsing namespace watch event. " \
+              "The connection might have been closed. Sleeping for " \
+              "#{Thread.current[:namespace_watch_retry_backoff_interval]} " \
+              "seconds and resetting the namespace watcher.", e)
+            sleep(Thread.current[:namespace_watch_retry_backoff_interval])
+            Thread.current[:namespace_watch_retry_count] += 1
+            Thread.current[:namespace_watch_retry_backoff_interval] *= @watch_retry_exponential_backoff_base
+            namespace_watcher = nil
+          else
+            # Since retries failed for many times, log as errors instead
+            # of info and raise exceptions and trigger Fluentd to restart.
+            message =
+              "Exception encountered parsing namespace watch event. The " \
+              "connection might have been closed. Retried " \
+              "#{@watch_retry_max_times} times yet still failing. Restarting."
+            log.error(message, e)
+            raise Fluent::UnrecoverableError.new(message)
+          end
         end
-        options[:resource_version] = namespaces.resourceVersion
-        watcher = @client.watch_namespaces(options)
-      rescue Exception=>e
-        message = "start_namespace_watch: Exception encountered setting up namespace watch from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}: #{e.message}"
-        message += " (#{e.response})" if e.respond_to?(:response)
-        log.debug(message)
+      end
+    end
+
+    def start_namespace_watch
+      return get_namespaces_and_start_watcher
+    rescue Exception => e
+      message = "start_namespace_watch: Exception encountered setting up " \
+                "namespace watch from Kubernetes API #{@apiVersion} endpoint " \
+                "#{@kubernetes_url}: #{e.message}"
+      message += " (#{e.response})" if e.respond_to?(:response)
+      log.debug(message)
+
+      raise Fluent::ConfigError, message
+    end
 
-        raise Fluent::ConfigError, message
+    # List all namespaces, record the resourceVersion and return a watcher
+    # starting from that resourceVersion.
+    def get_namespaces_and_start_watcher
+      options = {
+        resource_version: '0'  # Fetch from API server.
+      }
+      namespaces = @client.get_namespaces(options)
+      namespaces.each do |namespace|
+        cache_key = namespace.metadata['uid']
+        @namespace_cache[cache_key] = parse_namespace_metadata(namespace)
+        @stats.bump(:namespace_cache_host_updates)
       end
+      options[:resource_version] = namespaces.resourceVersion
+      watcher = @client.watch_namespaces(options)
+      watcher
+    end
+
+    # Process a watcher notice and potentially raise an exception.
+    def process_namespace_watcher_notices(watcher)
       watcher.each do |notice|
         case notice.type
           when 'MODIFIED'
@@ -55,7 +110,7 @@ def start_namespace_watch
               @stats.bump(:namespace_cache_watch_misses)
             end
           when 'DELETED'
-            # ignore and let age out for cases where 
+            # ignore and let age out for cases where
             # deleted but still processing logs
             @stats.bump(:namespace_cache_watch_deletes_ignored)
           else
@@ -65,6 +120,5 @@ def start_namespace_watch
         end
       end
     end
-
   end
 end

lib/fluent/plugin/kubernetes_metadata_watch_pods.rb

@@ -23,30 +23,84 @@ module WatchPods
 
     include ::KubernetesMetadata::Common
 
-    def start_pod_watch
-      begin
-        options = {
-          resource_version: '0'  # Fetch from API server.
-        }
-        if ENV['K8S_NODE_NAME']
-          options[:field_selector] = 'spec.nodeName=' + ENV['K8S_NODE_NAME']
-        end
-        pods = @client.get_pods(options)
-        pods.each do |pod|
-          cache_key = pod.metadata['uid']
-          @cache[cache_key] = parse_pod_metadata(pod)
-          @stats.bump(:pod_cache_host_updates)
+    def set_up_pod_thread
+      # Any failures / exceptions in the initial setup should raise
+      # Fluent:ConfigError, so that users can inspect potential errors in
+      # the configuration.
+      pod_watcher = start_pod_watch
+      Thread.current[:pod_watch_retry_backoff_interval] = @watch_retry_interval
+      Thread.current[:pod_watch_retry_count] = 0
+
+      # Any failures / exceptions in the followup watcher notice
+      # processing will be swallowed and retried. These failures /
+      # exceptions could be caused by Kubernetes API being temporarily
+      # down. We assume the configuration is correct at this point.
+      while thread_current_running?
+        begin
+          pod_watcher ||= get_pods_and_start_watcher
+          process_pod_watcher_notices(pod_watcher)
+        rescue Exception => e
+          @stats.bump(:pod_watch_failures)
+          if Thread.current[:pod_watch_retry_count] < @watch_retry_max_times
+            # Instead of raising exceptions and crashing Fluentd, swallow
+            # the exception and reset the watcher.
+            log.info(
+              "Exception encountered parsing pod watch event. The " \
+              "connection might have been closed. Sleeping for " \
+              "#{Thread.current[:pod_watch_retry_backoff_interval]} " \
+              "seconds and resetting the pod watcher.", e)
+            sleep(Thread.current[:pod_watch_retry_backoff_interval])
+            Thread.current[:pod_watch_retry_count] += 1
+            Thread.current[:pod_watch_retry_backoff_interval] *= @watch_retry_exponential_backoff_base
+            pod_watcher = nil
+          else
+            # Since retries failed for many times, log as errors instead
+            # of info and raise exceptions and trigger Fluentd to restart.
+            message =
+              "Exception encountered parsing pod watch event. The " \
+              "connection might have been closed. Retried " \
+              "#{@watch_retry_max_times} times yet still failing. Restarting."
+            log.error(message, e)
+            raise Fluent::UnrecoverableError.new(message)
+          end
         end
-        options[:resource_version] = pods.resourceVersion
-        watcher = @client.watch_pods(options)
-      rescue Exception => e
-        message = "start_pod_watch: Exception encountered setting up pod watch from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}: #{e.message}"
-        message += " (#{e.response})" if e.respond_to?(:response)
-        log.debug(message)
+      end
+    end
 
-        raise Fluent::ConfigError, message
+    def start_pod_watch
+      get_pods_and_start_watcher
+    rescue Exception => e
+      message = "start_pod_watch: Exception encountered setting up pod watch " \
+                "from Kubernetes API #{@apiVersion} endpoint " \
+                "#{@kubernetes_url}: #{e.message}"
+      message += " (#{e.response})" if e.respond_to?(:response)
+      log.debug(message)
+
+      raise Fluent::ConfigError, message
+    end
+
+    # List all pods, record the resourceVersion and return a watcher starting
+    # from that resourceVersion.
+    def get_pods_and_start_watcher
+      options = {
+        resource_version: '0'  # Fetch from API server.
+      }
+      if ENV['K8S_NODE_NAME']
+        options[:field_selector] = 'spec.nodeName=' + ENV['K8S_NODE_NAME']
+      end
+      pods = @client.get_pods(options)
+      pods.each do |pod|
+        cache_key = pod.metadata['uid']
+        @cache[cache_key] = parse_pod_metadata(pod)
+        @stats.bump(:pod_cache_host_updates)
       end
+      options[:resource_version] = pods.resourceVersion
+      watcher = @client.watch_pods(options)
+      watcher
+    end
 
+    # Process a watcher notice and potentially raise an exception.
+    def process_pod_watcher_notices(watcher)
       watcher.each do |notice|
         case notice.type
           when 'MODIFIED'

test/plugin/test_watch_namespaces.rb

@@ -74,7 +74,7 @@ class WatchNamespacesTestTest < WatchTest
 
     test 'namespace list caches namespaces' do
       @client.stub :get_namespaces, @initial do
-        start_namespace_watch
+        process_namespace_watcher_notices(start_namespace_watch)
         assert_equal(true, @namespace_cache.key?('initial_uid'))
         assert_equal(true, @namespace_cache.key?('modified_uid'))
         assert_equal(2, @stats[:namespace_cache_host_updates])
@@ -86,7 +86,7 @@ class WatchNamespacesTestTest < WatchTest
       ENV['K8S_NODE_NAME'] = 'aNodeName'
       @client.stub :get_namespaces, @initial do
         @client.stub :watch_namespaces, [@modified] do
-          start_namespace_watch
+          process_namespace_watcher_notices(start_namespace_watch)
           assert_equal(2, @stats[:namespace_cache_host_updates])
           assert_equal(1, @stats[:namespace_cache_watch_updates])
         end
@@ -96,15 +96,15 @@ class WatchNamespacesTestTest < WatchTest
 
     test 'namespace watch ignores CREATED' do
       @client.stub :watch_namespaces, [@created] do
-       start_namespace_watch
+       process_namespace_watcher_notices(start_namespace_watch)
        assert_equal(false, @namespace_cache.key?('created_uid'))
        assert_equal(1, @stats[:namespace_cache_watch_ignored])
       end
     end
 
     test 'namespace watch ignores MODIFIED when info not in cache' do
       @client.stub :watch_namespaces, [@modified] do
-       start_namespace_watch
+       process_namespace_watcher_notices(start_namespace_watch)
        assert_equal(false, @namespace_cache.key?('modified_uid'))
        assert_equal(1, @stats[:namespace_cache_watch_misses])
       end
@@ -113,7 +113,7 @@ class WatchNamespacesTestTest < WatchTest
     test 'namespace watch updates cache when MODIFIED is received and info is cached' do
       @namespace_cache['modified_uid'] = {}
       @client.stub :watch_namespaces, [@modified] do
-       start_namespace_watch
+       process_namespace_watcher_notices(start_namespace_watch)
        assert_equal(true, @namespace_cache.key?('modified_uid'))
        assert_equal(1, @stats[:namespace_cache_watch_updates])
       end
@@ -122,10 +122,22 @@ class WatchNamespacesTestTest < WatchTest
     test 'namespace watch ignores DELETED' do
       @namespace_cache['deleted_uid'] = {}
       @client.stub :watch_namespaces, [@deleted] do
-       start_namespace_watch
+       process_namespace_watcher_notices(start_namespace_watch)
        assert_equal(true, @namespace_cache.key?('deleted_uid'))
        assert_equal(1, @stats[:namespace_cache_watch_deletes_ignored])
       end
     end
 
+    test 'namespace watch retries when exceptions are encountered' do
+      @client.stub :get_namespaces, @initial do
+        @client.stub :watch_namespaces, [[@created, @exception_raised]] do
+          assert_raise Fluent::UnrecoverableError do
+            set_up_namespace_thread
+          end
+          assert_equal(3, @stats[:namespace_watch_failures])
+          assert_equal(2, Thread.current[:namespace_watch_retry_count])
+          assert_equal(4, Thread.current[:namespace_watch_retry_backoff_interval])
+        end
+      end
+    end
 end