def _determine_root_span()

in azext_edge/edge/providers/stats.py [0:0]

• 25 lines of code
• 10 McCabe index (conditional complexity)

def _determine_root_span(message_dict: dict) -> Tuple[str, str, Union[datetime, None]]:
    """
    Attempts to determine root span, and normalizes traceId, spanId and parentSpanId to hex.
    """
    import base64

    root_span = None
    resource_name = None
    timestamp = None

    for resource_span in message_dict.get("resourceSpans", []):
        for scope_span in resource_span.get("scopeSpans", []):
            for span in scope_span.get("spans", []):
                if "traceId" in span:
                    span["traceId"] = base64.b64decode(span["traceId"]).hex()
                if "spanId" in span:
                    span["spanId"] = base64.b64decode(span["spanId"]).hex()
                if "parentSpanId" in span:
                    span["parentSpanId"] = base64.b64decode(span["parentSpanId"]).hex()
                else:
                    root_span = span

                    if "startTimeUnixNano" in root_span:
                        timestamp_unix_nano = root_span["startTimeUnixNano"]
                        timestamp = datetime.utcfromtimestamp(float(timestamp_unix_nano) / 1e9)

                    # determine resource name
                    resource = resource_span.get("resource", {})
                    attributes = resource.get("attributes", [])
                    for a in attributes:
                        if a["key"] == "service.name":
                            resource_name = a["value"].get("stringValue", "unknown")

    return root_span, resource_name, timestamp