Solace Event Mesh Synthetic Delivery and Round-Trip Latency Sentinel

id: solace-event-mesh-synthetic-delivery-sentinel
namespace: company.team

description: |
  Audit Solace PubSub+ broker health and event mesh availability using synthetic end-to-end
  message publishing and consumption. Measures round-trip transit latency, detects delivery
  timeouts or dropped messages, and dispatches structured Slack alerts when latency SLAs breach.

inputs:
  - id: solace_host
    type: STRING
    displayName: Solace Broker Host
    description: Hostname and port for the Solace PubSub+ SMF messaging interface,
      for example localhost:55555.
    defaults: "localhost:55555"

  - id: solace_vpn
    type: STRING
    displayName: Solace Message VPN
    description: Message VPN name to connect to on the Solace event broker.
    defaults: "default"

  - id: canary_topic
    type: STRING
    displayName: Canary Topic Destination
    description: Solace topic string that the canary queue is subscribed to for
      message routing.
    defaults: "canary/health/probe"

  - id: queue_name
    type: STRING
    displayName: Canary Queue Name
    description: Dedicated queue provisioned on the Solace broker for synthetic
      health probing.
    defaults: "canary.health.queue"

  - id: queue_type
    type: SELECT
    displayName: Canary Queue Type
    description: Solace queue access type as provisioned on the broker.
    values:
      - "DURABLE_NON_EXCLUSIVE"
      - "DURABLE_EXCLUSIVE"
      - "NON_DURABLE_EXCLUSIVE"
    defaults: "DURABLE_NON_EXCLUSIVE"

  - id: latency_sla_ms
    type: INT
    displayName: Latency SLA Threshold (ms)
    description: Maximum acceptable round-trip transit latency in milliseconds
      before flagging degraded broker performance.
    defaults: 2000

  - id: alert_mode
    type: SELECT
    displayName: Sentinel Alert Mode
    description: Operating mode of the sentinel. ALERT dispatches a notification to
      Slack when thresholds breach; AUDIT_ONLY logs findings without paging
      on-call.
    values:
      - "ALERT"
      - "AUDIT_ONLY"
    defaults: "ALERT"

triggers:
  - id: periodic_canary_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Runs every 15 minutes to verify Solace broker end-to-end message
      routing and latency. Shipped disabled by default for safe initial setup.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      solace_host: "localhost:55555"
      solace_vpn: "default"
      canary_topic: "canary/health/probe"
      queue_name: "canary.health.queue"
      queue_type: "DURABLE_NON_EXCLUSIVE"
      latency_sla_ms: 2000
      alert_mode: "ALERT"

tasks:
  - id: record_probe_dispatch_time
    type: io.kestra.plugin.core.debug.Return
    description: Capture initial timestamp in ISO-8601 format to benchmark
      end-to-end message transit duration.
    format: "{{ now() }}"

  - id: publish_canary_probe
    type: io.kestra.plugin.solace.Produce
    description: Publish a synthetic canary probe payload with unique execution
      correlation ID to the canary topic destination.
    host: "{{ inputs.solace_host }}"
    vpn: "{{ inputs.solace_vpn }}"
    username: "{{ secret('SOLACE_USERNAME') }}"
    password: "{{ secret('SOLACE_PASSWORD') }}"
    topicDestination: "{{ inputs.canary_topic }}"
    from:
      payload: "{\"probe_id\": \"{{ execution.id }}\", \"dispatched_at\": \"{{
        outputs.record_probe_dispatch_time.value }}\", \"canary_origin\":
        \"kestra-sentinel\"}"
      properties:
        correlationId: "{{ execution.id }}"
        canaryOrigin: "kestra-sentinel"
    messageSerializer: JSON
    deliveryMode: PERSISTENT
    awaitAcknowledgementTimeout: PT10S

  - id: consume_canary_probe
    type: io.kestra.plugin.solace.Consume
    description: Pull the synthetic canary message from the target queue within a
      bounded poll window.
    host: "{{ inputs.solace_host }}"
    vpn: "{{ inputs.solace_vpn }}"
    username: "{{ secret('SOLACE_USERNAME') }}"
    password: "{{ secret('SOLACE_PASSWORD') }}"
    queueName: "{{ inputs.queue_name }}"
    queueType: "{{ inputs.queue_type }}"
    messageDeserializer: JSON
    maxMessages: 1
    maxDuration: PT10S

  - id: classify_delivery_health
    type: io.kestra.plugin.core.debug.Return
    description: Evaluate published versus consumed counts and compute round-trip
      transit duration in milliseconds.
    format: "{{ (outputs.publish_canary_probe.messagesCount ?? 0) == 0 ?
      'PUBLISH_FAILED' : ((outputs.consume_canary_probe.messagesCount ?? 0) == 0
      ? 'TIMEOUT_UNCONSUMED' : (((now() | timestampMilli) -
      (outputs.record_probe_dispatch_time.value | timestampMilli)) >
      inputs.latency_sla_ms ? 'SLA_BREACH' : 'HEALTHY')) }}"

  - id: route_delivery_health
    type: io.kestra.plugin.core.flow.Switch
    description: Branch execution based on canary delivery classification to page
      on-call or record healthy audit pass.
    value: "{{ outputs.classify_delivery_health.value | trim }}"
    cases:
      SLA_BREACH:
        - id: handle_sla_breach
          type: io.kestra.plugin.core.flow.If
          description: Verify alert_mode setting before dispatching degraded latency
            notification.
          condition: "{{ inputs.alert_mode == 'ALERT' }}"
          then:
            - id: send_sla_breach_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: Dispatch degraded transit latency notification to Slack.
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "Solace PubSub+ Transit Latency SLA Breach Detected",
                  "blocks": [
                    {
                      "type": "header",
                      "text": {
                        "type": "plain_text",
                        "text": "Solace Transit Latency SLA Breach"
                      }
                    },
                    {
                      "type": "section",
                      "fields": [
                        {
                          "type": "mrkdwn",
                          "text": "*Status:*\nDegraded Performance"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Broker Host:*\n`{{ inputs.solace_host }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Message VPN:*\n`{{ inputs.solace_vpn }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Queue:*\n`{{ inputs.queue_name }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Latency SLA:*\n{{ inputs.latency_sla_ms }} ms"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Measured Duration:*\n{{ (now() | timestampMilli) - (outputs.record_probe_dispatch_time.value | timestampMilli) }} ms"
                        }
                      ]
                    },
                    {
                      "type": "context",
                      "elements": [
                        {
                          "type": "mrkdwn",
                          "text": "Execution ID: `{{ execution.id }}` | Namespace: `{{ flow.namespace }}`"
                        }
                      ]
                    }
                  ]
                }
          else:
            - id: log_sla_breach_audit_only
              type: io.kestra.plugin.core.log.Log
              description: Record transit SLA breach in audit log without paging on-call.
              message: "AUDIT ONLY: Solace transit duration exceeded SLA threshold of {{
                inputs.latency_sla_ms }} ms on broker {{ inputs.solace_host }}
                for queue {{ inputs.queue_name }}."

      TIMEOUT_UNCONSUMED:
        - id: handle_timeout_unconsumed
          type: io.kestra.plugin.core.flow.If
          description: Verify alert_mode setting before dispatching unconsumed probe
            notification.
          condition: "{{ inputs.alert_mode == 'ALERT' }}"
          then:
            - id: send_timeout_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: Dispatch unconsumed canary probe timeout notification to Slack.
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "Solace PubSub+ Canary Message Unconsumed Timeout",
                  "blocks": [
                    {
                      "type": "header",
                      "text": {
                        "type": "plain_text",
                        "text": "Solace Canary Message Delivery Timeout"
                      }
                    },
                    {
                      "type": "section",
                      "fields": [
                        {
                          "type": "mrkdwn",
                          "text": "*Status:*\nMessage Lost or Stalled"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Broker Host:*\n`{{ inputs.solace_host }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Message VPN:*\n`{{ inputs.solace_vpn }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Queue:*\n`{{ inputs.queue_name }}`"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Published Count:*\n{{ outputs.publish_canary_probe.messagesCount }}"
                        },
                        {
                          "type": "mrkdwn",
                          "text": "*Consumed Count:*\n{{ outputs.consume_canary_probe.messagesCount }}"
                        }
                      ]
                    },
                    {
                      "type": "context",
                      "elements": [
                        {
                          "type": "mrkdwn",
                          "text": "Execution ID: `{{ execution.id }}` | Namespace: `{{ flow.namespace }}`"
                        }
                      ]
                    }
                  ]
                }
          else:
            - id: log_timeout_audit_only
              type: io.kestra.plugin.core.log.Log
              description: Record unconsumed probe timeout in audit log without paging
                on-call.
              message: "AUDIT ONLY: Solace canary message published but not consumed within
                poll window on broker {{ inputs.solace_host }} for queue {{
                inputs.queue_name }}."

      PUBLISH_FAILED:
        - id: handle_publish_failed
          type: io.kestra.plugin.core.flow.If
          description: Verify alert_mode setting before dispatching publish failure
            notification.
          condition: "{{ inputs.alert_mode == 'ALERT' }}"
          then:
            - id: send_publish_failed_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: Dispatch publish failure alert to Slack.
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "Solace PubSub+ Synthetic Probe Publish Failure",
                  "blocks": [
                    {
                      "type": "header",
                      "text": {
                        "type": "plain_text",
                        "text": "Solace Publish Failure"
                      }
                    },
                    {
                      "type": "section",
                      "text": {
                        "type": "mrkdwn",
                        "text": "The synthetic canary probe task failed to publish to topic `{{ inputs.canary_topic }}` on broker `{{ inputs.solace_host }}`. Check broker spool capacity and client connection limits."
                      }
                    }
                  ]
                }
          else:
            - id: log_publish_failed_audit_only
              type: io.kestra.plugin.core.log.Log
              description: Record publish failure in audit log without paging on-call.
              message: "AUDIT ONLY: Solace canary probe failed to publish to topic {{
                inputs.canary_topic }} on broker {{ inputs.solace_host }}."

      HEALTHY:
        - id: log_healthy_delivery
          type: io.kestra.plugin.core.log.Log
          description: Record nominal synthetic probe round-trip transit duration.
          message: "HEALTHY: Solace canary probe delivered and consumed in {{ (now() |
            timestampMilli) - (outputs.record_probe_dispatch_time.value |
            timestampMilli) }} ms (SLA limit: {{ inputs.latency_sla_ms }} ms).
            Broker {{ inputs.solace_host }} is fully operational."
    defaults:
      - id: log_default_delivery_status
        type: io.kestra.plugin.core.log.Log
        description: Record fallback status for unexpected canary delivery states.
        message: "Solace canary delivery audit encountered unclassified status: {{
          outputs.classify_delivery_health.value }}."

  - id: log_sentinel_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Record final audit entry maintaining an unbroken execution history
      of sentinel verification passes.
    message: "Solace synthetic delivery audit pass complete. Status: {{
      outputs.classify_delivery_health.value | trim }} | Execution: {{
      execution.id }} | Host: {{ inputs.solace_host }} | Topic: {{
      inputs.canary_topic }} | Queue: {{ inputs.queue_name }}."

errors:
  - id: on_failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Out-of-band error handler alerting on-call if the Solace broker is
      unreachable or authentication fails.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "Solace synthetic delivery sentinel execution {{ execution.id }} FAILED for broker {{ inputs.solace_host }} in namespace {{ flow.namespace }}. Check broker reachability, network ACLs, and credentials."
      }

Solace PubSub+ event brokers power mission-critical event-driven architectures across banking, aerospace, retail, and IoT ecosystems. In high-throughput event meshes, silent degradations such as message spool saturation, queue bind timeouts, network buffer bloat, or slow client acknowledgement loops can cause latency spikes or message stalls without fully dropping TCP connections.

Passive connection monitoring and port health checks fail to detect whether the broker can actively accept, persist, serialize, and dispatch messages to consumers within operational service level agreements (SLAs).

This blueprint implements an automated Day 2 operations canary sentinel for Solace PubSub+. On an automated schedule, Kestra publishes a persistent synthetic probe with a unique correlation ID to a topic destination (canary/health/probe) routed to a subscribed canary queue, consumes the message within a bounded poll window, measures round-trip transit duration in milliseconds, validates message delivery integrity, and pages on-call teams via Slack when transit latency breaches configured thresholds.

How it works

periodic_canary_schedule triggers the sentinel flow on a 15-minute schedule (disabled by default for initial broker setup).
record_probe_dispatch_time captures the baseline dispatch timestamp using io.kestra.plugin.core.debug.Return.
publish_canary_probe publishes a JSON canary payload with a unique execution correlation ID to the target Solace topic destination (canary/health/probe) using io.kestra.plugin.solace.Produce in PERSISTENT delivery mode with an explicit broker acknowledgement timeout.
consume_canary_probe polls the subscribed canary queue using io.kestra.plugin.solace.Consume with a 10-second bounded duration (maxDuration: PT10S) and single-record limit (maxMessages: 1).
classify_delivery_health evaluates message delivery and computes round-trip transit latency:
- PUBLISH_FAILED: If the producer failed to acknowledge the message write.
- TIMEOUT_UNCONSUMED: If the published message was not retrieved within the 10-second polling window.
- SLA_BREACH: If the measured transit latency exceeds latency_sla_ms.
- HEALTHY: If delivery succeeded within the latency threshold.
route_delivery_health uses io.kestra.plugin.core.flow.Switch to branch on the evaluated classification. When degraded or timed out, it checks alert_mode: ALERT dispatches structured notifications to Slack via io.kestra.plugin.slack.notifications.SlackIncomingWebhook, while AUDIT_ONLY logs findings without alerting.
log_sentinel_audit_complete records a concluding log entry on every run to maintain continuous telemetry in Kestra execution history.
A flow-level errors handler catches network partitions, DNS resolution failures, or broker authentication rejections and pages on-call operators.

What you get

Active end-to-end synthetic health verification of Solace PubSub+ event brokers without external monitoring agents.
Continuous round-trip transit latency measurement in milliseconds.
Early detection of message spool saturation, queue binding stalls, and consumer timeouts.
Dual operating modes: ALERT for active incident paging and AUDIT_ONLY for establishing baseline latency profiles.
Resilient error handling that alerts operators if broker endpoints become completely unreachable.

Who it's for

Messaging Platform Engineers and SREs maintaining production Solace PubSub+ broker clusters.
Enterprise Integration Architects operating high-volume event meshes for financial transactions and order processing.
DevOps teams standardizing Day 2 synthetic canary testing across event-driven infrastructure.

Why orchestrate this with Kestra

Solace PubSub+ provides CLI and SEMP monitoring tools, but does not natively execute recurring end-to-end synthetic publish-consume cycles or route SLA breaches to collaboration tools. Kestra orchestrates declarative, version-controlled workflows with built-in secret masking, precise scheduling, conditional branching, native Slack alerts, and complete execution audit logging.

Prerequisites

A running Solace PubSub+ software broker or Solace Cloud instance reachable from the Kestra worker environment.
A dedicated canary health queue provisioned on the Solace Message VPN (e.g. canary.health.queue).
Broker credentials with produce and consume permissions on the canary queue.
A Slack incoming webhook URL for dispatching alerts.

Secrets

SOLACE_USERNAME: Broker username authorized to produce and consume messages on the canary queue.
SOLACE_PASSWORD: Password for the Solace broker user.
SLACK_WEBHOOK: Slack incoming webhook URL for alerting channels.

Note for Kestra Open-Source (OSS): Secrets are configured as base64-encoded environment variables formatted as SECRET_<NAME> (e.g. SECRET_SOLACE_USERNAME, SECRET_SLACK_WEBHOOK).

Inputs

Name	Type	Default	Allowed Values	Description
`solace_host`	STRING	`"localhost:55555"`	Hostname and port	Solace SMF messaging interface host:port
`solace_vpn`	STRING	`"default"`	Valid VPN name	Message VPN name on the Solace broker
`canary_topic`	STRING	`"canary/health/probe"`	Valid topic path	Solace topic destination for synthetic canary publish
`queue_name`	STRING	`"canary.health.queue"`	Valid queue name	Dedicated queue subscribed to the canary topic
`queue_type`	SELECT	`"DURABLE_NON_EXCLUSIVE"`	Access types	Solace queue access type
`latency_sla_ms`	INT	`2000`	Positive integer	Max round-trip transit latency in ms
`alert_mode`	SELECT	`"ALERT"`	`"ALERT"`, `"AUDIT_ONLY"`	Operational alerting mode on breach

Outputs

{{ outputs.publish_canary_probe.messagesCount }}: Total number of synthetic messages published to the queue.
{{ outputs.consume_canary_probe.messagesCount }}: Total number of canary messages retrieved from the queue.
{{ outputs.classify_delivery_health.value }}: Health classification verdict (HEALTHY, SLA_BREACH, TIMEOUT_UNCONSUMED, PUBLISH_FAILED).

Quick start

Start a local Solace PubSub+ Standard broker container using Docker on host port 8180 and 55555 (avoiding port 8080 collision with Kestra UI):

docker run -d
--name kestra-solace
-p 8180:8080
-p 55555:55555
--shm-size=1g
--env username_admin_globalaccesslevel=admin
--env username_admin_password=admin
--ulimit nofile=2448:1048576
solace/solace-pubsub-standard:latest
Connect the broker container to your Kestra Docker network:

docker network connect YOUR_KESTRA_NETWORK kestra-solace

Verify your network name with docker network ls.
Provision the canary queue on the Solace broker using the Solace CLI or Web Admin interface:

docker exec -i kestra-solace /usr/sw/loads/currentload/bin/cli -A -s cliscripts/create_queue.cli << 'EOF' enable configure message-spool message-vpn default create queue canary.health.queue permission all consume subscription topic canary/health/probe no shutdown end EOF
Add the SOLACE_USERNAME, SOLACE_PASSWORD, and SLACK_WEBHOOK secrets to your Kestra environment.
Import this blueprint into your Kestra instance.
Run the flow manually with alert_mode: AUDIT_ONLY to verify connectivity and benchmark baseline transit latency.
Enable periodic_canary_schedule to begin automated 15-minute synthetic health sweeps.

How to extend

Add multi-region mesh probing by executing synthetic produce and consume cycles across distinct Solace broker nodes in separate availability zones.
Route degraded performance alerts to PagerDuty or Opsgenie using Kestra notification plugins based on consecutive failure counts.
Persist latency measurements into a time-series database such as InfluxDB or Prometheus using complementary Kestra database tasks.
Extend the probe payload to include binary or Avro-serialized data structures to validate multi-protocol gateway translation.

Pitfalls

Queue and topic subscription requirement: The target canary.health.queue must be pre-provisioned on the Solace broker with permission all consume, no shutdown, and an active topic subscription to canary/health/probe (e.g. subscription topic canary/health/probe). The Kestra Solace Produce task publishes to the topic destination, and the broker routes messages to the subscribed queue for the Consume task.
Web UI port collision: The Solace broker's default web admin management interface listens on port 8080, which conflicts with Kestra's web UI. Always map container port 8080 to host port 8180 (e.g. -p 8180:8080). The messaging interface uses port 55555.
Container network resolution: When Kestra and Solace run in separate Docker containers, localhost:55555 refers to the Kestra container. Use the container hostname kestra-solace:55555 on a shared Docker bridge network.
Inactive consumers and exclusive queues: When using DURABLE_EXCLUSIVE queue types, only one consumer can bind at a time. If an external client holds an active exclusive bind, the sentinel's Consume task will fail. Use DURABLE_NON_EXCLUSIVE for shared canary testing.