[IMPROVEMENT] Add Support for Custom Headers and HTTP Methods in Generic Webhook

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/generic/GenericPublisher.java

@@ -43,6 +43,8 @@
 public class GenericPublisher implements Destination<ChangeEvent> {
  private final Client client;
  private final Webhook webhook;
+ private static final String TEST_MESSAGE_JSON =
+ "This is a test message from OpenMetadata to confirm your webhook destination is configured correctly.";
 
  @Getter private final SubscriptionDestination subscriptionDestination;
  private final EventSubscription eventSubscription;
@@ -67,79 +69,89 @@ public GenericPublisher(
  public void sendMessage(ChangeEvent event) throws EventPublisherException {
  long attemptTime = System.currentTimeMillis();
  try {
- // Post Message to default
  String json = JsonUtils.pojoToJson(event);
- if (webhook.getEndpoint() != null) {
- if (webhook.getSecretKey() != null && !webhook.getSecretKey().isEmpty()) {
- String hmac =
- "sha256="
- + CommonUtil.calculateHMAC(decryptWebhookSecretKey(webhook.getSecretKey()), json);
- postWebhookMessage(this, getTarget().header(RestUtil.SIGNATURE_HEADER, hmac), json);
- } else {
- postWebhookMessage(this, getTarget(), json);
- }
- }
+
+ prepareAndSendMessage(json, getTarget());
 
  // Post to Generic Webhook with Actions
- String eventJson = JsonUtils.pojoToJson(event);
  List<Invocation.Builder> targets =
  getTargetsForWebhookAlert(
  webhook, subscriptionDestination.getCategory(), WEBHOOK, client, event);
+ String eventJson = JsonUtils.pojoToJson(event);
  for (Invocation.Builder actionTarget : targets) {
  postWebhookMessage(this, actionTarget, eventJson);
  }
  } catch (Exception ex) {
- Throwable cause = ex.getCause();
- String message;
- if (cause != null && cause.getClass() == UnknownHostException.class) {
- message =
- String.format(
- "Unknown Host Exception for Generic Publisher : %s , WebhookEndpoint : %s",
- subscriptionDestination.getId(), webhook.getEndpoint());
- LOG.warn(message);
- setErrorStatus(attemptTime, 400, "UnknownHostException");
- } else {
- message =
- CatalogExceptionMessage.eventPublisherFailedToPublish(WEBHOOK, event, ex.getMessage());
- LOG.error(message);
- }
- throw new EventPublisherException(message, Pair.of(subscriptionDestination.getId(), event));
+ handleException(attemptTime, event, ex);
  }
  }
 
  @Override
  public void sendTestMessage() throws EventPublisherException {
  long attemptTime = System.currentTimeMillis();
  try {
- // Post Message to default
- String json =
- "This is a test message from OpenMetadata to confirm your webhook destination is configured correctly.";
- if (webhook.getEndpoint() != null) {
- if (webhook.getSecretKey() != null && !webhook.getSecretKey().isEmpty()) {
- String hmac =
- "sha256="
- + CommonUtil.calculateHMAC(decryptWebhookSecretKey(webhook.getSecretKey()), json);
- postWebhookMessage(this, getTarget().header(RestUtil.SIGNATURE_HEADER, hmac), json);
- } else {
- postWebhookMessage(this, getTarget(), json);
- }
- }
+ prepareAndSendMessage(TEST_MESSAGE_JSON, getTarget());
  } catch (Exception ex) {
- Throwable cause = ex.getCause();
- String message;
- if (cause != null && cause.getClass() == UnknownHostException.class) {
- message =
- String.format(
- "Unknown Host Exception for Generic Publisher : %s , WebhookEndpoint : %s",
- subscriptionDestination.getId(), webhook.getEndpoint());
- LOG.warn(message);
- setErrorStatus(attemptTime, 400, "UnknownHostException");
- } else {
- message = CatalogExceptionMessage.eventPublisherFailedToPublish(WEBHOOK, ex.getMessage());
- LOG.error(message);
+ handleException(attemptTime, ex);
+ }
+ }
+
+ private void prepareAndSendMessage(String json, Invocation.Builder target) {
+ if (!CommonUtil.nullOrEmpty(webhook.getEndpoint())) {
+
+ // Add HMAC signature header if secret key is present
+ if (!CommonUtil.nullOrEmpty(webhook.getSecretKey())) {
+ String hmac =
+ "sha256="
+ + CommonUtil.calculateHMAC(decryptWebhookSecretKey(webhook.getSecretKey()), json);
+ target.header(RestUtil.SIGNATURE_HEADER, hmac);
  }
- throw new EventPublisherException(message);
+
+ // Add custom headers if they exist
+ Map<String, String> headers = webhook.getHeaders();
+ if (!CommonUtil.nullOrEmpty(headers)) {
+ headers.forEach(target::header);
+ }
+
+ Webhook.HttpMethod httpOperation = webhook.getHttpMethod();
+ postWebhookMessage(this, target, json, httpOperation);
+ }
+ }
+
+ private void handleException(long attemptTime, ChangeEvent event, Exception ex)
+ throws EventPublisherException {
+ Throwable cause = ex.getCause();
+ String message;
+ if (cause != null && cause.getClass() == UnknownHostException.class) {
+ message =
+ String.format(
+ "Unknown Host Exception for Generic Publisher : %s , WebhookEndpoint : %s",
+ subscriptionDestination.getId(), webhook.getEndpoint());
+ LOG.warn(message);
+ setErrorStatus(attemptTime, 400, "UnknownHostException");
+ } else {
+ message =
+ CatalogExceptionMessage.eventPublisherFailedToPublish(WEBHOOK, event, ex.getMessage());
+ LOG.error(message);
+ }
+ throw new EventPublisherException(message, Pair.of(subscriptionDestination.getId(), event));
+ }
+
+ private void handleException(long attemptTime, Exception ex) throws EventPublisherException {
+ Throwable cause = ex.getCause();
+ String message;
+ if (cause != null && cause.getClass() == UnknownHostException.class) {
+ message =
+ String.format(
+ "Unknown Host Exception for Generic Publisher : %s , WebhookEndpoint : %s",
+ subscriptionDestination.getId(), webhook.getEndpoint());
+ LOG.warn(message);
+ setErrorStatus(attemptTime, 400, "UnknownHostException");
+ } else {
+ message = CatalogExceptionMessage.eventPublisherFailedToPublish(WEBHOOK, ex.getMessage());
+ LOG.error(message);
  }
+ throw new EventPublisherException(message);
  }
 
  private Invocation.Builder getTarget() {

openmetadata-service/src/main/java/org/openmetadata/service/resources/analytics/WebAnalyticEventResource.java

@@ -42,18 +42,12 @@
 import org.openmetadata.schema.analytics.WebAnalyticEventData;
 import org.openmetadata.schema.analytics.type.WebAnalyticEventType;
 import org.openmetadata.schema.api.data.RestoreEntity;
-import org.openmetadata.schema.api.events.EventSubscriptionDestinationTestRequest;
 import org.openmetadata.schema.api.tests.CreateWebAnalyticEvent;
-import org.openmetadata.schema.entity.events.EventSubscription;
-import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.EntityHistory;
 import org.openmetadata.schema.type.Include;
 import org.openmetadata.schema.type.MetadataOperation;
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.OpenMetadataApplicationConfig;
-import org.openmetadata.service.apps.bundles.changeEvent.AlertFactory;
-import org.openmetadata.service.apps.bundles.changeEvent.Destination;
-import org.openmetadata.service.events.errors.EventPublisherException;
 import org.openmetadata.service.jdbi3.ListFilter;
 import org.openmetadata.service.jdbi3.WebAnalyticEventRepository;
 import org.openmetadata.service.limits.Limits;
@@ -492,41 +486,6 @@ public Response addReportResult(
  return repository.addWebAnalyticEventData(sanitizeWebAnalyticEventData(webAnalyticEventData));
  }
 
- @POST
- @Path("/sendTestAlert")
- @Operation(
- operationId = "testDestination",
- summary = "Send a test message alert to external destinations.",
- description = "Send a test message alert to external destinations of the alert.",
- responses = {
- @ApiResponse(
- responseCode = "200",
- description = "Test message sent successfully",
- content = @Content(schema = @Schema(implementation = Response.class)))
- })
- public Response sendTestMessageAlert(
- @Context UriInfo uriInfo,
- @Context SecurityContext securityContext,
- EventSubscriptionDestinationTestRequest request) {
- EventSubscription eventSubscription =
- new EventSubscription().withFullyQualifiedName(request.getAlertName());
-
- request
- .getDestinations()
- .forEach(
- (destination) -> {
- Destination<ChangeEvent> alert =
- AlertFactory.getAlert(eventSubscription, destination);
- try { // by-pass alertEventConsumer
- alert.sendTestMessage();
- } catch (EventPublisherException e) {
- LOG.error(e.getMessage());
- }
- });
-
- return Response.ok().build();
- }
-
  @DELETE
  @Path("/{name}/{timestamp}/collect")
  @Operation(

openmetadata-service/src/main/java/org/openmetadata/service/resources/events/subscription/EventSubscriptionResource.java

@@ -59,16 +59,21 @@
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.common.utils.CommonUtil;
 import org.openmetadata.schema.api.events.CreateEventSubscription;
+import org.openmetadata.schema.api.events.EventSubscriptionDestinationTestRequest;
 import org.openmetadata.schema.entity.events.EventFilterRule;
 import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.entity.events.SubscriptionStatus;
+import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.EntityHistory;
 import org.openmetadata.schema.type.FilterResourceDescriptor;
 import org.openmetadata.schema.type.MetadataOperation;
 import org.openmetadata.schema.type.NotificationResourceDescriptor;
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.OpenMetadataApplicationConfig;
+import org.openmetadata.service.apps.bundles.changeEvent.AlertFactory;
+import org.openmetadata.service.apps.bundles.changeEvent.Destination;
+import org.openmetadata.service.events.errors.EventPublisherException;
 import org.openmetadata.service.events.scheduled.EventSubscriptionScheduler;
 import org.openmetadata.service.events.subscription.AlertUtil;
 import org.openmetadata.service.events.subscription.EventsSubscriptionRegistry;
@@ -621,6 +626,42 @@ public void validateCondition(
  AlertUtil.validateExpression(expression, Boolean.class);
  }
 
+ @POST
+ @Path("/testDestination")
+ @Operation(
+ operationId = "testDestination",
+ summary = "Send a test message alert to external destinations.",
+ description = "Send a test message alert to external destinations of the alert.",
+ responses = {
+ @ApiResponse(
+ responseCode = "200",
+ description = "Test message sent successfully",
+ content = @Content(schema = @Schema(implementation = Response.class)))
+ })
+ public Response sendTestMessageAlert(
+ @Context UriInfo uriInfo,
+ @Context SecurityContext securityContext,
+ EventSubscriptionDestinationTestRequest request) {
+ EventSubscription eventSubscription =
+ new EventSubscription().withFullyQualifiedName(request.getAlertName());
+
+ // by-pass AbstractEventConsumer - covers external destinations as of now
+ request
+ .getDestinations()
+ .forEach(
+ (destination) -> {
+ Destination<ChangeEvent> alert =
+ AlertFactory.getAlert(eventSubscription, destination);
+ try {
+ alert.sendTestMessage();
+ } catch (EventPublisherException e) {
+ LOG.error(e.getMessage());
+ }
+ });
+
+ return Response.ok().build();
+ }
+
  private EventSubscription getEventSubscription(CreateEventSubscription create, String user) {
  return repository
  .copy(new EventSubscription(), create, user)

openmetadata-service/src/main/java/org/openmetadata/service/util/SubscriptionUtil.java

@@ -417,14 +417,31 @@ public static Invocation.Builder appendHeadersToTarget(Client client, String uri
 
  public static void postWebhookMessage(
  Destination<ChangeEvent> destination, Invocation.Builder target, Object message) {
+ postWebhookMessage(destination, target, message, Webhook.HttpMethod.POST);
+ }
+
+ public static void postWebhookMessage(
+ Destination<ChangeEvent> destination,
+ Invocation.Builder target,
+ Object message,
+ Webhook.HttpMethod httpMethod) {
  long attemptTime = System.currentTimeMillis();
- Response response =
- target.post(javax.ws.rs.client.Entity.entity(message, MediaType.APPLICATION_JSON_TYPE));
+ Response response;
+
+ if (httpMethod == Webhook.HttpMethod.PUT) {
+ response =
+ target.put(javax.ws.rs.client.Entity.entity(message, MediaType.APPLICATION_JSON_TYPE));
+ } else {
+ response =
+ target.post(javax.ws.rs.client.Entity.entity(message, MediaType.APPLICATION_JSON_TYPE));
+ }
+
  LOG.debug(
- "Subscription Destination Posted Message {}:{} received response {}",
+ "Subscription Destination HTTP Operation {}:{} received response {}",
+ httpMethod,
  destination.getSubscriptionDestination().getId(),
- message,
  response.getStatusInfo());
+
  if (response.getStatus() >= 300 && response.getStatus() < 400) {
  // 3xx response/redirection is not allowed for callback. Set the webhook state as in error
  destination.setErrorStatus(

openmetadata-spec/src/main/resources/json/schema/entity/events/webhook.json

@@ -35,6 +35,17 @@
  "description": "Secret set by the webhook client used for computing HMAC SHA256 signature of webhook payload and sent in `X-OM-Signature` header in POST requests to publish the events.",
  "type": "string"
  },
+ "headers": {
+ "description": "Custom headers to be sent with the webhook request.",
+ "type": "object",
+ "existingJavaType": "java.util.Map<String, String>"
+ },
+ "httpMethod": {
+ "description": "HTTP operation to send the webhook request. Supports POST or PUT.",
+ "type": "string",
+ "enum": ["POST", "PUT"],
+ "default": "POST"
+ },
  "sendToAdmins": {
  "description": "Send the Event to Admins",
  "type": "boolean",