Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, decision: "Decision"):
self._decision = decision
def should_sample(
self,
parent_context: Optional["SpanContext"],
trace_id: int,
span_id: int,
name: str,
attributes: Optional[Attributes] = None,
links: Sequence["Link"] = (),
) -> "Decision":
return self._decision
class ProbabilitySampler(Sampler):
def __init__(self, rate: float):
self._rate = rate
self._bound = self.get_bound_for_rate(self._rate)
# The sampler checks the last 8 bytes of the trace ID to decide whether to
# sample a given trace.
CHECK_BYTES = 0xFFFFFFFFFFFFFFFF
@classmethod
def get_bound_for_rate(cls, rate: float) -> int:
return round(rate * (cls.CHECK_BYTES + 1))
@property
def rate(self) -> float:
return self._rate
class Sampler(abc.ABC):
@abc.abstractmethod
def should_sample(
self,
parent_context: Optional["SpanContext"],
trace_id: int,
span_id: int,
name: str,
attributes: Optional[Attributes] = None,
links: Sequence["Link"] = (),
) -> "Decision":
pass
class StaticSampler(Sampler):
"""Sampler that always returns the same decision."""
def __init__(self, decision: "Decision"):
self._decision = decision
def should_sample(
self,
parent_context: Optional["SpanContext"],
trace_id: int,
span_id: int,
name: str,
attributes: Optional[Attributes] = None,
links: Sequence["Link"] = (),
) -> "Decision":
return self._decision