本文作者

本文作者：hutaishi， 快手软件工程师，Apache ShenYu Contributor

可观测性介绍

可观测性能力的应用离不开数据与信息，而日志（logs）、指标（metrics）与链路（trace）是最重要的数据信息源。
Apache ShenYu 利用 java agent 和 字节码增强 技术实现了无侵入的可观性，使得用户无需引入依赖即可接入第三方可观测性系统，
获取 Traces、Metrics 和 Logging。 
本文基于2.4.3-SNAPSHOT版本分析Apache ShenYu集成RocketMQ实现日志的可观测性。
分析Agent技术架构、Agent实现细节、响应式系统中日志采集、RocketMQ的集成

采集可观测性日志

从数据的采集到可视化的整体流程图如下：

 

Java Agent和ByteBuddy技术

ShenYu-Agent的可观测性实现模块尽管是放在ShenYu项目下，但实际上他和网关是独立的，对网关的监控是无代码侵入的。 

这种无代码侵入的自动埋点基于Java Agent技术。Java Agent是JDK1.5之后引入的新特性,该特性可以在JVM将字节码文件读入内存之后，使用对应的字节流在Java堆中生成一个Class对象之前，对其字节码进行修改(增强)的能力，而JVM也会使用用户修改过的字节码进行Class对象的创建。 

Shenyu-Agent利用JavaAgent技术拦截构造函数、实例方法、静态方法并进行AOP修改，使用ByteBuddy做字节码增强。

当执行到某个被ShenYu-agent代理过的方法时， 会执行自定义的增强逻辑来实现指标、日志、链路的收集、传递、上报，实现网关的可观测性。

Java Agent中的核心类

Instrumentation

public interface Instrumentation {



    //增加一个Class 文件的转换器，转换器用于改变 Class 二进制流的数据，// 参数 canRetransform 设置是否允许重新转换。    void addTransformer(ClassFileTransformer transformer, boolean canRetransform);



    void addTransformer(ClassFileTransformer transformer);



    //删除一个类转换器    boolean removeTransformer(ClassFileTransformer transformer);

    // 是否支持重新转换类    boolean isRetransformClassesSupported();



    //在类加载之后，重新定义 Class。这个很重要，该方法是1.6 之后加入的，事实上，该方法是 update 了一个类。    void retransformClasses(Class<?>... classes) throws UnmodifiableClassException;



    // 是否支持重新定义一个类    boolean isRedefineClassesSupported();



    // 重新定义一个类    void redefineClasses(ClassDefinition... definitions)throws  ClassNotFoundException, UnmodifiableClassException;



    boolean isModifiableClass(Class<?> theClass);



    @SuppressWarnings("rawtypes")

    Class[] getAllLoadedClasses();



    @SuppressWarnings("rawtypes")

    Class[] getInitiatedClasses(ClassLoader loader);

    //获取一个对象的大小long getObjectSize(Object objectToSize);    void appendToBootstrapClassLoaderSearch(JarFile jarfile);



    void appendToSystemClassLoaderSearch(JarFile jarfile);



    boolean isNativeMethodPrefixSupported();



    void setNativeMethodPrefix(ClassFileTransformer transformer, String prefix);

}

在类加载之前，重新定义 Class 文件，ClassDefinition 表示对一个类新的定义，如果在类加载之后，需要使用 retransformClasses 方法重新定义。 

addTransformer方法配置之后，后续的类加载都会被Transformer拦截。对于已经加载过的类，可以执行retransformClasses来重新触发,这个Transformer的拦截。类加载的字节码被修改后，除非再次被retransform，否则不会恢复。

ClassFileTransformer

这个接口的主要作用就是字节码转换，修改载入JVM的字节码数据，从而增加我们期望的功能逻辑。

public interface ClassFileTransformer {



    byte[]    transform(  ClassLoader         loader,

                String              className,

                Class<?>            classBeingRedefined,

                ProtectionDomain    protectionDomain,

                byte[]              classfileBuffer)

        throws IllegalClassFormatException;

}

ClassLoader：当前载入Class的ClassLoader。如果ClassLoader是bootstrap Loader，为null。

className：当前载入Class的类名。Java虚拟机规范中定义的完全限定类和接口名称的内部形式的类的名称，例如java/util/List。

byte[] classfileBuffer：当前类的以byte数组呈现的字节码数据。（可能跟class文件的数据不一致，因为此处的byte数据是此类最新的字节码数据，
即此数据可能是原始字节码数据被其他增强方法增强之后的字节码数据）

JavaAgent内部字节码修改流程图如下：

通过上面的分析

 

我们会发现ClassFileTransformer这个接口的实现很偏底层，我们可以利用ByteBuddy字节码增强框架来高效实现ClassFileTransformer接口。
调用java agent中的premain方法的Instrumentation参数的addTransformer方法，添加我们的ClassFileTransformer实现。

 

    public static void premain(String agentArgs, Instrumentation inst) {

        ClassFileTransformer myClassFileTransformer;  // 实现ClassFileTransformer接口        inst.addTransformer(myClassFileTransformer);

    }

ByteBuddy操作字节码的神器

通过前面的JavaAgent介绍，我们知道JavaAgent的关键是修改字节码，字节码编程框架或库有asm,javassist,byte-buddy。 

Byte Buddy 玩法上更加高级，你可以完全不需要了解一个类和方法块是如何通过 指令码 LDC、LOAD、STORE、IRETURN… 生成出来的。 

就像它的官网介绍；Byte Buddy 是一个代码生成和操作库，用于在 Java 应用程序运行时创建和修改 Java 类，而无需编译器的帮助。

除了 Java 类库附带的代码生成实用程序外，Byte Buddy 还允许创建任意类，并且不限于实现用于创建运行时代理的接口。 

此外，Byte Buddy 提供了一种方便的 API，可以使用 Java 代理或在构建过程中手动更改类。

下面是一个使用ByteBuddy的样例:

public class MyAgent {

    public static void premain(String agentArgs, Instrumentation inst) {

        System.out.println("this is my agent：" + agentArgs);

        // 拦截哪个方法，以及拦截逻辑        AgentBuilder.Transformer transformer = (builder, typeDescription, classLoader, javaModule) -> {

            return builder

                    .method(ElementMatchers.any()) // 拦截任意方法                    .intercept(MethodDelegation.to(MethodCostTime.class)); // 委托到MethodCostTime        };

        // 修改字节码的监听器，一般调式用到。这里由于空间，加个注释。        // AgentBuilder.Listener listener = new AgentBuilder.Listener() {}        // 通过byte-buddy构造ClassFileTransformer并添加到Instrumentation inst        new AgentBuilder

                .Default()

                .type(ElementMatchers.nameStartsWith("org.demo.test")) // 指定需要拦截的类                .transform(transformer)

                .with(listener)

                .installOn(inst); // 将ClassFileTransformer安装到Instrumentation        }

    }

}

2. ShenYuAgent插件化技术架构

ShenYu网关采用插件化设计思想，插件热插拔，方便扩展，而在ShenYu Agent的设计中同样采用了插件化设计思想，支持热插拔，非常容易扩展。 

分析采集可观测性数据metrics、tracing、logging，我们一般只需要在方法执行前、执行后以及执行异常这3个地方埋点，  

就可以说实现指标、日志、链路的收集、传递、上报。 

ShenYu在字节码增强方面提供了一个统一的AOP，AOP写好增强的各种细节以及对配置的解析和插件的管理。

上层插件只需要写增强的代码逻辑即可，类似spring中的AOP.

ByteBuddy实现AOP

ByteBuddy实现AOP，涉及4个关键点。

• 确定拦截的类的方法。根据配置的拦截的类和方法构建ByteBuddy的类型匹配器、方法匹配器

• 构建一个带有before、after、onThrowing方法的接口处理器，插件层实现这个接口来实现相应的业务逻辑

• 对每个被拦截的类的方法构建一个@RuntimeType的方法，这个方法用于实现拦截逻辑。这个逻辑里面会分别遍历第二步的插件处理器接口

• 定义一个获得和设置上下文的接口，让所有被拦截的类实现这个接口，实现了这个接口的被拦截的类会作为参数传入前面的处理方法中。这个主要用于链路追踪

下面分别分析ShenYu是如何实现上面4个关键点的。

确定拦截的类的方法。根据配置的拦截的类和方法构建ByteBuddy的类型匹配器、方法匹配器

ShenYu Agent采取配置化的方式，可以配置拦截的类和方法。如下是一个采集日志需要配置的拦截的类和方法。
logging-point.yaml:

pointCuts:

  - targetClass: org.springframework.web.server.adapter.HttpWebHandlerAdapter

    points:

      - type: instanceMethod

        name: createExchange

    handlers:

      rocketmq:

        - org.apache.shenyu.agent.plugin.logging.common.handler.DefaultLoggingPluginHandler

ShenYu Agent定义了一个获取节点集合的接口AgentPluginDefinition，采用ShenYu SPI加载各个插件的切点集合。

@SPIpublic interface AgentPluginDefinition {



    /**

     * Collector shenyu agent join point.

     *

     * @return the collection

     */    Collection<ShenyuAgentJoinPoint> collector();

}

一般这个接口的实现是利用ShenYu Agent提供的yaml工具类读取相关yaml配置，ShenYuAgent根据拦截的方法构建ElementMatcher matcher。

用于确定ByteBuddy需要拦截哪些方法。最终的构建结果便是：

public final class ShenyuAgentJoinPoint {



    private final String classTarget; // 拦截的目标类    private final List<ConstructorPointCut> constructorPoints;  // 拦截的目标类下的构造方法    private final List<InstanceMethodPointCut> instanceMethodPoints; // 拦截的目标类下的实例方法    private final List<StaticMethodPointCut> staticMethodPoints; // 拦截的目标类下的静态方法}

ShenyuAgentTypeMatcher继承ElementMatcher.Junction.AbstractBase用于确定拦截哪些类。

构建一个带有before、after、onThrowing方法的接口处理器，插件层实现这个接口来实现相应的业务逻辑

利用ByteBuddy的@RuntimeType注解可以实现拦截器逻辑。当执行到被拦截的方法时，就会执行@RuntimeType注解的方法，在这个方法里面可以实现我们的逻辑。
定义的拦截方法如下：

    @RuntimeType    public Object intercept(@This final Object target, @Origin final Method method, 

    @AllArguments final Object[] args, @SuperCall final Callable<?> callable) throws Exception;

注解含义如下：
@This    当前被拦截的、动态生成的那个对象  
@Origin  可以绑定到以下类型的参数：Method 被调用的原始方法， Constructor 被调用的原始构造器， Class 当前动态创建的类等  
@AllArguments    绑定所有参数的数组  
@SuperCall   用于调用父类版本的方法

实现的逻辑便是执行被拦截的方法前、方法后、出现异常的情况下，分别遍历每个插件处理器hander执行对应的before、after、onThrowing方法。

    @RuntimeType    public Object intercept(@This final Object target, @Origin final Method method, @AllArguments final Object[] args,

        @SuperCall final Callable<?> callable) throws Exception {



        TargetObject instance = (TargetObject) target;

        MethodResult methodResult = new MethodResult();

        for (InstanceMethodHandler handler : handlerList) {

            // handler before logic 执行before逻辑        }

        Object result;

        try {

            result = callable.call();

        } catch (final Throwable ex) {

            for (InstanceMethodHandler handler : handlerList) {

                // handler onThrowing logic 执行onThrowing逻辑            }

            throw ex;

        } finally {

            for (InstanceMethodHandler handler : handlerList) {

                // handler after logic 执行after逻辑            }

        }

        return result;

    }

定义一个获得和设置上下文的接口，让所有被拦截的类实现这个接口

上下文接口定义:

public interface TargetObject {

    Object getContext();



    void setContext(Object context);

}

ByteBuddy让每个被拦截的类实现这个方法。关键代码如下：

builder.defineField(EXTRA_DATA, Object.class, Opcodes.ACC_PRIVATE | Opcodes.ACC_VOLATILE)  // 定义一个上下文字段.implement(TargetObject.class)  // 实现上下文接口.intercept(FieldAccessor.ofField(EXTRA_DATA)); // 实现逻辑为这个上下文字段的get,set。

插件生命周期管理

ShenYu定义了插件启动服务，包含start和close方法。在premain方法中利用SPI技术启动每个插件，同时会注册一个关闭所有插件的钩子ShutdownHook。

插件启动流程图如下： 

 

 

2.1 自定义类加载器ShenyuAgentPluginLoader加载插件

由于插件会被打成一个个的独立的jar包，这些jar包会被统一放入到shenyu-agent-dist下，要访问这些jar中的插件，需要自定义一个类加载器
来加载。ShenyuAgentPluginLoader继承ClassLoader，采取双亲委派机制来加载类。ShenyuAgentPluginLoader在约定的插件目录下扫描所有的jar包，
在这些jar包中找class和Resources.

.

├── conf

│   ├── logback.xml

│   ├── logging-meta.yaml

│   ├── logging-point.yaml

│   ├── metrics-meta.yaml

│   ├── metrics-point.yaml

│   ├── shenyu-agent.yaml

│   └── tracing-point.yaml

├── plugins

│   ├── shenyu-agent-plugin-logging-api-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-logging-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-logging-rocketmq-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-api-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-prometheus-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-jaeger-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-opentelemetry-2.4.3-SNAPSHOT.jar

│   └── shenyu-agent-plugin-tracing-zipkin-2.4.3-SNAPSHOT.jar

└── shenyu-agent.jar

加载这些jar中的InstanceMethodHandler便是通过自定义类加载器实现的。

inst = Class.forName(className, true, ShenyuAgentPluginLoader.getInstance()).newInstance();

2.2插件需要实现的接口

要完成一个插件，只需实现3个接口。shenyu-agent基于SPI技术加载你的插件。

• AgentPluginDefinition：定义增强(拦截)哪个类的哪些方法以及对应的处理器

• AgentPluginBootService：用于启动插件

• MethodHandler：拦截的方法执行的增强逻辑(涉及拦截构造函数、实例方法、静态方法)
  如下图所示

   

  

 

2.2 ShenYuAgent配置文件

ShenYuAgent的可观测性是可配置化的，ShenYu采用maven-assembly-plugin进行打包.  
配置文件和打包输出在shenyu-dist/shenyu-agent-dist/target/shenyu-agent,如下图所示：

.

├── conf

│   ├── logback.xml

│   ├── logging-meta.yaml

│   ├── logging-point.yaml

│   ├── metrics-meta.yaml

│   ├── metrics-point.yaml

│   ├── shenyu-agent.yaml

│   └── tracing-point.yaml

├── plugins

│   ├── shenyu-agent-plugin-logging-api-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-logging-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-logging-rocketmq-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-api-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-metrics-prometheus-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-common-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-jaeger-2.4.3-SNAPSHOT.jar

│   ├── shenyu-agent-plugin-tracing-opentelemetry-2.4.3-SNAPSHOT.jar

│   └── shenyu-agent-plugin-tracing-zipkin-2.4.3-SNAPSHOT.jar

└── shenyu-agent.jar

其中shenyu-agent.yaml用于配置各个插件的属性，-point.yaml之类的文件用于配置拦截点(包含类名、方法类型、方法名)和对应的处理器

3.响应式网关请求日志采集

收集的日志字段

字段名称	含义	说明	备注
clientIp	客户端IP
timeLocal	请求时间字符串,  格式：yyyy-MM-dd HH:mm:ss.SSS
method	请求方法(不同rpc类型不一样，http类的为:get,post等待，rpc类的为接口名称)
requestHeader	请求头(json格式)
responseHeader	响应头(json格式)
queryParams	请求查询参数
requestBody	请求Body（二进制类型的body不会采集）
requestUri	请求uri
responseBody	响应body
responseContentLength	响应body大小
rpcType	rpc类型
status	响应码
upstreamIp	上游(提供服务的程序)IP
upstreamResponseTime	上游(提供服务的程序)响应请求的耗时(毫秒ms)
userAgent	请求的用户代理
host	请求的host
module	请求的模块
path	请求的路径path
traceId	请求的链路追踪ID	需要开启链路追踪插件

样例：

{

  "clientIp":"0:0:0:0:0:0:0:1%0",

  "timeLocal":"2022-03-09 21:39:04.373",

  "method":"POST",

  "requestHeader":"[{\"myvar\":[\"A\"]},{\"Content-Type\":[\"application/json\"]},{\"User-Agent\":[\"PostmanRuntime/7.29.0\"]},{\"Accept\":[\"*/*\"]},{\"Cache-Control\":[\"no-cache\"]},{\"Postman-Token\":[\"31a566b2-9682-47ca-b7f2-131fc6ad714b\"]},{\"Host\":[\"localhost:9195\"]},{\"Accept-Encoding\":[\"gzip, deflate, br\"]},{\"Connection\":[\"keep-alive\"]},{\"Content-Length\":[\"43\"]}]",

  "responseHeader":"[{\"transfer-encoding\":[\"chunked\"]},{\"Content-Length\":[\"44\"]},{\"Content-Type\":[\"application/json\"]}]",

  "queryParams":"a=1",

  "requestBody":"{\n    \"id\":666,\n    \"name\": \"hello world\"\n}",

  "requestUri":"http://localhost:9195/http/order/save?a=1",

  "responseBody":"{\"id\":\"666\",\"name\":\"hello world save order\"}",

  "responseContentLength":"44",

  "rpcType":"http",

  "status":200,

  "upstreamIp":"172.17.105.222",

  "upstreamResponseTime":19,

  "userAgent":"PostmanRuntime/7.29.0",

  "host":"localhost:9195",

  "module":null,

  "traceId":"097b826595c34f51",

  "path":"/http/order/save"}

日志收集对应的配置

日志元数据配置在logging-meta.yaml文件中。 

有时候我们可能不需要采集某些字段，特别是大body之类的，或者某些字段涉及敏感信息，那么可以关闭这些字段的采集。logFieldSwitchConfig便是控制字段是否采集。

有时候有些接口可能是不需要采集的，或者只需要采集小部分，可以针对接口或应用配置采样率，采样率配置0即为关闭采样。

同时支持接口或应用级别配置不同的topic。logApiSwitchConfigMap字段进行接口或应用级别配置。

3.1如何正确读取请求body和响应body

ShenYu是基于WebFlux实现的响应式网关，要记录请求的body和响应的body并不能简单的读取流，因为读取一下就会改变字节流的可读取位置，  
会导致下一个订阅者无法读到body。

所以需要设计一个无副作用的方案来读取body。 

我们需要在读取Flux(针对请求body)或写入Flux(针对响应body)时进行body记录，在每个DataBuffer被读或写的时候，  

读取一份只读的字节buffer写入到我们的内存字节流中，调用dataBuffer.asByteBuffer().asReadOnlyBuffer()方法读取，这个方法无副作用。

3.1.1BodyWriter的设计

dataBuffer.asByteBuffer().asReadOnlyBuffer()会返回ByteBuffer，我们需要将这个写入内存，并且需要保证body采集完之后，立即释放内存，可观测性需要尽可能的降低资源的消耗。

BodyWriter使用WritableByteChannel作为字节流容器，当读取完这个字节通道中的字节后就会关闭通道，释放资源。

3.1.2LoggingServerHttpRequest的设计

SpringWebFlux提供了ServerHttpRequestDecorator类用于ServerHttpRequest将原本的ServerHttpRequest作为委托类。

重新了getBody()方法，在super.getBody()被订阅的时候，在doOnNext方法中调用dataBuffer.asByteBuffer().asReadOnlyBuffer()，

将得到的ByteBuffer写入BodyWriter中的WritableByteChannel中。

public class LoggingServerHttpRequest extends ServerHttpRequestDecorator {

    private final ShenyuRequestLog logInfo;



    public LoggingServerHttpRequest(final ServerHttpRequest delegate, final ShenyuRequestLog logInfo) {

        super(delegate);

        this.logInfo = logInfo;

    }



    /**

     * get request body.

     *

     * @return Flux

     */    @Override    @NonNull    public Flux<DataBuffer> getBody() {

        BodyWriter writer = new BodyWriter();

        // 在super.getBody()被订阅的时候，在doOnNext方法中调用dataBuffer.asByteBuffer().asReadOnlyBuffer()        // 将得到的ByteBuffer写入BodyWriter中的WritableByteChannel中        return super.getBody().doOnNext(dataBuffer -> writer.write(dataBuffer.asByteBuffer().asReadOnlyBuffer()))

                .doFinally(signal -> {

                    int size = writer.size();

                    String body = writer.output();

                    if (size > 0 && LogCollectUtils.isNotBinaryType(getHeaders())

                            && !LogCollectConfigUtils.isRequestBodyTooLarge(size)) {

                        logInfo.setRequestBody(body);

                    }

                });

    }

}

3.1.2LoggingServerHttpResponse的设计

同读取ServerHttpRequest中的body一样，response中的body读取一样设计成在写入DataBuffer的时候，顺便读取一下，不产生副作用。

继承ServerHttpResponseDecorator类，将原本的ServerHttpResponse委托给父类。重写writeWith方法用于读取响应body。

body被订阅的时候，在doOnNext方法中调用dataBuffer.asByteBuffer().asReadOnlyBuffer()  ，将得到的ByteBuffer写入BodyWriter中的WritableByteChannel中，最终在Flux中的doFinally中收集完日志。

当我们读取完response的body，即可将收集到访问信息收集起来。这个collect方法会在LoggingServerHttpResponse中的logResponse方法中调用。

/**

 * Collect logs and save them.

 */public interface LogCollector extends AutoCloseable {



    /**

     * collect log.

     *

     * @param log access log

     */    void collect(Object log);

}

3.2如何埋点

ShenYu的插件设计是可以插拔的，而全局插件GlobalPlugin是必定存在的并且位于插件链的第一位置，最先被执行。

一般我们可以拦截全局插件GlobalPlugin的execute方法，在基于链式执行中，我们一般会修改ServerWebExchange的request和response。

然后将ServerWebExchange作为参数传给插件链ShenyuPluginChain。核心代码为：

    return chain.execute(

            exchange.mutate().request(new LoggingServerHttpRequest(request, requestInfo))

            .response(new LoggingServerHttpResponse(exchange.getResponse(), requestInfo)

            ).build()

    );

有2种方式实现埋点：

• 1.拦截网关全局插件GlobalPlugin#execute，修改第一个参数ServerWebExchange，这种方式需要配合ByteBuddy的注解@Morph来实现

• 2.拦截创建ServerWebExchange的方法做后置处理，也就是拦截HttpWebHandlerAdapter#createExchange方法。 

ShenYu Agent Logging采用的第二种方式埋点。
会在response的body被订阅的时候，完成一个完整请求日志的收集。

4.异步采集日志并采用Oneway方式推送到RocketMQ集群

当读取完response的body后，就会将日志收集。由于网关的请求量特别大，对性能要求很高，所以收集的操作不应该阻塞主流程。

同时也要求能快速消费日志。设计上引入了一个JVM缓冲队列(大小可配置)，请求日志会先放入缓冲队列，而不是直接发送到消息队列集群，  如果缓冲队列没有足够空间，则会直接丢弃日志(当网关处理请求速度超过异步消费请求日志的速度的极端情况会出现)，不阻塞。 

同时会启动一个消费线程，循环从缓存队列里面批量获取日志，解析出日志对应的topic，同时将日志序列化(采取JSON序列化)，这里考虑到对极端情况下日志丢失可以容忍，采取了效率最高的Oneway方式推送消息到RocketMQ集群，这种方式性能最高，消耗资源最低，但丢数据的可能性  相比同步和异步相对高一点。如果生产上是追求消息零丢失的，则不建议采用这种方式。

采集日志设计如下： 

 

日志收集时序图如下： 

 

日志可视化

日志写入到rocketmq后，需要可视化展示出来实现日志的可观测性。日志数据可以存储到clickhouse,ES,loki等等，可视化可以采用开源的grafana loki或kibana.
一个可供参考的grafana loki可视化界面：

 

参与Apache ShenYu

Apache ShenYu (incubating) 于2021年5月进入Apache基金会孵化器。它是一款多协议，高可用，易扩展的云原生API网关。
官网地址：https://shenyu.apache.org
Github地址：https://github.com/apache/incubator-shenyu
Gitee地址：https://gitee.com/Apache-ShenYu/incubator-shenyu

• 发送订阅邮件。用自己的邮箱向dev-subscribe@shenyu.apache.org发送一封邮件，主题和内容任意。

• 接收确认邮件并回复。完成步骤1后，您将收到一封来自dev-help@shenyu.apache.org的确认邮件（如未收到，请确认该邮件是否已被拦截，或已经被自动归入订阅邮件、垃圾邮件、推广邮件等文件夹）。直接回复该邮件，或点击邮件里的链接快捷回复即可，主题和内容任意。

• 接收欢迎邮件。完成以上步骤后，您会收到一封主题为WELCOME to dev@shenyu.apache.org的欢迎邮件，至此您已成功订阅Apache ShenYu的邮件列表。