grcp using with skywalking-toolkit-opentracing

[SeriousMa]

client 和server没有关联
拓补图上体现不出调用流程

client interceptor code

package cn.enncloud.ceres.core.interceptor;

import cn.enncloud.ceres.Constants;
import io.grpc.*;
import io.opentracing.Span;
import io.opentracing.Tracer;
import io.opentracing.contrib.ActiveSpanSource;
import io.opentracing.contrib.OperationNameConstructor;
import io.opentracing.propagation.Format;
import io.opentracing.propagation.TextMap;
import io.opentracing.tag.Tags;

import java.util.Iterator;
import java.util.Map;

/**
 * Created by Serious on 2017/4/25.
 */
public class SWTraceClientInterceptor implements ClientInterceptor {

    private final Tracer tracer;
    private final OperationNameConstructor operationNameConstructor;
    private final ActiveSpanSource activeSpanSource;

    /**
     * @param tracer to use to trace requests
     */
    public SWTraceClientInterceptor(Tracer tracer) {
        this.tracer = tracer;
        this.operationNameConstructor = OperationNameConstructor.DEFAULT;
        this.activeSpanSource = ActiveSpanSource.GRPC_CONTEXT;
    }


    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
        final String operationName = operationNameConstructor.constructOperationName(method);

        Span activeSpan = this.activeSpanSource.getActiveSpan();
        final Span span = createSpanFromParent(activeSpan, operationName);
        Tags.COMPONENT.set(span, Constants.CERES);
        Tags.SPAN_KIND.set(span, Tags.SPAN_KIND_CLIENT);
        Tags.PEER_HOSTNAME.set(span,"127.0.0.1");
        Tags.PEER_PORT.set(span, (short) 1234);
        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            @Override
            public void start(Listener<RespT> responseListener, final Metadata headers) {


                tracer.inject(span.context(), Format.Builtin.HTTP_HEADERS, new TextMap() {
                    @Override
                    public void put(String key, String value) {
                        Metadata.Key<String> headerKey = Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER);
                        headers.put(headerKey, value);
                    }

                    @Override
                    public Iterator<Map.Entry<String, String>> iterator() {
                        throw new UnsupportedOperationException(
                                "TextMapInjectAdapter should only be used with Tracer.inject()");
                    }
                });

                Listener<RespT> tracingResponseListener = new ForwardingClientCallListener
                        .SimpleForwardingClientCallListener<RespT>(responseListener) {

                    @Override
                    public void onHeaders(Metadata headers) {
                        delegate().onHeaders(headers);
                    }

                    @Override
                    public void onMessage(RespT message) {
                        delegate().onMessage(message);
                    }

                    @Override
                    public void onClose(Status status, Metadata trailers) {
                        span.log("Call failed", status.getDescription());
                        span.finish();
                        delegate().onClose(status, trailers);
                    }
                };

                delegate().start(tracingResponseListener, headers);
            }

        }

                ;
    }


    private Span createSpanFromParent(Span parentSpan, String operationName) {
        if (parentSpan == null) {
            return tracer.buildSpan(operationName).start();
        } else {
            return tracer.buildSpan(operationName).asChildOf(parentSpan).start();
        }
    }

}

server interceptor code

package cn.enncloud.ceres.core.interceptor;

import cn.enncloud.ceres.Constants;
import io.grpc.*;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.contrib.OpenTracingContextKey;
import io.opentracing.propagation.Format;
import io.opentracing.propagation.TextMapExtractAdapter;
import io.opentracing.tag.Tags;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Created by Serious on 2017/4/25.
 */
public class SWServerTracingInterceptor implements ServerInterceptor {
    private static final Logger logger = LoggerFactory.getLogger(SWServerTracingInterceptor.class);
    private final Tracer tracer;
    private final Set<ServerRequestAttribute> tracedAttributes;

    /**
     * @param tracer used to trace requests
     */
    public SWServerTracingInterceptor(Tracer tracer) {
        this.tracer = tracer;
        this.tracedAttributes = new HashSet<ServerRequestAttribute>();
    }

    public enum ServerRequestAttribute {
        HEADERS,
        METHOD_TYPE,
        METHOD_NAME,
        CALL_ATTRIBUTES
    }


    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
        Map<String, String> headerMap = new HashMap<String, String>();
        for (String key : headers.keys()) {
            if (!key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
                String value = headers.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER));
                headerMap.put(key, value);
            }

        }

        final String operationName = call.getMethodDescriptor().getFullMethodName();
        final Span span = getSpanFromHeaders(headerMap, operationName);
        Tags.COMPONENT.set(span, Constants.CERES);
        Tags.SPAN_KIND.set(span, Tags.SPAN_KIND_SERVER);
        Tags.PEER_HOSTNAME.set(span, "127.0.0.1");
        Tags.PEER_PORT.set(span, (short) 4321);
        for (ServerRequestAttribute attr : this.tracedAttributes) {
            switch (attr) {
                case METHOD_TYPE:
                    span.setTag("grpc.method_type", call.getMethodDescriptor().getType().toString());
                    break;
                case METHOD_NAME:
                    span.setTag("grpc.method_name", call.getMethodDescriptor().getFullMethodName());
                    break;
                case CALL_ATTRIBUTES:
                    span.setTag("grpc.call_attributes", call.getAttributes().toString());
                    break;
                case HEADERS:
                    span.setTag("grpc.headers", headers.toString());
                    break;
            }
        }

        Context ctxWithSpan = Context.current().withValue(OpenTracingContextKey.getKey(), span);
        ServerCall.Listener<ReqT> listenerWithContext = Contexts.interceptCall(ctxWithSpan, call, headers, next);
        logger.debug("span :{}", span);
        span.finish();
        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listenerWithContext) {

            @Override
            public void onMessage(ReqT message) {
                delegate().onMessage(message);
            }

            @Override
            public void onHalfClose() {
                delegate().onHalfClose();
            }

            @Override
            public void onCancel() {
                delegate().onCancel();
            }

            @Override
            public void onComplete() {
                delegate().onComplete();
            }
        };
    }

    private Span getSpanFromHeaders(Map<String, String> headers, String operationName) {
        Span span;
        try {
            SpanContext parentSpanCtx = tracer.extract(Format.Builtin.HTTP_HEADERS,
                    new TextMapExtractAdapter(headers));
            if (parentSpanCtx == null) {
                span = tracer.buildSpan(operationName).start();
            } else {
                span = tracer.buildSpan(operationName).asChildOf(parentSpanCtx).start();
            }
        } catch (IllegalArgumentException iae) {
            span = tracer.buildSpan(operationName)
                    .withTag("Error", "Extract failed and an IllegalArgumentException was thrown")
                    .start();
        }
        return span;
    }
}

[wu-sheng]

It is a bug for OT-toolkit.

[wu-sheng]

@ascrutae

• Accept tag value is short, convert to int, instead of string
• peer.hostname, peer.ipv4 and peer.ipv6 should map to peer.host in OT-activation.

[ascrutae]

@SeriousMa This issue has fixed and will release in 3.0.1-2017.