diff --git a/maestro-engine/src/main/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisher.java b/maestro-engine/src/main/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisher.java new file mode 100644 index 00000000..31d1a2ae --- /dev/null +++ b/maestro-engine/src/main/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisher.java @@ -0,0 +1,159 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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.maestro.engine.publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.netflix.maestro.exceptions.MaestroInternalError; +import com.netflix.maestro.exceptions.MaestroRetryableError; +import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException; +import com.netflix.maestro.models.events.MaestroEvent; +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.EnumSet; +import java.util.HexFormat; +import java.util.Set; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import lombok.extern.slf4j.Slf4j; + +/** + * Webhook notification publisher implementation for MaestroNotificationPublisher. It POSTs the + * JSON-serialized {@link MaestroEvent} to the configured endpoint whenever a step or workflow + * status (or workflow definition) changes. + * + *

Delivery semantics follow the other publishers: a non-2xx client error (4xx) is thrown as + * non-retryable ({@link MaestroUnprocessableEntityException}) since the receiver rejected the + * payload, while server errors (5xx) and I/O failures are thrown as {@link MaestroRetryableError} + * so the notification job event is retried by the queue system. + * + *

When a signing secret is configured, the request carries an {@code X-Maestro-Signature-256} + * header with the hex-encoded HMAC-SHA256 of the request body (GitHub-webhook style, {@code + * sha256=...}), so receivers can authenticate the sender. + */ +@Slf4j +public class WebhookNotificationPublisher implements MaestroNotificationPublisher { + private static final String EVENT_TYPE_HEADER = "X-Maestro-Event-Type"; + private static final String SIGNATURE_HEADER = "X-Maestro-Signature-256"; + private static final String SIGNATURE_PREFIX = "sha256="; + private static final String HMAC_ALGORITHM = "HmacSHA256"; + + private final HttpClient httpClient; + private final String url; + private final Set eventTypes; + private final Duration requestTimeout; + private final SecretKeySpec signingKey; + private final ObjectMapper objectMapper; + + /** + * Constructor. + * + * @param httpClient http client used to send webhook requests + * @param url webhook endpoint to POST events to + * @param eventTypes event types to publish; an empty set means all event types + * @param requestTimeout per-request timeout + * @param signingSecret secret for the HMAC-SHA256 signature header; null or empty disables it + * @param objectMapper object mapper to serialize events + */ + public WebhookNotificationPublisher( + HttpClient httpClient, + String url, + Set eventTypes, + Duration requestTimeout, + String signingSecret, + ObjectMapper objectMapper) { + this.httpClient = httpClient; + this.url = url; + this.eventTypes = + eventTypes == null || eventTypes.isEmpty() + ? EnumSet.allOf(MaestroEvent.Type.class) + : EnumSet.copyOf(eventTypes); + this.requestTimeout = requestTimeout; + this.signingKey = + signingSecret == null || signingSecret.isEmpty() + ? null + : new SecretKeySpec(signingSecret.getBytes(StandardCharsets.UTF_8), HMAC_ALGORITHM); + this.objectMapper = objectMapper; + } + + /** Send a Maestro event to the webhook endpoint. */ + @Override + public void send(MaestroEvent event) { + if (!this.eventTypes.contains(event.getType())) { + LOG.debug("Skipping maestro event of type [{}]: not in the configured set", event.getType()); + return; + } + + byte[] payload; + try { + payload = this.objectMapper.writeValueAsBytes(event); + } catch (JsonProcessingException je) { + throw new MaestroUnprocessableEntityException("cannot serialize maestro event: " + event, je); + } + + HttpRequest.Builder requestBuilder = + HttpRequest.newBuilder(URI.create(this.url)) + .timeout(this.requestTimeout) + .header("Content-Type", "application/json") + .header(EVENT_TYPE_HEADER, event.getType().name()) + .POST(HttpRequest.BodyPublishers.ofByteArray(payload)); + if (this.signingKey != null) { + requestBuilder.header(SIGNATURE_HEADER, SIGNATURE_PREFIX + sign(payload)); + } + + HttpResponse response; + try { + response = this.httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString()); + } catch (IOException e) { + throw new MaestroRetryableError( + e, "Failed to POST a maestro event to webhook [%s] and will retry.", this.url); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MaestroRetryableError( + e, "Interrupted while sending a maestro event to webhook [%s] and will retry.", this.url); + } + + int statusCode = response.statusCode(); + if (statusCode >= 200 && statusCode < 300) { + LOG.info( + "Published a maestro event of type [{}] to webhook [{}] with status code: [{}]", + event.getType(), + this.url, + statusCode); + } else if (statusCode >= 400 && statusCode < 500) { + // The receiver actively rejected the payload: retrying the same request will not help. + throw new MaestroUnprocessableEntityException( + "Webhook [%s] rejected a maestro event of type [%s] with status code [%s] and body [%s]", + this.url, event.getType(), statusCode, response.body()); + } else { + throw new MaestroRetryableError( + "Webhook [%s] returned status code [%s] for a maestro event of type [%s] and will retry.", + this.url, statusCode, event.getType()); + } + } + + private String sign(byte[] payload) { + try { + Mac mac = Mac.getInstance(HMAC_ALGORITHM); + mac.init(this.signingKey); + return HexFormat.of().formatHex(mac.doFinal(payload)); + } catch (Exception e) { + throw new MaestroInternalError(e, "Failed to compute the webhook signature"); + } + } +} diff --git a/maestro-engine/src/test/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisherTest.java b/maestro-engine/src/test/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisherTest.java new file mode 100644 index 00000000..44fd1cc0 --- /dev/null +++ b/maestro-engine/src/test/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisherTest.java @@ -0,0 +1,177 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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.maestro.engine.publisher; + +import com.netflix.maestro.AssertHelper; +import com.netflix.maestro.MaestroBaseTest; +import com.netflix.maestro.exceptions.MaestroRetryableError; +import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException; +import com.netflix.maestro.models.events.MaestroEvent; +import com.netflix.maestro.models.events.WorkflowVersionChangeEvent; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.http.HttpClient; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HexFormat; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class WebhookNotificationPublisherTest extends MaestroBaseTest { + private static final Duration TIMEOUT = Duration.ofSeconds(5); + private static final String SECRET = "test-secret"; + + private HttpServer server; + private String url; + private final AtomicInteger statusToReturn = new AtomicInteger(200); + private final AtomicInteger requestCount = new AtomicInteger(); + private final AtomicReference lastBody = new AtomicReference<>(); + private final Map lastHeaders = new ConcurrentHashMap<>(); + + @Before + public void setUp() throws IOException { + server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + server.createContext( + "/webhook", + exchange -> { + requestCount.incrementAndGet(); + lastBody.set(exchange.getRequestBody().readAllBytes()); + exchange + .getRequestHeaders() + .forEach((key, values) -> lastHeaders.put(key, values.getFirst())); + exchange.sendResponseHeaders(statusToReturn.get(), -1); + exchange.close(); + }); + server.start(); + url = "http://localhost:" + server.getAddress().getPort() + "/webhook"; + } + + @After + public void tearDown() { + server.stop(0); + } + + @Test + public void testPublishEventWithSignature() throws Exception { + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, SECRET, MAPPER); + MaestroEvent event = sampleEvent(); + + publisher.send(event); + + Assert.assertEquals(1, requestCount.get()); + Assert.assertEquals("application/json", lastHeaders.get("Content-type")); + Assert.assertEquals( + MaestroEvent.Type.WORKFLOW_VERSION_CHANGE_EVENT.name(), + lastHeaders.get("X-maestro-event-type")); + // the body round-trips as the same event + Assert.assertEquals( + MAPPER.readTree(MAPPER.writeValueAsBytes(event)), MAPPER.readTree(lastBody.get())); + // the signature is the HMAC-SHA256 of the exact bytes received + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + Assert.assertEquals( + "sha256=" + HexFormat.of().formatHex(mac.doFinal(lastBody.get())), + lastHeaders.get("X-maestro-signature-256")); + } + + @Test + public void testNoSignatureHeaderWithoutSecret() { + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER); + + publisher.send(sampleEvent()); + + Assert.assertEquals(1, requestCount.get()); + Assert.assertFalse(lastHeaders.containsKey("X-maestro-signature-256")); + } + + @Test + public void testEventTypeFiltering() { + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), + url, + EnumSet.of(MaestroEvent.Type.STEP_INSTANCE_STATUS_CHANGE_EVENT), + TIMEOUT, + null, + MAPPER); + + publisher.send(sampleEvent()); // WORKFLOW_VERSION_CHANGE_EVENT: filtered out + + Assert.assertEquals(0, requestCount.get()); + } + + @Test + public void testClientErrorIsNotRetryable() { + statusToReturn.set(400); + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER); + + AssertHelper.assertThrows( + "4xx means the receiver rejected the payload", + MaestroUnprocessableEntityException.class, + "rejected a maestro event", + () -> publisher.send(sampleEvent())); + } + + @Test + public void testServerErrorIsRetryable() { + statusToReturn.set(503); + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER); + + AssertHelper.assertThrows( + "5xx is retryable", + MaestroRetryableError.class, + "will retry", + () -> publisher.send(sampleEvent())); + } + + @Test + public void testConnectionFailureIsRetryable() { + server.stop(0); + WebhookNotificationPublisher publisher = + new WebhookNotificationPublisher( + HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER); + + AssertHelper.assertThrows( + "connection failures are retryable", + MaestroRetryableError.class, + "will retry", + () -> publisher.send(sampleEvent())); + } + + private MaestroEvent sampleEvent() { + return WorkflowVersionChangeEvent.builder() + .workflowId("sample-wf") + .workflowName("sample-wf-name") + .clusterName("test-cluster") + .build(); + } +} diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroServerConfiguration.java b/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroServerConfiguration.java index 03fe6338..cb0d1a94 100644 --- a/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroServerConfiguration.java +++ b/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroServerConfiguration.java @@ -12,6 +12,7 @@ */ package com.netflix.maestro.server.config; +import com.fasterxml.jackson.databind.ObjectMapper; import com.netflix.maestro.engine.concurrency.InstanceStepConcurrencyHandler; import com.netflix.maestro.engine.concurrency.TagPermitManager; import com.netflix.maestro.engine.dao.MaestroRunStrategyDao; @@ -29,8 +30,10 @@ import com.netflix.maestro.engine.processors.UpdateJobEventProcessor; import com.netflix.maestro.engine.publisher.MaestroNotificationPublisher; import com.netflix.maestro.engine.publisher.NoOpMaestroNotificationPublisher; +import com.netflix.maestro.engine.publisher.WebhookNotificationPublisher; import com.netflix.maestro.flow.runtime.FlowOperation; import com.netflix.maestro.metrics.MaestroMetrics; +import com.netflix.maestro.models.Constants; import com.netflix.maestro.models.definition.User; import com.netflix.maestro.queue.MaestroQueueSystem; import com.netflix.maestro.queue.dao.MaestroQueueDao; @@ -41,6 +44,9 @@ import com.netflix.maestro.queue.worker.MaestroQueueWorkerService; import com.netflix.maestro.server.interceptor.UserInfoInterceptor; import com.netflix.maestro.server.properties.MaestroProperties; +import com.netflix.maestro.server.properties.WebhookNotifierProperties; +import java.net.http.HttpClient; +import java.time.Duration; import java.util.EnumMap; import java.util.concurrent.BlockingQueue; import lombok.extern.slf4j.Slf4j; @@ -54,7 +60,7 @@ /** beans for maestro server related classes. */ @Configuration @Slf4j -@EnableConfigurationProperties(MaestroProperties.class) +@EnableConfigurationProperties({MaestroProperties.class, WebhookNotifierProperties.class}) @SuppressWarnings("PMD.LooseCoupling") public class MaestroServerConfiguration { private static final String EVENT_QUEUES_QUALIFIER = "EventQueues"; @@ -75,6 +81,25 @@ public MaestroNotificationPublisher notificationPublisher() { return new NoOpMaestroNotificationPublisher(); } + @ConditionalOnProperty(value = "maestro.notifier.type", havingValue = "webhook") + @Bean + public MaestroNotificationPublisher webhookNotificationPublisher( + WebhookNotifierProperties props, + @Qualifier(Constants.MAESTRO_QUALIFIER) ObjectMapper objectMapper) { + LOG.info( + "Creating Webhook MaestroNotificationPublisher within Spring boot for url [{}]...", + props.getUrl()); + return new WebhookNotificationPublisher( + HttpClient.newBuilder() + .connectTimeout(Duration.ofMillis(props.getRequestTimeoutMs())) + .build(), + props.getUrl(), + props.getEventTypes(), + Duration.ofMillis(props.getRequestTimeoutMs()), + props.getSigningSecret(), + objectMapper); + } + // below are beans for internal queue and processors. @Bean public StartWorkflowJobEventProcessor startWorkflowJobProcessor( diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/properties/WebhookNotifierProperties.java b/maestro-server/src/main/java/com/netflix/maestro/server/properties/WebhookNotifierProperties.java new file mode 100644 index 00000000..d28faae2 --- /dev/null +++ b/maestro-server/src/main/java/com/netflix/maestro/server/properties/WebhookNotifierProperties.java @@ -0,0 +1,45 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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.maestro.server.properties; + +import com.netflix.maestro.models.events.MaestroEvent; +import java.util.Collections; +import java.util.Set; +import lombok.Getter; +import lombok.Setter; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Webhook notifier config properties, used when {@code maestro.notifier.type} is {@code webhook}. + */ +@Getter +@Setter +@ConfigurationProperties(prefix = "maestro.notifier.webhook") +public class WebhookNotifierProperties { + private static final long DEFAULT_REQUEST_TIMEOUT_MS = 10_000L; + + /** Webhook endpoint URL to POST maestro events to. Required. */ + private String url; + + /** Event types to publish. An empty set publishes all event types. */ + private Set eventTypes = Collections.emptySet(); + + /** Per-request timeout in milliseconds. */ + private long requestTimeoutMs = DEFAULT_REQUEST_TIMEOUT_MS; + + /** + * Secret for the HMAC-SHA256 signature header ({@code X-Maestro-Signature-256}). Null or empty + * disables signing. + */ + private String signingSecret; +}