Skip to content

Commit

Permalink
Add conversion to TimeSeries methods
Browse files Browse the repository at this point in the history
  • Loading branch information
shovnik committed Nov 27, 2020
1 parent cf31639 commit 970212c
Show file tree
Hide file tree
Showing 3 changed files with 349 additions and 39 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -5,3 +5,5 @@
((#180)[https://github.com/open-telemetry/opentelemetry-python-contrib/pull/180])
- Add Exporter constructor validation methods
((#206)[https://github.com/open-telemetry/opentelemetry-python-contrib/pull/206])
- Add conversion to TimeSeries methods
((#207)[https://github.com/open-telemetry/opentelemetry-python-contrib/pull/207])
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import re
from typing import Dict, Sequence

from opentelemetry.exporter.prometheus_remote_write.gen.remote_pb2 import (
WriteRequest,
)
from opentelemetry.exporter.prometheus_remote_write.gen.types_pb2 import (
Label,
Sample,
Expand All @@ -24,6 +28,13 @@
MetricsExporter,
MetricsExportResult,
)
from opentelemetry.sdk.metrics.export.aggregate import (
HistogramAggregator,
LastValueAggregator,
MinMaxSumCountAggregator,
SumAggregator,
ValueObserverAggregator,
)


class PrometheusRemoteWriteMetricsExporter(MetricsExporter):
Expand Down Expand Up @@ -142,50 +153,144 @@ def export(
def shutdown(self) -> None:
raise NotImplementedError()

def convert_to_timeseries(
def _convert_to_timeseries(
self, export_records: Sequence[ExportRecord]
) -> Sequence[TimeSeries]:
raise NotImplementedError()

def convert_from_sum(self, sum_record: ExportRecord) -> TimeSeries:
raise NotImplementedError()

def convert_from_min_max_sum_count(
converter_map = {
MinMaxSumCountAggregator: self._convert_from_min_max_sum_count,
SumAggregator: self._convert_from_sum,
HistogramAggregator: self._convert_from_histogram,
LastValueAggregator: self._convert_from_last_value,
ValueObserverAggregator: self._convert_from_value_observer,
}
timeseries = []
for export_record in export_records:
aggregator_type = type(export_record.aggregator)
converter = converter_map.get(aggregator_type)
if converter is None:
raise ValueError(
str(aggregator_type) + " conversion is not supported"
)
timeseries.extend(converter(export_record))
return timeseries

def _convert_from_sum(self, sum_record: ExportRecord) -> TimeSeries:
return [
self._create_timeseries(
sum_record,
sum_record.instrument.name,
sum_record.aggregator.checkpoint,
)
]

def _convert_from_min_max_sum_count(
self, min_max_sum_count_record: ExportRecord
) -> TimeSeries:
raise NotImplementedError()

def convert_from_histogram(
timeseries = []
for agg_type in ["min", "max", "sum", "count"]:
name = min_max_sum_count_record.instrument.name + "_" + agg_type
value = getattr(
min_max_sum_count_record.aggregator.checkpoint, agg_type
)
timeseries.append(
self._create_timeseries(min_max_sum_count_record, name, value)
)
return timeseries

def _convert_from_histogram(
self, histogram_record: ExportRecord
) -> TimeSeries:
raise NotImplementedError()

def convert_from_last_value(
count = 0
timeseries = []
for bound in histogram_record.aggregator.checkpoint.keys():
bound_str = "+Inf" if bound == float("inf") else str(bound)
value = histogram_record.aggregator.checkpoint[bound]
timeseries.append(
self._create_timeseries(
histogram_record,
histogram_record.instrument.name,
value,
labels=[("le", bound_str)],
)
)
count += value
name = histogram_record.instrument.name + "_count"
timeseries.append(
self._create_timeseries(histogram_record, name, float(count))
)
return timeseries

def _convert_from_last_value(
self, last_value_record: ExportRecord
) -> TimeSeries:
raise NotImplementedError()

def convert_from_value_observer(
return [
self._create_timeseries(
last_value_record,
last_value_record.instrument.name,
last_value_record.aggregator.checkpoint,
)
]

def _convert_from_value_observer(
self, value_observer_record: ExportRecord
) -> TimeSeries:
raise NotImplementedError()
timeseries = []
for agg_type in ["min", "max", "sum", "count", "last"]:
timeseries.append(
self._create_timeseries(
value_observer_record,
value_observer_record.instrument.name + "_" + agg_type,
getattr(
value_observer_record.aggregator.checkpoint, agg_type
),
)
)
return timeseries

def convert_from_quantile(
# TODO: Implement convert from quantile once supported by SDK for Prometheus Summaries
def _convert_from_quantile(
self, summary_record: ExportRecord
) -> TimeSeries:
raise NotImplementedError()

# pylint: disable=no-member
def create_timeseries(
self, export_record: ExportRecord, name, value: float
def _create_timeseries(
self, export_record: ExportRecord, name, value: float, labels=None
) -> TimeSeries:
raise NotImplementedError()

def create_sample(self, timestamp: int, value: float) -> Sample:
raise NotImplementedError()

def create_label(self, name: str, value: str) -> Label:
raise NotImplementedError()
timeseries = TimeSeries()
seen = set()

def add_label(label_name, label_value):
# Label name must contain only alphanumeric characters and underscores
label_name = re.sub("[^\\w_]", "_", label_name)
if label_name not in seen:
label = Label()
label.name = label_name
label.value = label_value
timeseries.labels.append(label)
seen.add(label_name)

add_label("__name__", name)
if labels:
for [label_name, label_value] in labels:
add_label(label_name, label_value)
if export_record.resource.attributes:
for (
label_name,
label_value,
) in export_record.resource.attributes.items():
add_label(label_name, label_value)
if export_record.labels:
for [label_name, label_value] in export_record.labels:
add_label(label_name, label_value)

sample = Sample()
sample.timestamp = int(
export_record.aggregator.last_update_timestamp / 1000000
)
sample.value = value
timeseries.samples.append(sample)
return timeseries

def build_message(self, timeseries: Sequence[TimeSeries]) -> bytes:
raise NotImplementedError()
Expand Down
Loading

0 comments on commit 970212c

Please sign in to comment.