# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
# pylint: disable=unused-import
from collections.abc import Sequence
from dataclasses import asdict, dataclass, field
from json import dumps, loads
# This kind of import is needed to avoid Sphinx errors.
import opentelemetry.sdk.metrics._internal
from opentelemetry.sdk.metrics._internal.exemplar import Exemplar
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.util.instrumentation import InstrumentationScope
from opentelemetry.util.types import Attributes
[docs]
@dataclass(frozen=True)
class NumberDataPoint:
"""Single data point in a timeseries that describes the time-varying scalar
value of a metric.
"""
attributes: Attributes
start_time_unix_nano: int
time_unix_nano: int
value: int | float
exemplars: Sequence[Exemplar] = field(default_factory=list)
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(asdict(self), indent=indent)
[docs]
@dataclass(frozen=True)
class HistogramDataPoint:
"""Single data point in a timeseries that describes the time-varying scalar
value of a metric.
"""
attributes: Attributes
start_time_unix_nano: int
time_unix_nano: int
count: int
sum: int | float
bucket_counts: Sequence[int]
explicit_bounds: Sequence[float]
min: float
max: float
exemplars: Sequence[Exemplar] = field(default_factory=list)
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(asdict(self), indent=indent)
[docs]
@dataclass(frozen=True)
class Buckets:
offset: int
bucket_counts: Sequence[int]
[docs]
@dataclass(frozen=True)
class ExponentialHistogramDataPoint:
"""Single data point in a timeseries whose boundaries are defined by an
exponential function. This timeseries describes the time-varying scalar
value of a metric.
"""
attributes: Attributes
start_time_unix_nano: int
time_unix_nano: int
count: int
sum: int | float
scale: int
zero_count: int
positive: Buckets
negative: Buckets
flags: int
min: float
max: float
exemplars: Sequence[Exemplar] = field(default_factory=list)
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(asdict(self), indent=indent)
[docs]
@dataclass(frozen=True)
class ExponentialHistogram:
"""Represents the type of a metric that is calculated by aggregating as an
ExponentialHistogram of all reported measurements over a time interval.
"""
data_points: Sequence[ExponentialHistogramDataPoint]
aggregation_temporality: (
"opentelemetry.sdk.metrics.export.AggregationTemporality"
)
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"data_points": [
loads(data_point.to_json(indent=indent))
for data_point in self.data_points
],
"aggregation_temporality": self.aggregation_temporality,
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class Sum:
"""Represents the type of a scalar metric that is calculated as a sum of
all reported measurements over a time interval."""
data_points: Sequence[NumberDataPoint]
aggregation_temporality: (
"opentelemetry.sdk.metrics.export.AggregationTemporality"
)
is_monotonic: bool
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"data_points": [
loads(data_point.to_json(indent=indent))
for data_point in self.data_points
],
"aggregation_temporality": self.aggregation_temporality,
"is_monotonic": self.is_monotonic,
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class Gauge:
"""Represents the type of a scalar metric that always exports the current
value for every data point. It should be used for an unknown
aggregation."""
data_points: Sequence[NumberDataPoint]
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"data_points": [
loads(data_point.to_json(indent=indent))
for data_point in self.data_points
],
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class Histogram:
"""Represents the type of a metric that is calculated by aggregating as a
histogram of all reported measurements over a time interval."""
data_points: Sequence[HistogramDataPoint]
aggregation_temporality: (
"opentelemetry.sdk.metrics.export.AggregationTemporality"
)
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"data_points": [
loads(data_point.to_json(indent=indent))
for data_point in self.data_points
],
"aggregation_temporality": self.aggregation_temporality,
},
indent=indent,
)
# pylint: disable=invalid-name
DataT = Sum | Gauge | Histogram | ExponentialHistogram
DataPointT = (
NumberDataPoint | HistogramDataPoint | ExponentialHistogramDataPoint
)
[docs]
@dataclass(frozen=True)
class Metric:
"""Represents a metric point in the OpenTelemetry data model to be
exported."""
name: str
description: str | None
unit: str | None
data: DataT
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"name": self.name,
"description": self.description or "",
"unit": self.unit or "",
"data": loads(self.data.to_json(indent=indent)),
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class ScopeMetrics:
"""A collection of Metrics produced by a scope"""
scope: InstrumentationScope
metrics: Sequence[Metric]
schema_url: str
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"scope": loads(self.scope.to_json(indent=indent)),
"metrics": [
loads(metric.to_json(indent=indent))
for metric in self.metrics
],
"schema_url": self.schema_url,
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class ResourceMetrics:
"""A collection of ScopeMetrics from a Resource"""
resource: Resource
scope_metrics: Sequence[ScopeMetrics]
schema_url: str
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"resource": loads(self.resource.to_json(indent=indent)),
"scope_metrics": [
loads(scope_metrics.to_json(indent=indent))
for scope_metrics in self.scope_metrics
],
"schema_url": self.schema_url,
},
indent=indent,
)
[docs]
@dataclass(frozen=True)
class MetricsData:
"""An array of ResourceMetrics"""
resource_metrics: Sequence[ResourceMetrics]
[docs]
def to_json(self, indent: int | None = 4) -> str:
return dumps(
{
"resource_metrics": [
loads(resource_metrics.to_json(indent=indent))
for resource_metrics in self.resource_metrics
]
},
indent=indent,
)