Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_label_set(self, labels: Dict[str, str]):
"""See `opentelemetry.metrics.Meter.create_metric`.
This implementation encodes the labels to use as a map key.
Args:
labels: The dictionary of label keys to label values.
"""
if len(labels) == 0:
return EMPTY_LABEL_SET
# Use simple encoding for now until encoding API is implemented
encoded = tuple(sorted(labels.items()))
# If LabelSet exists for this meter in memory, use existing one
if encoded not in self.labels:
self.labels[encoded] = LabelSet(labels=labels)
return self.labels[encoded]
unit,
value_type,
label_keys=label_keys,
enabled=enabled,
monotonic=monotonic,
)
def record(self, label_set: LabelSet, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.Measure.record`."""
self.get_handle(label_set).record(value)
UPDATE_FUNCTION = record
# Used when getting a LabelSet with no key/values
EMPTY_LABEL_SET = LabelSet()
class Meter(metrics_api.Meter):
"""See `opentelemetry.metrics.Meter`."""
def __init__(self):
self.labels = {}
def record_batch(
self,
label_set: LabelSet,
record_tuples: Sequence[Tuple[metrics_api.Metric, metrics_api.ValueT]],
) -> None:
"""See `opentelemetry.metrics.Meter.record_batch`."""
for metric, value in record_tuples:
metric.UPDATE_FUNCTION(label_set, value)