handle connection error (#63)

Author: rockb1017

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    fluent-plugin-kubernetes-objects (1.1.6)
+    fluent-plugin-kubernetes-objects (1.1.7)
       fluentd (>= 1.9.1)
       http_parser.rb (= 0.5.3)
       kubeclient (~> 4.6.0)
@@ -106,4 +106,4 @@ DEPENDENCIES
   webmock (~> 3.5)
 
 BUNDLED WITH
-   2.2.23
+   2.2.24

VERSION

@@ -1 +1 @@
-1.1.6
+1.1.7

lib/fluent/plugin/in_kubernetes_objects.rb

@@ -245,15 +245,19 @@ def create_watcher_thread(conf)
         while thread_current_running?
           @client.public_send("watch_#{resource_name}", options).tap do |watcher|
             tag = generate_tag "#{resource_name}"
-            watcher.each do |entity|
-              begin
-                entity = JSON.parse(entity)
-                router.emit tag, Fluent::Engine.now, entity
-                options[:resource_version] = entity['object']['metadata']['resourceVersion']
-                @storage.put resource_name, entity['object']['metadata']['resourceVersion']
-              rescue => e
-                log.info "Got exception #{e} parsing entity #{entity}. Resetting watcher."
+            begin
+              watcher.each do |entity|
+                begin
+                  entity = JSON.parse(entity)
+                  router.emit tag, Fluent::Engine.now, entity
+                  options[:resource_version] = entity['object']['metadata']['resourceVersion']
+                  @storage.put resource_name, entity['object']['metadata']['resourceVersion']
+                rescue => e
+                  log.info "Got exception #{e} parsing entity #{entity}. Resetting watcher."
+                end
               end
+            rescue => e
+              log.info "Got exception #{e}. Resetting watcher."
             end
           end
         end