diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 499ced67c..719183c83 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -9,6 +9,11 @@ on: - main workflow_dispatch: +permissions: + contents: read + checks: write + pull-requests: write + concurrency: group: ${{ github.workflow }}-${{ github.ref }} cancel-in-progress: true diff --git a/INTERCEPTOR.md b/INTERCEPTOR.md index 14d603703..81fe78515 100644 --- a/INTERCEPTOR.md +++ b/INTERCEPTOR.md @@ -399,10 +399,14 @@ import com.netflix.conductor.client.http.ConductorClient; import com.netflix.conductor.client.http.TaskClient; import com.netflix.conductor.client.http.WorkflowClient; import com.netflix.conductor.client.metrics.prometheus.PrometheusMetricsCollector; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; -// 1. Create and start metrics -PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(); -metricsCollector.startServer(); // port 9991, /metrics +// 1. Create a registry and a collector bound to it. The SDK does not start a +// web server; expose registry.scrape() from an endpoint you own (or, in +// Spring Boot, hand in the app's MeterRegistry and use Actuator). +PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(registry); // 2. Create ConductorClient — withMetricsCollector installs the HTTP interceptor // and enables automatic listener registration on downstream clients @@ -427,12 +431,23 @@ For fine-grained control over which listeners are registered, create the `Conduc ### Custom Metrics Endpoint +The SDK no longer serves metrics itself. Expose the registry from an endpoint your application owns: + ```java -// Start Prometheus server on custom port and endpoint -PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(); -metricsCollector.startServer(8080, "/custom-metrics"); +PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(registry); + +var server = com.sun.net.httpserver.HttpServer.create(new java.net.InetSocketAddress(8080), 0); +server.createContext("/custom-metrics", exchange -> { + byte[] body = registry.scrape().getBytes(); + exchange.sendResponseHeaders(200, body.length); + try (var os = exchange.getResponseBody()) { os.write(body); } +}); +server.start(); ``` +In Spring Boot, prefer Actuator: hand the application's `MeterRegistry` to the collector (done automatically by `ConductorMetricsAutoConfiguration`) and configure the scrape path/port with `management.*` properties. + ### Registering Custom Event Listeners #### Approach 1: Using Builder's Listener API @@ -908,8 +923,9 @@ public class CloudWatchMetricsCollector implements MetricsCollector { ```java // Create multiple collectors -PrometheusMetricsCollector prometheus = new PrometheusMetricsCollector(); -prometheus.startServer(9991, "/metrics"); +PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector prometheus = new PrometheusMetricsCollector(prometheusRegistry); +// Expose prometheusRegistry.scrape() from an endpoint your application owns. DatadogMetricsCollector datadog = new DatadogMetricsCollector( System.getenv("DATADOG_API_KEY"), @@ -1293,9 +1309,9 @@ public class ConductorMonitoringSetup { new AnotherTaskWorker() ); - // 3. Setup Prometheus metrics - PrometheusMetricsCollector prometheus = new PrometheusMetricsCollector(); - prometheus.startServer(9991, "/metrics"); + // 3. Setup Prometheus metrics (expose the registry from an endpoint you own) + PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); + PrometheusMetricsCollector prometheus = new PrometheusMetricsCollector(prometheusRegistry); // 4. Setup custom monitoring SLAMonitor slaMonitor = new SLAMonitor( @@ -1443,12 +1459,11 @@ public class EventSystemTest { **Problem**: Prometheus endpoint returns no metrics **Solution**: -1. Verify server is started: - ```java - metricsCollector.startServer(); // Don't forget this! - ``` +1. Verify the registry is exposed. The SDK does not start a server: + - Spring Boot: confirm `spring-boot-starter-actuator` + `micrometer-registry-prometheus` are present and the `prometheus` endpoint is exposed (`/actuator/prometheus` by default). + - Plain Java: confirm you are serving `registry.scrape()` from your own HTTP endpoint, and that the same `registry` was passed to `new PrometheusMetricsCollector(registry)`. -2. Check port availability: +2. Check port availability for whatever port your app exposes, e.g.: ```bash lsof -i :9991 ``` diff --git a/README.md b/README.md index 106833cb6..737efbf53 100644 --- a/README.md +++ b/README.md @@ -298,7 +298,7 @@ executor.initWorkers("com.mycompany.workers"); // Package to scan for @WorkerTa ## Monitoring Workers -Enable Prometheus metrics collection for monitoring workers: +Collect Micrometer/Prometheus metrics for monitoring workers. The SDK publishes into a `MeterRegistry` you provide; it does not start a metrics web server. ```groovy // Using conductor-client-metrics module @@ -307,11 +307,25 @@ dependencies { } ``` +**Spring Boot:** add Actuator and the Prometheus registry, and SDK metrics appear on `/actuator/prometheus` automatically (no wiring code needed): + +```groovy +dependencies { + implementation 'org.conductoross:conductor-client-metrics:5.1.0' + implementation 'org.springframework.boot:spring-boot-starter-actuator' + implementation 'io.micrometer:micrometer-registry-prometheus' +} +``` + +**Plain Java:** create a registry, pass it to the collector, and expose it yourself: + ```java import com.netflix.conductor.client.metrics.prometheus.PrometheusMetricsCollector; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; -PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(); -metricsCollector.startServer(); // http://localhost:9991/metrics +PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(registry); ConductorClient client = ConductorClient.builder() .basePath("...") @@ -324,6 +338,8 @@ WorkflowClient workflowClient = new WorkflowClient(client); TaskRunnerConfigurer configurer = new TaskRunnerConfigurer.Builder(taskClient, workers) .withThreadCount(10) .build(); + +// Expose registry.scrape() from an HTTP endpoint your application owns. ``` When a `MetricsCollector` is attached to the `ConductorClient`, downstream clients (`TaskClient`, `WorkflowClient`, `TaskRunnerConfigurer`) automatically register themselves as event listeners. diff --git a/conductor-client-metrics/README.md b/conductor-client-metrics/README.md index 9ef6e767a..463f7eacf 100644 --- a/conductor-client-metrics/README.md +++ b/conductor-client-metrics/README.md @@ -1,6 +1,13 @@ # Conductor Client Metrics -The `conductor-client-metrics` module provides Prometheus metrics for Java SDK clients and workers. It helps operators monitor worker polling, task execution, task result updates, payload sizes, workflow starts, and HTTP client latency. +The `conductor-client-metrics` module records Micrometer metrics for Java SDK clients and workers. It helps operators monitor worker polling, task execution, task result updates, payload sizes, workflow starts, and HTTP client latency. + +The SDK **should not start a metrics web server**. Metrics are published into a Micrometer `MeterRegistry` that you supply, and exposing them for scraping is the responsibility of the embedding application: + +- **Spring Boot apps** hand in the application's `MeterRegistry` (the one backing `/actuator/prometheus`), so SDK metrics appear alongside the app's own metrics. This module ships a Spring Boot auto-configuration that does this automatically. +- **Plain Java apps** create a `PrometheusMeterRegistry` (or any other registry), pass it to the collector, and expose it however they like (e.g. serve `registry.scrape()` from an HTTP endpoint they own). + +> **Deprecated:** the no-argument `new PrometheusMetricsCollector()` plus `startServer()` still spin up a minimal embedded scrape server for backward compatibility, but both are deprecated and will be removed in a future major release. Prefer supplying a `MeterRegistry`. ## Installation @@ -14,13 +21,48 @@ dependencies { ## Usage -Create a `PrometheusMetricsCollector`, start the scrape server, and pass the collector to `ConductorClient.Builder`. All downstream clients and the task runner auto-register themselves as listeners. +### Spring Boot (recommended) + +Add this module and Spring Boot Actuator with the Prometheus registry to a Spring Boot worker app: + +```groovy +dependencies { + implementation 'org.conductoross:conductor-client-metrics:5.1.0' + implementation 'org.springframework.boot:spring-boot-starter-actuator' + implementation 'io.micrometer:micrometer-registry-prometheus' +} +``` + +That's it. When a `MeterRegistry` bean is present, `ConductorMetricsAutoConfiguration` creates a `MetricsCollector` bound to it, and the Conductor Spring auto-configuration wires that collector into the client and task runner. SDK metrics then show up on the app's existing scrape endpoint (by default `/actuator/prometheus`). + +To serve them at a custom path/port, use the standard Actuator properties, e.g.: + +```yaml +server: + port: 9991 +management: + endpoints: + web: + base-path: / + exposure: + include: health,prometheus + path-mapping: + prometheus: metrics # scrape endpoint at /metrics +``` + +To provide your own collector (or a non-Prometheus registry), just declare a `MetricsCollector` bean; the auto-configuration backs off (`@ConditionalOnMissingBean`). + +### Plain Java + +Create a registry, pass it to the collector, and expose the registry yourself. Passing the collector to `ConductorClient.Builder` auto-registers all downstream clients and the task runner as listeners. ```java import com.netflix.conductor.client.metrics.prometheus.PrometheusMetricsCollector; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; -PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(); -metricsCollector.startServer(); // http://localhost:9991/metrics +PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(registry); ConductorClient client = ConductorClient.builder() .basePath("http://conductor-server:8080/api") @@ -34,9 +76,18 @@ TaskRunnerConfigurer configurer = new TaskRunnerConfigurer.Builder(taskClient, w .withThreadCount(10) .build(); configurer.init(); + +// Expose the metrics from an HTTP endpoint you own, e.g.: +// var server = com.sun.net.httpserver.HttpServer.create(new InetSocketAddress(9991), 0); +// server.createContext("/metrics", exchange -> { +// byte[] body = registry.scrape().getBytes(); +// exchange.sendResponseHeaders(200, body.length); +// try (var os = exchange.getResponseBody()) { os.write(body); } +// }); +// server.start(); ``` -`startServer()` also accepts `(port, endpoint)` for custom scrape configurations. The client builder accepts the usual timeouts, SSL, authentication, and other options alongside `withMetricsCollector` -- none of them change how metrics wiring works. +The client builder accepts the usual timeouts, SSL, authentication, and other options alongside `withMetricsCollector` -- none of them change how metrics wiring works. ### How Auto-Registration Works @@ -57,9 +108,11 @@ For advanced use cases where you need fine-grained control over which listeners ```java import com.netflix.conductor.client.metrics.prometheus.PrometheusMetricsCollector; +import io.micrometer.prometheusmetrics.PrometheusConfig; +import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; -PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(); -metricsCollector.startServer(); // http://localhost:9991/metrics +PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); +PrometheusMetricsCollector metricsCollector = new PrometheusMetricsCollector(registry); ConductorClient client = ConductorClient.builder() .basePath("http://conductor-server:8080/api") @@ -168,9 +221,9 @@ Micrometer publishes a `*_max` Gauge alongside every Timer and DistributionSumma ### Metrics Are Empty -- Verify that the collector is wired into the client. The simplest check: was `withMetricsCollector` called on `ConductorClient.Builder`? +- Verify that the collector is wired into the client. In Spring Boot, confirm a `MeterRegistry` bean exists (add `spring-boot-starter-actuator` + `micrometer-registry-prometheus`). In plain Java, confirm `withMetricsCollector` was called on `ConductorClient.Builder`. - Verify workers have polled or executed tasks. Metrics are created lazily when the relevant event occurs. -- Confirm the scrape endpoint is reachable at the expected host and port. +- Confirm the scrape endpoint is reachable at the expected host and port (Actuator's `/actuator/prometheus` by default, or whatever your app exposes). ### Missing HTTP or Size Metrics diff --git a/conductor-client-metrics/build.gradle b/conductor-client-metrics/build.gradle index 9a4e2e3e0..940f426c0 100644 --- a/conductor-client-metrics/build.gradle +++ b/conductor-client-metrics/build.gradle @@ -17,9 +17,20 @@ dependencies { implementation project(":conductor-client") implementation "com.squareup.okhttp3:okhttp:${versions.okHttp}" + // Optional Spring Boot auto-configuration. Present only at compile time so + // plain-Java consumers pull in no Spring dependency; the auto-config only + // activates when a Spring Boot app already has spring-boot-autoconfigure + // (and a Micrometer MeterRegistry bean) on its runtime classpath. + compileOnly 'org.springframework.boot:spring-boot-autoconfigure:3.5.13' + testImplementation 'org.mockito:mockito-core:5.12.0' testImplementation 'org.junit.jupiter:junit-jupiter-api:5.10.3' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.10.3' + + // Exercises the Spring Boot auto-configuration in isolation. + testImplementation 'org.springframework.boot:spring-boot-autoconfigure:3.5.13' + testImplementation 'org.springframework.boot:spring-boot-test:3.5.13' + testImplementation 'org.assertj:assertj-core:3.25.3' } java { diff --git a/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfiguration.java b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfiguration.java new file mode 100644 index 000000000..d1ad0994d --- /dev/null +++ b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfiguration.java @@ -0,0 +1,52 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package com.netflix.conductor.client.metrics.prometheus; + +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; + +import com.netflix.conductor.client.metrics.MetricsCollector; + +import io.micrometer.core.instrument.MeterRegistry; + +/** + * Spring Boot auto-configuration that exposes the SDK's {@link MetricsCollector} + * bound to the application's Micrometer {@link MeterRegistry}. + * + *

When a {@code MeterRegistry} bean exists (e.g. the one Spring Boot Actuator + * creates to back {@code /actuator/prometheus}), the SDK's metrics are published + * into it and therefore show up alongside the application's own metrics — + * no separate metrics web server is started by the SDK. + * + *

This class is only loaded in a Spring Boot application (it is referenced + * from {@code META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports}). + * Plain-Java consumers never trigger it and pull in no Spring dependency. + */ +@AutoConfiguration( + afterName = { + "org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration", + "org.springframework.boot.actuate.autoconfigure.metrics.export.prometheus.PrometheusMetricsExportAutoConfiguration" + }) +@ConditionalOnClass(MeterRegistry.class) +public class ConductorMetricsAutoConfiguration { + + @Bean + @ConditionalOnBean(MeterRegistry.class) + @ConditionalOnMissingBean(MetricsCollector.class) + public MetricsCollector conductorMetricsCollector(MeterRegistry meterRegistry) { + return new PrometheusMetricsCollector(meterRegistry); + } +} diff --git a/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusApiClientMetrics.java b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusApiClientMetrics.java index 7ebd403ec..252579ea2 100644 --- a/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusApiClientMetrics.java +++ b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusApiClientMetrics.java @@ -17,8 +17,8 @@ import com.netflix.conductor.client.metrics.ApiClientMetrics; import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Timer; -import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; /** * Prometheus-backed implementation of {@link ApiClientMetrics} that emits the @@ -47,9 +47,9 @@ public final class PrometheusApiClientMetrics implements ApiClientMetrics { 100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000 }; - private final PrometheusMeterRegistry registry; + private final MeterRegistry registry; - public PrometheusApiClientMetrics(PrometheusMeterRegistry registry) { + public PrometheusApiClientMetrics(MeterRegistry registry) { this.registry = registry; } diff --git a/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollector.java b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollector.java index 1d99e6df1..4a84f8d59 100644 --- a/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollector.java +++ b/conductor-client-metrics/src/main/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollector.java @@ -47,13 +47,36 @@ import com.sun.net.httpserver.HttpServer; import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Timer; import io.micrometer.prometheusmetrics.PrometheusConfig; import io.micrometer.prometheusmetrics.PrometheusMeterRegistry; /** - * Prometheus metrics implementation emitting the harmonized metric names - * defined in the cross-SDK metrics catalog. + * Micrometer-backed {@link MetricsCollector} that records the harmonized metric + * names defined in the cross-SDK metrics catalog into a caller-supplied + * {@link MeterRegistry}. + * + *

The recommended usage is to not let the SDK start any HTTP server and + * instead let the embedding application expose the metrics: + * + *

+ * + *

For backward compatibility, the no-argument constructor plus + * {@link #startServer()} still spin up a minimal embedded scrape server, but + * both are deprecated — a library should not run a web server. Prefer the + * {@link #PrometheusMetricsCollector(MeterRegistry)} constructor. */ public class PrometheusMetricsCollector implements MetricsCollector { @@ -85,10 +108,33 @@ public class PrometheusMetricsCollector implements MetricsCollector { private static final String STATUS_SUCCESS = "SUCCESS"; private static final String STATUS_FAILURE = "FAILURE"; - private final PrometheusMeterRegistry registry; + private final MeterRegistry registry; private final PrometheusApiClientMetrics apiClientMetrics; private final ConcurrentHashMap activeWorkerGauges = new ConcurrentHashMap<>(); + /** + * Create a collector that records into the supplied {@link MeterRegistry}. + * + * @param registry the Micrometer registry to publish metrics into; in a + * Spring Boot app this is typically the application's + * registry backing {@code /actuator/prometheus}. + */ + public PrometheusMetricsCollector(MeterRegistry registry) { + this.registry = registry; + this.apiClientMetrics = new PrometheusApiClientMetrics(registry); + } + + /** + * Create a collector backed by a shared internal {@link PrometheusMeterRegistry} + * and expose it via {@link #startServer()}. + * + * @deprecated A library should not run its own web server. Pass in the + * application's {@link MeterRegistry} via + * {@link #PrometheusMetricsCollector(MeterRegistry)} instead (in Spring + * Boot this is auto-configured), and let the application expose the + * scrape endpoint (e.g. Actuator's {@code /actuator/prometheus}). + */ + @Deprecated public PrometheusMetricsCollector() { this(SHARED_REGISTRY); if (instantiated.getAndSet(true)) { @@ -99,29 +145,54 @@ public PrometheusMetricsCollector() { } } - /** Package-private constructor for test isolation. */ - PrometheusMetricsCollector(PrometheusMeterRegistry registry) { - this.registry = registry; - this.apiClientMetrics = new PrometheusApiClientMetrics(registry); - } - @Override public ApiClientMetrics getApiClientMetrics() { return apiClientMetrics; } - public PrometheusMeterRegistry getRegistry() { + public MeterRegistry getRegistry() { return registry; } + /** + * Start a minimal embedded HTTP server exposing the Prometheus scrape + * endpoint at {@code http://localhost:9991/metrics}. + * + * @deprecated The SDK should not run a web server. Expose metrics from the + * embedding application instead (Spring Boot Actuator, or serve + * {@link #getRegistry()}'s {@code scrape()} from your own endpoint). + * @throws IOException if the server cannot bind + * @throws IllegalStateException if this collector was constructed with a + * non-Prometheus {@link MeterRegistry} (such a registry cannot be + * scraped by this server; expose it through your application instead) + */ + @Deprecated public void startServer() throws IOException { startServer(DEFAULT_PORT, DEFAULT_ENDPOINT); } + /** + * Start a minimal embedded HTTP server exposing the Prometheus scrape + * endpoint on the given port and path. + * + * @deprecated The SDK should not run a web server. Expose metrics from the + * embedding application instead (Spring Boot Actuator, or serve + * {@link #getRegistry()}'s {@code scrape()} from your own endpoint). + * @throws IOException if the server cannot bind + * @throws IllegalStateException if this collector was constructed with a + * non-Prometheus {@link MeterRegistry} + */ + @Deprecated public void startServer(int port, String endpoint) throws IOException { + if (!(registry instanceof PrometheusMeterRegistry prometheusRegistry)) { + throw new IllegalStateException( + "startServer() requires a PrometheusMeterRegistry, but this collector was " + + "constructed with " + registry.getClass().getName() + ". Expose the " + + "registry through your application (e.g. Spring Boot Actuator) instead."); + } var server = HttpServer.create(new InetSocketAddress(port), 0); server.createContext(endpoint, exchange -> { - var body = registry.scrape(); + var body = prometheusRegistry.scrape(); exchange.getResponseHeaders().set("Content-Type", "text/plain"); exchange.sendResponseHeaders(200, body.getBytes().length); try (var os = exchange.getResponseBody()) { diff --git a/conductor-client-metrics/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/conductor-client-metrics/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 000000000..38e9b4d7b --- /dev/null +++ b/conductor-client-metrics/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +com.netflix.conductor.client.metrics.prometheus.ConductorMetricsAutoConfiguration diff --git a/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfigurationTest.java b/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfigurationTest.java new file mode 100644 index 000000000..a95434cf3 --- /dev/null +++ b/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/ConductorMetricsAutoConfigurationTest.java @@ -0,0 +1,58 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package com.netflix.conductor.client.metrics.prometheus; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import com.netflix.conductor.client.metrics.MetricsCollector; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class ConductorMetricsAutoConfigurationTest { + + private final ApplicationContextRunner runner = new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(ConductorMetricsAutoConfiguration.class)); + + @Test + void registersCollectorBoundToApplicationRegistry() { + SimpleMeterRegistry registry = new SimpleMeterRegistry(); + runner.withBean(MeterRegistry.class, () -> registry).run(ctx -> { + assertTrue(ctx.containsBean("conductorMetricsCollector")); + MetricsCollector collector = ctx.getBean(MetricsCollector.class); + assertSame(registry, ((PrometheusMetricsCollector) collector).getRegistry()); + }); + } + + @Test + void doesNotRegisterCollectorWithoutMeterRegistry() { + runner.run(ctx -> assertFalse(ctx.containsBean("conductorMetricsCollector"))); + } + + @Test + void backsOffWhenApplicationSuppliesItsOwnCollector() { + runner.withBean(MeterRegistry.class, SimpleMeterRegistry::new) + .withBean("customCollector", MetricsCollector.class, + () -> new PrometheusMetricsCollector(new SimpleMeterRegistry())) + .run(ctx -> { + assertTrue(ctx.containsBean("customCollector")); + assertFalse(ctx.containsBean("conductorMetricsCollector")); + }); + } +} diff --git a/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollectorTest.java b/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollectorTest.java index c39c7c268..07b3c7d75 100644 --- a/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollectorTest.java +++ b/conductor-client-metrics/src/test/java/com/netflix/conductor/client/metrics/prometheus/PrometheusMetricsCollectorTest.java @@ -297,6 +297,25 @@ void getRegistryReturnsNonNull() { assertNotNull(collector.getRegistry()); } + // --- Deprecated backward-compatible embedded-server path --- + + @Test + @SuppressWarnings("deprecation") + void deprecatedNoArgConstructorStillRecordsIntoPrometheusRegistry() { + PrometheusMetricsCollector legacy = new PrometheusMetricsCollector(); + assertInstanceOf(io.micrometer.prometheusmetrics.PrometheusMeterRegistry.class, legacy.getRegistry()); + assertNotNull(legacy.getApiClientMetrics()); + assertDoesNotThrow(() -> legacy.consume(new PollStarted("HTTP"))); + } + + @Test + @SuppressWarnings("deprecation") + void startServerRejectsNonPrometheusRegistry() { + PrometheusMetricsCollector simple = + new PrometheusMetricsCollector(new io.micrometer.core.instrument.simple.SimpleMeterRegistry()); + assertThrows(IllegalStateException.class, () -> simple.startServer(0, "/metrics")); + } + // --- Multiple increments accumulate --- @Test diff --git a/conductor-client-spring-boot4/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java b/conductor-client-spring-boot4/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java index 0baf28177..00309c051 100644 --- a/conductor-client-spring-boot4/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java +++ b/conductor-client-spring-boot4/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java @@ -49,7 +49,8 @@ public class ConductorClientAutoConfiguration { @ConditionalOnMissingBean @ConditionalOnExpression( "'${conductor.client.root-uri:}' != '' or '${conductor.client.base-path:}' != ''") - public ConductorClient conductorClient(ClientProperties properties) { + public ConductorClient conductorClient(ClientProperties properties, + Optional metricsCollector) { var basePath = StringUtils.isBlank(properties.getRootUri()) ? properties.getBasePath() : properties.getRootUri(); var builder = ConductorClient.builder() @@ -58,6 +59,7 @@ public ConductorClient conductorClient(ClientProperties properties) { .readTimeout(properties.getTimeout().getRead()) .writeTimeout(properties.getTimeout().getWrite()) .verifyingSsl(properties.isVerifyingSsl()); + metricsCollector.ifPresent(builder::withMetricsCollector); return builder.build(); } diff --git a/conductor-client-spring-boot4/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java b/conductor-client-spring-boot4/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java index be37cf1b6..eb477fd0c 100644 --- a/conductor-client-spring-boot4/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java +++ b/conductor-client-spring-boot4/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java @@ -12,6 +12,8 @@ */ package io.orkes.conductor.client.spring; +import java.util.Optional; + import org.apache.commons.lang3.StringUtils; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; @@ -22,6 +24,7 @@ import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; +import com.netflix.conductor.client.metrics.MetricsCollector; import com.netflix.conductor.client.spring.ClientProperties; import io.orkes.conductor.client.ApiClient; @@ -49,7 +52,8 @@ public class OrkesConductorClientAutoConfiguration { "'${conductor.client.root-uri:}' != '' or '${conductor.client.base-path:}' != ''" + " or '${conductor.server.url:}' != ''") public ApiClient orkesConductorClient(ClientProperties clientProperties, - OrkesClientProperties orkesClientProperties) { + OrkesClientProperties orkesClientProperties, + Optional metricsCollector) { var basePath = StringUtils.isBlank(clientProperties.getRootUri()) ? clientProperties.getBasePath() : clientProperties.getRootUri(); if (basePath == null) { basePath = orkesClientProperties.getConductorServerUrl(); @@ -69,6 +73,8 @@ public ApiClient orkesConductorClient(ClientProperties clientProperties, builder.credentials(orkesClientProperties.getSecurityKeyId(), orkesClientProperties.getSecuritySecret()); } + metricsCollector.ifPresent(builder::withMetricsCollector); + return builder.build(); } diff --git a/conductor-client-spring-boot4/src/test/java/com/netflix/conductor/client/spring/ConductorClientAutoConfigurationTest.java b/conductor-client-spring-boot4/src/test/java/com/netflix/conductor/client/spring/ConductorClientAutoConfigurationTest.java index 67686f4ad..1a6fd7a02 100644 --- a/conductor-client-spring-boot4/src/test/java/com/netflix/conductor/client/spring/ConductorClientAutoConfigurationTest.java +++ b/conductor-client-spring-boot4/src/test/java/com/netflix/conductor/client/spring/ConductorClientAutoConfigurationTest.java @@ -14,6 +14,7 @@ import java.lang.reflect.Method; import java.time.Duration; +import java.util.Optional; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; @@ -136,7 +137,7 @@ void clientProperties_areBoundFromEnvironment() { void conductorClient_beanMethod_hasConditionalOnExpressionGuard() throws Exception { Method method = ConductorClientAutoConfiguration.class.getDeclaredMethod( - "conductorClient", ClientProperties.class); + "conductorClient", ClientProperties.class, Optional.class); assertThat(method.isAnnotationPresent(ConditionalOnExpression.class)).isTrue(); } diff --git a/conductor-client-spring-boot4/src/test/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfigurationTest.java b/conductor-client-spring-boot4/src/test/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfigurationTest.java index 65553fd05..869b70d40 100644 --- a/conductor-client-spring-boot4/src/test/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfigurationTest.java +++ b/conductor-client-spring-boot4/src/test/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfigurationTest.java @@ -13,6 +13,7 @@ package io.orkes.conductor.client.spring; import java.lang.reflect.Method; +import java.util.Optional; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; @@ -89,7 +90,10 @@ void apiClient_isNotRegistered_whenNoUrlConfigured() { void orkesConductorClient_beanMethod_hasConditionalOnExpressionGuard() throws Exception { Method method = OrkesConductorClientAutoConfiguration.class.getDeclaredMethod( - "orkesConductorClient", ClientProperties.class, OrkesClientProperties.class); + "orkesConductorClient", + ClientProperties.class, + OrkesClientProperties.class, + Optional.class); assertThat(method.isAnnotationPresent(ConditionalOnExpression.class)).isTrue(); } } diff --git a/conductor-client-spring/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java b/conductor-client-spring/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java index 28e2150e9..fa34e1ac5 100644 --- a/conductor-client-spring/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java +++ b/conductor-client-spring/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java @@ -49,7 +49,8 @@ public class ConductorClientAutoConfiguration { @Bean @ConditionalOnMissingBean - public ConductorClient conductorClient(ClientProperties properties) { + public ConductorClient conductorClient(ClientProperties properties, + Optional metricsCollector) { var basePath = StringUtils.isBlank(properties.getRootUri()) ? properties.getBasePath() : properties.getRootUri(); if (basePath == null) { return null; @@ -61,6 +62,7 @@ public ConductorClient conductorClient(ClientProperties properties) { .readTimeout(properties.getTimeout().getRead()) .writeTimeout(properties.getTimeout().getWrite()) .verifyingSsl(properties.isVerifyingSsl()); + metricsCollector.ifPresent(builder::withMetricsCollector); return builder.build(); } diff --git a/conductor-client-spring/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java b/conductor-client-spring/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java index 6e23f48ff..cf2b9f00e 100644 --- a/conductor-client-spring/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java +++ b/conductor-client-spring/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java @@ -12,6 +12,8 @@ */ package io.orkes.conductor.client.spring; +import java.util.Optional; + import org.apache.commons.lang3.StringUtils; import org.conductoross.conductor.client.FileClient; import org.conductoross.conductor.client.FileClientProperties; @@ -24,6 +26,7 @@ import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; +import com.netflix.conductor.client.metrics.MetricsCollector; import com.netflix.conductor.client.spring.ClientProperties; import io.orkes.conductor.client.ApiClient; @@ -48,7 +51,8 @@ public class OrkesConductorClientAutoConfiguration { @Primary @ConditionalOnMissingBean public ApiClient orkesConductorClient(ClientProperties clientProperties, - OrkesClientProperties orkesClientProperties) { + OrkesClientProperties orkesClientProperties, + Optional metricsCollector) { var basePath = StringUtils.isBlank(clientProperties.getRootUri()) ? clientProperties.getBasePath() : clientProperties.getRootUri(); if (basePath == null) { basePath = orkesClientProperties.getConductorServerUrl(); @@ -72,6 +76,8 @@ public ApiClient orkesConductorClient(ClientProperties clientProperties, builder.credentials(orkesClientProperties.getSecurityKeyId(), orkesClientProperties.getSecuritySecret()); } + metricsCollector.ifPresent(builder::withMetricsCollector); + return builder.build(); } diff --git a/harness/Dockerfile b/harness/Dockerfile index 36740c980..f35ba7476 100644 --- a/harness/Dockerfile +++ b/harness/Dockerfile @@ -1,10 +1,10 @@ FROM eclipse-temurin:21-jdk AS build WORKDIR /app COPY . . -RUN ./gradlew :harness:installDist -x test --no-daemon +RUN ./gradlew :harness:bootJar -x test --no-daemon FROM eclipse-temurin:21-jre WORKDIR /app -COPY --from=build /app/harness/build/install/harness/lib/ ./lib/ +COPY --from=build /app/harness/build/libs/harness.jar ./harness.jar -ENTRYPOINT ["java", "-cp", "lib/*", "io.orkes.conductor.harness.HarnessMain"] +ENTRYPOINT ["java", "-jar", "harness.jar"] diff --git a/harness/README.md b/harness/README.md index 9a5ecdd80..4e8ce6a3f 100644 --- a/harness/README.md +++ b/harness/README.md @@ -1,12 +1,24 @@ # Java SDK Worker Harness -A self-feeding worker harness that runs indefinitely. On startup it registers five simulated tasks (`java_worker_0` through `java_worker_4`) and the `java_simulated_tasks_workflow`, then runs two background services: +A self-feeding worker harness that runs indefinitely. It is a **Spring Boot** application, which is how it demonstrates the recommended way to expose SDK metrics: the SDK publishes into the app's Micrometer registry and Spring Boot Actuator serves the scrape endpoint. The SDK itself starts no metrics web server. + +On startup it registers five simulated tasks (`java_worker_0` through `java_worker_4`) and the `java_simulated_tasks_workflow`, then runs two background services: - **WorkflowGovernor** -- starts a configurable number of `java_simulated_tasks_workflow` instances per second (default 2), indefinitely. - **SimulatedTaskWorkers** -- five task handlers, each with a codename and a default sleep duration. Each worker supports configurable delay types, failure simulation, and output generation via task input parameters. The workflow chains them in sequence: quickpulse (1s) → whisperlink (2s) → shadowfetch (3s) → ironforge (4s) → deepcrawl (5s). All resource names use a `java_` prefix so multiple SDK harnesses (C#, Python, Go, etc.) can coexist on the same cluster. +## Metrics + +The harness listens on a single HTTP port (`HARNESS_METRICS_PORT`, default `9991`) and serves Actuator endpoints at the root: + +- `GET /metrics` -- Prometheus text exposition (the scrape endpoint; Actuator's `prometheus` endpoint remapped to `/metrics`). This matches the cross-SDK PodMonitor contract (port name `metrics` → 9991, path `/metrics`). +- `GET /metrics-json` -- Actuator's JSON metrics browser (the `metrics` endpoint renamed to avoid colliding with the Prometheus path). +- `GET /health` -- Actuator health. + +Both SDK metrics (e.g. `task_poll_total`) and standard JVM/process metrics are exposed on the same registry. See `src/main/resources/application.yml`. + ## Local Run ```bash @@ -14,7 +26,7 @@ All resource names use a `java_` prefix so multiple SDK harnesses (C#, Python, G CONDUCTOR_SERVER_URL=https://your-cluster.example.com/api \ CONDUCTOR_AUTH_KEY=your-key \ CONDUCTOR_AUTH_SECRET=your-secret \ -./gradlew :harness:run +./gradlew :harness:bootRun ``` ## Docker @@ -67,6 +79,9 @@ kubectl rollout status deployment/java-sdk-harness-worker -n $NS | `HARNESS_WORKFLOWS_PER_SEC` | no | 2 | Workflows to start per second | | `HARNESS_BATCH_SIZE` | no | 20 | Thread count per worker (controls polling concurrency) | | `HARNESS_POLL_INTERVAL_MS` | no | 100 | Milliseconds between poll cycles | +| `HARNESS_METRICS_PORT` | no | 9991 | HTTP port for the Actuator/metrics endpoints | +| `HARNESS_PROBE_RATE_PER_SEC` | no | 0 | Control-plane probe rate; 0 disables the probe | +| `HARNESS_WORKERS_ONLY` | no | false | Run workers only (skip registration, governor, and probe) | ## Kubernetes diff --git a/harness/build.gradle b/harness/build.gradle index d78185b5b..281863598 100644 --- a/harness/build.gradle +++ b/harness/build.gradle @@ -1,19 +1,42 @@ plugins { - id 'java' - id 'application' -} - -application { - mainClass = 'io.orkes.conductor.harness.HarnessMain' + id 'org.springframework.boot' version '3.5.13' + id 'io.spring.dependency-management' version '1.1.7' } repositories { mavenCentral() } +// The root build pins slf4j-api 1.7.36 on every subproject, which is an +// slf4j 1.x binding and is incompatible with Spring Boot's Logback (slf4j 2.x). +// Align slf4j with the version Spring Boot's Logback expects. +configurations.all { + resolutionStrategy.eachDependency { details -> + if (details.requested.group == 'org.slf4j' && details.requested.name == 'slf4j-api') { + details.useVersion '2.0.17' + details.because 'Align slf4j with Spring Boot Logback (slf4j 2.x)' + } + } +} + dependencies { - implementation project(':conductor-client') + // Brings conductor-client plus the base + Orkes Spring auto-configurations. + implementation project(':conductor-client-spring') + // Provides the MetricsCollector and its Spring Boot auto-configuration, + // which binds SDK metrics to the application's Micrometer MeterRegistry. implementation project(':conductor-client-metrics') - implementation project(':orkes-client') - implementation "ch.qos.logback:logback-classic:1.5.6" + + implementation 'org.springframework.boot:spring-boot-starter-web' + implementation 'org.springframework.boot:spring-boot-starter-actuator' + // Backs the Actuator /metrics (Prometheus) endpoint with a + // PrometheusMeterRegistry bean that the SDK collector publishes into. + implementation 'io.micrometer:micrometer-registry-prometheus' +} + +springBoot { + mainClass = 'io.orkes.conductor.harness.HarnessApplication' +} + +bootJar { + archiveFileName = 'harness.jar' } diff --git a/harness/manifests/deployment.yaml b/harness/manifests/deployment.yaml index 23a02a0de..6de889f74 100644 --- a/harness/manifests/deployment.yaml +++ b/harness/manifests/deployment.yaml @@ -40,7 +40,7 @@ spec: name: conductor-credentials key: auth-secret - # === HARNESS TUNING (defaults match HarnessMain.java) === + # === HARNESS TUNING (defaults match application.yml) === # Adjust these per-cluster if needed, or move to ConfigMap. - name: HARNESS_WORKFLOWS_PER_SEC diff --git a/harness/src/main/java/io/orkes/conductor/harness/HarnessApplication.java b/harness/src/main/java/io/orkes/conductor/harness/HarnessApplication.java new file mode 100644 index 000000000..3ba0fe7e0 --- /dev/null +++ b/harness/src/main/java/io/orkes/conductor/harness/HarnessApplication.java @@ -0,0 +1,34 @@ +/* + * Copyright 2024 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package io.orkes.conductor.harness; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.EnableConfigurationProperties; + +/** + * Spring Boot entry point for the Java SDK worker harness. + * + *

The SDK no longer starts its own metrics web server. Instead, the metrics + * are published into the Spring-managed Micrometer registry and exposed through + * Spring Boot Actuator. See {@code application.yml} for the endpoint mapping + * (Prometheus scrape at {@code /metrics} on the configured port). + */ +@SpringBootApplication +@EnableConfigurationProperties(HarnessProperties.class) +public class HarnessApplication { + + public static void main(String[] args) { + SpringApplication.run(HarnessApplication.class, args); + } +} diff --git a/harness/src/main/java/io/orkes/conductor/harness/HarnessBootstrap.java b/harness/src/main/java/io/orkes/conductor/harness/HarnessBootstrap.java new file mode 100644 index 000000000..cf6aa535b --- /dev/null +++ b/harness/src/main/java/io/orkes/conductor/harness/HarnessBootstrap.java @@ -0,0 +1,129 @@ +/* + * Copyright 2024 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package io.orkes.conductor.harness; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; +import org.springframework.stereotype.Component; + +import com.netflix.conductor.client.http.ConductorClient; +import com.netflix.conductor.client.http.MetadataClient; +import com.netflix.conductor.client.http.WorkflowClient; +import com.netflix.conductor.common.metadata.tasks.TaskDef; +import com.netflix.conductor.common.metadata.tasks.TaskType; +import com.netflix.conductor.common.metadata.workflow.WorkflowDef; +import com.netflix.conductor.common.metadata.workflow.WorkflowTask; + +import jakarta.annotation.PreDestroy; + +/** + * Drives the self-feeding harness once the Spring context (and the + * auto-configured, already-polling {@code TaskRunnerConfigurer}) is up: + * registers metadata, then starts the workflow governor and the optional + * status probe. + */ +@Component +public class HarnessBootstrap implements ApplicationRunner { + + private static final Logger log = LoggerFactory.getLogger(HarnessBootstrap.class); + + private static final String WORKFLOW_NAME = "java_simulated_tasks_workflow"; + + private final ConductorClient client; + private final WorkflowClient workflowClient; + private final HarnessProperties props; + private final List workers; + + private WorkflowGovernor governor; + private WorkflowStatusProbe probe; + + public HarnessBootstrap(ConductorClient client, + WorkflowClient workflowClient, + HarnessProperties props, + List workers) { + this.client = client; + this.workflowClient = workflowClient; + this.props = props; + // Sort by task name (java_worker_0..N) so the workflow chain order is + // deterministic regardless of bean-injection order. + this.workers = workers.stream() + .sorted(Comparator.comparing(SimulatedTaskWorker::getTaskDefName)) + .toList(); + } + + @Override + public void run(ApplicationArguments args) { + if (props.isWorkersOnly()) { + log.info("Running in workers-only mode (no registration, governor, or probe)"); + return; + } + + registerMetadata(new MetadataClient(client)); + + probe = new WorkflowStatusProbe(workflowClient, props.getProbeRatePerSec()); + governor = new WorkflowGovernor(workflowClient, WORKFLOW_NAME, props.getWorkflowsPerSec(), probe::offer); + governor.start(); + probe.start(); + } + + @PreDestroy + public void shutdown() { + log.info("Shutting down harness..."); + if (governor != null) { + governor.shutdown(); + } + if (probe != null) { + probe.shutdown(); + } + } + + private void registerMetadata(MetadataClient metadataClient) { + List taskDefs = new ArrayList<>(); + for (SimulatedTaskWorker worker : workers) { + TaskDef td = new TaskDef(worker.getTaskDefName()); + td.setDescription("Java SDK harness simulated task (" + worker.getCodename() + + ", default delay " + worker.getDelaySeconds() + "s)"); + td.setRetryCount(1); + td.setTimeoutSeconds(300); + td.setResponseTimeoutSeconds(300); + taskDefs.add(td); + } + metadataClient.registerTaskDefs(taskDefs); + log.info("Registered {} task definitions", taskDefs.size()); + + WorkflowDef workflowDef = new WorkflowDef(); + workflowDef.setName(WORKFLOW_NAME); + workflowDef.setVersion(1); + workflowDef.setDescription("Java SDK harness simulated task workflow"); + workflowDef.setOwnerEmail("java-sdk-harness@conductor.io"); + + List wfTasks = new ArrayList<>(); + for (SimulatedTaskWorker worker : workers) { + WorkflowTask wt = new WorkflowTask(); + wt.setName(worker.getTaskDefName()); + wt.setTaskReferenceName(worker.getCodename()); + wt.setType(TaskType.SIMPLE.name()); + wfTasks.add(wt); + } + workflowDef.setTasks(wfTasks); + + metadataClient.updateWorkflowDefs(List.of(workflowDef)); + log.info("Registered workflow definition: {}", WORKFLOW_NAME); + } +} diff --git a/harness/src/main/java/io/orkes/conductor/harness/HarnessMain.java b/harness/src/main/java/io/orkes/conductor/harness/HarnessMain.java deleted file mode 100644 index 5cd12d3c6..000000000 --- a/harness/src/main/java/io/orkes/conductor/harness/HarnessMain.java +++ /dev/null @@ -1,182 +0,0 @@ -/* - * Copyright 2024 Conductor Authors. - *

- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - *

- * http://www.apache.org/licenses/LICENSE-2.0 - *

- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ -package io.orkes.conductor.harness; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.netflix.conductor.client.automator.TaskRunnerConfigurer; -import com.netflix.conductor.client.http.ConductorClient; -import com.netflix.conductor.client.http.MetadataClient; -import com.netflix.conductor.client.http.TaskClient; -import com.netflix.conductor.client.http.WorkflowClient; -import com.netflix.conductor.client.metrics.prometheus.PrometheusMetricsCollector; -import com.netflix.conductor.client.worker.Worker; -import com.netflix.conductor.common.metadata.tasks.TaskDef; -import com.netflix.conductor.common.metadata.tasks.TaskType; -import com.netflix.conductor.common.metadata.workflow.WorkflowDef; -import com.netflix.conductor.common.metadata.workflow.WorkflowTask; - -import io.orkes.conductor.client.ApiClient; - -public class HarnessMain { - - private static final Logger log = LoggerFactory.getLogger(HarnessMain.class); - - private static final String WORKFLOW_NAME = "java_simulated_tasks_workflow"; - - private static final String[][] SIMULATED_WORKERS = { - {"java_worker_0", "quickpulse", "1"}, - {"java_worker_1", "whisperlink", "2"}, - {"java_worker_2", "shadowfetch", "3"}, - {"java_worker_3", "ironforge", "4"}, - {"java_worker_4", "deepcrawl", "5"}, - }; - - public static void main(String[] args) throws Exception { - boolean workersOnly = Arrays.asList(args).contains("--workers-only"); - - int workflowsPerSec = envInt("HARNESS_WORKFLOWS_PER_SEC", 2); - int batchSize = envInt("HARNESS_BATCH_SIZE", 20); - int pollIntervalMs = envInt("HARNESS_POLL_INTERVAL_MS", 100); - int metricsPort = envInt("HARNESS_METRICS_PORT", 9991); - int probeRatePerSec = envInt("HARNESS_PROBE_RATE_PER_SEC", 0); - - List workers = new ArrayList<>(); - for (String[] entry : SIMULATED_WORKERS) { - workers.add(new SimulatedTaskWorker(entry[0], entry[1], Integer.parseInt(entry[2]), batchSize, - pollIntervalMs)); - } - Map threadCounts = - workers.stream().collect(Collectors.toMap(Worker::getTaskDefName, w -> batchSize)); - - PrometheusMetricsCollector metricsCollector = null; - if (!workersOnly) { - metricsCollector = new PrometheusMetricsCollector(); - } - - var clientBuilder = ApiClient.builder() - .useEnvVariables(true) - .readTimeout(10_000) - .connectTimeout(10_000) - .writeTimeout(10_000); - if (metricsCollector != null) { - clientBuilder.withMetricsCollector(metricsCollector); - } - - ConductorClient client = clientBuilder.build(); - - TaskClient taskClient = new TaskClient(client); - - TaskRunnerConfigurer configurer = - new TaskRunnerConfigurer.Builder(taskClient, workers) - .withTaskThreadCount(threadCounts) - .build(); - - if (!workersOnly) { - try { - registerMetadata(client); - } catch (Exception e) { - log.error("Failed to register metadata (bad credentials?): {}", e.getMessage()); - System.exit(1); - return; - } - metricsCollector.startServer(metricsPort, "/metrics"); - log.info("Prometheus metrics server started on port {}", metricsPort); - } - - configurer.init(); - - if (workersOnly) { - log.info("Running in --workers-only mode (no metrics, no registration, no governor)"); - Runtime.getRuntime().addShutdownHook(new Thread(() -> { - log.info("Shutting down harness..."); - configurer.shutdown(); - })); - } else { - WorkflowClient workflowClient = new WorkflowClient(client); - WorkflowStatusProbe probe = new WorkflowStatusProbe(workflowClient, probeRatePerSec); - WorkflowGovernor governor = new WorkflowGovernor( - workflowClient, WORKFLOW_NAME, workflowsPerSec, probe::offer); - governor.start(); - probe.start(); - - Runtime.getRuntime().addShutdownHook(new Thread(() -> { - log.info("Shutting down harness..."); - governor.shutdown(); - probe.shutdown(); - configurer.shutdown(); - })); - } - - Thread.currentThread().join(); - } - - private static void registerMetadata(ConductorClient client) { - MetadataClient metadataClient = new MetadataClient(client); - - List taskDefs = new ArrayList<>(); - for (String[] entry : SIMULATED_WORKERS) { - String taskName = entry[0]; - String codename = entry[1]; - int sleepSeconds = Integer.parseInt(entry[2]); - - TaskDef td = new TaskDef(taskName); - td.setDescription( - "Java SDK harness simulated task (" + codename + ", default delay " + sleepSeconds + "s)"); - td.setRetryCount(1); - td.setTimeoutSeconds(300); - td.setResponseTimeoutSeconds(300); - taskDefs.add(td); - } - metadataClient.registerTaskDefs(taskDefs); - log.info("Registered {} task definitions", taskDefs.size()); - - WorkflowDef workflowDef = new WorkflowDef(); - workflowDef.setName(WORKFLOW_NAME); - workflowDef.setVersion(1); - workflowDef.setDescription("Java SDK harness simulated task workflow"); - workflowDef.setOwnerEmail("java-sdk-harness@conductor.io"); - - List wfTasks = new ArrayList<>(); - for (String[] entry : SIMULATED_WORKERS) { - WorkflowTask wt = new WorkflowTask(); - wt.setName(entry[0]); - wt.setTaskReferenceName(entry[1]); - wt.setType(TaskType.SIMPLE.name()); - wfTasks.add(wt); - } - workflowDef.setTasks(wfTasks); - - metadataClient.updateWorkflowDefs(List.of(workflowDef)); - log.info("Registered workflow definition: {}", WORKFLOW_NAME); - } - - private static int envInt(String name, int defaultValue) { - String value = System.getenv(name); - if (value == null || value.isBlank()) { - return defaultValue; - } - try { - return Integer.parseInt(value.trim()); - } catch (NumberFormatException e) { - return defaultValue; - } - } -} diff --git a/harness/src/main/java/io/orkes/conductor/harness/HarnessProperties.java b/harness/src/main/java/io/orkes/conductor/harness/HarnessProperties.java new file mode 100644 index 000000000..52039446b --- /dev/null +++ b/harness/src/main/java/io/orkes/conductor/harness/HarnessProperties.java @@ -0,0 +1,44 @@ +/* + * Copyright 2024 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package io.orkes.conductor.harness; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +import lombok.Getter; +import lombok.Setter; + +/** + * Tuning knobs for the harness, bound from the {@code harness.*} namespace (see + * {@code application.yml}, which maps the {@code HARNESS_*} environment variables + * onto these properties for backwards compatibility with existing deployments). + */ +@Getter +@Setter +@ConfigurationProperties("harness") +public class HarnessProperties { + + /** Workflows to start per second (governor). */ + private int workflowsPerSec = 2; + + /** Thread count per worker (controls polling concurrency). */ + private int batchSize = 20; + + /** Milliseconds between poll cycles. */ + private int pollIntervalMs = 100; + + /** Control-plane probe rate; 0 disables the probe. */ + private int probeRatePerSec = 0; + + /** When true, run workers only (no metadata registration, governor, or probe). */ + private boolean workersOnly = false; +} diff --git a/harness/src/main/java/io/orkes/conductor/harness/HarnessWorkerConfiguration.java b/harness/src/main/java/io/orkes/conductor/harness/HarnessWorkerConfiguration.java new file mode 100644 index 000000000..5008ce24c --- /dev/null +++ b/harness/src/main/java/io/orkes/conductor/harness/HarnessWorkerConfiguration.java @@ -0,0 +1,56 @@ +/* + * Copyright 2024 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package io.orkes.conductor.harness; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Registers the five simulated task workers as beans. The auto-configured + * {@code TaskRunnerConfigurer} collects every {@code Worker} bean and starts + * polling for them. Thread counts are taken from + * {@code conductor.worker..threadCount} (wired to + * {@code HARNESS_BATCH_SIZE} in {@code application.yml}). + * + *

The workflow chains the workers in sequence: + * quickpulse (1s) → whisperlink (2s) → shadowfetch (3s) → + * ironforge (4s) → deepcrawl (5s). + */ +@Configuration +public class HarnessWorkerConfiguration { + + @Bean + public SimulatedTaskWorker quickpulseWorker(HarnessProperties props) { + return new SimulatedTaskWorker("java_worker_0", "quickpulse", 1, props.getBatchSize(), props.getPollIntervalMs()); + } + + @Bean + public SimulatedTaskWorker whisperlinkWorker(HarnessProperties props) { + return new SimulatedTaskWorker("java_worker_1", "whisperlink", 2, props.getBatchSize(), props.getPollIntervalMs()); + } + + @Bean + public SimulatedTaskWorker shadowfetchWorker(HarnessProperties props) { + return new SimulatedTaskWorker("java_worker_2", "shadowfetch", 3, props.getBatchSize(), props.getPollIntervalMs()); + } + + @Bean + public SimulatedTaskWorker ironforgeWorker(HarnessProperties props) { + return new SimulatedTaskWorker("java_worker_3", "ironforge", 4, props.getBatchSize(), props.getPollIntervalMs()); + } + + @Bean + public SimulatedTaskWorker deepcrawlWorker(HarnessProperties props) { + return new SimulatedTaskWorker("java_worker_4", "deepcrawl", 5, props.getBatchSize(), props.getPollIntervalMs()); + } +} diff --git a/harness/src/main/java/io/orkes/conductor/harness/SimulatedTaskWorker.java b/harness/src/main/java/io/orkes/conductor/harness/SimulatedTaskWorker.java index f633b8f59..818db3a04 100644 --- a/harness/src/main/java/io/orkes/conductor/harness/SimulatedTaskWorker.java +++ b/harness/src/main/java/io/orkes/conductor/harness/SimulatedTaskWorker.java @@ -65,6 +65,16 @@ public String getTaskDefName() { return taskName; } + /** Human-friendly alias, also used as the workflow task reference name. */ + public String getCodename() { + return codename; + } + + /** Default simulated delay in seconds (drives the task-def description). */ + public int getDelaySeconds() { + return defaultDelayMs / 1000; + } + @Override public String getIdentity() { return workerId; diff --git a/harness/src/main/java/io/orkes/conductor/harness/WorkflowStatusProbe.java b/harness/src/main/java/io/orkes/conductor/harness/WorkflowStatusProbe.java index c663d7841..9ac77909c 100644 --- a/harness/src/main/java/io/orkes/conductor/harness/WorkflowStatusProbe.java +++ b/harness/src/main/java/io/orkes/conductor/harness/WorkflowStatusProbe.java @@ -28,7 +28,7 @@ * endpoints, so {@code http_api_client_request_seconds} on the canonical * Prometheus surface picks up entries with {@code uri=/api/workflow/<uuid>} * and {@code uri=/api/workflow/<uuid>/status}. The default - * {@link HarnessMain} traffic only ever hits bounded, no-path-param URLs + * {@link HarnessBootstrap} traffic only ever hits bounded, no-path-param URLs * ({@code /api/tasks/poll/batch/<taskType>}, {@code /api/tasks}, * {@code /api/workflow}, etc.), so the high-cardinality concern on the * {@code uri} label is invisible without something like this probe. diff --git a/harness/src/main/resources/application.yml b/harness/src/main/resources/application.yml new file mode 100644 index 000000000..1b173c528 --- /dev/null +++ b/harness/src/main/resources/application.yml @@ -0,0 +1,50 @@ +# The harness listens on a single HTTP port that serves the Actuator +# endpoints. The Prometheus scrape endpoint is remapped to /metrics so it +# matches the cross-SDK PodMonitor contract (port name "metrics" -> 9991, +# path /metrics) without any SDK-owned web server. +server: + port: ${HARNESS_METRICS_PORT:9991} + +management: + endpoints: + web: + # Serve endpoints at the root (e.g. /metrics, /health) instead of the + # default /actuator/* prefix. + base-path: / + exposure: + include: health,prometheus,metrics + path-mapping: + # Prometheus text exposition (what Prometheus scrapes) -> /metrics + prometheus: metrics + # Actuator's JSON metrics browser -> /metrics-json (renamed to avoid + # colliding with the Prometheus endpoint above). + metrics: metrics-json + endpoint: + health: + probes: + enabled: true + +conductor: + client: + base-path: ${CONDUCTOR_SERVER_URL:} + key-id: ${CONDUCTOR_AUTH_KEY:} + secret: ${CONDUCTOR_AUTH_SECRET:} + # Thread count per worker (polling concurrency). Mirrors HARNESS_BATCH_SIZE. + worker: + java_worker_0: + threadCount: ${HARNESS_BATCH_SIZE:20} + java_worker_1: + threadCount: ${HARNESS_BATCH_SIZE:20} + java_worker_2: + threadCount: ${HARNESS_BATCH_SIZE:20} + java_worker_3: + threadCount: ${HARNESS_BATCH_SIZE:20} + java_worker_4: + threadCount: ${HARNESS_BATCH_SIZE:20} + +harness: + workflows-per-sec: ${HARNESS_WORKFLOWS_PER_SEC:2} + batch-size: ${HARNESS_BATCH_SIZE:20} + poll-interval-ms: ${HARNESS_POLL_INTERVAL_MS:100} + probe-rate-per-sec: ${HARNESS_PROBE_RATE_PER_SEC:0} + workers-only: ${HARNESS_WORKERS_ONLY:false}