conductor-oss / conductor-oss/conductor
[FEATURE] Support user-defined metadata in webhook notification payloads via `WorkflowDef.metadata`
- Dominant language
- Java
- Stars
- 32.2k
- Forks
- 1k
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 37
Description
## Motivation
Webhook notification payloads currently have no mechanism to carry user-defined metadata. Contextual information like team, service, or environment can only exist inside the serialized `input` string, forcing consumers to parse raw strings for routing and filtering. A definition-level mechanism to surface this metadata as structured top-level fields would solve this cleanly.
## Use Cases
- **Consumer-side routing**: Webhook consumers can route or fan-out notifications to different channels (Slack, PagerDuty, etc.) based on `meta_team` or `meta_service` without parsing the serialized `input` string.
- **Alert filtering**: Filter or group alerts in monitoring systems (PagerDuty, Opsgenie, etc.) using `meta_environment` or `meta_severity`.
- **Audit and Observability**: Tag workflow/task events in log aggregators or dashboards (Datadog, Splunk, etc.) for per-team or per-service drill-down.
## Scope
Webhook-based publishers only:
- `TaskStatusPublisher` (task status -> REST webhook)
- `StatusChangePublisher` (workflow status -> REST webhook)
Other publishers (Kafka, Queue, archiving) are not affected.
## Proposal
Leverage the existing `WorkflowDef.metadata` field to enrich webhook notification payloads with user-defined fields. Users define a `notificationMeta` key inside `metadata` in the workflow definition. When enabled, every workflow and task notification from that definition automatically includes the metadata as top-level `meta_`-prefixed fields.
### Usage
**Step 1**: Enable in server configuration:
```properties
conductor.status-notifier.notification.publish-webhook-notification-meta-enabled=true
```
**Step 2**: Define `notificationMeta` in the workflow definition:
```json
{
"name": "order_processing",
"version": 1,
"metadata": {
"notificationMeta": {
"team": "payments",
"service": "order-processor",
"environment": "production"
}
},
"tasks": [...]
}
```
Define once, applied to every execution's notifications automatically.
### How it works
- **Workflow notifications**: `StatusChangeNotification` reads `workflow.getWorkflowDefinition().getMetadata().get("notificationMeta")` and promotes entries to top-level fields with a `meta_` prefix using `@JsonAnyGetter`. No extra DB calls — `WorkflowDef` is already on the `Workflow` object.
- **Task notifications**: `TaskStatusPublisher` looks up the parent workflow via `executionDAOFacade.getWorkflowModel(task.getWorkflowInstanceId(), false)` and extracts `notificationMeta` from the workflow definition. One extra async DB call per task notification.
- **Feature flag**: Both publishers check `publish-webhook-notification-meta-enabled` before extracting metadata. Disabled by default.
### Code Changes
The feature flag `publishWebhookNotificationMetaEnabled` (default `false`) is added to `StatusNotifierNotificationProperties` and passed through both `StatusChangePublisherConfiguration` and `TaskStatusPublisherConfiguration` to their respective publishers.
When the flag is enabled, publishers extract `notificationMeta` from the workflow definition. For workflow notifications, `WorkflowDef` is already on the `Workflow` object. For task notifications, the parent workflow is looked up via `executionDAOFacade.getWorkflowModel()`.
The metadata extraction logic (shared pattern for both publishers):
```java
WorkflowDef def = workflow.getWorkflowDefinition();
if (def != null && def.getMetadata() != null) {
Object notifMeta = def.getMetadata().get("notificationMeta");
if (notifMeta instanceof Map) {
return (Map) notifMeta;
}
}
```
Both notification classes (`StatusChangeNotification`, `TaskNotification`) receive the extracted map and promote entries as top-level JSON fields:
```java
private Map ioMetaFields = new LinkedHashMap<>();
// In constructor:
if (notificationMeta != null) {
notificationMeta.forEach((key, value) ->
ioMetaFields.put("meta_" + key, value));
}
@JsonAnyGetter
public Map getIoMetaFields() {
return ioMetaFields;
}
```
### Resulting Payloads
Workflow webhook payload:
```json
{
"workflowType": "order_processing",
"workflowId": "d5f8a2b1-...",
"status": "COMPLETED",
"meta_team": "payments",
"meta_service": "order-processor",
"meta_environment": "production",
"input": "...",
"output": "..."
}
```
Task webhook payload:
```json
{
"taskId": "a1b2c3d4-...",
"taskDefName": "process_order",
"status": "SCHEDULED",
"workflowId": "d5f8a2b1-...",
"meta_team": "payments",
"meta_service": "order-processor",
"meta_environment": "production",
"input": "...",
"output": null
}
```
### Design Decisions
- **`WorkflowDef.metadata` as source**: Metadata is defined once in the workflow definition — not per execution. Keeps infrastructure concerns separate from business input data.
- **`notificationMeta` subkey**: Only this specific key is promoted to payloads. Prevents sensitive or unrelated data already in `metadata` from leaking into webhooks after an upgrade.
- **`meta_` prefix**: Prevents collisions with existing Conductor payload fields.
- **`@JsonAnyGetter`**: Promotes entries as top-level JSON fields for easy consumer access.
- **Single feature flag**: One flag (`publish-webhook-notification-meta-enabled`, default `false`) controls both publishers. The `webhook` in the name scopes it clearly to webhook publishers only.
- **Extra DB read for task notifications only**: One `getWorkflowModel()` call per task notification. Async processing makes this acceptable. Workflow notifications need zero extra calls.
### Backward Compatibility
- **Feature flag defaults to `false`** — upgrading Conductor changes nothing.
- **Subkey is opt-in** — even with the flag enabled, only workflows with `metadata.notificationMeta` get `meta_*` fields.
- **Existing `metadata` is safe** — only the `notificationMeta` key is read; other data is never exposed.
- **Additive only** — no fields renamed or removed. Other publishers unaffected.
Contributor guide
Assessment
This issue has not been assessed yet.