SUMO-115364 Handle exceptions thrwon from kubeclient calls

fluent-plugin-events/lib/fluent/plugin/in_events.rb

@@ -119,20 +119,26 @@ def start_watcher_thread
             log.debug "Closing watch stream"
           end
         end
+      rescue => e
+        log.error "Got exception #{e} watching for resource #{resource_name}. Skipping."
       end
 
       def create_config_map
         @configmap.data = { "resource-version-#{resource_name}": "#{@resource_version}" }
         @clients['v1'].public_send("create_config_map", @configmap).tap do |map|
           log.debug "Created config map: #{map}"
         end
+      rescue => e
+        log.error "Got exception #{e} creating config map. Skipping."
       end
 
       def patch_config_map
         pull_resource_version
         @clients['v1'].public_send("patch_config_map", "fluentd-config-resource-version", {data: { "resource-version-#{resource_name}": "#{@resource_version}"}}, @deploy_namespace).tap do |map|
           log.debug "Patched config map for #{@resource_name}: #{map}"
         end
+      rescue => e
+        log.error "Got exception #{e} patching config map. Skipping."
       end
 
       def initialize_resource_version
@@ -152,6 +158,8 @@ def initialize_resource_version
           end
         rescue Kubeclient::ResourceNotFoundError
           create_config_map
+        rescue => e
+          log.error "Got exception #{e} getting config map. Skipping."
         end
       end
 
@@ -167,6 +175,8 @@ def pull_resource_version
         end
 
         @resource_version = resource_version
+      rescue => e
+        log.error "Got exception #{e} pulling resource version #{resource_version} for resource #{resource_name}. Skipping."
       end
 
       def normalize_param

fluent-plugin-events/test/helper.rb

@@ -63,6 +63,36 @@ def mock_patch_config_map(rv)
     .returns(object.to_json)
 end
 
+def mock_create_config_map_execution(configmap)
+  Kubeclient::Client.any_instance.stubs(:public_send)
+    .with("create_config_map", configmap)
+    .raises(StandardError.new 'Error occurred when creating config map.')
+end
+
+def mock_patch_config_map_exception(rv)
+  Kubeclient::Client.any_instance.stubs(:public_send)
+    .with("patch_config_map", "fluentd-config-resource-version",
+    {data: { "resource-version-events": rv.to_s}}, 'sumologic')
+    .raises(StandardError.new 'Error occurred when patching config map.')
+end
+
+def mock_get_config_map_exception
+  Kubeclient::Client.any_instance.stubs(:public_send)
+    .with("get_config_map", "fluentd-config-resource-version", "sumologic")
+    .raises(StandardError.new 'Error occurred when getting config map.')
+end
+
+def mock_get_events_exception
+  Kubeclient::Client.any_instance.stubs(:public_send).with("get_events", {:as=>:raw})
+    .raises(StandardError.new 'Error occurred when getting events.')
+end
+
+def mock_watch_events_exception
+  Kubeclient::Client.any_instance.stubs(:public_send)
+    .with("watch_events", {:as=>:raw, :field_selector=>nil, :label_selector=>nil, :namespace=>nil, :resource_version=>nil, :timeout_seconds=>360})
+    .raises(StandardError.new 'Error occurred when watching events.')
+end
+
 def get_watch_resources_count_by_type_selector(type_selector, file_name)
   text = File.read(test_resource(file_name))
   objects = text.split(/\n+/).map {|line| JSON.parse(line)}

fluent-plugin-events/test/plugin/test_in_events.rb

@@ -66,7 +66,7 @@ def connect_kubernetes_with_api_version(driver)
     assert_equal 'dummy-services-rv', resource.data['resource-version-services']
   end
 
-  test 'initialize_resource_verion correctly for different client' do
+  test 'initialize_resource_version correctly for different client' do
     config = %([
       api_version "events.k8s.io/v1beta1"
     ])
@@ -186,4 +186,65 @@ def connect_kubernetes_with_api_version(driver)
 
     driver.start_monitor
   end
+
+  sub_test_case 'exception from kubeclient will not interrupt execution of start monitor' do
+    test 'exception from kubeclient call in create_config_map' do
+      config = %([])
+      driver = create_driver(config).instance
+      connect_kubernetes_with_api_version(driver)
+      driver.instance_variable_set(:@clients, @clients)
+      driver.instance_variable_set(:@last_recreated, 0)
+
+      mock_get_config_map
+      driver.initialize_resource_version
+      mock_create_config_map_execution(driver.instance_variable_get(:@configmap))
+      assert_nothing_raised { driver.create_config_map }
+    end
+
+    test 'exception from kubeclient call in patch_config_map' do
+      config = %([])
+      driver = create_driver(config).instance
+      connect_kubernetes_with_api_version(driver)
+      driver.instance_variable_set(:@clients, @clients)
+      driver.instance_variable_set(:@last_recreated, 0)
+
+      mock_get_events('api_list_events_v1.json')
+      mock_watch_events('api_watch_events_v1.txt')
+      mock_patch_config_map_exception(2346293)
+      assert_nothing_raised { driver.start_monitor }
+    end
+
+    test 'exception from kubeclient call in initialize_resource_version' do
+      config = %([])
+      driver = create_driver(config).instance
+      connect_kubernetes_with_api_version(driver)
+      driver.instance_variable_set(:@clients, @clients)
+      driver.instance_variable_set(:@last_recreated, 0)
+
+      mock_get_config_map_exception
+      assert_nothing_raised { driver.initialize_resource_version }
+    end
+
+    test 'exception from kubeclient call in pull_resource_version' do
+      config = %([])
+      driver = create_driver(config).instance
+      connect_kubernetes_with_api_version(driver)
+      driver.instance_variable_set(:@clients, @clients)
+      driver.instance_variable_set(:@last_recreated, 0)
+
+      mock_get_events_exception
+      assert_nothing_raised { driver.pull_resource_version }
+    end
+
+    test 'exception from kubclient call in start_watcher_thread' do
+      config = %([])
+      driver = create_driver(config).instance
+      connect_kubernetes_with_api_version(driver)
+      driver.instance_variable_set(:@clients, @clients)
+      driver.instance_variable_set(:@last_recreated, 0)
+
+      mock_watch_events_exception
+      assert_nothing_raised { driver.start_watcher_thread}
+    end
+  end
 end