Publish work queue metrics for the job queues

src/main/groovy/io/seqera/wave/service/data/workqueue/BaseWorkQueue.groovy

@@ -18,8 +18,10 @@
 
 package io.seqera.wave.service.data.workqueue
 
+import io.micronaut.core.annotation.Nullable
 import io.seqera.data.workqueue.AbstractWorkQueue
 import io.seqera.data.workqueue.WorkQueue
+import io.seqera.data.workqueue.metrics.QueueMetrics
 import io.seqera.lang.type.TypeHelper
 import io.seqera.serde.encode.StringEncodingStrategy
 import io.seqera.serde.moshi.MoshiEncodeStrategy
@@ -48,9 +50,11 @@ abstract class BaseWorkQueue<M> extends AbstractWorkQueue<M> {
      *
      * @param target the underlying string-based work queue that handles
      *               the actual message transport and lease management
+     * @param metrics the {@link QueueMetrics} recording the queue activity,
+     *               or {@code null} when metrics are disabled
      */
-    BaseWorkQueue(WorkQueue<String> target) {
-        super(target)
+    BaseWorkQueue(WorkQueue<String> target, @Nullable QueueMetrics metrics) {
+        super(target, metrics)
     }
 
     /**

src/main/groovy/io/seqera/wave/service/data/workqueue/QueueMetricsFactory.groovy

@@ -0,0 +1,45 @@
+/*
+ *  Wave, containers provisioning service
+ *  Copyright (c) 2023-2025, Seqera Labs
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU Affero General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Affero General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Affero General Public License
+ *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.seqera.wave.service.data.workqueue
+
+import groovy.transform.CompileStatic
+import io.micrometer.core.instrument.MeterRegistry
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Requires
+import io.seqera.data.workqueue.metrics.MicrometerQueueMetrics
+import io.seqera.data.workqueue.metrics.QueueMetrics
+import jakarta.inject.Singleton
+
+/**
+ * Provide the {@link QueueMetrics} injected into the Redis work queue, which records the
+ * message lease metrics (leased entries, lease renewal, lost and leaked leases). Without
+ * this bean the Redis work queue records no lease metrics
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+@Factory
+@CompileStatic
+class QueueMetricsFactory {
+
+    @Singleton
+    @Requires(beans = MeterRegistry)
+    QueueMetrics queueMetrics(MeterRegistry registry) {
+        return new MicrometerQueueMetrics(registry, 'wave-job-queue')
+    }
+}

src/main/groovy/io/seqera/wave/service/job/JobPendingQueue.groovy

@@ -22,9 +22,12 @@ import java.time.Duration
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import io.micrometer.core.instrument.MeterRegistry
 import io.micronaut.context.annotation.Requires
+import io.micronaut.core.annotation.Nullable
 import io.seqera.data.workqueue.MessageConsumer
 import io.seqera.data.workqueue.WorkQueue
+import io.seqera.data.workqueue.metrics.MicrometerQueueMetrics
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
 import io.seqera.wave.service.data.workqueue.BaseWorkQueue
@@ -42,19 +45,21 @@ import jakarta.inject.Singleton
 @CompileStatic
 class JobPendingQueue extends BaseWorkQueue<JobSpec> {
 
+    private final static String QUEUE_NAME = 'jobs-pending'
+
     private final static String QUEUE_ID = 'jobs-pending/v2'
 
     private JobManagerConfig config
 
-    JobPendingQueue(WorkQueue<String> target, JobManagerConfig config) {
-        super(target)
+    JobPendingQueue(WorkQueue<String> target, JobManagerConfig config, @Nullable MeterRegistry registry) {
+        super(target, registry ? new MicrometerQueueMetrics(registry, QUEUE_NAME) : null)
         this.config = config
         log.info "Created jobs pending queue - config=${config}"
     }
 
     @Override
     protected String name() {
-        return "jobs-pending"
+        return QUEUE_NAME
     }
 
     @Override

src/main/groovy/io/seqera/wave/service/job/JobProcessingQueue.groovy

@@ -22,9 +22,12 @@ import java.time.Duration
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import io.micrometer.core.instrument.MeterRegistry
 import io.micronaut.context.annotation.Requires
+import io.micronaut.core.annotation.Nullable
 import io.seqera.data.workqueue.MessageConsumer
 import io.seqera.data.workqueue.WorkQueue
+import io.seqera.data.workqueue.metrics.MicrometerQueueMetrics
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
 import io.seqera.wave.service.data.workqueue.BaseWorkQueue
@@ -47,8 +50,8 @@ class JobProcessingQueue extends BaseWorkQueue<JobSpec> {
 
     private final JobManagerConfig config
 
-    JobProcessingQueue(WorkQueue<String> target, JobManagerConfig config) {
-        super(target)
+    JobProcessingQueue(WorkQueue<String> target, JobManagerConfig config, @Nullable MeterRegistry registry) {
+        super(target, registry ? new MicrometerQueueMetrics(registry, QUEUE_NAME) : null)
         this.config = config
         log.info "Created jobs processing queue - config=${config}"
     }