py_zipkin/encoding/__init__.py
def detect_span_version_and_encoding(message):
    """Returns the span type and encoding for the message provided.

    The logic in this function is a Python port of
    https://github.com/openzipkin/zipkin/blob/master/zipkin/src/main/java/zipkin/internal/DetectingSpanDecoder.java

    :param message: span to perform operations on.
    :type message: byte array
    :returns: span encoding.
    :rtype: Encoding
    """
    # In case message is sent in as non-bytearray format,
    # safeguard convert to bytearray before handling
    if isinstance(message, six.string_types):
        # Even six.b is not enough to handle the py2/3 difference since
        # it uses latin-1 as default encoding and not utf-8.
        if six.PY2:
            message = six.b(message)  # pragma: no cover
        else:
            message = message.encode('utf-8')  # pragma: no cover

    if len(message) < 2:
        raise ZipkinError("Invalid span format. Message too short.")

    # Check for binary format
    if six.byte2int(message) <= 16:
        if six.byte2int(message) == 10 and six.byte2int(message[1:2]) != 0:
            return Encoding.V2_PROTO3
        return Encoding.V1_THRIFT

    str_msg = message.decode('utf-8')

    # JSON case for list of spans
    if str_msg[0] == '[':
        span_list = json.loads(str_msg)
        if len(span_list) > 0:
            # Assumption: All spans in a list are the same version
            # Logic: Search for identifying fields in all spans, if any span can
            # be strictly identified to a version, return that version.
            # Otherwise, if no spans could be strictly identified, default to V2.
            for span in span_list:
                if any(word in span for word in _V2_ATTRIBUTES):
                    return Encoding.V2_JSON
                elif (
                    'binaryAnnotations' in span or
                    (
                        'annotations' in span and
                        'endpoint' in span['annotations']
                    )
                ):
                    return Encoding.V1_JSON
            return Encoding.V2_JSON

    raise ZipkinError("Unknown or unsupported span encoding")
def detect_span_version_and_encoding(message):
    """Returns the span type and encoding for the message provided.

    The logic in this function is a Python port of
    https://github.com/openzipkin/zipkin/blob/master/zipkin/src/main/java/zipkin/internal/DetectingSpanDecoder.java

    :param message: span to perform operations on.
    :type message: byte array
    :returns: span encoding.
    :rtype: Encoding
    """
    # In case message is sent in as non-bytearray format,
    # safeguard convert to bytearray before handling
    if isinstance(message, six.string_types):
        # Even six.b is not enough to handle the py2/3 difference since
        # it uses latin-1 as default encoding and not utf-8.
        if six.PY2:
            message = six.b(message)  # pragma: no cover
        else:
            message = message.encode('utf-8')  # pragma: no cover

    if len(message) < 2:
        raise ZipkinError("Invalid span format. Message too short.")

    # Check for binary format
    if six.byte2int(message) <= 16:
        if six.byte2int(message) == 10 and six.byte2int(message[1:2]) != 0:
            return Encoding.V2_PROTO3
        return Encoding.V1_THRIFT

    str_msg = message.decode('utf-8')

    # JSON case for list of spans
    if str_msg[0] == '[':
        span_list = json.loads(str_msg)
        if len(span_list) > 0:
            # Assumption: All spans in a list are the same version
            # Logic: Search for identifying fields in all spans, if any span can
            # be strictly identified to a version, return that version.
            # Otherwise, if no spans could be strictly identified, default to V2.
            for span in span_list:
                if any(word in span for word in _V2_ATTRIBUTES):
                    return Encoding.V2_JSON
                elif (
                    'binaryAnnotations' in span or
                    (
                        'annotations' in span and
                        'endpoint' in span['annotations']
                    )
                ):
                    return Encoding.V1_JSON
            return Encoding.V2_JSON

    raise ZipkinError("Unknown or unsupported span encoding")
Returns the span type and encoding for the message provided.

    The logic in this function is a Python port of
    https://github.com/openzipkin/zipkin/blob/master/zipkin/src/main/java/zipkin/internal/DetectingSpanDecoder.java

    :param message: span to perform operations on.
    :type message: byte array
    :returns: span encoding.
    :rtype: Encoding