Skip to content

Commit dd8521e

Browse files
hectorhdzgc24t
hectorhdzg
and
c24t
authored
Adding OT Collector metrics exporter (#454)
Current implementation use OpenCensus receiver available in Collector, this will need to be updated eventually to use OT receiver when is ready. Fixes #344 Co-authored-by: Chris Kleinknecht <[email protected]>
1 parent 4b6a52d commit dd8521e

File tree

8 files changed

+508
-3
lines changed

8 files changed

+508
-3
lines changed

‎examples/metrics/collector.py

+53
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
# Copyright 2020, OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
#
15+
"""
16+
This module serves as an example for a simple application using metrics
17+
exporting to Collector
18+
"""
19+
20+
from opentelemetry import metrics
21+
from opentelemetry.ext.otcollector.metrics_exporter import (
22+
CollectorMetricsExporter,
23+
)
24+
from opentelemetry.sdk.metrics import Counter, MeterProvider
25+
from opentelemetry.sdk.metrics.export.controller import PushController
26+
27+
# Meter is responsible for creating and recording metrics
28+
metrics.set_preferred_meter_provider_implementation(lambda _: MeterProvider())
29+
meter = metrics.get_meter(__name__)
30+
# exporter to export metrics to OT Collector
31+
exporter = CollectorMetricsExporter(
32+
service_name="basic-service", endpoint="localhost:55678"
33+
)
34+
# controller collects metrics created from meter and exports it via the
35+
# exporter every interval
36+
controller = PushController(meter, exporter, 5)
37+
38+
counter = meter.create_metric(
39+
"requests",
40+
"number of requests",
41+
"requests",
42+
int,
43+
Counter,
44+
("environment",),
45+
)
46+
47+
# Labelsets are used to identify key-values that are associated with a specific
48+
# metric that you want to record. These are useful for pre-aggregation and can
49+
# be used to store custom dimensions pertaining to a metric
50+
label_set = meter.get_label_set({"environment": "staging"})
51+
52+
counter.add(25, label_set)
53+
input("Press any key to exit...")

‎examples/metrics/docker/collector-config.yaml

+18
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
receivers:
2+
opencensus:
3+
endpoint: "0.0.0.0:55678"
4+
5+
exporters:
6+
prometheus:
7+
endpoint: "0.0.0.0:8889"
8+
logging: {}
9+
10+
processors:
11+
batch:
12+
queued_retry:
13+
14+
service:
15+
pipelines:
16+
metrics:
17+
receivers: [opencensus]
18+
exporters: [logging, prometheus]

‎examples/metrics/docker/docker-compose.yaml

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
version: "2"
2+
services:
3+
4+
otel-collector:
5+
image: omnition/opentelemetry-collector-contrib:latest
6+
command: ["--config=/conf/collector-config.yaml", "--log-level=DEBUG"]
7+
volumes:
8+
- ./collector-config.yaml:/conf/collector-config.yaml
9+
ports:
10+
- "8889:8889" # Prometheus exporter metrics
11+
- "55678:55678" # OpenCensus receiver
12+
13+
prometheus:
14+
container_name: prometheus
15+
image: prom/prometheus:latest
16+
volumes:
17+
- ./prometheus.yaml:/etc/prometheus/prometheus.yml
18+
ports:
19+
- "9090:9090"

‎examples/metrics/docker/prometheus.yaml

+5
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
scrape_configs:
2+
- job_name: 'otel-collector'
3+
scrape_interval: 5s
4+
static_configs:
5+
- targets: ['otel-collector:8889']

‎ext/opentelemetry-ext-otcollector/README.rst

+46-3
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ OpenTelemetry Collector Exporter
66
.. |pypi| image:: https://badge.fury.io/py/opentelemetry-ext-otcollector.svg
77
:target: https://pypi.org/project/opentelemetry-ext-otcollector/
88

9-
This library allows to export data to `OpenTelemetry Collector <https://github.com/open-telemetry/opentelemetry-collector/>`_.
9+
This library allows to export data to `OpenTelemetry Collector <https://github.com/open-telemetry/opentelemetry-collector/>`_ , currently using OpenCensus receiver in Collector side.
1010

1111
Installation
1212
------------
@@ -16,8 +16,8 @@ Installation
1616
pip install opentelemetry-ext-otcollector
1717

1818

19-
Usage
20-
-----
19+
Traces Usage
20+
------------
2121

2222
The **OpenTelemetry Collector Exporter** allows to export `OpenTelemetry`_ traces to `OpenTelemetry Collector`_.
2323

@@ -48,6 +48,49 @@ The **OpenTelemetry Collector Exporter** allows to export `OpenTelemetry`_ trace
4848
with tracer.start_as_current_span("foo"):
4949
print("Hello world!")
5050
51+
Metrics Usage
52+
-------------
53+
54+
The **OpenTelemetry Collector Exporter** allows to export `OpenTelemetry`_ metrics to `OpenTelemetry Collector`_.
55+
56+
.. code:: python
57+
58+
from opentelemetry import metrics
59+
from opentelemetry.ext.otcollector.metrics_exporter import CollectorMetricsExporter
60+
from opentelemetry.sdk.metrics import Counter, MeterProvider
61+
from opentelemetry.sdk.metrics.export.controller import PushController
62+
63+
64+
# create a CollectorMetricsExporter
65+
collector_exporter = CollectorMetricsExporter(
66+
# optional:
67+
# endpoint="myCollectorUrl:55678",
68+
# service_name="test_service",
69+
# host_name="machine/container name",
70+
)
71+
72+
# Meter is responsible for creating and recording metrics
73+
metrics.set_preferred_meter_provider_implementation(lambda _: MeterProvider())
74+
meter = metrics.get_meter(__name__)
75+
# controller collects metrics created from meter and exports it via the
76+
# exporter every interval
77+
controller = PushController(meter, collector_exporter, 5)
78+
counter = meter.create_metric(
79+
"requests",
80+
"number of requests",
81+
"requests",
82+
int,
83+
Counter,
84+
("environment",),
85+
)
86+
# Labelsets are used to identify key-values that are associated with a specific
87+
# metric that you want to record. These are useful for pre-aggregation and can
88+
# be used to store custom dimensions pertaining to a metric
89+
label_set = meter.get_label_set({"environment": "staging"})
90+
91+
counter.add(25, label_set)
92+
93+
5194
References
5295
----------
5396

‎ext/opentelemetry-ext-otcollector/src/opentelemetry/ext/otcollector/metrics_exporter/__init__.py

+165
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
# Copyright 2020, OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
"""OpenTelemetry Collector Metrics Exporter."""
16+
17+
import logging
18+
from typing import Sequence
19+
20+
import grpc
21+
from opencensus.proto.agent.metrics.v1 import (
22+
metrics_service_pb2,
23+
metrics_service_pb2_grpc,
24+
)
25+
from opencensus.proto.metrics.v1 import metrics_pb2
26+
27+
import opentelemetry.ext.otcollector.util as utils
28+
from opentelemetry.sdk.metrics import Counter, Metric
29+
from opentelemetry.sdk.metrics.export import (
30+
MetricRecord,
31+
MetricsExporter,
32+
MetricsExportResult,
33+
aggregate,
34+
)
35+
36+
DEFAULT_ENDPOINT = "localhost:55678"
37+
38+
logger = logging.getLogger(__name__)
39+
40+
41+
# pylint: disable=no-member
42+
class CollectorMetricsExporter(MetricsExporter):
43+
"""OpenTelemetry Collector metrics exporter.
44+
45+
Args:
46+
endpoint: OpenTelemetry Collector OpenCensus receiver endpoint.
47+
service_name: Name of Collector service.
48+
host_name: Host name.
49+
client: MetricsService client stub.
50+
"""
51+
52+
def __init__(
53+
self,
54+
endpoint: str = DEFAULT_ENDPOINT,
55+
service_name: str = None,
56+
host_name: str = None,
57+
client: metrics_service_pb2_grpc.MetricsServiceStub = None,
58+
):
59+
self.endpoint = endpoint
60+
if client is None:
61+
channel = grpc.insecure_channel(self.endpoint)
62+
self.client = metrics_service_pb2_grpc.MetricsServiceStub(
63+
channel=channel
64+
)
65+
else:
66+
self.client = client
67+
68+
self.node = utils.get_node(service_name, host_name)
69+
70+
def export(
71+
self, metric_records: Sequence[MetricRecord]
72+
) -> MetricsExportResult:
73+
try:
74+
responses = self.client.Export(
75+
self.generate_metrics_requests(metric_records)
76+
)
77+
78+
# Read response
79+
for _ in responses:
80+
pass
81+
82+
except grpc.RpcError:
83+
return MetricsExportResult.FAILED_RETRYABLE
84+
85+
return MetricsExportResult.SUCCESS
86+
87+
def shutdown(self) -> None:
88+
pass
89+
90+
def generate_metrics_requests(
91+
self, metrics: Sequence[MetricRecord]
92+
) -> metrics_service_pb2.ExportMetricsServiceRequest:
93+
collector_metrics = translate_to_collector(metrics)
94+
service_request = metrics_service_pb2.ExportMetricsServiceRequest(
95+
node=self.node, metrics=collector_metrics
96+
)
97+
yield service_request
98+
99+
100+
# pylint: disable=too-many-branches
101+
def translate_to_collector(
102+
metric_records: Sequence[MetricRecord],
103+
) -> Sequence[metrics_pb2.Metric]:
104+
collector_metrics = []
105+
for metric_record in metric_records:
106+
107+
label_values = []
108+
label_keys = []
109+
for label_tuple in metric_record.label_set.labels:
110+
label_keys.append(metrics_pb2.LabelKey(key=label_tuple[0]))
111+
label_values.append(
112+
metrics_pb2.LabelValue(
113+
has_value=label_tuple[1] is not None, value=label_tuple[1]
114+
)
115+
)
116+
117+
metric_descriptor = metrics_pb2.MetricDescriptor(
118+
name=metric_record.metric.name,
119+
description=metric_record.metric.description,
120+
unit=metric_record.metric.unit,
121+
type=get_collector_metric_type(metric_record.metric),
122+
label_keys=label_keys,
123+
)
124+
125+
timeseries = metrics_pb2.TimeSeries(
126+
label_values=label_values,
127+
points=[get_collector_point(metric_record)],
128+
)
129+
collector_metrics.append(
130+
metrics_pb2.Metric(
131+
metric_descriptor=metric_descriptor, timeseries=[timeseries]
132+
)
133+
)
134+
return collector_metrics
135+
136+
137+
# pylint: disable=no-else-return
138+
def get_collector_metric_type(metric: Metric) -> metrics_pb2.MetricDescriptor:
139+
if isinstance(metric, Counter):
140+
if metric.value_type == int:
141+
return metrics_pb2.MetricDescriptor.CUMULATIVE_INT64
142+
elif metric.value_type == float:
143+
return metrics_pb2.MetricDescriptor.CUMULATIVE_DOUBLE
144+
return metrics_pb2.MetricDescriptor.UNSPECIFIED
145+
146+
147+
def get_collector_point(metric_record: MetricRecord) -> metrics_pb2.Point:
148+
point = metrics_pb2.Point(
149+
timestamp=utils.proto_timestamp_from_time_ns(
150+
metric_record.metric.bind(
151+
metric_record.label_set
152+
).last_update_timestamp
153+
)
154+
)
155+
if metric_record.metric.value_type == int:
156+
point.int64_value = metric_record.aggregator.checkpoint
157+
elif metric_record.metric.value_type == float:
158+
point.double_value = metric_record.aggregator.checkpoint
159+
else:
160+
raise TypeError(
161+
"Unsupported metric type: {}".format(
162+
metric_record.metric.value_type
163+
)
164+
)
165+
return point

0 commit comments

Comments
 (0)