context.py

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import logging
import threading
from typing import List

from skywalking import agent, config
from skywalking.trace import ID
from skywalking.trace.carrier import Carrier
from skywalking.trace.segment import Segment, SegmentRef
from skywalking.trace.snapshot import Snapshot
from skywalking.trace.span import Span, Kind, NoopSpan, EntrySpan, ExitSpan
from skywalking.utils.ant_matcher import fast_path_match
from skywalking.utils.counter import Counter

logger = logging.getLogger(__name__)


class SpanContext(object):
    def __init__(self):
        self.spans = []  # type: List[Span]
        self.segment = Segment()  # type: Segment
        self._sid = Counter()
        self._correlation = {}  # type: dict

    def new_local_span(self, op: str) -> Span:
        span = self.ignore_check(op, Kind.Local)
        if span is not None:
            return span

        parent = self.spans[-1] if self.spans else None  # type: Span

        return Span(
            context=self,
            sid=self._sid.next(),
            pid=parent.sid if parent else -1,
            op=op,
            kind=Kind.Local,
        )

    def new_entry_span(self, op: str, carrier: 'Carrier' = None) -> Span:
        span = self.ignore_check(op, Kind.Entry)
        if span is not None:
            return span

        parent = self.spans[-1] if self.spans else None  # type: Span

        span = parent if parent is not None and parent.kind.is_entry else EntrySpan(
            context=self,
            sid=self._sid.next(),
            pid=parent.sid if parent else -1,
        )
        span.op = op

        if carrier is not None and carrier.is_valid:
            span.extract(carrier=carrier)

        return span

    def new_exit_span(self, op: str, peer: str, carrier: 'Carrier' = None) -> Span:
        span = self.ignore_check(op, Kind.Exit)
        if span is not None:
            return span

        parent = self.spans[-1] if self.spans else None  # type: Span

        span = parent if parent is not None and parent.kind.is_exit else ExitSpan(
            context=self,
            sid=self._sid.next(),
            pid=parent.sid if parent else -1,
            op=op,
            peer=peer,
        )

        if carrier is not None:
            span.inject(carrier=carrier)

        return span

    def ignore_check(self, op: str, kind: Kind):
        suffix_idx = op.rfind(".")
        if suffix_idx > -1 and config.ignore_suffix.find(op[suffix_idx:]) > -1:
            return NoopSpan(
                context=NoopContext(),
                kind=kind,
            )
        if config.trace_ignore:
            for pattern in config.trace_ignore_path:
                if fast_path_match(pattern, op):
                    return NoopSpan(
                        context=NoopContext(),
                        kind=kind,
                    )

        return None

    def start(self, span: Span):
        if span not in self.spans:
            self.spans.append(span)

    def stop(self, span: Span) -> bool:
        assert span is self.spans[-1]

        span.finish(self.segment) and self.spans.pop()

        if len(self.spans) == 0:
            _thread_local.context = None
            agent.archive(self.segment)

        return len(self.spans) == 0

    def active_span(self):
        if self.spans:
            return self.spans[len(self.spans) - 1]

        return None

    def get_correlation(self, key):
        if key in self._correlation:
            return self._correlation[key]
        return None

    def put_correlation(self, key, value):
        if key is None:
            return
        if value is None:
            self._correlation.pop(key, value)
            return
        if len(value) > config.correlation_value_max_length:
            return
        if len(self._correlation) > config.correlation_element_max_number:
            return

        self._correlation[key] = value

    def capture(self):
        if len(self.spans) == 0:
            return None

        return Snapshot(
            segment_id=str(self.segment.segment_id),
            span_id=self.active_span().sid,
            trace_id=self.segment.related_traces[0],
            endpoint=self.spans[0].op,
            correlation=self._correlation,
        )

    def continued(self, snapshot: 'Snapshot'):
        if snapshot is None:
            return None
        if not snapshot.is_from_current(self) and snapshot.is_valid():
            ref = SegmentRef.build_ref(snapshot)
            span = self.active_span()
            span.refs.append(ref)
            self.segment.relate(ID(ref.trace_id))
            self._correlation.update(snapshot.correlation)


class NoopContext(SpanContext):
    def __init__(self):
        super().__init__()
        self._depth = 0
        self._noop_span = NoopSpan(self, kind=Kind.Local)
        self.correlation = {}  # type: dict

    def new_local_span(self, op: str) -> Span:
        self._depth += 1
        return self._noop_span

    def new_entry_span(self, op: str, carrier: 'Carrier' = None) -> Span:
        self._depth += 1
        if carrier is not None:
            self._noop_span.extract(carrier)
        return self._noop_span

    def new_exit_span(self, op: str, peer: str, carrier: 'Carrier' = None) -> Span:
        self._depth += 1
        if carrier is not None:
            self._noop_span.inject(carrier)

        return self._noop_span

    def stop(self, span: Span) -> bool:
        self._depth -= 1
        return self._depth == 0

    def active_span(self):
        return self._noop_span

    def capture(self):
        return Snapshot(
            segment_id=None,
            span_id=-1,
            trace_id=None,
            endpoint=None,
            correlation=self._correlation,
        )

    def continued(self, snapshot: 'Snapshot'):
        self._correlation.update(snapshot.correlation)


_thread_local = threading.local()
_thread_local.context = None


def get_context() -> SpanContext:
    if not hasattr(_thread_local, 'context'):
        _thread_local.context = None
    _thread_local.context = _thread_local.context or (SpanContext() if agent.connected() else NoopContext())

    return _thread_local.context