You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2022/09/11 08:33:43 UTC
[skywalking-python] branch master updated: feat: add support for MeterReportService (#231)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking-python.git
The following commit(s) were added to refs/heads/master by this push:
new 473ece7 feat: add support for MeterReportService (#231)
473ece7 is described below
commit 473ece77e4a59a83606e6f083f976d54b498505a
Author: jiang1997 <ji...@live.cn>
AuthorDate: Sun Sep 11 16:33:39 2022 +0800
feat: add support for MeterReportService (#231)
---
docs/en/setup/advanced/MeterReporter.md | 110 +++++++++++++
skywalking/agent/__init__.py | 33 +++-
skywalking/agent/protocol/__init__.py | 3 +
skywalking/agent/protocol/grpc.py | 36 ++++-
skywalking/client/__init__.py | 5 +
skywalking/client/grpc.py | 14 +-
skywalking/config.py | 5 +
skywalking/{client => meter}/__init__.py | 29 +---
skywalking/meter/counter.py | 100 ++++++++++++
skywalking/{client/__init__.py => meter/gauge.py} | 35 ++---
skywalking/meter/histogram.py | 120 ++++++++++++++
skywalking/meter/meter.py | 114 ++++++++++++++
skywalking/meter/meter_service.py | 54 +++++++
tests/unit/test_meter.py | 183 ++++++++++++++++++++++
14 files changed, 796 insertions(+), 45 deletions(-)
diff --git a/docs/en/setup/advanced/MeterReporter.md b/docs/en/setup/advanced/MeterReporter.md
new file mode 100644
index 0000000..f9cceb2
--- /dev/null
+++ b/docs/en/setup/advanced/MeterReporter.md
@@ -0,0 +1,110 @@
+# Python Agent Meter Reporter
+
+To enable or disable this feature, you will need to set some environment variables.
+
+
+## Enabling the feature (default)
+```Python
+os.environ['SW_AGENT_METER_REPORTER_ACTIVE'] = 'True'
+```
+or
+```bash
+export SW_AGENT_METER_REPORTER_ACTIVE=True
+```
+
+## Disable the feature
+```Python
+os.environ['SW_AGENT_METER_REPORTER_ACTIVE'] = 'False'
+```
+or
+```bash
+export SW_AGENT_METER_REPORTER_ACTIVE=False
+```
+## Counter
+* `Counter` API represents a single monotonically increasing counter, automatic collect data and report to backend.
+```python
+tg_ls = [MeterTag("key", "value")]
+c = Counter('c2', CounterMode.INCREMENT, tg_ls)
+# or with more compact way
+# Counter('c2', CounterMode.INCREMENT).tag('key1', 'value1').tag('key2', 'value2')
+c.build()
+c.increment(2)
+```
+### Syntactic sugars
+```python
+c = Counter('c2', CounterMode.INCREMENT)
+c.build()
+
+# increase Counter c by the time the with-wrapped codes consumed
+with c.create_timer():
+ # some codes may consume a certain time
+```
+
+```python
+c = Counter('c3', CounterMode.INCREMENT)
+c.build()
+
+# increase Counter c by num once counter_decorator_test gets called
+@Counter.increase(name='c3', num=2)
+def counter_decorator_test():
+ # some codes
+```
+
+```python
+c = Counter('c4', CounterMode.INCREMENT)
+c.build()
+
+# increase Counter c by the time counter_decorator_test consumed
+@Counter.timer(name='c4')
+def counter_decorator_test(s):
+ # some codes may consume a certain time
+```
+
+1. `Counter(name, tags)` Create a new counter with the meter name and optional tags.
+1. `Counter.tag(key: str, value)` Mark a tag key/value pair.
+1. `Counter.mode(mode: CounterMode)` Change the counter mode, RATE mode means reporting rate to the backend.
+1. `Counter.increment(count)` Increment count to the `Counter`, It could be a positive value.
+
+## Gauge
+* `Gauge` API represents a single numerical value.
+```python
+tg_ls = [MeterTag("key", "value")]
+# producer: iterable object
+g = Gauge('g1', producer, tg_ls)
+g.build()
+```
+1. `Gauge(name, tags)` Create a new gauge with the meter name and iterable object, this iterable object need to produce numeric value.
+1. `Gauge.tag(key: str, value)` Mark a tag key/value pair.
+1. `Gauge.build()` Build a new `Gauge` which is collected and reported to the backend.
+
+## Histogram
+* `Histogram` API represents a summary sample observations with customize buckets.
+```python
+tg_ls = [MeterTag("key", "value")]
+h = Histogram('h2', [i / 10 for i in range(10)], tg_ls)
+h.build()
+```
+### Syntactic sugars
+```python
+h = Histogram('h3', [i / 10 for i in range(10)])
+h.build()
+
+# Histogram h will record the time the with-wrapped codes consumed
+with h.create_timer():
+ # some codes may consume a certain time
+```
+
+```python
+h = Histogram('h2', [i / 10 for i in range(10)])
+h.build()
+
+# Histogram h will record the time histogram_decorator_test consumed
+@Histogram.timer(name='h2')
+def histogram_decorator_test(s):
+ time.sleep(s)
+```
+1. `Histogram(name, tags)` Create a new histogram with the meter name and optional tags.
+1. `Histogram.tag(key: str, value)` Mark a tag key/value pair.
+1. `Histogram.minValue(value)` Set up the minimal value of this histogram, default is `0`.
+1. `Histogram.build()` Build a new `Histogram` which is collected and reported to the backend.
+1. `Histogram.addValue(value)` Add value into the histogram, automatically analyze what bucket count needs to be increment. rule: count into [step1, step2).
\ No newline at end of file
diff --git a/skywalking/agent/__init__.py b/skywalking/agent/__init__.py
index d2fd393..c0dfddb 100644
--- a/skywalking/agent/__init__.py
+++ b/skywalking/agent/__init__.py
@@ -24,12 +24,14 @@ from typing import TYPE_CHECKING
from skywalking import config, plugins
from skywalking import loggings
from skywalking import profile
+from skywalking import meter
from skywalking.agent.protocol import Protocol
from skywalking.command import command_service
from skywalking.loggings import logger
from skywalking.profile.profile_task import ProfileTask
from skywalking.profile.snapshot import TracingThreadSnapshot
from skywalking.protocol.logging.Logging_pb2 import LogData
+from skywalking.protocol.language_agent.Meter_pb2 import MeterData
if TYPE_CHECKING:
from skywalking.trace.context import Segment
@@ -37,7 +39,7 @@ if TYPE_CHECKING:
__started = False
__protocol = None # type: Protocol
__heartbeat_thread = __report_thread = __log_report_thread = __query_profile_thread = __command_dispatch_thread \
- = __send_profile_thread = __queue = __log_queue = __snapshot_queue = __finished = None
+ = __send_profile_thread = __queue = __log_queue = __snapshot_queue = __meter_queue = __finished = None
def __heartbeat():
@@ -110,6 +112,20 @@ def __query_profile_command():
__finished.wait(wait)
+def __report_meter():
+ wait = base = 1
+
+ while not __finished.is_set():
+ try:
+ __protocol.report_meter(__meter_queue) # is blocking actually, blocks for max config.QUEUE_TIMEOUT seconds
+ wait = base
+ except Exception as exc:
+ logger.error(str(exc))
+ wait = min(60, wait * 2 or 1)
+
+ __finished.wait(wait)
+
+
def __command_dispatch():
# command dispatch will stuck when there are no commands
command_service.dispatch()
@@ -117,7 +133,7 @@ def __command_dispatch():
def __init_threading():
global __heartbeat_thread, __report_thread, __log_report_thread, __query_profile_thread, \
- __command_dispatch_thread, __send_profile_thread, __queue, __log_queue, __snapshot_queue, __finished
+ __command_dispatch_thread, __send_profile_thread, __queue, __log_queue, __snapshot_queue, __meter_queue, __finished
__queue = Queue(maxsize=config.max_buffer_size)
__finished = Event()
@@ -129,6 +145,11 @@ def __init_threading():
__report_thread.start()
__command_dispatch_thread.start()
+ if config.meter_reporter_active:
+ __meter_queue = Queue(maxsize=config.meter_reporter_max_buffer_size)
+ __meter_report_thread = Thread(name='MeterReportThread', target=__report_meter, daemon=True)
+ __meter_report_thread.start()
+
if config.log_reporter_active:
__log_queue = Queue(maxsize=config.log_reporter_max_buffer_size)
__log_report_thread = Thread(name='LogReportThread', target=__report_log, daemon=True)
@@ -217,6 +238,7 @@ def start():
loggings.init()
config.finalize()
profile.init()
+ meter.init()
__init()
@@ -254,6 +276,13 @@ def archive_log(log_data: 'LogData'):
logger.warning('the queue is full, the log will be abandoned')
+def archive_meter(meterdata: 'MeterData'):
+ try:
+ __meter_queue.put(meterdata, block=False)
+ except Full:
+ logger.warning('the queue is full, the meter will be abandoned')
+
+
def add_profiling_snapshot(snapshot: TracingThreadSnapshot):
try:
__snapshot_queue.put(snapshot)
diff --git a/skywalking/agent/protocol/__init__.py b/skywalking/agent/protocol/__init__.py
index ffe09ed..0ec8c2e 100644
--- a/skywalking/agent/protocol/__init__.py
+++ b/skywalking/agent/protocol/__init__.py
@@ -41,6 +41,9 @@ class Protocol(ABC):
def report_log(self, queue: Queue, block: bool = True):
raise NotImplementedError()
+ def report_meter(self, queue: Queue, block: bool = True):
+ raise NotImplementedError()
+
def query_profile_commands(self):
pass
diff --git a/skywalking/agent/protocol/grpc.py b/skywalking/agent/protocol/grpc.py
index d15e62a..1da8114 100644
--- a/skywalking/agent/protocol/grpc.py
+++ b/skywalking/agent/protocol/grpc.py
@@ -26,13 +26,14 @@ from skywalking import config
from skywalking.agent import Protocol
from skywalking.agent.protocol.interceptors import header_adder_interceptor
from skywalking.client.grpc import GrpcServiceManagementClient, GrpcTraceSegmentReportService, \
- GrpcProfileTaskChannelService, GrpcLogDataReportService
+ GrpcProfileTaskChannelService, GrpcLogDataReportService, GrpcMeterReportService
from skywalking.loggings import logger, logger_debug_enabled
from skywalking.profile.profile_task import ProfileTask
from skywalking.profile.snapshot import TracingThreadSnapshot
from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
from skywalking.protocol.language_agent.Tracing_pb2 import SegmentObject, SpanObject, Log, SegmentReference
from skywalking.protocol.logging.Logging_pb2 import LogData
+from skywalking.protocol.language_agent.Meter_pb2 import MeterData
from skywalking.protocol.profile.Profile_pb2 import ThreadSnapshot, ThreadStack
from skywalking.trace.segment import Segment
@@ -57,6 +58,7 @@ class GrpcProtocol(Protocol):
self.traces_reporter = GrpcTraceSegmentReportService(self.channel)
self.profile_channel = GrpcProfileTaskChannelService(self.channel)
self.log_reporter = GrpcLogDataReportService(self.channel)
+ self.meter_reporter = GrpcMeterReportService(self.channel)
def _cb(self, state):
if logger_debug_enabled:
@@ -189,6 +191,38 @@ class GrpcProtocol(Protocol):
self.on_error()
raise
+ def report_meter(self, queue: Queue, block: bool = True):
+ start = None
+
+ def generator():
+ nonlocal start
+
+ while True:
+ try:
+ timeout = config.QUEUE_TIMEOUT # type: int
+ if not start: # make sure first time through queue is always checked
+ start = time()
+ else:
+ timeout -= int(time() - start)
+ if timeout <= 0: # this is to make sure we exit eventually instead of being fed continuously
+ return
+ meter_data = queue.get(block=block, timeout=timeout) # type: MeterData
+ except Empty:
+ return
+
+ queue.task_done()
+
+ if logger_debug_enabled:
+ logger.debug('Reporting Meter')
+
+ yield meter_data
+
+ try:
+ self.meter_reporter.report(generator())
+ except grpc.RpcError:
+ self.on_error()
+ raise
+
def send_snapshot(self, queue: Queue, block: bool = True):
start = None
diff --git a/skywalking/client/__init__.py b/skywalking/client/__init__.py
index 562e3a4..cdd4519 100644
--- a/skywalking/client/__init__.py
+++ b/skywalking/client/__init__.py
@@ -29,6 +29,11 @@ class TraceSegmentReportService(object):
raise NotImplementedError()
+class MeterReportService(object):
+ def report(self, generator):
+ raise NotImplementedError()
+
+
class LogDataReportService(object):
def report(self, generator):
raise NotImplementedError()
diff --git a/skywalking/client/grpc.py b/skywalking/client/grpc.py
index ca115fe..e6e39b6 100644
--- a/skywalking/client/grpc.py
+++ b/skywalking/client/grpc.py
@@ -20,7 +20,7 @@ import grpc
from skywalking import config
from skywalking.client import ServiceManagementClient, TraceSegmentReportService, ProfileTaskChannelService, \
- LogDataReportService
+ LogDataReportService, MeterReportService
from skywalking.command import command_service
from skywalking.loggings import logger, logger_debug_enabled
from skywalking.profile import profile_task_execution_service
@@ -29,6 +29,7 @@ from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
from skywalking.protocol.language_agent.Tracing_pb2_grpc import TraceSegmentReportServiceStub
from skywalking.protocol.logging.Logging_pb2_grpc import LogReportServiceStub
from skywalking.protocol.management.Management_pb2 import InstancePingPkg, InstanceProperties
+from skywalking.protocol.language_agent.Meter_pb2_grpc import MeterReportServiceStub
from skywalking.protocol.management.Management_pb2_grpc import ManagementServiceStub
from skywalking.protocol.profile.Profile_pb2 import ProfileTaskCommandQuery, ProfileTaskFinishReport
from skywalking.protocol.profile.Profile_pb2_grpc import ProfileTaskStub
@@ -73,6 +74,17 @@ class GrpcTraceSegmentReportService(TraceSegmentReportService):
self.report_stub.collect(generator)
+class GrpcMeterReportService(MeterReportService):
+ def __init__(self, channel: grpc.Channel):
+ self.report_stub = MeterReportServiceStub(channel)
+
+ def report_batch(self, generator):
+ self.report_stub.collectBatch(generator)
+
+ def report(self, generator):
+ self.report_stub.collect(generator)
+
+
class GrpcLogDataReportService(LogDataReportService):
def __init__(self, channel: grpc.Channel):
self.report_stub = LogReportServiceStub(channel)
diff --git a/skywalking/config.py b/skywalking/config.py
index b3c33fd..edd560c 100644
--- a/skywalking/config.py
+++ b/skywalking/config.py
@@ -86,6 +86,11 @@ log_reporter_layout: str = os.getenv('SW_AGENT_LOG_REPORTER_LAYOUT') or \
# This configuration is shared by log reporter and tracer
cause_exception_depth: int = int(os.getenv('SW_AGENT_CAUSE_EXCEPTION_DEPTH') or '10')
+# meter reporter configurations
+meter_reporter_active: bool = os.getenv('SW_AGENT_METER_REPORTER_ACTIVE') == 'True'
+meter_reporter_max_buffer_size: int = int(os.getenv('SW_AGENT_METER_REPORTER_BUFFER_SIZE') or '10000')
+meter_reporter_peroid: int = int(os.getenv('SW_AGENT_METER_REPORTER_PEROID') or '20')
+
options = {key for key in globals() if key not in options} # THIS MUST FOLLOW DIRECTLY AFTER LIST OF CONFIG OPTIONS!
diff --git a/skywalking/client/__init__.py b/skywalking/meter/__init__.py
similarity index 58%
copy from skywalking/client/__init__.py
copy to skywalking/meter/__init__.py
index 562e3a4..eacfe5f 100644
--- a/skywalking/client/__init__.py
+++ b/skywalking/meter/__init__.py
@@ -15,28 +15,15 @@
# limitations under the License.
#
+_meter_service = None
-class ServiceManagementClient(object):
- def send_instance_props(self):
- raise NotImplementedError()
- def send_heart_beat(self):
- raise NotImplementedError()
+def init():
+ from skywalking.meter.meter_service import MeterService
+ global _meter_service
+ if _meter_service:
+ return
-class TraceSegmentReportService(object):
- def report(self, generator):
- raise NotImplementedError()
-
-
-class LogDataReportService(object):
- def report(self, generator):
- raise NotImplementedError()
-
-
-class ProfileTaskChannelService(object):
- def do_query(self):
- raise NotImplementedError()
-
- def send(self, generator):
- raise NotImplementedError()
+ _meter_service = MeterService()
+ _meter_service.start()
diff --git a/skywalking/meter/counter.py b/skywalking/meter/counter.py
new file mode 100644
index 0000000..123b9d2
--- /dev/null
+++ b/skywalking/meter/counter.py
@@ -0,0 +1,100 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+import timeit
+from enum import Enum
+from skywalking.meter.meter import BaseMeter, MeterType
+from skywalking.protocol.language_agent.Meter_pb2 import MeterData, MeterSingleValue
+
+
+class CounterMode(Enum):
+ INCREMENT = 1
+ RATE = 2
+
+
+class Counter(BaseMeter):
+ def __init__(self, name: str, mode: CounterMode, tags=None):
+ super().__init__(name, tags)
+ self.count = 0
+ self.previous = 0
+ self.mode = mode
+
+ def increment(self, value):
+ self.count += value
+
+ def get(self):
+ return self.count
+
+ def transform(self):
+ current_value = self.get()
+ if self.mode == CounterMode.RATE:
+ self.previous = current_value
+ count = current_value - self.previous
+ else:
+ count = current_value
+
+ meterdata = MeterData(singleValue=MeterSingleValue(name=self.get_name(), labels=self.transform_tags(), value=count))
+ return meterdata
+
+ def mode(self, mode):
+ self.mode = mode
+ return self
+
+ def get_type(self):
+ return MeterType.COUNTER
+
+ def create_timer(self):
+ return Counter.Timer(self)
+
+ class Timer():
+ def __init__(self, metrics):
+ self.metrics = metrics
+
+ def __enter__(self):
+ self.start = timeit.default_timer()
+
+ def __exit__(self, exc_type, exc_value, exc_tb):
+ self.stop = timeit.default_timer()
+ duration = self.stop - self.start
+ self.metrics.increment(duration)
+
+ @staticmethod
+ def timer(name: str):
+ def inner(func):
+ def wrapper(*args, **kwargs):
+ start = timeit.default_timer()
+ func(*args, **kwargs)
+ stop = timeit.default_timer()
+ duration = stop - start
+ counter = Counter.meter_service.get_meter(name)
+ counter.increment(duration)
+
+ return wrapper
+
+ return inner
+
+ @staticmethod
+ def increase(name: str, num=1):
+ def inner(func):
+ def wrapper(*args, **kwargs):
+ func(*args, **kwargs)
+ counter = Counter.meter_service.get_meter(name)
+ counter.increment(num)
+
+ return wrapper
+
+ return inner
diff --git a/skywalking/client/__init__.py b/skywalking/meter/gauge.py
similarity index 55%
copy from skywalking/client/__init__.py
copy to skywalking/meter/gauge.py
index 562e3a4..b004fed 100644
--- a/skywalking/client/__init__.py
+++ b/skywalking/meter/gauge.py
@@ -15,28 +15,23 @@
# limitations under the License.
#
+from skywalking.meter.meter import BaseMeter, MeterType
+from skywalking.protocol.language_agent.Meter_pb2 import MeterData, MeterSingleValue
-class ServiceManagementClient(object):
- def send_instance_props(self):
- raise NotImplementedError()
- def send_heart_beat(self):
- raise NotImplementedError()
+class Gauge(BaseMeter):
+ def __init__(self, name: str, generator, tags=None):
+ super().__init__(name, tags)
+ self.generator = generator
+ def get(self):
+ data = next(self.generator, None)
+ return data if data else 0
-class TraceSegmentReportService(object):
- def report(self, generator):
- raise NotImplementedError()
+ def transform(self):
+ count = self.get()
+ meterdata = MeterData(singleValue=MeterSingleValue(name=self.get_name(), labels=self.transform_tags(), value=count))
+ return meterdata
-
-class LogDataReportService(object):
- def report(self, generator):
- raise NotImplementedError()
-
-
-class ProfileTaskChannelService(object):
- def do_query(self):
- raise NotImplementedError()
-
- def send(self, generator):
- raise NotImplementedError()
+ def get_type(self):
+ return MeterType.GAUGE
diff --git a/skywalking/meter/histogram.py b/skywalking/meter/histogram.py
new file mode 100644
index 0000000..3a81cf2
--- /dev/null
+++ b/skywalking/meter/histogram.py
@@ -0,0 +1,120 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+import timeit
+from skywalking.meter.meter import BaseMeter, MeterType
+from skywalking.protocol.language_agent.Meter_pb2 import MeterBucketValue, MeterData, MeterHistogram
+
+
+class Histogram(BaseMeter):
+ def __init__(self, name: str, steps, min_value=0, tags=None):
+ super().__init__(name, tags)
+ self.min_value = min_value
+
+ # https://stackoverflow.com/a/53522/9845190
+ if not steps:
+ raise Exception('steps must not be empty')
+
+ # https://stackoverflow.com/a/2931683/9845190
+ steps = sorted(set(steps))
+
+ if steps[0] < self.min_value:
+ raise Exception('steps are invalid')
+ elif steps[0] != self.min_value:
+ steps.insert(0, self.min_value)
+ self.init_bucks(steps)
+
+
+ def add_value(self, value):
+ bucket = self.find_bucket(value)
+ if bucket is None:
+ return
+
+ bucket.increment(1)
+
+ def find_bucket(self, value):
+ left = 0
+ right = len(self.buckets)
+
+ # find the first bucket greater than or equal the value
+ while left < right:
+ mid = (left + right) // 2
+ if self.buckets[mid].bucket < value:
+ left = mid + 1
+ else:
+ right = mid
+
+ left -= 1
+
+ return self.buckets[left] if left < len(self.buckets) and left >= 0 else None
+
+
+ def init_bucks(self, steps):
+ self.buckets = [Histogram.Bucket(step) for step in steps]
+
+ def transform(self):
+ values = [bucket.transform() for bucket in self.buckets]
+ return MeterData(histogram=MeterHistogram(name=self.get_name(), labels=self.transform_tags(), values=values))
+
+ def get_type(self):
+ return MeterType.HISTOGRAM
+
+ def create_timer(self):
+ return Histogram.Timer(self)
+
+ class Bucket():
+
+ def __init__(self, bucket):
+ self.bucket = bucket
+ self.count = 0
+
+ def increment(self, count):
+ self.count += count
+
+ def transform(self):
+ return MeterBucketValue(bucket=self.bucket, count=self.count)
+
+ def __hash__(self) -> int:
+ return hash((self.bucket, self.count))
+
+ class Timer():
+ def __init__(self, metrics):
+ self.metrics = metrics
+
+ def __enter__(self):
+ self.start = timeit.default_timer()
+
+
+ def __exit__(self, exc_type, exc_value, exc_tb):
+ self.stop = timeit.default_timer()
+ duration = self.stop - self.start
+ self.metrics.add_value(duration)
+
+ @staticmethod
+ def timer(name: str):
+ def inner(func):
+ def wrapper(*args, **kwargs):
+ start = timeit.default_timer()
+ func(*args, **kwargs)
+ stop = timeit.default_timer()
+ duration = stop - start
+ histogram = Histogram.meter_service.get_meter(name)
+ histogram.add_value(duration)
+
+ return wrapper
+
+ return inner
diff --git a/skywalking/meter/meter.py b/skywalking/meter/meter.py
new file mode 100644
index 0000000..5eba338
--- /dev/null
+++ b/skywalking/meter/meter.py
@@ -0,0 +1,114 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 enum import Enum
+from skywalking.protocol.language_agent.Meter_pb2 import Label
+import skywalking.meter as meter
+
+
+class MeterType(Enum):
+ GAUGE = 1
+ COUNTER = 2
+ HISTOGRAM = 3
+
+
+class MeterTag():
+ def __init__(self, key: str, value: str):
+ self.key = key
+ self.value = value
+
+ def __lt__(self, other):
+ if self.key != other.key:
+ return self.key < other.key
+ else:
+ return self.value < other.value
+
+ def get_key(self):
+ return self.key
+
+ def get_value(self):
+ return self.value
+
+ def __hash__(self) -> int:
+ return hash((self.key, self.value))
+
+
+class MeterId():
+ def __init__(self, name: str, type, tags: list) -> None:
+ if tags is None:
+ tags = []
+
+ self.name = name
+ self.type = type
+ self.tags = tags
+ self.labels = None
+
+ def transform_tags(self):
+ if self.labels is not None:
+ return self.labels
+
+ self.labels = [Label(tag.key, tag.value) for tag in self.tags]
+ return self.labels
+
+ def get_name(self):
+ return self.name
+
+ def get_tags(self):
+ return self.tags
+
+ def get_type(self):
+ return self.type
+
+ def __hash__(self):
+ return hash((self.name, self.type, tuple(self.tags)))
+
+
+class BaseMeter(ABC):
+ meter_service = None
+
+ def __init__(self, name: str, tags=None):
+ if BaseMeter.meter_service is None:
+ BaseMeter.meter_service = meter._meter_service
+
+ self.meterId = MeterId(name, self.get_type(), tags)
+
+ def get_name(self):
+ return self.meterId.get_name()
+
+ def get_tag(self, tag_key):
+ for tag in self.meterId.get_tags():
+ if tag.get_key() == tag_key:
+ return tag.get_value()
+
+ def get_id(self):
+ return self.meterId
+
+ def transform_tags(self):
+ return self.get_id().transform_tags()
+
+ def tag(self, name: str, value):
+ self.meterId.get_tags().append(MeterTag(name, value))
+ return self
+
+ def build(self):
+ self.meterId.get_tags().sort()
+ BaseMeter.meter_service.register(self)
+
+ @abstractmethod
+ def get_type(self):
+ pass
diff --git a/skywalking/meter/meter_service.py b/skywalking/meter/meter_service.py
new file mode 100644
index 0000000..2f15873
--- /dev/null
+++ b/skywalking/meter/meter_service.py
@@ -0,0 +1,54 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+import time
+from concurrent.futures import ThreadPoolExecutor
+from threading import Thread
+from skywalking import config
+from skywalking import agent
+from skywalking.meter.meter import BaseMeter
+from skywalking.utils.time import current_milli_time
+from skywalking.config import meter_reporter_peroid
+
+
+class MeterService(Thread):
+ def __init__(self):
+ super().__init__(daemon=True)
+ self.meter_map = {}
+
+ def register(self, meter: BaseMeter):
+ self.meter_map[meter.get_id().get_name()] = meter
+
+ def get_meter(self, name: str):
+ return self.meter_map.get(name)
+
+ def send(self):
+
+ def archive(meterdata):
+ meterdata = meterdata.transform()
+ meterdata.service = config.service_name
+ meterdata.serviceInstance = config.service_instance
+ meterdata.timestamp = current_milli_time()
+ agent.archive_meter(meterdata)
+
+ with ThreadPoolExecutor(max_workers=1) as executor:
+ executor.map(archive, self.meter_map.values())
+
+ def run(self):
+ while True:
+ time.sleep(meter_reporter_peroid)
+ self.send()
diff --git a/tests/unit/test_meter.py b/tests/unit/test_meter.py
new file mode 100644
index 0000000..1569fbf
--- /dev/null
+++ b/tests/unit/test_meter.py
@@ -0,0 +1,183 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+import unittest
+import random
+import time
+
+from skywalking.meter.counter import Counter, CounterMode
+from skywalking.meter.histogram import Histogram
+from skywalking.meter.gauge import Gauge
+from skywalking.meter.meter import BaseMeter
+
+
+class MockMeterService():
+ def __init__(self):
+ self.meter_map = {}
+
+ def register(self, meter: BaseMeter):
+ self.meter_map[meter.get_id().get_name()] = meter
+
+ def get_meter(self, name: str):
+ return self.meter_map.get(name)
+
+ def transform(self, meter):
+ meterdata = meter.transform()
+ meterdata.service = 'service_name'
+ meterdata.serviceInstance = 'service_instance'
+ meterdata.timestamp = int(0)
+ return meterdata
+
+
+meter_service = MockMeterService()
+BaseMeter.meter_service = meter_service
+
+# picked empirically
+tolerance = 5e-2
+
+
+class TestMeter(unittest.TestCase):
+ def test_counter(self):
+ c = Counter('c1', CounterMode.INCREMENT)
+ c.build()
+
+ @Counter.increase(name='c1')
+ def increase_by_one():
+ # do nothing
+ pass
+
+ for i in range(1, 52):
+ increase_by_one()
+ self.assertEqual(i, c.count)
+ meterdata = meter_service.transform(c)
+ self.assertEqual(meterdata.singleValue.value, c.count)
+ self.assertEqual(i, meterdata.singleValue.value)
+
+ def test_counter_with_satement(self):
+ c = Counter('c2', CounterMode.INCREMENT)
+ c.build()
+
+ ls = [i / 10 for i in range(10)]
+ random.shuffle(ls)
+
+ pre = 0
+ for i in ls:
+ with c.create_timer():
+ time.sleep(i)
+
+ meterdata = meter_service.transform(c)
+ self.assertEqual(meterdata.singleValue.value, c.count)
+ self.assertLess(abs(i - (meterdata.singleValue.value - pre)), tolerance)
+ pre = meterdata.singleValue.value
+
+
+ def test_counter_increase_decarator(self):
+ c = Counter('c3', CounterMode.INCREMENT)
+ c.build()
+
+ @Counter.increase(name='c3', num=2)
+ def counter_decorator_test():
+ # do nothing
+ pass
+
+ for i in range(1, 10):
+ counter_decorator_test()
+ meterdata = meter_service.transform(c)
+ self.assertEqual(meterdata.singleValue.value, c.count)
+ self.assertEqual(i * 2, meterdata.singleValue.value)
+
+ def test_counter_timer_decarator(self):
+ c = Counter('c4', CounterMode.INCREMENT)
+ c.build()
+
+ ls = [i / 10 for i in range(10)]
+
+ @Counter.timer(name='c4')
+ def counter_decorator_test(s):
+ time.sleep(s)
+
+ total = 0
+ for _ in range(1, 5):
+ random.shuffle(ls)
+ for i in ls:
+ counter_decorator_test(i)
+ total += i
+ meterdata = meter_service.transform(c)
+ self.assertEqual(meterdata.singleValue.value, c.count)
+ self.assertLessEqual(abs(total - meterdata.singleValue.value), tolerance)
+
+ def test_histogram(self):
+ h = Histogram('h1', list(range(0, 10)))
+ h.build()
+
+ for repeat in range(1, 10):
+ ls = list(range(1, 10))
+ random.shuffle(ls)
+
+ for i in ls:
+ h.add_value(i)
+ self.assertEqual(repeat, h.find_bucket(i).count)
+ meterdata = meter_service.transform(h)
+ self.assertEqual(repeat, meterdata.histogram.values[i - 1].count)
+
+ def test_histogram_timer_decarator(self):
+ h = Histogram('h2', [i / 10 for i in range(10)])
+ h.build()
+
+ ls = [i / 10 for i in range(10)]
+
+ @Histogram.timer(name='h2')
+ def histogram_decorator_test(s):
+ time.sleep(s)
+
+ for repeat in range(1, 5):
+ random.shuffle(ls)
+ for i in ls:
+ histogram_decorator_test(i)
+ idx = int(i * 10)
+ meterdata = meter_service.transform(h)
+ self.assertEqual(repeat, meterdata.histogram.values[idx].count)
+
+ def test_histogram_with_satement(self):
+ h = Histogram('h3', [i / 10 for i in range(10)])
+ h.build()
+
+ ls = [i / 10 for i in range(10)]
+
+ for repeat in range(1, 5):
+ random.shuffle(ls)
+ for i in ls:
+ with h.create_timer():
+ time.sleep(i)
+ idx = int(i * 10)
+ meterdata = meter_service.transform(h)
+ self.assertEqual(repeat, meterdata.histogram.values[idx].count)
+
+
+ def test_gauge(self):
+ ls = list(range(1, 10))
+ random.shuffle(ls)
+ g = Gauge('g1', iter(ls))
+ g.build()
+
+ for i in ls:
+ meterdata = meter_service.transform(g)
+ self.assertEqual(i, meterdata.singleValue.value)
+
+
+if __name__ == '__main__':
+ unittest.main()