Adaptive worker polling and concurrency

conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollector.java

@@ -25,6 +25,7 @@
 import com.netflix.conductor.client.events.task.TaskPayloadUsedEvent;
 import com.netflix.conductor.client.events.task.TaskResultPayloadSizeEvent;
 import com.netflix.conductor.client.events.taskrunner.ActiveWorkersChanged;
+import com.netflix.conductor.client.events.taskrunner.AdaptiveConcurrencyChanged;
 import com.netflix.conductor.client.events.taskrunner.PollCompleted;
 import com.netflix.conductor.client.events.taskrunner.PollFailure;
 import com.netflix.conductor.client.events.taskrunner.PollStarted;
@@ -88,6 +89,7 @@ public class PrometheusMetricsCollector implements MetricsCollector {
     private final PrometheusMeterRegistry registry;
     private final PrometheusApiClientMetrics apiClientMetrics;
     private final ConcurrentHashMap<String, AtomicInteger> activeWorkerGauges = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, AtomicInteger> adaptiveGauges = new ConcurrentHashMap<>();
 
     public PrometheusMetricsCollector() {
         this(SHARED_REGISTRY);
@@ -196,6 +198,31 @@ public void consume(ActiveWorkersChanged e) {
                 .set(Math.max(0, e.getCount()));
     }
 
+    // ----- Adaptive concurrency -----
+
+    @Override
+    public void consume(AdaptiveConcurrencyChanged e) {
+        String taskType = nullToEmpty(e.getTaskType());
+        adaptiveGauge("adaptive_worker_concurrency_target", "Adaptive concurrency share of the worker",
+                taskType).set(e.getTargetConcurrency());
+        adaptiveGauge("adaptive_worker_held_slots", "Slots held by the worker's polls and tasks",
+                taskType).set(e.getHeldSlots());
+        adaptiveGauge("adaptive_worker_task_slot_age_seconds",
+                "Seconds since the worker last released a task slot, while any are held; a wedged worker grows this",
+                taskType).set((int) Math.min(Integer.MAX_VALUE, e.getTaskSlotAgeSeconds()));
+        adaptiveGauge("adaptive_queue_depth", "Last sampled queue depth of the worker's task queue",
+                taskType).set(e.getQueueDepth());
+        adaptiveGauge("adaptive_process_concurrency_target", "Adaptive concurrency target of the process",
+                null).set(e.getProcessTargetConcurrency());
+        adaptiveGauge("adaptive_process_held_slots", "Slots held by all adaptive workers of the process",
+                null).set(e.getProcessHeldSlots());
+        adaptiveGauge("adaptive_admission_paused", "1 while task acquisition is paused by heap pressure",
+                null).set(e.isAdmissionPaused() ? 1 : 0);
+        counter("adaptive_concurrency_changes_total", "Adaptive concurrency observations by reason",
+                "taskType", taskType,
+                "reason", nullToEmpty(e.getReason())).increment();
+    }
+
     // ----- Task update -----
 
     @Override
@@ -291,6 +318,19 @@ public void consume(WorkflowStartedEvent event) {
 
     // ----- Helpers -----
 
+    private AtomicInteger adaptiveGauge(String name, String description, String taskType) {
+        return adaptiveGauges.computeIfAbsent(taskType == null ? name : name + "|" + taskType, k -> {
+            AtomicInteger val = new AtomicInteger(0);
+            Gauge.Builder<AtomicInteger> builder = Gauge.builder(name, val, AtomicInteger::doubleValue)
+                    .description(description);
+            if (taskType != null) {
+                builder.tag("taskType", taskType);
+            }
+            builder.register(registry);
+            return val;
+        });
+    }
+
     private Counter counter(String name, String description, String... tagKv) {
         return Counter.builder(name)
                 .description(description)

conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollectorTest.java

@@ -23,6 +23,7 @@
 import com.netflix.conductor.client.events.task.TaskPayloadUsedEvent;
 import com.netflix.conductor.client.events.task.TaskResultPayloadSizeEvent;
 import com.netflix.conductor.client.events.taskrunner.ActiveWorkersChanged;
+import com.netflix.conductor.client.events.taskrunner.AdaptiveConcurrencyChanged;
 import com.netflix.conductor.client.events.taskrunner.PollCompleted;
 import com.netflix.conductor.client.events.taskrunner.PollFailure;
 import com.netflix.conductor.client.events.taskrunner.PollStarted;
@@ -134,6 +135,21 @@ void activeWorkersChangedSetsGauge() {
         assertEquals(7.0, val);
     }
 
+    @Test
+    void adaptiveConcurrencyChangedUpdatesGauges() {
+        collector.consume(new AdaptiveConcurrencyChanged("SIMPLE", null, 29, 12, 3, 48, 32, 30, false, "DEPTH_SAMPLE"));
+        collector.consume(new AdaptiveConcurrencyChanged("SIMPLE", null, 24, 12, 7, 48, 24, 30, true, "HEAP_PAUSE"));
+
+        assertEquals(24.0, registry.get("adaptive_worker_concurrency_target").tag("taskType", "SIMPLE").gauge().value());
+        assertEquals(12.0, registry.get("adaptive_worker_held_slots").tag("taskType", "SIMPLE").gauge().value());
+        assertEquals(7.0, registry.get("adaptive_worker_task_slot_age_seconds").tag("taskType", "SIMPLE").gauge().value());
+        assertEquals(48.0, registry.get("adaptive_queue_depth").tag("taskType", "SIMPLE").gauge().value());
+        assertEquals(24.0, registry.get("adaptive_process_concurrency_target").gauge().value());
+        assertEquals(1.0, registry.get("adaptive_admission_paused").gauge().value());
+        assertEquals(1.0, registry.get("adaptive_concurrency_changes_total")
+                .tag("taskType", "SIMPLE").tag("reason", "HEAP_PAUSE").counter().count());
+    }
+
     @Test
     void activeWorkersChangedClampsNegativeToZero() {
         collector.consume(new ActiveWorkersChanged("SIMPLE", -1));

conductor-client-spring-boot4/src/main/java/com/netflix/conductor/client/spring/ClientProperties.java

@@ -18,6 +18,8 @@
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
+import com.netflix.conductor.client.automator.AdaptiveConcurrencyConfig;
+
 @ConfigurationProperties("conductor.client")
 public class ClientProperties {
 
@@ -51,6 +53,229 @@ public void setWrite(int write) {
         }
     }
 
+    public static class AdaptiveConcurrency {
+
+        // Enables adaptive concurrency for workers started by this client
+        private boolean enabled = false;
+
+        // Aggregate admission ceiling for all adaptive workers
+        private int maxProcessConcurrency = 256;
+
+        // Initial process target; 0 uses min(logical queues, available processors)
+        private int initialProcessConcurrency = 0;
+
+        // Lowest process target an idle process decays to; CPU and heap pressure may go below it
+        private int minProcessConcurrency = 1;
+
+        // Per-worker ceiling for workers without their own
+        private int defaultMaxWorkerConcurrency = 64;
+
+        // Ceiling for the number of tasks requested by one poll
+        private int maxPollBatch = 16;
+
+        private Duration resourceSampleInterval = Duration.ofSeconds(1);
+
+        private Duration metricFreshness = Duration.ofSeconds(5);
+
+        private Duration depthSampleInterval = Duration.ofSeconds(1);
+
+        private Duration depthFreshness = Duration.ofSeconds(5);
+
+        private Duration heartbeatInterval = Duration.ofSeconds(5);
+
+        private Duration standbyPollTimeout = Duration.ofSeconds(1);
+
+        private Duration controlInterval = Duration.ofSeconds(5);
+
+        private double cpuSmoothingAlpha = 0.2;
+
+        private double growthThreshold = 0.75;
+
+        private double cpuPressureThreshold = 0.90;
+
+        private double heapPauseThreshold = 0.90;
+
+        private double heapResumeThreshold = 0.80;
+
+        // Allow running without CPU and heap metrics
+        private boolean demandOnlyEnabled = false;
+
+        public boolean isEnabled() {
+            return enabled;
+        }
+
+        public void setEnabled(boolean enabled) {
+            this.enabled = enabled;
+        }
+
+        public int getMaxProcessConcurrency() {
+            return maxProcessConcurrency;
+        }
+
+        public void setMaxProcessConcurrency(int maxProcessConcurrency) {
+            this.maxProcessConcurrency = maxProcessConcurrency;
+        }
+
+        public int getInitialProcessConcurrency() {
+            return initialProcessConcurrency;
+        }
+
+        public void setInitialProcessConcurrency(int initialProcessConcurrency) {
+            this.initialProcessConcurrency = initialProcessConcurrency;
+        }
+
+        public int getMinProcessConcurrency() {
+            return minProcessConcurrency;
+        }
+
+        public void setMinProcessConcurrency(int minProcessConcurrency) {
+            this.minProcessConcurrency = minProcessConcurrency;
+        }
+
+        public int getDefaultMaxWorkerConcurrency() {
+            return defaultMaxWorkerConcurrency;
+        }
+
+        public void setDefaultMaxWorkerConcurrency(int defaultMaxWorkerConcurrency) {
+            this.defaultMaxWorkerConcurrency = defaultMaxWorkerConcurrency;
+        }
+
+        public int getMaxPollBatch() {
+            return maxPollBatch;
+        }
+
+        public void setMaxPollBatch(int maxPollBatch) {
+            this.maxPollBatch = maxPollBatch;
+        }
+
+        public Duration getResourceSampleInterval() {
+            return resourceSampleInterval;
+        }
+
+        public void setResourceSampleInterval(Duration resourceSampleInterval) {
+            this.resourceSampleInterval = resourceSampleInterval;
+        }
+
+        public Duration getMetricFreshness() {
+            return metricFreshness;
+        }
+
+        public void setMetricFreshness(Duration metricFreshness) {
+            this.metricFreshness = metricFreshness;
+        }
+
+        public Duration getDepthSampleInterval() {
+            return depthSampleInterval;
+        }
+
+        public void setDepthSampleInterval(Duration depthSampleInterval) {
+            this.depthSampleInterval = depthSampleInterval;
+        }
+
+        public Duration getDepthFreshness() {
+            return depthFreshness;
+        }
+
+        public void setDepthFreshness(Duration depthFreshness) {
+            this.depthFreshness = depthFreshness;
+        }
+
+        public Duration getHeartbeatInterval() {
+            return heartbeatInterval;
+        }
+
+        public void setHeartbeatInterval(Duration heartbeatInterval) {
+            this.heartbeatInterval = heartbeatInterval;
+        }
+
+        public Duration getStandbyPollTimeout() {
+            return standbyPollTimeout;
+        }
+
+        public void setStandbyPollTimeout(Duration standbyPollTimeout) {
+            this.standbyPollTimeout = standbyPollTimeout;
+        }
+
+        public Duration getControlInterval() {
+            return controlInterval;
+        }
+
+        public void setControlInterval(Duration controlInterval) {
+            this.controlInterval = controlInterval;
+        }
+
+        public double getCpuSmoothingAlpha() {
+            return cpuSmoothingAlpha;
+        }
+
+        public void setCpuSmoothingAlpha(double cpuSmoothingAlpha) {
+            this.cpuSmoothingAlpha = cpuSmoothingAlpha;
+        }
+
+        public double getGrowthThreshold() {
+            return growthThreshold;
+        }
+
+        public void setGrowthThreshold(double growthThreshold) {
+            this.growthThreshold = growthThreshold;
+        }
+
+        public double getCpuPressureThreshold() {
+            return cpuPressureThreshold;
+        }
+
+        public void setCpuPressureThreshold(double cpuPressureThreshold) {
+            this.cpuPressureThreshold = cpuPressureThreshold;
+        }
+
+        public double getHeapPauseThreshold() {
+            return heapPauseThreshold;
+        }
+
+        public void setHeapPauseThreshold(double heapPauseThreshold) {
+            this.heapPauseThreshold = heapPauseThreshold;
+        }
+
+        public double getHeapResumeThreshold() {
+            return heapResumeThreshold;
+        }
+
+        public void setHeapResumeThreshold(double heapResumeThreshold) {
+            this.heapResumeThreshold = heapResumeThreshold;
+        }
+
+        public boolean isDemandOnlyEnabled() {
+            return demandOnlyEnabled;
+        }
+
+        public void setDemandOnlyEnabled(boolean demandOnlyEnabled) {
+            this.demandOnlyEnabled = demandOnlyEnabled;
+        }
+
+        public AdaptiveConcurrencyConfig toConfig() {
+            return new AdaptiveConcurrencyConfig.Builder()
+                    .withMaxProcessConcurrency(maxProcessConcurrency)
+                    .withInitialProcessConcurrency(initialProcessConcurrency)
+                    .withMinProcessConcurrency(minProcessConcurrency)
+                    .withDefaultMaxWorkerConcurrency(defaultMaxWorkerConcurrency)
+                    .withMaxPollBatch(maxPollBatch)
+                    .withResourceSampleInterval(resourceSampleInterval)
+                    .withMetricFreshness(metricFreshness)
+                    .withDepthSampleInterval(depthSampleInterval)
+                    .withDepthFreshness(depthFreshness)
+                    .withHeartbeatInterval(heartbeatInterval)
+                    .withStandbyPollTimeout(standbyPollTimeout)
+                    .withControlInterval(controlInterval)
+                    .withCpuSmoothingAlpha(cpuSmoothingAlpha)
+                    .withGrowthThreshold(growthThreshold)
+                    .withCpuPressureThreshold(cpuPressureThreshold)
+                    .withHeapPauseThreshold(heapPauseThreshold)
+                    .withHeapResumeThreshold(heapResumeThreshold)
+                    .withDemandOnlyEnabled(demandOnlyEnabled)
+                    .build();
+        }
+    }
+
     private String rootUri;
 
     private String basePath;
@@ -75,6 +300,36 @@ public void setWrite(int write) {
 
     private boolean verifyingSsl = true;
 
+    private boolean useVirtualThreads = false;
+
+    private boolean updateV2 = false;
+
+    private AdaptiveConcurrency adaptiveConcurrency = new AdaptiveConcurrency();
+
+    public boolean isUseVirtualThreads() {
+        return useVirtualThreads;
+    }
+
+    public void setUseVirtualThreads(boolean useVirtualThreads) {
+        this.useVirtualThreads = useVirtualThreads;
+    }
+
+    public boolean isUpdateV2() {
+        return updateV2;
+    }
+
+    public void setUpdateV2(boolean updateV2) {
+        this.updateV2 = updateV2;
+    }
+
+    public AdaptiveConcurrency getAdaptiveConcurrency() {
+        return adaptiveConcurrency;
+    }
+
+    public void setAdaptiveConcurrency(AdaptiveConcurrency adaptiveConcurrency) {
+        this.adaptiveConcurrency = adaptiveConcurrency;
+    }
+
     public void setVerifyingSsl(boolean verifyingSsl) {
         this.verifyingSsl = verifyingSsl;
     }