[FS-172266, FS-199069] : Conductor Monitor Changes for queue depth metrics

core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java

@@ -100,10 +100,13 @@ public void sweep(String workflowId) {
         long workflowOffsetTimeout =
                 workflowOffsetWithJitter(properties.getWorkflowOffsetTimeout().getSeconds());
         if (workflow != null) {
-            long startTime = Instant.now().toEpochMilli();
+            // long startTime = Instant.now().toEpochMilli();
             unack(workflow, workflowOffsetTimeout);
-            long endTime = Instant.now().toEpochMilli();
-            Monitors.recordUnackTime(workflow.getWorkflowName(), endTime - startTime);
+            // long endTime = Instant.now().toEpochMilli();
+            // This metrics(recordUnackTime) has workflowName in our prod account we have more number of workflows.
+            // So we may get cardinality problems in haystack due to this.
+            // Disabling this temporarily.
+            // Monitors.recordUnackTime(workflow.getWorkflowName(), endTime - startTime);
         } else {
             LOGGER.warn(
                     "Workflow with {} id can not be found. Attempting to unack using the id",

core/src/main/java/com/netflix/conductor/metrics/WorkflowMonitor.java

@@ -77,48 +77,57 @@ public WorkflowMonitor(
             initialDelayString = "${conductor.workflow-monitor.stats.initial-delay:120000}",
             fixedDelayString = "${conductor.workflow-monitor.stats.delay:60000}")
     public void reportMetrics() {
-        try {
-            if (refreshCounter <= 0) {
+        if (refreshCounter <= 0) {
                 workflowDefs = metadataService.getWorkflowDefs();
                 taskDefs = new ArrayList<>(metadataService.getTaskDefs());
                 refreshCounter = metadataRefreshInterval;
-            }
-
-            getPendingWorkflowToOwnerAppMap(workflowDefs)
-                    .forEach(
-                            (workflowName, ownerApp) -> {
-                                long count =
-                                        executionDAOFacade.getPendingWorkflowCount(workflowName);
-                                Monitors.recordRunningWorkflows(count, workflowName, ownerApp);
-                            });
-
-            taskDefs.forEach(
-                    taskDef -> {
+        }
+
+        // Commented out as we don't have use case for Pending Workflows as of now and hence we don't have pending workflow implementation from
+        // scylla persistence.
+        // try {
+        //         getPendingWorkflowToOwnerAppMap(workflowDefs)
+        //                 .forEach(
+        //                         (workflowName, ownerApp) -> {
+        //                         long count =
+        //                                 executionDAOFacade.getPendingWorkflowCount(workflowName);
+        //                         Monitors.recordRunningWorkflows(count, workflowName, ownerApp);
+        //                         });
+        // } catch (Exception e) {
+        //     LOGGER.error("Error while publishing scheduled metrics", e);
+        // }
+
+        try {
+                taskDefs.forEach(
+                        taskDef -> {
                         long size = queueDAO.getSize(taskDef.getName());
                         long inProgressCount =
                                 executionDAOFacade.getInProgressTaskCount(taskDef.getName());
                         Monitors.recordQueueDepth(taskDef.getName(), size, taskDef.getOwnerApp());
                         if (taskDef.concurrencyLimit() > 0) {
-                            Monitors.recordTaskInProgress(
-                                    taskDef.getName(), inProgressCount, taskDef.getOwnerApp());
+                                Monitors.recordTaskInProgress(
+                                        taskDef.getName(), inProgressCount, taskDef.getOwnerApp());
                         }
-                    });
+                        });
+        } catch (Exception e) {
+                LOGGER.error("Error while publishing scheduled metrics", e);
+        }
 
-            asyncSystemTasks.forEach(
-                    workflowSystemTask -> {
+        try {
+                asyncSystemTasks.forEach(
+                        workflowSystemTask -> {
                         long size = queueDAO.getSize(workflowSystemTask.getTaskType());
                         long inProgressCount =
                                 executionDAOFacade.getInProgressTaskCount(
                                         workflowSystemTask.getTaskType());
                         Monitors.recordQueueDepth(workflowSystemTask.getTaskType(), size, "system");
                         Monitors.recordTaskInProgress(
                                 workflowSystemTask.getTaskType(), inProgressCount, "system");
-                    });
-
-            refreshCounter--;
+                        });
         } catch (Exception e) {
-            LOGGER.error("Error while publishing scheduled metrics", e);
+                LOGGER.error("Error while publishing scheduled metrics", e);
         }
+        refreshCounter--;
     }
 
     /**

server/build.gradle

@@ -45,6 +45,7 @@ dependencies {
     implementation 'com.netflix.conductor:conductor-kafka:3.13.3'
 
     implementation 'io.micrometer:micrometer-registry-prometheus:1.11.0'
+    implementation "com.netflix.spectator:spectator-reg-micrometer:${revSpectator}"
 
     implementation "io.opentelemetry:opentelemetry-extension-annotations:1.18.0"
     implementation "io.opentelemetry:opentelemetry-api:1.35.0"

server/src/main/java/com/netflix/conductor/config/PrometheusIntegrationConfig.java

@@ -0,0 +1,46 @@
+package com.netflix.conductor.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.context.annotation.Configuration;
+
+import com.netflix.spectator.api.Registry;
+import com.netflix.spectator.api.Spectator;
+import com.netflix.spectator.micrometer.MicrometerRegistry;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
+import io.micrometer.prometheus.PrometheusRenameFilter;
+import io.prometheus.client.CollectorRegistry;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.Metrics;
+
+// This class loads all the configurations related to prometheus. 
+@Configuration
+public class PrometheusIntegrationConfig
+        implements CommandLineRunner {
+
+    private static final Logger log = LoggerFactory.getLogger(PrometheusIntegrationConfig.class);
+    private static PrometheusMeterRegistry prometheusRegistry;
+
+    public PrometheusIntegrationConfig(PrometheusMeterRegistry prometheusRegistry) {
+        this.prometheusRegistry = prometheusRegistry;
+    }
+
+    @Override
+    public void run(String... args) throws Exception {
+        setupPrometheusRegistry();
+    }
+
+    /**
+     * To Register PrometheusRegistry
+    */
+    private static void setupPrometheusRegistry() {
+        log.info("Registered PrometheusRegistry");
+        final MicrometerRegistry metricsRegistry = new MicrometerRegistry(prometheusRegistry);
+        prometheusRegistry.config().meterFilter(new PrometheusRenameFilter());
+        Spectator.globalRegistry().add(metricsRegistry);
+    }
+
+}