Improve Kubernetes pod lookups (#189)

This PR contains the following related fixes to pod lookup performance:

Cache results from get_pods.
Restrict get_pods and watch_pods requests to the current host, if provided with the K8S_NODE_NAME environment variable.
Force lookup from API server cache by setting resourceVersion=0 in get_pods request.

Author: jkohen

fluent-plugin-kubernetes_metadata_filter.gemspec

@@ -23,7 +23,7 @@ Gem::Specification.new do |gem|
   gem.add_runtime_dependency "lru_redux"
   gem.add_runtime_dependency "kubeclient", '< 5'
 
-  gem.add_development_dependency "bundler", "~> 1.3"
+  gem.add_development_dependency "bundler", "~> 2.0.2"
   gem.add_development_dependency "rake"
   gem.add_development_dependency "minitest", "~> 4.0"
   gem.add_development_dependency "test-unit", "~> 3.0.2"

lib/fluent/plugin/kubernetes_metadata_watch_pods.rb

@@ -25,8 +25,20 @@ module WatchPods
 
     def start_pod_watch
       begin
-        resource_version = @client.get_pods.resourceVersion
-        watcher          = @client.watch_pods(resource_version)
+        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)
       rescue Exception => e
         message = "Exception encountered fetching metadata from Kubernetes API endpoint: #{e.message}"
         message += " (#{e.response})" if e.respond_to?(:response)

test/plugin/test_watch_pods.rb

@@ -25,6 +25,47 @@ class DefaultPodWatchStrategyTest < WatchTest
      include KubernetesMetadata::WatchPods
 
      setup do
+       @initial = Kubeclient::Common::EntityList.new(
+         'PodList',
+         '123',
+         [
+           Kubeclient::Resource.new({
+                                      'metadata' => {
+                                        'name' => 'initial',
+                                        'namespace' => 'initial_ns',
+                                        'uid' => 'initial_uid',
+                                        'labels' => {},
+                                      },
+                                      'spec' => {
+                                        'nodeName' => 'aNodeName',
+                                        'containers' => [{
+                                                           'name' => 'foo',
+                                                           'image' => 'bar',
+                                                         }, {
+                                                           'name' => 'bar',
+                                                           'image' => 'foo',
+                                                         }]
+                                      }
+                                    }),
+           Kubeclient::Resource.new({
+                                      'metadata' => {
+                                        'name' => 'modified',
+                                        'namespace' => 'create',
+                                        'uid' => 'modified_uid',
+                                        'labels' => {},
+                                      },
+                                      'spec' => {
+                                        'nodeName' => 'aNodeName',
+                                        'containers' => [{
+                                                           'name' => 'foo',
+                                                           'image' => 'bar',
+                                                         }, {
+                                                           'name' => 'bar',
+                                                           'image' => 'foo',
+                                                         }]
+                                      }
+                                    }),
+         ])
        @created = OpenStruct.new(
          type: 'CREATED',
          object: {
@@ -97,11 +138,38 @@ class DefaultPodWatchStrategyTest < WatchTest
        )
      end
 
+    test 'pod list caches pods' do
+      orig_env_val = ENV['K8S_NODE_NAME']
+      ENV['K8S_NODE_NAME'] = 'aNodeName'
+      @client.stub :get_pods, @initial do
+        start_pod_watch
+        assert_equal(true, @cache.key?('initial_uid'))
+        assert_equal(true, @cache.key?('modified_uid'))
+        assert_equal(2, @stats[:pod_cache_host_updates])
+      end
+      ENV['K8S_NODE_NAME'] = orig_env_val
+    end
+
+    test 'pod list caches pods and watch updates' do
+      orig_env_val = ENV['K8S_NODE_NAME']
+      ENV['K8S_NODE_NAME'] = 'aNodeName'
+      @client.stub :get_pods, @initial do
+        @client.stub :watch_pods, [@modified] do
+          start_pod_watch
+          assert_equal(2, @stats[:pod_cache_host_updates])
+          assert_equal(1, @stats[:pod_cache_watch_updates])
+        end
+      end
+      ENV['K8S_NODE_NAME'] = orig_env_val
+    end
+
     test 'pod watch notice ignores CREATED' do
-      @client.stub :watch_pods, [@created] do
-       start_pod_watch
-       assert_equal(false, @cache.key?('created_uid'))
-       assert_equal(1, @stats[:pod_cache_watch_ignored])
+      @client.stub :get_pods, @initial do
+        @client.stub :watch_pods, [@created] do
+          start_pod_watch
+          assert_equal(false, @cache.key?('created_uid'))
+          assert_equal(1, @stats[:pod_cache_watch_ignored])
+        end
       end
     end
 

test/plugin/watch_test.rb

@@ -30,16 +30,16 @@ class WatchTest < Test::Unit::TestCase
       def @client.resourceVersion
         '12345'
       end
-      def @client.watch_pods(value)
+      def @client.watch_pods(options = {})
          []
       end
-      def @client.watch_namespaces(value)
+      def @client.watch_namespaces(options = {})
          []
       end
-      def @client.get_namespaces 
+      def @client.get_namespaces(options = {})
           self
       end
-      def @client.get_pods
+      def @client.get_pods(options = {})
           self
       end
     end