1 # ================================================================================
2 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
3 # ================================================================================
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 # ============LICENSE_END=========================================================
17 """generic class to keep track of app health"""
20 from datetime import datetime
21 from threading import Lock
24 class HealthStats(object):
25 """keep track of stats for calls"""
26 def __init__(self, name):
27 """keep track of stats for metrics calls"""
28 self._name = name or "stats_" + str(uuid.uuid4())
32 self._longest_timer = 0
34 self._last_success = None
35 self._last_error = None
38 """returns dict of stats"""
42 "call_count" : self._call_count,
43 "error_count" : self._error_count,
44 "last_success" : str(self._last_success),
45 "last_error" : str(self._last_error),
46 "longest_timer_millisecs" : self._longest_timer,
47 "ave_timer_millisecs" : (float(self._total_timer)/self._call_count \
48 if self._call_count else 0)
52 def success(self, timer):
53 """records the successful execution"""
56 self._last_success = datetime.now()
57 self._total_timer += timer
58 if not self._longest_timer or self._longest_timer < timer:
59 self._longest_timer = timer
61 def error(self, timer):
62 """records the errored execution"""
65 self._error_count += 1
66 self._last_error = datetime.now()
67 self._total_timer += timer
68 if not self._longest_timer or self._longest_timer < timer:
69 self._longest_timer = timer
72 """Health stats for multiple requests"""
74 """Health stats for application"""
78 def _add_or_get_stats(self, stats_name):
79 """add to or get from the ever growing dict of HealthStats"""
82 stats = self._all_stats.get(stats_name)
84 self._all_stats[stats_name] = stats = HealthStats(stats_name)
87 def success(self, stats_name, timer):
88 """records the successful execution on stats_name"""
89 stats = self._add_or_get_stats(stats_name)
92 def error(self, stats_name, timer):
93 """records the error execution on stats_name"""
94 stats = self._add_or_get_stats(stats_name)
98 """returns dict of stats"""
100 stats = dict((k, v.dump()) for (k, v) in self._all_stats.items())