初探灰度发布系列--AB Test以及栗子

Posted 2022-04-22 go大鸡腿

文章目录

• 灰度发布
• AB Test实践
• • 逻辑架构图
  • • 这个是外部访问内部，内部怎么访问外部？
• 实现demo
• • 解析
  • 关键修改点
• 实践
• 优化点
• ReactiveLoadBalancerClientFilter源码

灰度发布

灰度发布分几种：蓝绿、ABTest以及金丝雀
灰度作用：为了减少灰度版本对生产环境的影响

灰度策略	内容	缺点
蓝绿	会有两套环境，灰度发布之后，流量从绿切到蓝，然后进行验证	环境比较浪费，而且生产直接路由到灰度版本
ABTest	在网关层对不同用户进行路由，只有特定用户切流量到灰度版本，这样的话不会影响生产用户	相对流量权重比较复杂，需要对用户进行区分对待
金丝雀	将部分流量切换过去，然后进行验证灰度的准确性，如果没有问题则流量全部切换	还是那个生产用户直接测试灰度版本

AB Test实践

逻辑架构图

这是第一版本，我们实现一个自定义loadBalance，filter拿到对应header头，或者说被动打标，通过不同的域名，来进行路由
流量路由的时候，基于nacos注册服务里面metaData标识，来决定路由到哪台服务
灰度测试完之后滚动更新生产pod

这个是外部访问内部，内部怎么访问外部？

fegin或者http请求，灰度版本服务通过域名来访问
dubbo请求，本身也有负载均衡器，需要拿到对应的标识，比如说版本号来负载

进阶第二版
在第一版我们自定义了loadBalance，以及路由标识，比如说网关配置lb:xxx，为啥是lb开头，大家可以看下ReactiveLoadBalancerClientFilter源码，我们其实是需要定义另一种标识
那么问题来了，当灰度测试完之后，负载怎么换成正常的lb？

这个涉及到网关route动态配置，publushEvent即可（这个不在本章介绍）

实现demo

参照另一篇文章

• gateway灰度发布

GrayGatewayReactiveLoadBalancerClientAutoConfiguration

import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author javachen
 * @description GrayGatewayReactiveLoadBalancerClientAutoConfiguration
 */
@Configuration
public class GrayGatewayReactiveLoadBalancerClientAutoConfiguration 

    public GrayGatewayReactiveLoadBalancerClientAutoConfiguration() 
    

    @Bean
    @ConditionalOnMissingBean(GrayReactiveLoadBalancerClientFilter.class)
    public GrayReactiveLoadBalancerClientFilter grayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) 
        return new GrayReactiveLoadBalancerClientFilter(clientFactory, properties);
    

GrayLoadBalancer

import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.loadbalancer.core.NoopServiceInstanceListSupplier;
import org.springframework.cloud.loadbalancer.core.ReactorServiceInstanceLoadBalancer;
import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier;
import org.springframework.http.HttpHeaders;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.*;

/**
 * @author javachen
 * @description GrayReactiveLoadBalancerClientFilter
 */
public class GrayLoadBalancer implements ReactorServiceInstanceLoadBalancer 

    private static final Log log = LogFactory.getLog(GrayLoadBalancer.class);
    private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
    private  String serviceId;

    public GrayLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId) 
        this.serviceId = serviceId;
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
    

    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) 
        HttpHeaders headers = (HttpHeaders) request.getContext();
        if (this.serviceInstanceListSupplierProvider != null) 
            ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
            return ((Flux)supplier.get()).next().map(list->getInstanceResponse((List<ServiceInstance>)list,headers));
        

        return null;
    

    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances, HttpHeaders headers) 
        if (instances.isEmpty()) 
            return getServiceInstanceEmptyResponse();
         else 
            return getServiceInstanceResponseWithWeight(instances);
        
    

    /**
     * 根据版本进行分发
     * @param instances
     * @param headers
     * @return
     */
    private Response<ServiceInstance> getServiceInstanceResponseByVersion(List<ServiceInstance> instances, HttpHeaders headers) 
        String versionNo = headers.getFirst("version");
        //System.out.println(versionNo);
        Map<String,String> versionMap = new HashMap<>();
        versionMap.put("version",versionNo);
        final Set<Map.Entry<String,String>> attributes =
                Collections.unmodifiableSet(versionMap.entrySet());
        ServiceInstance serviceInstance = null;
        for (ServiceInstance instance : instances) 
            Map<String,String> metadata = instance.getMetadata();
            if(metadata.entrySet().containsAll(attributes))
                serviceInstance = instance;
                break;
            
        

        if(ObjectUtils.isEmpty(serviceInstance))
            return getServiceInstanceEmptyResponse();
        
        return new DefaultResponse(serviceInstance);
    

    /**
     *
     * 根据在nacos中配置的权重值，进行分发
     * @param instances
     *
     * @return
     */
    private Response<ServiceInstance> getServiceInstanceResponseWithWeight(List<ServiceInstance> instances) 
        Map<ServiceInstance,Integer> weightMap = new HashMap<>();
        for (ServiceInstance instance : instances) 
            Map<String,String> metadata = instance.getMetadata();
            //System.out.println(metadata.get("version")+"-->weight:"+metadata.get("weight"));
            if(metadata.containsKey("gray"))
                weightMap.put(instance,1000);
            else 
                weightMap.put(instance,1);
            
        
        WeightMeta<ServiceInstance> weightMeta = WeightRandomUtils.buildWeightMeta(weightMap);
        if(ObjectUtils.isEmpty(weightMeta))
            return getServiceInstanceEmptyResponse();
        
        ServiceInstance serviceInstance = weightMeta.random();
        if(ObjectUtils.isEmpty(serviceInstance))
            return getServiceInstanceEmptyResponse();
        
        return new DefaultResponse(serviceInstance);
    

    private Response<ServiceInstance> getServiceInstanceEmptyResponse() 
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    

GrayReactiveLoadBalancerClientFilter

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.*;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.filter.ReactiveLoadBalancerClientFilter;
import org.springframework.cloud.gateway.support.DelegatingServiceInstance;
import org.springframework.cloud.gateway.support.NotFoundException;
import org.springframework.cloud.gateway.support.ServerWebExchangeUtils;
import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier;
import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
import org.springframework.core.Ordered;
import org.springframework.http.HttpHeaders;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import javax.servlet.http.HttpServletRequest;
import java.net.URI;

/**
 * @author javachen
 * @description GrayReactiveLoadBalancerClientFilter
 */
public class GrayReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered 

    private static final Log log = LogFactory.getLog(ReactiveLoadBalancerClientFilter.class);
    private static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10150;
    private final LoadBalancerClientFactory clientFactory;
    private LoadBalancerProperties properties;

    public GrayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) 
        this.clientFactory = clientFactory;
        this.properties = properties;
    

    @Override
    public int getOrder() 
        return LOAD_BALANCER_CLIENT_FILTER_ORDER;
    

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) 

        URI url = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
        String schemePrefix = (String)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR);
        if (url != null && ("grayLb".equals(url.getScheme()) || "grayLb".equals(schemePrefix))) 
            ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
            if (log.isTraceEnabled()) 
                log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url);
            

            return this.choose(exchange).doOnNext((response) -> 
                if (!response.hasServer()) 
                    throw NotFoundException.create(true, "Unable to find instance for " + url.getHost());
                 else 
                    URI uri = exchange.getRequest().getURI();
                    String overrideScheme = null;
                    if (schemePrefix != null) 
                        overrideScheme = url.getScheme();
                    

                    DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance((ServiceInstance)response.getServer(), overrideScheme);
                    URI requestUrl = this.reconstructURI(serviceInstance, uri);
                    if (log.isTraceEnabled()) 
                        log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
                    

                    exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl);
                
            ).then(chain.filter(exchange));
         else 
            return chain.filter(exchange);
        
    

    protected URI reconstructURI(ServiceInstance serviceInstance, URI original) 
        return LoadBalancerUriTools.reconstructURI(serviceInstance, original);
    

    private Mono<Response<ServiceInstance>> choose(ServerWebExchange exchange) 
        URI uri = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
        GrayLoadBalancer loadBalancer = new GrayLoadBalancer(clientFactory.getLazyProvider(uri.getHost(), ServiceInstanceListSupplier.class), uri.getHost());
        if (loadBalancer == null) 
            throw new NotFoundException("No loadbalancer available for " + uri.getHost());
         else 
            return loadBalancer.choose(this.createRequest(exchange));
        
    

    private Request createRequest(ServerWebExchange exchange) 
        HttpHeaders headers = exchange.getRequest().getHeaders();
        Request<HttpHeaders> request = new DefaultRequest<>(headers);
        return request;
    

WeightMeta

import java.util.Arrays;
import java.util.Random;

/**
 * @author javachen
 * @description 权重元数据对象
 */
public class WeightMeta<T> 
    private final Random ran = new Random();
    private final T[] nodes;
    private final int[] weights;
    private final int maxW;

    public WeightMeta<
以上是关于初探灰度发布系列--AB Test以及栗子的主要内容，如果未能解决你的问题，请参考以下文章
 
 蓝绿发布金丝雀发布灰度发布滚动发布AB测试
 蓝绿部署金丝雀发布（灰度发布）AB测试……
 Spring Cloud实践：降级限流滚动灰度AB金丝雀的实现思路
 Spring Cloud 优雅下线以及灰度发布
 全链路灰度之 RocketMQ 灰度
 蓝绿部署滚动部署灰度发布金丝雀发布