Add a memory and cpu limits on deployments

app/models/miq_queue_worker_base/runner.rb

@@ -2,14 +2,9 @@
 
 class MiqQueueWorkerBase::Runner < MiqWorker::Runner
   def after_sync_config
-    sync_cpu_usage_threshold
     sync_dequeue_method
   end
 
-  def sync_cpu_usage_threshold
-    @cpu_usage_threshold = worker_settings[:cpu_usage_threshold]
-  end
-
   def sync_dequeue_method
     @dequeue_method = (worker_settings[:dequeue_method] || :sql).to_sym
   end
@@ -18,20 +13,6 @@ def dequeue_method_via_drb?
     @dequeue_method == :drb && drb_dequeue_available?
   end
 
-  def thresholds_exceeded?
-    return false if @cpu_usage_threshold == 0
-
-    usage = MiqSystem.cpu_usage
-    return false if usage.nil?
-
-    if usage > @cpu_usage_threshold
-      _log.info("#{log_prefix} [#{Process.pid}] System CPU usage [#{usage}] exceeded threshold [#{@cpu_usage_threshold}], sleeping")
-      return true
-    end
-
-    false
-  end
-
   def get_message_via_drb
     loop do
       begin
@@ -136,7 +117,6 @@ def do_work
     #   so we don't sleep in between messages
     loop do
       heartbeat
-      break if thresholds_exceeded?
       msg = get_message
       break if msg.nil?
       deliver_message(msg)

app/models/miq_worker/container_common.rb

@@ -21,6 +21,7 @@ def configure_worker_deployment(definition, replicas = 0)
       container[:image] = container_image
       container[:env] << {:name => "WORKER_CLASS_NAME", :value => self.class.name}
       container[:env] << {:name => "BUNDLER_GROUPS", :value => self.class.bundler_groups.join(",")}
+      container[:resources] = resource_constraints
     end
 
     def scale_deployment
@@ -40,6 +41,21 @@ def default_image
       "#{container_image_namespace}/#{container_image_name}:#{container_image_tag}"
     end
 
+    def resource_constraints
+      mem_threshold = self.class.worker_settings[:memory_threshold]
+      cpu_threshold = self.class.worker_settings[:cpu_threshold_percent]
+
+      return {} if !Settings.server.worker_monitor.enforce_resource_constraints || (mem_threshold.nil? && cpu_threshold.nil?)
+
+      {:limits => {}}.tap do |h|
+        h[:limits][:memory] = "#{mem_threshold / 1.megabyte}Mi" if mem_threshold
+        if cpu_threshold
+          millicores = ((cpu_threshold / 100.0) * 1000).to_i
+          h[:limits][:cpu] = "#{millicores}m"
+        end
+      end
+    end
+
     def container_image_namespace
       ENV["CONTAINER_IMAGE_NAMESPACE"]
     end

config/settings.yml

@@ -988,6 +988,7 @@
   :worker_messaging_frequency: 5.seconds
   :worker_monitor_frequency: 15.seconds
   :worker_monitor:
+    :enforce_resource_constraints: false
     :kill_algorithm:
       :name: :used_swap_percent_gt_value
       :value: 80
@@ -1079,6 +1080,7 @@
   :worker_base:
     :defaults:
       :count: 1
+      :cpu_threshold_percent: 50
       :gc_interval: 15.minutes
       :heartbeat_freq: 10.seconds
       :heartbeat_timeout: 2.minutes
@@ -1117,7 +1119,6 @@
         :poll: 10.seconds
     :queue_worker_base:
       :defaults:
-        :cpu_usage_threshold: 100.percent
         :dequeue_method: :drb
         :memory_threshold: 500.megabytes
         :poll_method: :normal
@@ -1151,7 +1152,6 @@
         :ems_refresh_worker_microsoft: {}
         :ems_refresh_worker_nuage_network: {}
       :event_handler:
-        :cpu_usage_threshold: 0.percent
         :nice_delta: 7
       :generic_worker:
         :count: 2

spec/models/miq_worker/container_common_spec.rb

@@ -133,4 +133,55 @@ def deployment_name_for(name)
       end
     end
   end
+
+  describe "#resource_constraints" do
+    context "when allowing resource constraints" do
+      before { stub_settings(:server => {:worker_monitor => {:enforce_resource_constraints => true}}) }
+
+      it "returns an empty hash when no thresholds are set" do
+        allow(MiqGenericWorker).to receive(:worker_settings).and_return({})
+        expect(MiqGenericWorker.new.resource_constraints).to eq({})
+      end
+
+      it "returns the correct hash when both values are set" do
+        allow(MiqGenericWorker).to receive(:worker_settings).and_return(:memory_threshold => 500.megabytes, :cpu_threshold_percent => 50)
+        constraints = {
+          :limits => {
+            :memory => "500Mi",
+            :cpu    => "500m"
+          }
+        }
+        expect(MiqGenericWorker.new.resource_constraints).to eq(constraints)
+      end
+
+      it "returns only memory when memory is set" do
+        allow(MiqGenericWorker).to receive(:worker_settings).and_return(:memory_threshold => 500.megabytes)
+        constraints = {
+          :limits => {
+            :memory => "500Mi",
+          }
+        }
+        expect(MiqGenericWorker.new.resource_constraints).to eq(constraints)
+      end
+
+      it "returns only cpu when cpu is set" do
+        allow(MiqGenericWorker).to receive(:worker_settings).and_return(:cpu_threshold_percent => 80)
+        constraints = {
+          :limits => {
+            :cpu => "800m"
+          }
+        }
+        expect(MiqGenericWorker.new.resource_constraints).to eq(constraints)
+      end
+    end
+
+    context "when not allowing resource constraints" do
+      before { stub_settings(:server => {:worker_monitor => {:enforce_resource_constraints => false}}) }
+
+      it "always returns an empty hash" do
+        allow(MiqGenericWorker).to receive(:worker_settings).and_return(:memory_threshold => 500.megabytes, :cpu_threshold => 50)
+        expect(MiqGenericWorker.new.resource_constraints).to eq({})
+      end
+    end
+  end
 end