EventMesh Workflow

Business Problem

Imaging you are building a simple Order Management System for an E-Commerce Store. The system should be able to receive and provision new orders from a store website. The provisioning process should be able to process all orders, handle payments, as well as process shipments.

For high availability and high performance, you architect the system using event-driven architecture (EDA), and build microservice apps to handle store frontend, order management, payment processing, and shipment management. You deploy the whole system in a cloud environment. To handle high workloads, you leverage a messaging system to buffer the loads, and scale up multiple instances of microservices. The architecture could look similar to:

Workflow Use Case

While each microservice is acting on its own event channels, EventMesh plays a crucial role of doing Event Orchestration.

We use CNCF Serverless Workflow to describe this Event Workflow Orchestration.

CNCF Serverless Workflow

CNCF Serverless Workflow defines a vendor-neutral, open-source, and fully community-driven ecosystem for defining and running DSL-based workflows that target the Serverless technology domain.

Serverless Workflow defines a Domain Specific Language (DSL) to describe stateful and stateless workflow-based orchestrations of serverless functions and microservices.

More details could be found in its official github site

EventMesh Workflow

We leverage Serverless Workflow DSL to describe the EventMesh workflow. Based on its spec, the workflow is consists of a series of workflow states used to describe the control-flow logic. At this time we only support event related workflow states. See the supported states in Workflow DSL Design.

A workflow state can include applicable actions, or services/functions that should be invoked during workflow execution. These actions can reference reusable function definitions which define how these functions/services should be invoked. They can also reference events that trigger event-based service invocations, and events to wait for that denote completion of such event-based service invocation completion.

In EDA solution, we usually defined our event-driven microservice using AsyncAPI. Serverless workflow function definitions support defining invocation semantics using AsyncAPI. See Using Funtions for AsyncAPI Service for more information.

AsyncAPI

AsyncAPI is an open source initiative that seeks to improve the current state of Event-Driven Architectures (EDA). Our long-term goal is to make working with EDAs as easy as it is to work with REST APIs. That goes from documentation to code generation, discovery to event management. Most of the processes you apply to your REST APIs nowadays would be applicable to your event-driven/asynchronous APIs too.

See AsyncAPI detail in the official site

Workflow Example

In this example, we build the event-driven workflow of the Order management system above.

First, we need to define AsyncAPI definitions for our microservice apps.

  • Online Store App
  1. asyncapi: 2.2.0
  2. info:
  3.   title: Online Store application
  4.   version: '0.1.0'
  5. channels:
  6.   store/order:
  7.     subscribe:
  8.       operationId: newStoreOrder
  9.       message:
  10.         $ref : '#/components/NewOrder'
  • Order Service
  1. asyncapi: 2.2.0
  2. info:
  3.   title: Order Service
  4.   version: '0.1.0'
  5. channels:
  6.   order/inbound:
  7.     publish:
  8.       operationId: sendOrder
  9.       message:
  10.         $ref : '#/components/Order'
  11.   order/outbound:
  12.     subscribe:
  13.       operationId: processedOrder
  14.       message:
  15.         $ref : '#/components/Order'
  • Payment Service
  1. asyncapi: 2.2.0
  2. info:
  3.   title: Payment Service
  4.   version: '0.1.0'
  5. channels:
  6.   payment/inbound:
  7.     publish:
  8.       operationId: sendPayment
  9.       message:
  10.         $ref : '#/components/OrderPayment'
  11.   payment/outbound:
  12.     subscribe:
  13.       operationId: paymentReceipt
  14.       message:
  15.         $ref : '#/components/OrderPayment'
  • Shipment Service
  1. asyncapi: 2.2.0
  2. info:
  3.   title: Shipment Service
  4.   version: '0.1.0'
  5. channels:
  6.   shipment/inbound:
  7.     publish:
  8.       operationId: sendShipment
  9.       message:
  10.         $ref : '#/components/OrderShipment'

Once that is defined, we define the order workflow that describes our Order Management business logic.

  1. id: storeorderworkflow
  2. version: '1.0'
  3. specVersion: '0.8'
  4. name: Store Order Management Workflow
  5. states:
  6.   - name: Receive New Order Event
  7.     type: event
  8.     onEvents:
  9.       - eventRefs:
  10.           - NewOrderEvent
  11.         actions:
  12.           - eventRef:
  13.               triggerEventRef: OrderServiceSendEvent
  14.               resultEventRef: OrderServiceResultEvent
  15.           - eventRef:
  16.               triggerEventRef: PaymentServiceSendEvent
  17.               resultEventRef: PaymentServiceResultEvent
  18.     transition: Check Payment Status
  19.   - name: Check Payment Status
  20.     type: switch
  21.     dataConditions:
  22.       - name: Payment Successfull
  23.         condition: "${ .payment.status == 'success' }"
  24.         transition: Send Order Shipment
  25.       - name: Payment Denied
  26.         condition: "${ .payment.status == 'denied' }"
  27.         end: true
  28.     defaultCondition:
  29.       end: true
  30.   - name: Send Order Shipment
  31.     type: operation
  32.     actions:
  33.       - eventRef:
  34.           triggerEventRef: ShipmentServiceSendEvent
  35.     end: true
  36. events:
  37.   - name: NewOrderEvent
  38.     source: file://onlineStoreApp.yaml#newStoreOrder
  39.     type: asyncapi
  40.     kind: consumed
  41.   - name: OrderServiceSendEvent
  42.     source: file://orderService.yaml#sendOrder
  43.     type: asyncapi
  44.     kind: produced
  45.   - name: OrderServiceResultEvent
  46.     source: file://orderService.yaml#processedOrder
  47.     type: asyncapi
  48.     kind: consumed
  49.   - name: PaymentServiceSendEvent
  50.     source: file://paymentService.yaml#sendPayment
  51.     type: asyncapi
  52.     kind: produced
  53.   - name: PaymentServiceResultEvent
  54.     source: file://paymentService.yaml#paymentReceipt
  55.     type: asyncapi
  56.     kind: consumed
  57.   - name: ShipmentServiceSendEvent
  58.     source: file://shipmentService.yaml#sendShipment
  59.     type: asyncapi
  60.     kind: produced

The corresponding workflow diagram is the following:

Workflow Diagram

EventMesh Workflow Engine

In the following architecture diagram, the EventMesh Catalog, EventMesh Workflow Engine and EventMesh Runtime are running in three different processors.

Workflow Architecture

The steps running the workflow is the followings:

  1. Deploy the Publisher and Subscriber Apps in the environment. Describe the App APIs using AsyncAPI, generate the asyncAPI yaml. Register the Publisher and Subscriber Apps in EventMesh Catalog using AsyncAPI.

  2. Register the Serverless Workflow DSL in EventMesh Workflow Engine.

  3. EventMesh Workflow Engine query the EventMesh Catalog for Publisher and Subscribers required in Workflow DSL function

  4. Event-driven Apps are publish events to EventMesh Runtime to trigger the Workflow. EventMesh Workflow Engine also publish and subscribe events for orchestrating the events.

EventMesh Catalog Design

EventMesh Catalog store the Publisher, Subscriber and Channel metadata. consists of the following modules:

  • AsyncAPI Parser

    Using the SDK provided by AsyncAPI community (see tool list), parse and validated the AsyncAPI yaml inputs, and generate the AsyncAPI definition.

  • Publisher, Channel, Subscriber Modules

    From the AsyncAPI definition store the Publisher, Subscriber and Channel information.

EventMesh Workflow Engine Design

EventMesh Workflow Engine consists of the following modules:

  • Workflow Parser

    Using the SDK provided by Serverless Workflow community (see supported SDKs), parse and validated the workflow DSL inputs, and generate workflow definition.

  • Workflow Module

    It manages a workflow instance life cycle, from create, start, stop to destroy.

  • State Module

    It manages workflow state life cycle. We support the event-related states, and the supported state list below is Work-in-Progress.

    Workflow StateDescription
    OperationExecute the AsyncAPI functions defined in the Actions
    EventCheck if the defined Event matched, if so execute the defined AsyncAPI functions
    SwitchCheck the event is matched with the event-conditions, and execute teh defined AsyncAPI functions
    ParallelExecute the defined AsyncAPI functions in parallel
    ForEachIterate the inputCollection and execute the defined AsyncAPI functions

  • Action Module

    It managed the functions inside the action.

  • Function Module

    It manages the AsyncAPI functions by creating the publisher and/or subscriber in EventMesh Runtime, and manage the publisher/subscriber life cycle.

    AsyncAPI OperationEventMesh Runtime
    PublishPublisher
    SubscribeSubscriber

  • Event Module

    It manages the CloudEvents data model, including event filter, correlation and transformation using the rules defined in the workflow DSL.

  • Retry Module

    It manages the retry logic of the event publishing into EventMesh Runtime.