Skip to content

Commit

Permalink
Add Aggregation to the metrics SDK
Browse files Browse the repository at this point in the history
  • Loading branch information
ocelotl committed Oct 22, 2021
1 parent 94f4915 commit 58ff131
Show file tree
Hide file tree
Showing 4 changed files with 486 additions and 0 deletions.
117 changes: 117 additions & 0 deletions opentelemetry-sdk/src/opentelemetry/sdk/metrics/aggregation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,117 @@
# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from abc import ABC, abstractmethod
from collections import OrderedDict
from math import inf

from opentelemetry.proto.metrics.v1.metrics_pb2 import (
AGGREGATION_TEMPORALITY_CUMULATIVE,
AGGREGATION_TEMPORALITY_DELTA
)


class Aggregation(ABC):

def __init__(self):
self._value = 0

@property
def value(self):
return self._value

@abstractmethod
def aggregate(self, value):
pass

@abstractmethod
def collect(self):
pass


class NoneAggregation(Aggregation):
"""
This aggregation drops all instrument measurements.
"""

def aggregate(self, value):
pass

def collect(self):
return self._value


class SumAggregation(Aggregation):
"""
This aggregation collects data for the SDK sum metric point.
"""

def __init__(self, temporality=AGGREGATION_TEMPORALITY_CUMULATIVE):
super().__init__()
self._temporality = temporality

def aggregate(self, value):
self._value = self._value + value

def collect(self):
value = self._value
if self._temporality == AGGREGATION_TEMPORALITY_DELTA:
self._value = 0

return value


class LastValueAggregation(Aggregation):

"""
This aggregation collects data for the SDK sum metric point.
"""

def aggregate(self, value):
self._value = value

def collect(self):
return self._value


class ExplicitBucketHistogramAggregation(Aggregation):

"""
This aggregation collects data for the SDK sum metric point.
"""

def __init__(
self,
temporality=AGGREGATION_TEMPORALITY_CUMULATIVE,
boundaries=(0, 5, 10, 25, 50, 75, 100, 250, 500, 1000, inf)
):
super().__init__()
self._temporality = temporality
self._boundaries = boundaries
self._value = OrderedDict([(key, 0) for key in boundaries])

def aggregate(self, value):
for key in self._value.keys():

if value < key:
self._value[key] = self._value[key] + value

break

def collect(self):
value = self._value
if self._temporality == AGGREGATION_TEMPORALITY_DELTA:
self._value = OrderedDict([(key, 0) for key in self._boundaries])

return value
103 changes: 103 additions & 0 deletions opentelemetry-sdk/src/opentelemetry/sdk/metrics/instrument.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from opentelemetry.metrics.instrument import (
Counter,
ObservableCounter,
UpDownCounter,
ObservableUpDownCounter,
Histogram,
ObservableGauge
)

from opentelemetry.sdk.metrics.aggregation import (
SumAggregation,
LastValueAggregation,
ExplicitBucketHistogramAggregation,
)


class Counter(Counter):

def __init__(
self,
aggregation=SumAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()

def add(self, amount, attributes=None):
pass


class UpDownCounter(UpDownCounter):

def __init__(
self,
aggregation=SumAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()

def add(self, amount, attributes=None):
pass


class ObservableCounter(ObservableCounter):

def __init__(
self,
aggregation=SumAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()


class ObservableUpDownCounter(ObservableUpDownCounter):

def __init__(
self,
aggregation=SumAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()


class Histogram(Histogram):

def __init__(
self,
aggregation=ExplicitBucketHistogramAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()

def add(self, amount, attributes=None):
pass


class ObservableGauge(ObservableGauge):

def __init__(
self,
aggregation=LastValueAggregation,
aggregation_config={}
):
self._aggregation = aggregation(**aggregation_config)
super().__init__()
Loading

0 comments on commit 58ff131

Please sign in to comment.