聊聊skywalking的spring-cloud-gateway-plugin

序

本文主要研究一下skywalking的spring-cloud-gateway-plugin

NettyRoutingFilterInstrumentation

skywalking-6.6.0/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v21x/define/NettyRoutingFilterInstrumentation.java

public class NettyRoutingFilterInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {

    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[]{
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("filter").and(takesArgumentWithType(0, "org.springframework.web.server.ServerWebExchange"));
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.NettyRoutingFilterInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return false;
                }
            }
        };
    }

    @Override
    public ClassMatch enhanceClass() {
        return byName("org.springframework.cloud.gateway.filter.NettyRoutingFilter");
    }

    @Override
    protected final String[] witnessClasses() {
        return new String[]{"org.springframework.cloud.gateway.handler.FilteringWebHandler", "reactor.netty.http.client.HttpClientOperations"};
    }
}

NettyRoutingFilterInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.NettyRoutingFilterInterceptor拦截org.springframework.cloud.gateway.filter.NettyRoutingFilter带有org.springframework.web.server.ServerWebExchange参数的filter方法

NettyRoutingFilterInterceptor

public class NettyRoutingFilterInterceptor implements InstanceMethodsAroundInterceptor {


    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        EnhancedInstance instance = NettyRoutingFilterInterceptor.getInstance(allArguments[0]);
        if (instance != null) {
            SWTransmitter swTransmitter = (SWTransmitter) instance.getSkyWalkingDynamicField();
            ContextManager.getRuntimeContext().put(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER, swTransmitter);
        }
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
                              Class<?>[] argumentsTypes, Object ret) throws Throwable {
        if (ContextManager.getRuntimeContext().get(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER) != null) {
            ContextManager.getRuntimeContext().remove(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER);
        }
        return ret;
    }


    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
    }

    public static EnhancedInstance getInstance(Object o) {
        EnhancedInstance instance = null;
        if (o instanceof ServerWebExchangeDecorator) {
            instance = getEnhancedInstance((ServerWebExchangeDecorator) o);
        } else if (o instanceof DefaultServerWebExchange) {
            instance = (EnhancedInstance) o;
        }
        return instance;
    }


    private static EnhancedInstance getEnhancedInstance(ServerWebExchangeDecorator serverWebExchangeDecorator) {
        Object o = serverWebExchangeDecorator.getDelegate();
        if (o instanceof ServerWebExchangeDecorator) {
            return getEnhancedInstance((ServerWebExchangeDecorator) o);
        } else if (o instanceof DefaultServerWebExchange) {
            return (EnhancedInstance) o;
        } else if (o == null) {
            throw new NullPointerException("The expected class DefaultServerWebExchange is null");
        } else {
            throw new RuntimeException("Unknown parameter types:" + o.getClass());
        }
    }
}

NettyRoutingFilterInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法获取swTransmitter，然后以名为Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER的key放入到ContextManager.getRuntimeContext()；其afterMethod方法执行ContextManager.getRuntimeContext().remove(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER)

HttpClientOperationsInstrumentation

public class HttpClientOperationsInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {

    @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[]{
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("headers").and(takesArgumentWithType(0, "io.netty.handler.codec.http.HttpHeaders"));
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsHeadersInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return false;
                }
            },new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("send").and(takesArgumentWithType(0, "org.reactivestreams.Publisher"));
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsSendInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return false;
                }
            },
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("status");
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsStatusInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return false;
                }
            },
        };
    }

    @Override
    public ClassMatch enhanceClass() {
        return byName("reactor.netty.http.client.HttpClientOperations");
    }
}

HttpClientOperationsInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine
它使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsHeadersInterceptor拦截reactor.netty.http.client.HttpClientOperations的带有io.netty.handler.codec.http.HttpHeaders参数的headers方法
它还使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsSendInterceptor拦截reactor.netty.http.client.HttpClientOperations的带有org.reactivestreams.Publisher参数的send方法
它还使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.HttpClientOperationsStatusInterceptor拦截reactor.netty.http.client.HttpClientOperations的status方法

HttpClientOperationsHeadersInterceptor

public class HttpClientOperationsHeadersInterceptor implements InstanceMethodsAroundInterceptor {

    private static final ILog logger = LogManager.getLogger(HttpClientOperationsHeadersInterceptor.class);

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
                              Class<?>[] argumentsTypes, Object ret) throws Throwable {
        Object transmitter = ((EnhancedInstance) allArguments[0]).getSkyWalkingDynamicField();
        if (transmitter != null) {
            objInst.setSkyWalkingDynamicField(transmitter);
            ((EnhancedInstance) allArguments[0]).setSkyWalkingDynamicField(null);
        }
        return ret;
    }


    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
    }
}

HttpClientOperationsHeadersInterceptor实现了InstanceMethodsAroundInterceptor接口，其afterMethod方法执行objInst.setSkyWalkingDynamicField(transmitter)

HttpClientOperationsSendInterceptor

public class HttpClientOperationsSendInterceptor implements InstanceMethodsAroundInterceptor {

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        SWTransmitter transmitter = (SWTransmitter) objInst.getSkyWalkingDynamicField();
        if (transmitter != null) {
            HttpClientRequest request = (HttpClientRequest) objInst;

            HttpHeaders header = request.requestHeaders();
            ChannelOperations channelOpt = (ChannelOperations) objInst;
            InetSocketAddress remote = (InetSocketAddress) (channelOpt.channel().remoteAddress());
            String peer = remote.getHostName() + ":" + remote.getPort();

            AbstractSpan span = ContextManager.createExitSpan(transmitter.getOperationName(), peer);
            ContextManager.continued(transmitter.getSnapshot());
            ContextCarrier contextCarrier = new ContextCarrier();
            ContextManager.inject(contextCarrier);

            span.setComponent(ComponentsDefine.SPRING_CLOUD_GATEWAY);
            Tags.URL.set(span, peer + request.uri());
            Tags.HTTP.METHOD.set(span, request.method().name());
            SpanLayer.asHttp(span);

            CarrierItem next = contextCarrier.items();
            while (next.hasNext()) {
                next = next.next();
                header.set(next.getHeadKey(), next.getHeadValue());
            }
            transmitter.setSpanGateway(span.prepareForAsync());
            ContextManager.stopSpan(span);
        }
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
                              Class<?>[] argumentsTypes, Object ret) throws Throwable {
        return ret;
    }


    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
        ContextManager.activeSpan().errorOccurred().log(t);
    }
}

HttpClientOperationsSendInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法获取request header、uri、method等信息设置到span中，最后执行ContextManager.stopSpan(span)；其handleMethodException方法执行ContextManager.activeSpan().errorOccurred().log(t)

HttpClientOperationsStatusInterceptor

public class HttpClientOperationsStatusInterceptor implements InstanceMethodsAroundInterceptor {

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                              Object ret) throws Throwable {

        SWTransmitter transmitter = (SWTransmitter) objInst.getSkyWalkingDynamicField();
        if (transmitter != null) {
            HttpResponseStatus response = (HttpResponseStatus) ret;
            if (response.code() >= 400) {
                Tags.STATUS_CODE.set(transmitter.getSpanGateway().errorOccurred(), String.valueOf(response.code()));
            }
            transmitter.getSpanGateway().asyncFinish();
            objInst.setSkyWalkingDynamicField(null);
        }
        return ret;
    }

    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
        ContextManager.activeSpan().errorOccurred().log(t);
    }
}

HttpClientOperationsStatusInterceptor实现了InstanceMethodsAroundInterceptor接口，其afterMethod方法获取transmitter，在response.code()大于等于400时设置statusCode的tag，然后执行transmitter.getSpanGateway().asyncFinish()；其handleMethodException执行ContextManager.activeSpan().errorOccurred().log(t)

FilteringWebHandlerInstrumentation

public class FilteringWebHandlerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {

    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[]{
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("handle");
                }

                @Override
                public String getMethodsInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.FilteringWebHandlerInterceptor";
                }

                @Override
                public boolean isOverrideArgs() {
                    return false;
                }
            }
        };
    }

    @Override
    public ClassMatch enhanceClass() {
        return byName("org.springframework.cloud.gateway.handler.FilteringWebHandler");
    }

}

FilteringWebHandlerInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.FilteringWebHandlerInterceptor拦截org.springframework.cloud.gateway.handler.FilteringWebHandler的handle方法

FilteringWebHandlerInterceptor

public class FilteringWebHandlerInterceptor implements InstanceMethodsAroundInterceptor {

    private static final String SPRING_CLOUD_GATEWAY_ROUTE_PREFIX = "GATEWAY/";

    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        EnhancedInstance instance = NettyRoutingFilterInterceptor.getInstance(allArguments[0]);
        if (instance == null) {
            return;
        }
        ContextSnapshot contextSnapshot = (ContextSnapshot) instance.getSkyWalkingDynamicField();
        if (contextSnapshot == null) {
            return;
        }

        ServerWebExchange exchange = (ServerWebExchange) allArguments[0];
        String operationName = SPRING_CLOUD_GATEWAY_ROUTE_PREFIX;
        Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);
        operationName = operationName + route.getId();
        SWTransmitter transmitter = new SWTransmitter(contextSnapshot, operationName);
        instance.setSkyWalkingDynamicField(transmitter);
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
                              Class<?>[] argumentsTypes, Object ret) throws Throwable {
        EnhancedInstance instance = NettyRoutingFilterInterceptor.getInstance(allArguments[0]);
        if (instance == null) {
            return ret;
        }
        SWTransmitter swTransmitter = (SWTransmitter) instance.getSkyWalkingDynamicField();
        if (swTransmitter == null) {
            return ret;
        }
        Mono<Void> mono = (Mono) ret;
        return mono.doFinally(d -> {
            ServerWebExchange exchange = (ServerWebExchange) allArguments[0];
            HttpStatus statusCode = exchange.getResponse().getStatusCode();
            if (statusCode == HttpStatus.TOO_MANY_REQUESTS) {
                AbstractSpan localSpan = ContextManager.createLocalSpan(swTransmitter.getOperationName());
                Tags.STATUS_CODE.set(localSpan,statusCode.toString());
                SpanLayer.asHttp(localSpan);
                localSpan.setComponent(ComponentsDefine.SPRING_CLOUD_GATEWAY);
                ContextManager.continued(swTransmitter.getSnapshot());
                ContextManager.stopSpan(localSpan);
            }
        });
    }


    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
    }

}

FilteringWebHandlerInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法从exchange中获取route信息最后形成operationName创建SWTransmitter并执行instance.setSkyWalkingDynamicField(transmitter)；其afterMethod方法获取SWTransmitter，然后注册mono的doFinally回调，在里头获取statusCode更细span，然后执行ContextManager.continued(swTransmitter.getSnapshot())及ContextManager.stopSpan(localSpan)

DefaultHttpHeadersInstrumentation

public class DefaultHttpHeadersInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {

    @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[] {
            new ConstructorInterceptPoint() {
                @Override public ElementMatcher<MethodDescription> getConstructorMatcher() {
                    return takesArgumentWithType(0, "io.netty.handler.codec.DefaultHeaders");
                }

                @Override public String getConstructorInterceptor() {
                    return "org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.DefaultHttpHeadersInterceptor";
                }
            }
        };
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[0];
    }

    @Override
    public ClassMatch enhanceClass() {
        return byName("io.netty.handler.codec.http.DefaultHttpHeaders");
    }
}

DefaultHttpHeadersInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它使用org.apache.skywalking.apm.plugin.spring.cloud.gateway.v21x.DefaultHttpHeadersInterceptor拦截io.netty.handler.codec.http.DefaultHttpHeaders的第一个参数为io.netty.handler.codec.DefaultHeaders的方法

DefaultHttpHeadersInterceptor

public class DefaultHttpHeadersInterceptor implements InstanceConstructorInterceptor {
    @Override
    public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
        Object transmitter = ContextManager.getRuntimeContext().get(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER);
        if (transmitter != null) {
            objInst.setSkyWalkingDynamicField(transmitter);
            ContextManager.getRuntimeContext().remove(Constants.SPRING_CLOUD_GATEWAY_TRANSMITTER);
        }
    }
}

DefaultHttpHeadersInterceptor实现了InstanceConstructorInterceptor接口，其onConstruct方法主要是将ContextManager.getRuntimeContext()中的transmitter设置到objInst中，然后从ContextManager.getRuntimeContext()移除该transmitter

小结

skywalking的spring-cloud-gateway-plugin主要有四个instrument，分别是NettyRoutingFilterInstrumentation、HttpClientOperationsInstrumentation、FilteringWebHandlerInstrumentation、DefaultHttpHeadersInstrumentation

doc

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 215,539评论 6赞 497
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 91,911评论 3赞 391
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 161,337评论 0赞 351
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 57,723评论 1赞 290
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,795评论 6赞 388
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,762评论 1赞 294
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,742评论 3赞 416
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 38,508评论 0赞 271
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,954评论 1赞 308
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,247评论 2赞 331
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 39,404评论 1赞 345
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,104评论 5赞 340
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,736评论 3赞 324
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 31,352评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,557评论 1赞 268
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 47,371评论 2赞 368
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 44,292评论 2赞 352