Skip to content

Commit 9f775f7

Browse files
nikhiln64nikhiln64
committed
Add webhook notification publisher for step/workflow status change events
Implements the webhook support requested in #111. A new WebhookNotificationPublisher POSTs the JSON-serialized MaestroEvent to a configured endpoint, following the same error-classification contract as the SNS publisher: 4xx responses are non-retryable (the receiver rejected the payload), 5xx and I/O failures throw MaestroRetryableError so the queue system retries delivery. Features: - event-type allowlist (empty = all event types) - optional HMAC-SHA256 request signing via X-Maestro-Signature-256 (GitHub-webhook style, sha256=<hex>), so receivers can authenticate the sender - X-Maestro-Event-Type header for cheap receiver-side routing - built on the JDK HttpClient: no new dependencies Wired behind maestro.notifier.type=webhook with a maestro.notifier.webhook.* properties block, mirroring the existing noop/sns conditional wiring.
1 parent 441149c commit 9f775f7

4 files changed

Lines changed: 407 additions & 1 deletion

File tree

‎maestro-engine/src/main/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisher.java‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
/*
2+
* Copyright 2026 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
5+
* the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
10+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
11+
* specific language governing permissions and limitations under the License.
12+
*/
13+
package com.netflix.maestro.engine.publisher;
14+
15+
import com.fasterxml.jackson.core.JsonProcessingException;
16+
import com.fasterxml.jackson.databind.ObjectMapper;
17+
import com.netflix.maestro.exceptions.MaestroInternalError;
18+
import com.netflix.maestro.exceptions.MaestroRetryableError;
19+
import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException;
20+
import com.netflix.maestro.models.events.MaestroEvent;
21+
import java.io.IOException;
22+
import java.net.URI;
23+
import java.net.http.HttpClient;
24+
import java.net.http.HttpRequest;
25+
import java.net.http.HttpResponse;
26+
import java.nio.charset.StandardCharsets;
27+
import java.time.Duration;
28+
import java.util.EnumSet;
29+
import java.util.HexFormat;
30+
import java.util.Set;
31+
import javax.crypto.Mac;
32+
import javax.crypto.spec.SecretKeySpec;
33+
import lombok.extern.slf4j.Slf4j;
34+
35+
/**
36+
* Webhook notification publisher implementation for MaestroNotificationPublisher. It POSTs the
37+
* JSON-serialized {@link MaestroEvent} to the configured endpoint whenever a step or workflow
38+
* status (or workflow definition) changes.
39+
*
40+
* <p>Delivery semantics follow the other publishers: a non-2xx client error (4xx) is thrown as
41+
* non-retryable ({@link MaestroUnprocessableEntityException}) since the receiver rejected the
42+
* payload, while server errors (5xx) and I/O failures are thrown as {@link MaestroRetryableError}
43+
* so the notification job event is retried by the queue system.
44+
*
45+
* <p>When a signing secret is configured, the request carries an {@code X-Maestro-Signature-256}
46+
* header with the hex-encoded HMAC-SHA256 of the request body (GitHub-webhook style, {@code
47+
* sha256=...}), so receivers can authenticate the sender.
48+
*/
49+
@Slf4j
50+
public class WebhookNotificationPublisher implements MaestroNotificationPublisher {
51+
private static final String EVENT_TYPE_HEADER = "X-Maestro-Event-Type";
52+
private static final String SIGNATURE_HEADER = "X-Maestro-Signature-256";
53+
private static final String SIGNATURE_PREFIX = "sha256=";
54+
private static final String HMAC_ALGORITHM = "HmacSHA256";
55+
56+
private final HttpClient httpClient;
57+
private final String url;
58+
private final Set<MaestroEvent.Type> eventTypes;
59+
private final Duration requestTimeout;
60+
private final SecretKeySpec signingKey;
61+
private final ObjectMapper objectMapper;
62+
63+
/**
64+
* Constructor.
65+
*
66+
* @param httpClient http client used to send webhook requests
67+
* @param url webhook endpoint to POST events to
68+
* @param eventTypes event types to publish; an empty set means all event types
69+
* @param requestTimeout per-request timeout
70+
* @param signingSecret secret for the HMAC-SHA256 signature header; null or empty disables it
71+
* @param objectMapper object mapper to serialize events
72+
*/
73+
public WebhookNotificationPublisher(
74+
HttpClient httpClient,
75+
String url,
76+
Set<MaestroEvent.Type> eventTypes,
77+
Duration requestTimeout,
78+
String signingSecret,
79+
ObjectMapper objectMapper) {
80+
this.httpClient = httpClient;
81+
this.url = url;
82+
this.eventTypes =
83+
eventTypes == null || eventTypes.isEmpty()
84+
? EnumSet.allOf(MaestroEvent.Type.class)
85+
: EnumSet.copyOf(eventTypes);
86+
this.requestTimeout = requestTimeout;
87+
this.signingKey =
88+
signingSecret == null || signingSecret.isEmpty()
89+
? null
90+
: new SecretKeySpec(signingSecret.getBytes(StandardCharsets.UTF_8), HMAC_ALGORITHM);
91+
this.objectMapper = objectMapper;
92+
}
93+
94+
/** Send a Maestro event to the webhook endpoint. */
95+
@Override
96+
public void send(MaestroEvent event) {
97+
if (!this.eventTypes.contains(event.getType())) {
98+
LOG.debug("Skipping maestro event of type [{}]: not in the configured set", event.getType());
99+
return;
100+
}
101+
102+
byte[] payload;
103+
try {
104+
payload = this.objectMapper.writeValueAsBytes(event);
105+
} catch (JsonProcessingException je) {
106+
throw new MaestroUnprocessableEntityException("cannot serialize maestro event: " + event, je);
107+
}
108+
109+
HttpRequest.Builder requestBuilder =
110+
HttpRequest.newBuilder(URI.create(this.url))
111+
.timeout(this.requestTimeout)
112+
.header("Content-Type", "application/json")
113+
.header(EVENT_TYPE_HEADER, event.getType().name())
114+
.POST(HttpRequest.BodyPublishers.ofByteArray(payload));
115+
if (this.signingKey != null) {
116+
requestBuilder.header(SIGNATURE_HEADER, SIGNATURE_PREFIX + sign(payload));
117+
}
118+
119+
HttpResponse<String> response;
120+
try {
121+
response = this.httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
122+
} catch (IOException e) {
123+
throw new MaestroRetryableError(
124+
e, "Failed to POST a maestro event to webhook [%s] and will retry.", this.url);
125+
} catch (InterruptedException e) {
126+
Thread.currentThread().interrupt();
127+
throw new MaestroRetryableError(
128+
e, "Interrupted while sending a maestro event to webhook [%s] and will retry.", this.url);
129+
}
130+
131+
int statusCode = response.statusCode();
132+
if (statusCode >= 200 && statusCode < 300) {
133+
LOG.info(
134+
"Published a maestro event of type [{}] to webhook [{}] with status code: [{}]",
135+
event.getType(),
136+
this.url,
137+
statusCode);
138+
} else if (statusCode >= 400 && statusCode < 500) {
139+
// The receiver actively rejected the payload: retrying the same request will not help.
140+
throw new MaestroUnprocessableEntityException(
141+
"Webhook [%s] rejected a maestro event of type [%s] with status code [%s] and body [%s]",
142+
this.url, event.getType(), statusCode, response.body());
143+
} else {
144+
throw new MaestroRetryableError(
145+
"Webhook [%s] returned status code [%s] for a maestro event of type [%s] and will retry.",
146+
this.url, statusCode, event.getType());
147+
}
148+
}
149+
150+
private String sign(byte[] payload) {
151+
try {
152+
Mac mac = Mac.getInstance(HMAC_ALGORITHM);
153+
mac.init(this.signingKey);
154+
return HexFormat.of().formatHex(mac.doFinal(payload));
155+
} catch (Exception e) {
156+
throw new MaestroInternalError(e, "Failed to compute the webhook signature");
157+
}
158+
}
159+
}

‎maestro-engine/src/test/java/com/netflix/maestro/engine/publisher/WebhookNotificationPublisherTest.java‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,177 @@
1+
/*
2+
* Copyright 2026 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
5+
* the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
10+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
11+
* specific language governing permissions and limitations under the License.
12+
*/
13+
package com.netflix.maestro.engine.publisher;
14+
15+
import com.netflix.maestro.AssertHelper;
16+
import com.netflix.maestro.MaestroBaseTest;
17+
import com.netflix.maestro.exceptions.MaestroRetryableError;
18+
import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException;
19+
import com.netflix.maestro.models.events.MaestroEvent;
20+
import com.netflix.maestro.models.events.WorkflowVersionChangeEvent;
21+
import com.sun.net.httpserver.HttpServer;
22+
import java.io.IOException;
23+
import java.net.InetSocketAddress;
24+
import java.net.http.HttpClient;
25+
import java.nio.charset.StandardCharsets;
26+
import java.time.Duration;
27+
import java.util.Collections;
28+
import java.util.EnumSet;
29+
import java.util.HexFormat;
30+
import java.util.Map;
31+
import java.util.concurrent.ConcurrentHashMap;
32+
import java.util.concurrent.atomic.AtomicInteger;
33+
import java.util.concurrent.atomic.AtomicReference;
34+
import javax.crypto.Mac;
35+
import javax.crypto.spec.SecretKeySpec;
36+
import org.junit.After;
37+
import org.junit.Assert;
38+
import org.junit.Before;
39+
import org.junit.Test;
40+
41+
public class WebhookNotificationPublisherTest extends MaestroBaseTest {
42+
private static final Duration TIMEOUT = Duration.ofSeconds(5);
43+
private static final String SECRET = "test-secret";
44+
45+
private HttpServer server;
46+
private String url;
47+
private final AtomicInteger statusToReturn = new AtomicInteger(200);
48+
private final AtomicInteger requestCount = new AtomicInteger();
49+
private final AtomicReference<byte[]> lastBody = new AtomicReference<>();
50+
private final Map<String, String> lastHeaders = new ConcurrentHashMap<>();
51+
52+
@Before
53+
public void setUp() throws IOException {
54+
server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
55+
server.createContext(
56+
"/webhook",
57+
exchange -> {
58+
requestCount.incrementAndGet();
59+
lastBody.set(exchange.getRequestBody().readAllBytes());
60+
exchange
61+
.getRequestHeaders()
62+
.forEach((key, values) -> lastHeaders.put(key, values.getFirst()));
63+
exchange.sendResponseHeaders(statusToReturn.get(), -1);
64+
exchange.close();
65+
});
66+
server.start();
67+
url = "http://localhost:" + server.getAddress().getPort() + "/webhook";
68+
}
69+
70+
@After
71+
public void tearDown() {
72+
server.stop(0);
73+
}
74+
75+
@Test
76+
public void testPublishEventWithSignature() throws Exception {
77+
WebhookNotificationPublisher publisher =
78+
new WebhookNotificationPublisher(
79+
HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, SECRET, MAPPER);
80+
MaestroEvent event = sampleEvent();
81+
82+
publisher.send(event);
83+
84+
Assert.assertEquals(1, requestCount.get());
85+
Assert.assertEquals("application/json", lastHeaders.get("Content-type"));
86+
Assert.assertEquals(
87+
MaestroEvent.Type.WORKFLOW_VERSION_CHANGE_EVENT.name(),
88+
lastHeaders.get("X-maestro-event-type"));
89+
// the body round-trips as the same event
90+
Assert.assertEquals(
91+
MAPPER.readTree(MAPPER.writeValueAsBytes(event)), MAPPER.readTree(lastBody.get()));
92+
// the signature is the HMAC-SHA256 of the exact bytes received
93+
Mac mac = Mac.getInstance("HmacSHA256");
94+
mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
95+
Assert.assertEquals(
96+
"sha256=" + HexFormat.of().formatHex(mac.doFinal(lastBody.get())),
97+
lastHeaders.get("X-maestro-signature-256"));
98+
}
99+
100+
@Test
101+
public void testNoSignatureHeaderWithoutSecret() {
102+
WebhookNotificationPublisher publisher =
103+
new WebhookNotificationPublisher(
104+
HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER);
105+
106+
publisher.send(sampleEvent());
107+
108+
Assert.assertEquals(1, requestCount.get());
109+
Assert.assertFalse(lastHeaders.containsKey("X-maestro-signature-256"));
110+
}
111+
112+
@Test
113+
public void testEventTypeFiltering() {
114+
WebhookNotificationPublisher publisher =
115+
new WebhookNotificationPublisher(
116+
HttpClient.newHttpClient(),
117+
url,
118+
EnumSet.of(MaestroEvent.Type.STEP_INSTANCE_STATUS_CHANGE_EVENT),
119+
TIMEOUT,
120+
null,
121+
MAPPER);
122+
123+
publisher.send(sampleEvent()); // WORKFLOW_VERSION_CHANGE_EVENT: filtered out
124+
125+
Assert.assertEquals(0, requestCount.get());
126+
}
127+
128+
@Test
129+
public void testClientErrorIsNotRetryable() {
130+
statusToReturn.set(400);
131+
WebhookNotificationPublisher publisher =
132+
new WebhookNotificationPublisher(
133+
HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER);
134+
135+
AssertHelper.assertThrows(
136+
"4xx means the receiver rejected the payload",
137+
MaestroUnprocessableEntityException.class,
138+
"rejected a maestro event",
139+
() -> publisher.send(sampleEvent()));
140+
}
141+
142+
@Test
143+
public void testServerErrorIsRetryable() {
144+
statusToReturn.set(503);
145+
WebhookNotificationPublisher publisher =
146+
new WebhookNotificationPublisher(
147+
HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER);
148+
149+
AssertHelper.assertThrows(
150+
"5xx is retryable",
151+
MaestroRetryableError.class,
152+
"will retry",
153+
() -> publisher.send(sampleEvent()));
154+
}
155+
156+
@Test
157+
public void testConnectionFailureIsRetryable() {
158+
server.stop(0);
159+
WebhookNotificationPublisher publisher =
160+
new WebhookNotificationPublisher(
161+
HttpClient.newHttpClient(), url, Collections.emptySet(), TIMEOUT, null, MAPPER);
162+
163+
AssertHelper.assertThrows(
164+
"connection failures are retryable",
165+
MaestroRetryableError.class,
166+
"will retry",
167+
() -> publisher.send(sampleEvent()));
168+
}
169+
170+
private MaestroEvent sampleEvent() {
171+
return WorkflowVersionChangeEvent.builder()
172+
.workflowId("sample-wf")
173+
.workflowName("sample-wf-name")
174+
.clusterName("test-cluster")
175+
.build();
176+
}
177+
}

0 commit comments

Comments
 (0)