You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2022/12/20 04:15:10 UTC
[skywalking-python] branch master updated: Support primary endpoint and add to log reporter (#261)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking-python.git
The following commit(s) were added to refs/heads/master by this push:
new d62a6fc Support primary endpoint and add to log reporter (#261)
d62a6fc is described below
commit d62a6fc629adb99760df9731ea3f7f4bb9028cb5
Author: Superskyyy <Su...@outlook.com>
AuthorDate: Mon Dec 19 23:14:54 2022 -0500
Support primary endpoint and add to log reporter (#261)
---
CHANGELOG.md | 1 +
skywalking/log/sw_logging.py | 7 ++-
skywalking/plugins/sw_django.py | 2 +-
skywalking/plugins/sw_flask.py | 4 +-
skywalking/plugins/sw_sanic.py | 4 +-
skywalking/plugins/sw_tornado.py | 2 +-
skywalking/trace/context.py | 93 +++++++++++++++++++++++------------
skywalking/trace/span.py | 7 ++-
skywalking/utils/exception.py | 7 +++
tests/e2e/case/expected/logs-list.yml | 12 ++---
10 files changed, 93 insertions(+), 46 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9682899..e8469b5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,7 @@
- Make sure `span.component` initialized as Unknown rather than 0 (#242)
- Ignore websocket connections inside fastapi temporarily (#244, issue#9724)
- Fix Kafka-python plugin SkyWalking self reporter ignore condition (#249)
+ - Add primary endpoint in tracing context and endpoint info to log reporter (#256)
- Docs:
- New documentation on how to test locally (#222)
diff --git a/skywalking/log/sw_logging.py b/skywalking/log/sw_logging.py
index 893362a..b24a62a 100644
--- a/skywalking/log/sw_logging.py
+++ b/skywalking/log/sw_logging.py
@@ -81,10 +81,15 @@ def install():
traceContext=TraceContext(
traceId=str(context.segment.related_traces[0]),
traceSegmentId=str(context.segment.segment_id),
- spanId=context.active_span().sid if context.active_span() else -1
+ spanId=context.active_span.sid if context.active_span else -1
),
tags=build_log_tags(),
)
+
+ primary_endpoint_name = context.primary_endpoint.get_name()
+ if primary_endpoint_name:
+ log_data.endpoint = primary_endpoint_name
+
_handle(self=self, record=record)
agent.archive_log(log_data)
diff --git a/skywalking/plugins/sw_django.py b/skywalking/plugins/sw_django.py
index 33e56d0..efe8bcd 100644
--- a/skywalking/plugins/sw_django.py
+++ b/skywalking/plugins/sw_django.py
@@ -77,7 +77,7 @@ def install():
def _sw_handle_uncaught_exception(request, resolver, exc_info):
if exc_info is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None:
entry_span.raised()
diff --git a/skywalking/plugins/sw_flask.py b/skywalking/plugins/sw_flask.py
index 5305307..1edac53 100644
--- a/skywalking/plugins/sw_flask.py
+++ b/skywalking/plugins/sw_flask.py
@@ -72,7 +72,7 @@ def install():
def _sw_handle_user_exception(this: Flask, e):
if e is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None and type(entry_span) is not NoopSpan:
entry_span.raised()
@@ -80,7 +80,7 @@ def install():
def _sw_handle_exception(this: Flask, e):
if e is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None and type(entry_span) is not NoopSpan:
entry_span.raised()
diff --git a/skywalking/plugins/sw_sanic.py b/skywalking/plugins/sw_sanic.py
index a4b867c..b29c422 100644
--- a/skywalking/plugins/sw_sanic.py
+++ b/skywalking/plugins/sw_sanic.py
@@ -49,7 +49,7 @@ def install():
def _sw_format_http1_response(status: int, headers, body=b''):
if status is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None and type(entry_span) is not NoopSpan:
if status >= 400:
entry_span.error_occurred = True
@@ -59,7 +59,7 @@ def install():
def _sw_handlers_ErrorHandler_response(self: handlers.ErrorHandler, req, e): # noqa
if e is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None and type(entry_span) is not NoopSpan:
entry_span.raised()
diff --git a/skywalking/plugins/sw_tornado.py b/skywalking/plugins/sw_tornado.py
index 61db9b0..fd3337e 100644
--- a/skywalking/plugins/sw_tornado.py
+++ b/skywalking/plugins/sw_tornado.py
@@ -44,7 +44,7 @@ def install():
def _sw_handler_uncaught_exception(self: RequestHandler, ty, value, tb, *args, **kwargs):
if value is not None:
- entry_span = get_context().active_span()
+ entry_span = get_context().active_span
if entry_span is not None:
entry_span.raised()
diff --git a/skywalking/trace/context.py b/skywalking/trace/context.py
index ec359d9..53dafc2 100644
--- a/skywalking/trace/context.py
+++ b/skywalking/trace/context.py
@@ -14,6 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
+from typing import Optional
from skywalking import Component, agent, config
from skywalking import profile
@@ -25,6 +26,7 @@ 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.counter import Counter
+from skywalking.utils.exception import IllegalStateError
from skywalking.utils.time import current_milli_time
try: # attempt to use async-local instead of thread-local context and spans
@@ -71,23 +73,44 @@ except ImportError:
_spans_dup = _spans
-class SpanContext(object):
+class PrimaryEndpoint:
+ """
+ Behavior mocks Java agent's PrimaryEndpoint.
+ Primary endpoint name is used for endpoint dependency. The name pick policy according to priority is
+ 1. Use the first entry span's operation name
+ 2. Use the first span's operation name
+ """
+
+ def __init__(self, span: Span):
+ self.span: Span = span
+
+ def set_primary_endpoint(self, span):
+ if self.span.kind != Kind.Entry and span.kind == Kind.Entry:
+ self.span = span
+
+ def get_name(self):
+ return self.span.op
+
+
+class SpanContext:
def __init__(self):
- self.segment = Segment() # type: Segment
- self._sid = Counter()
- self._correlation = {} # type: dict
- self._nspans = 0
- self.profile_status = None # type: ProfileStatusReference
+ self.segment: Segment = Segment()
+ self._sid: Counter = Counter()
+ self._correlation: dict = {}
+ self._nspans: int = 0
+ self.profile_status: Optional[ProfileStatusReference] = None
self.create_time = current_milli_time()
+ self.primary_endpoint: Optional[PrimaryEndpoint] = None
- def ignore_check(self, op: str, kind: Kind, carrier: 'Carrier' = None):
+ @staticmethod
+ def ignore_check(op: str, kind: Kind, carrier: Optional[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_span(self, parent: Span, SpanType: type, **kwargs) -> Span: # noqa
- finished = parent and not parent._depth
+ def new_span(self, parent: Optional[Span], SpanType: type, **kwargs) -> Span: # noqa
+ finished = parent and not parent.depth
context = SpanContext() if finished else self
span = SpanType(context=context,
sid=context._sid.next(),
@@ -116,19 +139,15 @@ class SpanContext(object):
if span is not None:
return span
- spans = _spans()
- parent = spans[-1] if spans else None # type: Span
-
+ parent = self.peek()
return self.new_span(parent, Span, op=op, kind=Kind.Local)
- def new_entry_span(self, op: str, carrier: 'Carrier' = None, inherit: Component = None) -> Span:
+ def new_entry_span(self, op: str, carrier: Optional[Carrier] = None, inherit: Optional[Component] = None) -> Span:
span = self.ignore_check(op, Kind.Entry, carrier)
if span is not None:
return span
- spans = _spans()
- parent = spans[-1] if spans else None # type: Span
-
+ parent = self.peek()
# start profiling if profile_context is set
if self.profile_status is None:
self.profile_status = profile.profile_task_execution_service.add_profiling(self,
@@ -136,7 +155,7 @@ class SpanContext(object):
op)
if parent is not None and parent.kind.is_entry and inherit == parent.component:
- # Span's operation name could be override, recheck here
+ # Span's operation name could be overridden, recheck here
# if the op name now is being profiling, start profile it here
self.profiling_recheck(parent, op)
@@ -151,14 +170,13 @@ class SpanContext(object):
return span
- def new_exit_span(self, op: str, peer: str, component: Component = None, inherit: Component = None) -> Span:
+ def new_exit_span(self, op: str, peer: str,
+ component: Optional[Component] = None, inherit: Optional[Component] = None) -> Span:
span = self.ignore_check(op, Kind.Exit)
if span is not None:
return span
- spans = _spans()
- parent = spans[-1] if spans else None # type: Span
-
+ parent = self.peek()
if parent is not None and parent.kind.is_exit and component == parent.inherit:
span = parent
span.op = op
@@ -184,6 +202,11 @@ class SpanContext(object):
spans = _spans_dup()
if span not in spans:
spans.append(span)
+ # check primary endpoint is set
+ if not self.primary_endpoint:
+ self.primary_endpoint = PrimaryEndpoint(span)
+ else:
+ self.primary_endpoint.set_primary_endpoint(span)
def stop(self, span: Span) -> bool:
spans = _spans_dup()
@@ -191,7 +214,7 @@ class SpanContext(object):
try:
spans.remove(span)
- except Exception:
+ except ValueError:
pass
self._nspans -= 1
@@ -201,12 +224,17 @@ class SpanContext(object):
return False
- def active_span(self):
+ @staticmethod
+ def peek() -> Optional[Span]:
spans = _spans()
- if spans:
- return spans[len(spans) - 1]
+ return spans[-1] if spans else None
- return None
+ @property
+ def active_span(self):
+ active_span = self.peek()
+ if not active_span:
+ raise IllegalStateError('No active span')
+ return active_span
def get_correlation(self, key):
if key in self._correlation:
@@ -233,7 +261,7 @@ class SpanContext(object):
return Snapshot(
segment_id=str(self.segment.segment_id),
- span_id=self.active_span().sid,
+ span_id=self.active_span.sid,
trace_id=self.segment.related_traces[0],
endpoint=spans[0].op,
correlation=self._correlation,
@@ -244,7 +272,7 @@ class SpanContext(object):
return None
if not snapshot.is_from_current(self) and snapshot.is_valid():
ref = SegmentRef.build_ref(snapshot)
- span = self.active_span()
+ span = self.active_span
span.refs.append(ref)
self.segment.relate(ID(ref.trace_id))
self._correlation.update(snapshot.correlation)
@@ -257,10 +285,11 @@ class NoopContext(SpanContext):
def new_local_span(self, op: str) -> Span:
return NoopSpan(self)
- def new_entry_span(self, op: str, carrier: 'Carrier' = None, inherit: Component = None) -> Span:
+ def new_entry_span(self, op: str, carrier: Optional[Carrier] = None, inherit: Optional[Component] = None) -> Span:
return NoopSpan(self)
- def new_exit_span(self, op: str, peer: str, component: Component = None, inherit: Component = None) -> Span:
+ def new_exit_span(self, op: str, peer: str,
+ component: Optional[Component] = None, inherit: Optional[Component] = None) -> Span:
return NoopSpan(self)
def stop(self, span: Span) -> bool:
@@ -268,7 +297,7 @@ class NoopContext(SpanContext):
try:
spans.remove(span)
- except Exception:
+ except ValueError:
pass
self._nspans -= 1
@@ -292,6 +321,6 @@ def get_context() -> SpanContext:
spans = _spans()
if spans:
- return spans[len(spans) - 1].context
+ return spans[-1].context
return SpanContext()
diff --git a/skywalking/trace/span.py b/skywalking/trace/span.py
index 96a0fba..3d016a2 100644
--- a/skywalking/trace/span.py
+++ b/skywalking/trace/span.py
@@ -26,6 +26,7 @@ from skywalking.trace.carrier import Carrier
from skywalking.trace.segment import SegmentRef, Segment
from skywalking.trace.tags import Tag
from skywalking.utils.lang import tostring
+from skywalking.utils.exception import IllegalStateError
if TYPE_CHECKING:
from skywalking.trace.context import SpanContext
@@ -62,6 +63,10 @@ class Span:
self.end_time = 0 # type: int
self.error_occurred = False # type: bool
+ @property
+ def depth(self):
+ return self._depth
+
def start(self):
self._depth += 1
if self._depth != 1:
@@ -111,7 +116,7 @@ class Span:
yield from tag
def inject(self) -> 'Carrier':
- raise RuntimeWarning(
+ raise IllegalStateError(
'can only inject context carrier into ExitSpan, this may be a potential bug in the agent, '
'please report this in https://github.com/apache/skywalking/issues if you encounter this. '
)
diff --git a/skywalking/utils/exception.py b/skywalking/utils/exception.py
index 5eb5d0d..b4913a3 100644
--- a/skywalking/utils/exception.py
+++ b/skywalking/utils/exception.py
@@ -18,3 +18,10 @@
class VersionRuleException(Exception):
def __init__(self, message):
self.message = message
+
+
+class IllegalStateError(RuntimeError):
+ """
+ Raised when tracing context falls into unexpected state.
+ """
+ pass
diff --git a/tests/e2e/case/expected/logs-list.yml b/tests/e2e/case/expected/logs-list.yml
index 121cb7d..f2d2487 100644
--- a/tests/e2e/case/expected/logs-list.yml
+++ b/tests/e2e/case/expected/logs-list.yml
@@ -23,8 +23,8 @@ logs:
serviceid: {{ b64enc "e2e-service-provider" }}.1
serviceinstancename: provider1
serviceinstanceid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
- endpointname: null
- endpointid: null
+ endpointname: /artist
+ endpointid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "/artist" }}
traceid: {{ notEmpty .traceid }}
timestamp: {{ gt .timestamp 0 }}
contenttype: TEXT
@@ -42,8 +42,8 @@ logs:
serviceid: {{ b64enc "e2e-service-provider" }}.1
serviceinstancename: provider1
serviceinstanceid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
- endpointname: null
- endpointid: null
+ endpointname: /artist
+ endpointid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "/artist" }}
traceid: {{ notEmpty .traceid }}
timestamp: {{ gt .timestamp 0 }}
contenttype: TEXT
@@ -61,8 +61,8 @@ logs:
serviceid: {{ b64enc "e2e-service-provider" }}.1
serviceinstancename: provider1
serviceinstanceid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
- endpointname: null
- endpointid: null
+ endpointname: /artist
+ endpointid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "/artist" }}
traceid: {{ notEmpty .traceid }}
timestamp: {{ gt .timestamp 0 }}
contenttype: TEXT