allow async concurrency tuning

.gitignore

@@ -14,3 +14,4 @@ old-index.yaml
 client
 .vscode/settings.json
 .env
+.vscode/launch.json

helm/kubetruth/templates/deployment.yaml

@@ -54,6 +54,10 @@ spec:
             {{- if not .Values.appSettings.async }}
             - --no-async
             {{- end }}
+            {{- if .Values.appSettings.concurrency }}
+            - --concurrency
+            - "{{ .Values.appSettings.concurrency }}"
+            {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

helm/kubetruth/values.yaml

@@ -86,6 +86,7 @@ appSettings:
   pollingInterval:
   debug: false
   async: true
+  concurrency:
 
 # If secret.create == false, secret generation is left to the user
 # outside of the chart, and a name should be specified
@@ -182,7 +183,7 @@ projectMappings:
           {%- if name_parts.size > 1 %}
           namespace: "{{ name_parts | first }}"
           {%- endif %}
-        
+
         spec: {{ templates[template] | nindent: 2 }}
 
         {%- endif %}

lib/kubetruth/cli.rb

@@ -43,6 +43,10 @@ class CLI < CLIBase
            :flag, "Run using async I/O (or not)",
            default: true
 
+    option ["-c", "--concurrency"], "CONCURRENCY", "Concurrency limit for async I/O", default: 3 do |a|
+      Integer(a)
+    end
+
     # TODO: option to map template to configmap?
 
     def execute
@@ -51,7 +55,7 @@ def execute
       Kubetruth::CtApi.configure(api_key: api_key, api_url: api_url)
       Kubetruth::KubeApi.configure(namespace: kube_namespace, token: kube_token, api_url: kube_url)
 
-      etl = ETL.new(dry_run: dry_run?, async: async?)
+      etl = ETL.new(dry_run: dry_run?, async: async?, async_concurrency: concurrency)
 
       Signal.trap("HUP") do
         puts "Handling HUP signal - waking up ETL poller" # logger cant be called from trap

lib/kubetruth/etl.rb

@@ -1,6 +1,7 @@
 require 'benchmark'
 require 'yaml'
 require 'async'
+require 'async/semaphore'
 require 'yaml/safe_load_stream'
 using YAMLSafeLoadStream
 
@@ -14,9 +15,11 @@ module Kubetruth
   class ETL
     include GemLogger::LoggerSupport
 
-    def initialize(dry_run: false, async: true)
+    def initialize(dry_run: false, async: true, async_concurrency: 3)
       @dry_run = dry_run
       @async = async
+      @async_concurrency = async_concurrency
+      @async_semaphore = nil
       @wrote_crds = false
     end
 
@@ -88,14 +91,14 @@ def with_polling(interval, &block)
 
     def async_task_tree(task)
       msg = ""
-    
+
       if task.parent
         # The root task seems to always be nil, so exclude it from name
         unless task.parent.parent.nil? && task.parent.annotation.blank?
           msg << async_task_tree(task.parent) << " -> "
         end
       end
-    
+
       msg << (task.annotation ? task.annotation : "unnamed")
       msg
     end
@@ -106,7 +109,7 @@ def wait
 
     def async(*args, **kwargs)
       if @async
-        Async(*args, **kwargs) do |task|
+        blk = ->(task) {
           task_name = async_task_tree(task)
           begin
             logger.info "Starting async task: #{task_name}"
@@ -116,7 +119,8 @@ def async(*args, **kwargs)
             logger.log_exception(e, "Failure in async task: #{task_name}")
             task.stop
           end
-        end
+        }
+        @async_semaphore.nil? ? Async(*args, **kwargs, &blk) : @async_semaphore.async(*args, **kwargs, &blk)
       else
         task_name = kwargs[:annotation] || "unnamed"
         begin
@@ -179,7 +183,7 @@ def initialize(project)
       def params_to_hash(param_list)
         Hash[param_list.collect {|param| [param.key, param.value]}]
       end
-  
+
       def params
         @param_data ||= begin
           # constructing the hash will cause any overrides to happen in the right
@@ -216,6 +220,7 @@ def params
 
     def apply
       async(annotation: "ETL Event Loop") do
+        @async_semaphore = Async::Semaphore.new(@async_concurrency) if @async
         logger.warn("Performing dry-run") if @dry_run
 
         load_config do |namespace, config|
@@ -280,7 +285,7 @@ def apply
                     parsed_ymls.each do |parsed_yml|
                       if parsed_yml.present?
                         async(annotation: "Apply Template: #{template_id}") do
-                          kube_apply(parsed_yml) 
+                          kube_apply(parsed_yml)
                         end
                       else
                         logger.debug {"Skipping empty stream template"}
@@ -293,7 +298,9 @@ def apply
             end
           end
         end
-      end.wait
+    ensure
+      @async_semaphore = nil
+    end.wait
     end
 
     def kube_apply(parsed_yml)

spec/kubetruth/cli_spec.rb

@@ -27,6 +27,7 @@ module Kubetruth
             --kube-url ku
             --dry-run
             --no-async
+            --concurrency 5
             --polling-interval 27
         ]
 
@@ -42,7 +43,7 @@ module Kubetruth
                                                              })
 
         etl = double(ETL)
-        expect(ETL).to receive(:new).with(dry_run: true, async: false).and_return(etl)
+        expect(ETL).to receive(:new).with(dry_run: true, async: false, async_concurrency: 5).and_return(etl)
         expect(etl).to receive(:apply)
         expect(etl).to receive(:with_polling).with(27).and_yield
         cli.run(args)

spec/kubetruth/etl_spec.rb

@@ -428,7 +428,7 @@ class ForceExit < Exception; end
       let(:config) {
         Kubetruth::Config.new([root_spec_crd])
       }
-      let(:collection) { ProjectCollection.new(config) }
+      let(:collection) { ProjectCollection.new(config.root_spec) }
 
       before(:each) do
 
@@ -447,7 +447,7 @@ class ForceExit < Exception; end
 
       it "renders multiple templates" do
         allow(etl).to receive(:load_config).and_yield(@ns, config)
-        
+
         expect(collection).to receive(:names).and_return(["proj1"])
 
         expect(etl).to receive(:kube_apply).with(hash_including("kind" => "ConfigMap"))
@@ -629,7 +629,7 @@ class ForceExit < Exception; end
       let(:config) {
         Kubetruth::Config.new([root_spec_crd])
       }
-      let(:collection) { ProjectCollection.new(config) }
+      let(:collection) { ProjectCollection.new(config.root_spec) }
 
       before(:each) do
 
@@ -683,6 +683,7 @@ class ForceExit < Exception; end
       it "sets log level when supplied by root pm in config" do
         Kubetruth::Logging.root_log_level = "error" # undo debug logging set by test harness
         config.root_spec.log_level = "error"
+        Logging.clear
 
         allow(etl).to receive(:load_config).and_yield(@ns, config)
         allow(config.root_spec.resource_templates["configmap"]).to receive(:render).and_return("")
@@ -698,6 +699,7 @@ class ForceExit < Exception; end
 
       it "sets log level when supplied by override pm in config" do
         Kubetruth::Logging.root_log_level = "error" # undo debug logging set by test harness
+        Logging.clear
 
         override_crd = {scope: "override", project_selector: "proj1", log_level: "debug"}
         config = Kubetruth::Config.new([root_spec_crd, override_crd])