Release/1.1.2 Watcher Fix (#36)

Author: chaitanyaphalak

.circleci/config.yml

@@ -50,10 +50,6 @@ jobs:
       - checkout
       - setup_remote_docker:
           reusable: true
-      - run:
-          name: Push rubygem to s3
-          command: |
-            .circleci/push_gem.sh
       - run:
           name: Build and push docker image to ecr
           command: |

.github/ISSUE_TEMPLATE/bug_report.md

@@ -0,0 +1,29 @@
+---
+name: Bug report
+about: Report a bug encountered while operating fluent-plugin-kubernetes-objects
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
+
+Please do not report security vulnerabilities with public GitHub issue reports. Please report security issues here: https://www.splunk.com/goto/report_vulnerabilities_prodsec
+-->
+
+
+**What happened**:
+
+**What you expected to happen**:
+
+**How to reproduce it (as minimally and precisely as possible)**:
+
+**Anything else we need to know?**:
+
+**Environment**:
+- Kubernetes version (use `kubectl version`):
+- Ruby version (use `ruby --version`):
+- OS (e.g: `cat /etc/os-release`):
+- Splunk version:
+- Others:

.github/ISSUE_TEMPLATE/enhancement_request.md

@@ -0,0 +1,14 @@
+---
+name: Enhancement Request
+about: Suggest an enhancement to the fluent-plugin-kubernetes-objects project
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+<<!-- Please only use this template for submitting enhancement requests -->
+
+**What would you like to be added**:
+
+**Why is this needed**:

.github/ISSUE_TEMPLATE/failing_test.md

@@ -0,0 +1,18 @@
+---
+name: Failing Test
+about: Report test failures in fluent-plugin-kubernetes-objects
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+<!-- Please only use this template for submitting reports about failing tests -->
+
+**Which test(s) are failing**:
+
+**Since when has it been failing**:
+
+**Reason for failure**:
+
+**Anything else we need to know**:

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,23 @@
+## Proposed changes
+
+Describe the big picture of your changes here to communicate to the maintainers why we should accept this pull request. If it fixes a bug or resolves a feature request, be sure to link to that issue.
+
+## Types of changes
+
+What types of changes does your code introduce?
+_Put an `x` in the boxes that apply_
+
+- [ ] Bugfix (non-breaking change which fixes an issue)
+- [ ] New feature (non-breaking change which adds functionality)
+- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
+
+## Checklist
+
+_Put an `x` in the boxes that apply._
+
+- [ ] I have read the [CONTRIBUTING](https://github.com/splunk/fluent-plugin-kubernetes-objects/blob/develop/CONTRIBUTING.md) doc
+- [ ] I have read the [CLA](https://github.com/splunk/fluent-plugin-kubernetes-objects/blob/develop/CLA.md)
+- [ ] I have added tests that prove my fix is effective or that my feature works
+- [ ] I have added necessary documentation (if appropriate)
+- [ ] Any dependent changes have been merged and published in downstream modules
+

CLA.md

@@ -0,0 +1,18 @@
+By submitting a Contribution to this Work, You agree that Your Contribution is made subject to the primary LICENSE 
+file applicable to this Work.  In addition, You represent that: (i) You are the copyright owner of the Contribution 
+or (ii) You have the requisite rights to make the Contribution.
+
+Definitions:
+
+“You” shall mean: (i) yourself if you are making a Contribution on your own behalf; or (ii) your company, 
+if you are making a Contribution on behalf of your company.  If you are making a Contribution on behalf of your 
+company, you represent that you have the requisite authority to do so.    
+ 
+"Contribution" shall mean any original work of authorship, including any modifications or additions to an existing 
+work, that is intentionally submitted by You for inclusion in, or documentation of, this project/repository.  For the 
+purposes of this definition, "submitted" means any form of electronic, verbal, or written communication submitted for 
+inclusion in this project/repository, including but not limited to communication on electronic mailing lists, source 
+code control systems, and issue tracking systems that are managed by, or on behalf of, the maintainers of 
+the project/repository. 
+
+“Work” shall mean the collective software, content, and documentation in this project/repository.

VERSION

@@ -1 +1 @@
-1.1.1
+1.1.2

lib/fluent/plugin/in_kubernetes_objects.rb

@@ -5,7 +5,7 @@
 
 module Fluent::Plugin
   class KubernetesObjectsInput < Fluent::Plugin::Input
-    VERSION = '1.1.1'.freeze
+    VERSION = '1.1.2'.freeze
 
     Fluent::Plugin.register_input('kubernetes_objects', self)
 
@@ -102,7 +102,6 @@ def start
     end
 
     def close
-      @watchers.each &:finish if @watchers
       super
     end
 
@@ -118,6 +117,21 @@ def generate_tag(item_name)
       [@tag_prefix, item_name, @tag_suffix].join
     end
 
+    def init_with_kubeconfig()
+      options = {}
+      config = Kubeclient::Config.read @kubeconfig
+      current_context = config.context
+
+      @client = Kubeclient::Client.new(
+        current_context.api_endpoint,
+        current_context.api_version,
+        options.merge(
+          ssl_options: current_context.ssl_options,
+          auth_options: current_context.auth_options
+        )
+      )
+    end
+
     def initialize_client
       # mostly borrowed from Fluentd Kubernetes Metadata Filter Plugin
       if @kubernetes_url.nil?
@@ -173,18 +187,7 @@ def start_pullers
     end
 
     def start_watchers
-      @watchers = @watch_objects.map do |o|
-        o = o.to_h.dup
-        o[:as] = :raw
-        resource_name = o.delete(:resource_name)
-        watch_interval = o.delete(:interval)
-
-        version = @storage.get(resource_name)
-        o[:resource_version] = version if version
-        @client.public_send("watch_#{resource_name}", o).tap do |watcher|
-          create_watcher_thread resource_name, watcher, watch_interval
-        end
-      end
+      @watch_objects.each(&method(:create_watcher_thread))
     end
 
     def create_pull_thread(conf)
@@ -227,19 +230,35 @@ def create_pull_thread(conf)
       end
     end
 
-    def create_watcher_thread(object_name, watcher, interval)
-      thread_create(:"watch_#{object_name}") do
-        tag = generate_tag "#{object_name}.watch"
+    def create_watcher_thread(conf)
+      options = conf.to_h.dup
+      options[:as] = :raw
+      resource_name = options[:resource_name]
+      version = @storage.get(resource_name)
+      if version
+        options[:resource_version] = version
+      else
+        options[:resource_version] = 0
+      end
+
+      thread_create :"watch_#{resource_name}" do
         while thread_current_running?
-          watcher.each do |entity|
-            log.trace { "Received new object from watching #{object_name}" }
-            entity = JSON.parse(entity)
-            router.emit tag, Fluent::Engine.now, entity
-            @storage.put object_name, entity['object']['metadata']['resourceVersion']
-            sleep(interval)
+          @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."
+              end
+            end
           end
         end
       end
     end
   end
 end
+

test/fluent/plugin/in_kubernetes_objects_test.rb

@@ -109,9 +109,12 @@
       </watch>
       CONF
 
+      stub_k8s_events params: {resourceVersion: "0"}
+      stub_k8s_events params: {resourceVersion: "6621683"}
+
       d.run expect_emits: 1, timeout: 3
       events = d.events
-      expect(events.all? { |e| e[0] == 'kubernetes.events.watch'}).must_equal true
+      expect(events.all? { |e| e[0] == 'kubernetes.events'}).must_equal true
     end
 
     it "should use checkpoints for watching" do
@@ -133,6 +136,7 @@
         CONF
 
         stub_k8s_events params: {resourceVersion: "123456"}
+        stub_k8s_events params: {resourceVersion: "6621683"}
 
         d.run expect_emits: 1, timeout: 3
       ensure