diff --git a/skywalking/agent/__init__.py b/skywalking/agent/__init__.py index 4587e5eb..b473783a 100644 --- a/skywalking/agent/__init__.py +++ b/skywalking/agent/__init__.py @@ -168,6 +168,10 @@ def started(): return __started +def isfull(): + return __queue.full() + + def archive(segment: 'Segment'): try: # unlike checking __queue.full() then inserting, this is atomic __queue.put(segment, block=False) diff --git a/skywalking/plugins/sw_urllib_request.py b/skywalking/plugins/sw_urllib_request.py index 6b0a64e0..9ce904d0 100644 --- a/skywalking/plugins/sw_urllib_request.py +++ b/skywalking/plugins/sw_urllib_request.py @@ -41,7 +41,8 @@ def _sw_open(this: OpenerDirector, fullurl, data=None, timeout=socket._GLOBAL_DE span.layer = Layer.Http code = None - [fullurl.add_header(item.key, item.val) for item in carrier] + for item in carrier: + fullurl.add_header(item.key, item.val) try: res = _open(this, fullurl, data, timeout) diff --git a/skywalking/trace/carrier.py b/skywalking/trace/carrier.py index 0119a9ee..fff0a24b 100644 --- a/skywalking/trace/carrier.py +++ b/skywalking/trace/carrier.py @@ -48,6 +48,7 @@ def __init__(self, trace_id: str = '', segment_id: str = '', span_id: str = '', service_instance: str = '', endpoint: str = '', client_address: str = '', correlation: dict = None): # pyre-ignore super(Carrier, self).__init__(key='sw8') + self.__val = None self.trace_id = trace_id # type: str self.segment_id = segment_id # type: str self.span_id = span_id # type: str @@ -101,6 +102,10 @@ def is_valid(self): len(self.client_address) > 0 and \ self.span_id.isnumeric() + @property + def is_suppressed(self): # if is invalid from previous set, ignored or suppressed status propagation downstream + return self.__val and not self.is_valid + def __iter__(self): self.__iter_index = 0 return self diff --git a/skywalking/trace/context.py b/skywalking/trace/context.py index 5d65a6e1..cdd57c43 100644 --- a/skywalking/trace/context.py +++ b/skywalking/trace/context.py @@ -16,6 +16,7 @@ # from skywalking import Component, agent, config +from skywalking.agent import isfull from skywalking.trace import ID from skywalking.trace.carrier import Carrier from skywalking.trace.segment import Segment, SegmentRef @@ -75,12 +76,18 @@ def __init__(self): self._correlation = {} # type: dict self._nspans = 0 + def ignore_check(self, op: str, kind: Kind, carrier: 'Carrier' = None): + if config.RE_IGNORE_PATH.match(op) or isfull() or (carrier is not None and carrier.is_suppressed): + return NoopSpan(context=NoopContext()) + + return None + def new_local_span(self, op: str) -> Span: span = self.ignore_check(op, Kind.Local) if span is not None: return span - spans = _spans_dup() + spans = _spans() parent = spans[-1] if spans else None # type: Span return Span( @@ -92,11 +99,11 @@ def new_local_span(self, op: str) -> Span: ) def new_entry_span(self, op: str, carrier: 'Carrier' = None, inherit: Component = None) -> Span: - span = self.ignore_check(op, Kind.Entry) + span = self.ignore_check(op, Kind.Entry, carrier) if span is not None: return span - spans = _spans_dup() + spans = _spans() parent = spans[-1] if spans else None # type: Span if parent is not None and parent.kind.is_entry and inherit == parent.component: @@ -120,12 +127,13 @@ def new_exit_span(self, op: str, peer: str, component: Component = None, inherit if span is not None: return span - spans = _spans_dup() + spans = _spans() parent = spans[-1] if spans else None # type: Span if parent is not None and parent.kind.is_exit and component == parent.inherit: span = parent span.op = op + span.peer = peer span.component = component else: span = ExitSpan( @@ -142,23 +150,14 @@ def new_exit_span(self, op: str, peer: str, component: Component = None, inherit return span - def ignore_check(self, op: str, kind: Kind): - if config.RE_IGNORE_PATH.match(op): - return NoopSpan( - context=NoopContext(), - kind=kind, - ) - - return None - def start(self, span: Span): self._nspans += 1 - spans = _spans() + spans = _spans_dup() if span not in spans: spans.append(span) def stop(self, span: Span) -> bool: - spans = _spans() + spans = _spans_dup() span.finish(self.segment) try: @@ -225,30 +224,27 @@ def continued(self, snapshot: 'Snapshot'): 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: - return self._noop_span + return NoopSpan(self) def new_entry_span(self, op: str, carrier: 'Carrier' = None, inherit: Component = None) -> Span: - if carrier is not None: - self._noop_span.extract(carrier) - return self._noop_span + return NoopSpan(self) def new_exit_span(self, op: str, peer: str, component: Component = None, inherit: Component = None) -> Span: - return self._noop_span - - def start(self, span: Span): - self._depth += 1 + return NoopSpan(self) def stop(self, span: Span) -> bool: - self._depth -= 1 - return self._depth == 0 + spans = _spans_dup() - def active_span(self): - return self._noop_span + try: + spans.remove(span) + except Exception: + pass + + self._nspans -= 1 + + return self._nspans == 0 def capture(self): return Snapshot( diff --git a/skywalking/trace/span.py b/skywalking/trace/span.py index 80f000a8..8d265f30 100644 --- a/skywalking/trace/span.py +++ b/skywalking/trace/span.py @@ -220,12 +220,11 @@ def inject(self) -> 'Carrier': @tostring class NoopSpan(Span): - def __init__(self, context: 'SpanContext' = None, kind: 'Kind' = None): - Span.__init__(self, context=context, kind=kind) + def __init__(self, context: 'SpanContext' = None): + Span.__init__(self, context=context, op='', kind=Kind.Local) def extract(self, carrier: 'Carrier'): - if carrier is not None: - self.context._correlation = carrier.correlation_carrier.correlation + return def inject(self) -> 'Carrier': - return Carrier(correlation=self.context._correlation) + return Carrier()