2.微服务灰度发布落地实践(agent实现)

前言

据上一篇，设计方案的分析，综合考虑，最终决定,客户端采用agent方案实现，具本原因不再赘述，
感觉兴趣的小伙伴可以回头了解一下.该篇主要讲java agent的实现,灰度agent客户端的基础框架实现

java agent的介绍

java Agent 是一种允许你在 Java 应用程序启动时或运行时修改其字节码的技术。
它通过 JVM 提供的 java.lang.instrument 包来实现，可以用于各种用途，
如性能监控、代码增强、AOP（面向切面编程）、日志记录等。
Java Agent 的核心功能是能够在类加载到 JVM 之前对其进行修改，
而无需修改应用程序的源代码。启动时代理（Premain Agent）：在 JVM 启动时加载的代理。你需要通过 -javaagent 参数指定代理的 JAR 文件路径，并提供一个 premain 方法作为入口点。

public static void premain(String agentArgs, Instrumentation inst) {
    // 在这里注册 ClassFileTransformer 或执行其他初始化逻辑
}

基础实现

agent的实现的基础，是基于开源项目skywalking的插件核心实现，没必要重复造轮子,要做的是专注
去实现灰度相关的功能.

agent端 http服务实现

为了方便灰度管理端，给agent端实时推送数据，agent端，基于netty实现单简的http服务，； 会提供接口，供服务端对实例状态的控制，以及其它服务实例状态的推送;实例在收发消息或路由时，依赖这些信息作相应处理。
如果不太了解这块功能的作用，可以去第一篇看一下整个系统的架构图;下需要agent http server代码片段

SimpleChannelInboundHandler agentServerHttpInboundHandler = new AgentServerHttpInboundHandler();
agentChannelInitializer = new AgentServerChannelInitializer(agentServerHttpInboundHandler);
Thread serverThread = new Thread(new RunnableWithExceptionProtection(() -> {
    LOGGER.info("dbcat gray agent netty server.");
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    bossGroup = new NioEventLoopGroup(1, new DefaultNamedThreadFactory("server-boss"));
    workerGroup = new NioEventLoopGroup(4, new DefaultNamedThreadFactory("server-worker"));
    serverBootstrap.group(bossGroup, workerGroup)
           .channel(NioServerSocketChannel.class)
           .option(ChannelOption.SO_REUSEADDR, true)
           .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
           .childOption(ChannelOption.TCP_NODELAY, true)
           .childOption(ChannelOption.SO_KEEPALIVE, true)
           .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
           .childHandler(agentChannelInitializer);
    int port = Config.Agent.SERVER_PORT;
    try {
         ChannelFuture f = serverBootstrap.bind(port).sync();
         LOGGER.info("dbcat gray agent server started, port is {}.", port);
         f.channel().closeFuture().sync();
         LOGGER.info("dbcat gray agent server closed, port is {}.", port);
    } catch (InterruptedException e) {
         LOGGER.error("dbcat gray agent server start failed", e);
         Thread.currentThread().interrupt();
    }
}, t -> {
}), "gray-agent-server");
serverThread.setDaemon(true);
serverThread.start();

agent端api接口

上面实现agent端的http服务，灰度服务端如果需要控制agent客户端，必需还要提供api接口，才能进行交互
http 接口实现片段

   public class AgentServerHttpInboundHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
       private static final ILog log = LogManager.getLogger(AgentServerHttpInboundHandler.class);
       private static final Gson GSON = new GsonBuilder().disableHtmlEscaping().create();
       @Override
       public void channelReadComplete(ChannelHandlerContext ctx) {
           ctx.flush();
       }
       @Override
       protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
           //收到http请求
           QueryStringDecoder queryStringDecoder = new QueryStringDecoder(request.uri());
           String requestUri = queryStringDecoder.rawPath();
           String method = request.method().name();
           if (!"POST".equals(method.toUpperCase())) {
               complete(ctx, request,  RestResult.buildFailure(HttpResponseStatus.METHOD_NOT_ALLOWED.toString(),HttpResponseStatus.METHOD_NOT_ALLOWED.code()));
               return;
           }
           //通过url配置接口对应用实现
           Endpoint endpoint = EndpointManager.getEndpoint(requestUri);
           if (endpoint == null) {
               complete(ctx, request, RestResult.buildFailure(HttpResponseStatus.NOT_FOUND.toString(),HttpResponseStatus.NOT_FOUND.code()));
               return;
           }
           Throwable throwable = null;
           RestResult result = null;
           try {
               ByteBuf content = request.content();
               byte[] requestBodyBytes = new byte[content.readableBytes()];
               content.readBytes(requestBodyBytes);
               String requestBodyString = new String(requestBodyBytes, "UTF-8");
               ParameterizedType parameterizedType = ((ParameterizedType) endpoint.getClass().getGenericInterfaces()[0]);
               Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
               Object data =  GSON.fromJson(requestBodyString,actualTypeArguments[0]);
               //调用相应的接口
               result = endpoint.invoke(data);
           } catch (Throwable e) {
               log.error(String.format("接口响应异常%s", requestUri), e);
               throwable = e;
           } finally {
               if (throwable == null) {
                   complete(ctx, request, result);
               } else {
                   complete(ctx, request, RestResult.buildFailure(throwable.getMessage(),HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
               }
           }
       }
   }

Endpoint api 接口类，通过spi 方式加载实现

    public class EndpointManager {
        private static final Map<String, Endpoint> ENDPOINT_INVOKER_MAP = new HashMap<>();
        static {
            for (final Endpoint endpointInvoker : ServiceLoader.load(Endpoint.class, AgentClassLoader.getDefault())) {
                ENDPOINT_INVOKER_MAP.put(endpointInvoker.path(), endpointInvoker);
            }
        }
        public static Endpoint<?, ?> getEndpoint(String requestUri) {
            return ENDPOINT_INVOKER_MAP.get(requestUri);
        }
    }

Endpoint 的实某个实现，如更新实例状态接口

public class EnvStatusUpdateEndpoint implements Endpoint<InstanceEnvUpdateRequest, List<ExecuteResponse>> {
    private static final ILog log = LogManager.getLogger(EnvStatusUpdateEndpoint.class);
    @Override
    public String path() {
        return "/env-status/update";
    }

    @Override
    public RestResult<List<ExecuteResponse>> invoke(InstanceEnvUpdateRequest request) throws Exception {
        log.info("更新实例信息");
        ServerInstance.getInstance().setEnvStatus(request.getEnvStatus());
        List<ExecuteResponse> executeResponses = MQConnectionManager.restartAll(request.getStrategy());
        return RestResult.buildSuccess(executeResponses);
    }
}

到此，agent客户端的基础框加已实现，下一篇将开始介绍相关核心组件灰度的增强实现

最后，给大家安利一款mysql监控软件: 安装方便，消耗低，可视化，傻瓜式操作，可以监控慢日志详情、cpu、内存、连接数、tps 等信息
体验演示
下载地址