oti/event-handler/otihandler/onap/health.py

   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
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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=========================================================
  16
  17 """generic class to keep track of app health"""
  18
  19 import uuid
  20 from datetime import datetime
  21 from threading import Lock
  22
  23
  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())
  29         self._lock = Lock()
  30         self._call_count = 0
  31         self._error_count = 0
  32         self._longest_timer = 0
  33         self._total_timer = 0
  34         self._last_success = None
  35         self._last_error = None
  36
  37     def dump(self):
  38         """returns dict of stats"""
  39         dump = None
  40         with self._lock:
  41             dump = {
  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)
  49             }
  50         return dump
  51
  52     def success(self, timer):
  53         """records the successful execution"""
  54         with self._lock:
  55             self._call_count += 1
  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
  60
  61     def error(self, timer):
  62         """records the errored execution"""
  63         with self._lock:
  64             self._call_count += 1
  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
  70
  71 class Health(object):
  72     """Health stats for multiple requests"""
  73     def __init__(self):
  74         """Health stats for application"""
  75         self._all_stats = {}
  76         self._lock = Lock()
  77
  78     def _add_or_get_stats(self, stats_name):
  79         """add to or get from the ever growing dict of HealthStats"""
  80         stats = None
  81         with self._lock:
  82             stats = self._all_stats.get(stats_name)
  83             if not stats:
  84                 self._all_stats[stats_name] = stats = HealthStats(stats_name)
  85         return stats
  86
  87     def success(self, stats_name, timer):
  88         """records the successful execution on stats_name"""
  89         stats = self._add_or_get_stats(stats_name)
  90         stats.success(timer)
  91
  92     def error(self, stats_name, timer):
  93         """records the error execution on stats_name"""
  94         stats = self._add_or_get_stats(stats_name)
  95         stats.error(timer)
  96
  97     def dump(self):
  98         """returns dict of stats"""
  99         with self._lock:
 100             stats = dict((k, v.dump()) for (k, v) in self._all_stats.items())
 101
 102         return stats