Skip to content

EventBridge Runtime

Jump to bottom
shenlin edited this page Jan 27, 2023 · 19 revisions

Status

  • Current State: Proposed
  • Authors: shenlin
  • Mailing List discussion: [email protected]
  • Pull Request: #PR_NUMBER
  • Released: <released_version>

Background & Motivation

What do we need to do

  • Will we add a new module? -- Yes.
  • Will we add new APIs? -- Yes.
  • Will we add new features? -- Yes.

Why should we do that

  • Are there any problems with our current project? There are several problems in current RocketMQ EventBridge:

    1. Too much dependence：When using EventBridge, additional deployment of rocketmq and rocketmq-connect is required;
    2. Without its own Runtime, EventBridge cannot provide better services for EDA scenarios, including:
    • Tracing：Comparing with the traditional synchronous call, in the EDA scenario, it is an essential function to support full link tracking query. In addition, Tracing needs to be well integrated with the open source ecosystem (like Zipkin, Skywalking, OpenTelemetry, etc.), which cannot be supported by the existing RocketMQ Connect.
    • Exception handling, retry, dead letter queue：Configure flexible exception handling, retry, dead letter queue.
    • Monitoring & Alarm：Whether there are rich metrics, whether to support monitoring and alarm.
    • Back pressure and flow control: Better protect the downstream.
    • User-friendly：A richer SDK and better integration with the Spring boot ecosystem allow users to use it at a lower cost;

  • What can we benefit from the proposed changes?

    1. Reduce dependence on external services.
    2. Provide better EDA integration experience.

Goals

  • What problem is this proposal designed to solve?

    1. 设计 Runtime的运行链路；
    2. 设计 Runtime的SPI；
    3. 设计 Runtime的管控；

Non-Goals

  • What problem is this proposal NOT designed to solve?

    1. Runtime的生命周期管理，包括如何创建/修改/删除；

  • Are there any limits to this proposal?

    Nothing specific.

Changes

Architecture

  • 事件网关，会接收用户发送给EventBridge的事件，并将事件投递给内置的存储，例如：RocketMQ.
  • 当用户配置Rule来订阅事件总线上的事件时，Push Runtime则会监听Rule关联的BUS，一旦BUS上收到事件，则将事件推送到Rule指定的Target；
  • 如果Rule上配置了过滤和转换，则在推送前，对事件进行前置处理；

image

Push Runtime 运行链路分析

  • 监听当前Runtime需要处理的EventRule,以及EventRule关联的BUS；
  • 通过long pooling监听MQ组件，
  • 当监听到新的事件产生，则立即将事件写入本地Buffer，监听组件继续监听后续的事件；（事件推送不阻塞监听线程）
  • Poll组件从Buffer中获取事件，并且根据上下文信息（Rule），对事件进行过滤和转换，然后发送给SinkConnector;
  • Put可以是异步操作，也可以是同步操作；Runtime每次提交位点之前，会调用PreCommit获取当前事件成功提交的offset，然后通过MQ监听组件，提交位点；

image

Push Runtime 如何监听EventBus

这里存在两种方式：

  • 方案1-独立模式：每一个Rule单独创建一个MQ的订阅，去监听Rule关联的BUS是否有事件产生；这种模式下，每一个Rule订阅都是一个独立的任务；

image

  • 方案2-共享模式：Push Worker中由一个统一的组件，去监听分配给当前Worker的所有Rule关联的BUS；如果监听的BUS有事件产生，则将事件分发给指定的Rule，进行过滤和转换，最终通过Sink Connector推送给Target端；

image

由于EventRule不同于Streaming的场景，大部分时候，Rule监听的BUS可能没有事件；为了降低监听MQ的成本，推荐方案2: Push Runtime中由一个统一的组件，去监听分配给当前Worker的所有Rule，共享资源。

Push Runtime Manager

问题分析

  • 如果每一个运行的Push Runtime需要加载所有的Rule，则当Rule越来越多时，Push runtime无法水平扩容，会成为瓶颈。 解决方案
  • Push Runtime Cluster只需要加载分配给他的Rule，按照Rule的定义，监听指定的BUS，一旦BUS上有事件产生，则将事件按照过滤规则进行过滤，并推送到目标端Target。 Push Runtime Manager 的职责
  • 所有的Push Runtime 生命周期由Push Runtime Manger统一管理；
  • Rule如何分配给Push Runtime，由Push Runtime Manger统一来管理；

image

Push Runtime Manager如何给Push Runtime分配Rule

  • 通过现有的event_target_runner表格，将EventRule分配给指定的Push Runtime运行

image

  • 增加event_push_worker表管理所有的worker节点

image

Push Manager 根据每个Worker的负载（如CPU、Memeory、Push延迟时间、线程池剩余大小）等Metrics信息给Rule-Target分配Worker，前期可以根据每个Worker分配的Rule- Target数量均匀分配；

Push Worker如何获取分配给他的Rule-Target

  • Worker可以提前创建好，然后配置到event_push_worker表中。如果底层基础设置支持自动创建容器的能力，则可以通过PushManager调用基础设施服务，动态创建Worker，并自动注册到event_push_worker表。
  • 所有创建好的Worker都有一个系统变量：worker-name，来指定当前worker的信息。Worker在启动的时候，可以根据worker-name来获取分配给他的Rule-Target。
  • 另外，运行过程中Rule-Target信息会发生变化，这个时候，PushManger需要将最新的变化的Rule-Target信息，通知给对应的Worker，Worker收到通知后，重新加载最新的Rule-Target。

Interface Design/Change

  • Method signature changes -- Add some async function in the Store module.
  • Method behavior changes -- Nothing specific.
  • CLI command changes -- Nothing specific.
  • Log format or content changes -- Nothing specific.

问题分析：

  • EventBridge是一个开放的架构，事件网关接收到的事件不一定会存储到RocketMQ，所以PushWorker在设计的时候，需要支持监听不同的MQ，因此这里我们需要定义一套标准的SPI，基于SPI实现不同的MQ监听组件；
  • PushWorker在BUS上监听到事件之后，会对事件进行过滤和转换，最终将事件推送到Rule中指定的目标端。推送之前，还会对事件进行过滤和转换，这些过程都需要定义好SPI，以便后期进行扩展，同时方便生态合作伙伴进行共建。

解决方案： 目前，OpenMessaging Connect API 已经定义了完整的SPI，并拥有一定的生态。所以，我们考虑复用OpenMessaging Connect API作为我们标准的SPI。

image

Compatibility, Deprecation, and Migration Plan

This proposal needs to modify the API specification of MessageStore to add asynchronous interfaces. This only involves changes inside the server side, without relying on a specific client version or changing the way that client is used. It is completely transparent to applications connected to RocketMQ.

Implementation Outline

We split this proposal into several tasks:

  • Task1: Push Runtime 实现；
  • Task2: Push Runtime Manger 实现；
  • Task3: Push Runtime 与 Openmessaging Connect 生态集成；

Rejected Alternatives

None.

How do alternatives solve the issue you proposed?

None.

Clone this wiki locally