base.py (8528B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\datadog\threadstats\base.py 2 import logging 3 import os 4 from functools import wraps 5 from contextlib import contextmanager 6 from time import time 7 from datadog.api.exceptions import ApiNotInitialized 8 from datadog.threadstats.constants import MetricType 9 from datadog.threadstats.metrics import MetricsAggregator, Counter, Gauge, Histogram, Timing 10 from datadog.threadstats.events import EventsAggregator 11 from datadog.threadstats.reporters import HttpReporter 12 log = logging.getLogger('datadog.threadstats') 13 14 class ThreadStats(object): 15 16 def __init__(self, namespace = '', constant_tags = None): 17 self.namespace = namespace 18 env_tags = [ tag for tag in os.environ.get('DATADOG_TAGS', '').split(',') if tag ] 19 if constant_tags is None: 20 constant_tags = [] 21 self.constant_tags = constant_tags + env_tags 22 self._disabled = True 23 24 def start(self, flush_interval = 10, roll_up_interval = 10, device = None, flush_in_thread = True, flush_in_greenlet = False, disabled = False): 25 self.flush_interval = flush_interval 26 self.roll_up_interval = roll_up_interval 27 self.device = device 28 self._disabled = disabled 29 self._is_auto_flushing = False 30 self._metric_aggregator = MetricsAggregator(self.roll_up_interval) 31 self._event_aggregator = EventsAggregator() 32 self.reporter = HttpReporter() 33 self._is_flush_in_progress = False 34 self.flush_count = 0 35 if self._disabled: 36 log.info('ThreadStats instance is disabled. No metrics will flush.') 37 elif flush_in_greenlet: 38 self._start_flush_greenlet() 39 elif flush_in_thread: 40 self._start_flush_thread() 41 42 def stop(self): 43 if not self._is_auto_flushing: 44 return True 45 if self._flush_thread: 46 self._flush_thread.end() 47 self._is_auto_flushing = False 48 return True 49 50 def event(self, title, text, alert_type = None, aggregation_key = None, source_type_name = None, date_happened = None, priority = None, tags = None, hostname = None): 51 if not self._disabled: 52 event_tags = tags 53 if self.constant_tags: 54 if tags: 55 event_tags = tags + self.constant_tags 56 else: 57 event_tags = self.constant_tags 58 self._event_aggregator.add_event(title=title, text=text, alert_type=alert_type, aggregation_key=aggregation_key, source_type_name=source_type_name, date_happened=date_happened, priority=priority, tags=event_tags, host=hostname) 59 60 def gauge(self, metric_name, value, timestamp = None, tags = None, sample_rate = 1, host = None): 61 if not self._disabled: 62 self._metric_aggregator.add_point(metric_name, tags, timestamp or time(), value, Gauge, sample_rate=sample_rate, host=host) 63 64 def increment(self, metric_name, value = 1, timestamp = None, tags = None, sample_rate = 1, host = None): 65 if not self._disabled: 66 self._metric_aggregator.add_point(metric_name, tags, timestamp or time(), value, Counter, sample_rate=sample_rate, host=host) 67 68 def decrement(self, metric_name, value = 1, timestamp = None, tags = None, sample_rate = 1, host = None): 69 if not self._disabled: 70 self._metric_aggregator.add_point(metric_name, tags, timestamp or time(), -value, Counter, sample_rate=sample_rate, host=host) 71 72 def histogram(self, metric_name, value, timestamp = None, tags = None, sample_rate = 1, host = None): 73 if not self._disabled: 74 self._metric_aggregator.add_point(metric_name, tags, timestamp or time(), value, Histogram, sample_rate=sample_rate, host=host) 75 76 def timing(self, metric_name, value, timestamp = None, tags = None, sample_rate = 1, host = None): 77 if not self._disabled: 78 self._metric_aggregator.add_point(metric_name, tags, timestamp or time(), value, Timing, sample_rate=sample_rate, host=host) 79 80 @contextmanager 81 def timer(self, metric_name, sample_rate = 1, tags = None, host = None): 82 start = time() 83 try: 84 yield 85 finally: 86 end = time() 87 self.timing(metric_name, end - start, end, tags=tags, sample_rate=sample_rate, host=host) 88 89 def timed(self, metric_name, sample_rate = 1, tags = None, host = None): 90 91 def wrapper(func): 92 93 @wraps(func) 94 def wrapped(*args, **kwargs): 95 with self.timer(metric_name, sample_rate, tags, host): 96 result = func(*args, **kwargs) 97 return result 98 99 return wrapped 100 101 return wrapper 102 103 def flush(self, timestamp = None): 104 try: 105 if self._is_flush_in_progress: 106 log.debug('A flush is already in progress. Skipping this one.') 107 return False 108 if self._disabled: 109 log.info("Not flushing because we're disabled.") 110 return False 111 self._is_flush_in_progress = True 112 metrics = self._get_aggregate_metrics(timestamp or time()) 113 count_metrics = len(metrics) 114 if count_metrics: 115 self.flush_count += 1 116 log.debug('Flush #%s sending %s metrics' % (self.flush_count, count_metrics)) 117 self.reporter.flush_metrics(metrics) 118 else: 119 log.debug('No metrics to flush. Continuing.') 120 events = self._get_aggregate_events() 121 count_events = len(events) 122 if count_events: 123 self.flush_count += 1 124 log.debug('Flush #%s sending %s events' % (self.flush_count, count_events)) 125 self.reporter.flush_events(events) 126 else: 127 log.debug('No events to flush. Continuing.') 128 except ApiNotInitialized: 129 raise 130 except: 131 try: 132 log.exception('Error flushing metrics and events') 133 except: 134 pass 135 136 finally: 137 self._is_flush_in_progress = False 138 139 def _get_aggregate_metrics(self, flush_time = None): 140 rolled_up_metrics = self._metric_aggregator.flush(flush_time) 141 metrics = [] 142 for timestamp, value, name, tags, host in rolled_up_metrics: 143 metric_tags = tags 144 metric_name = name 145 if self.constant_tags: 146 if tags: 147 metric_tags = tags + self.constant_tags 148 else: 149 metric_tags = self.constant_tags 150 if self.namespace: 151 metric_name = self.namespace + '.' + name 152 metric = {'metric': metric_name, 153 'points': [[timestamp, value]], 154 'type': MetricType.Gauge, 155 'host': host, 156 'device': self.device, 157 'tags': metric_tags} 158 metrics.append(metric) 159 160 return metrics 161 162 def _get_aggregate_events(self): 163 events = self._event_aggregator.flush() 164 return events 165 166 def _start_flush_thread(self): 167 from datadog.threadstats.periodic_timer import PeriodicTimer 168 if self._is_auto_flushing: 169 log.info('Autoflushing already started.') 170 return 171 self._is_auto_flushing = True 172 173 def flush(): 174 try: 175 log.debug('Flushing metrics in thread') 176 self.flush() 177 except: 178 try: 179 log.exception('Error flushing in thread') 180 except: 181 pass 182 183 log.info('Starting flush thread with interval %s.' % self.flush_interval) 184 self._flush_thread = PeriodicTimer(self.flush_interval, flush) 185 self._flush_thread.start() 186 187 def _start_flush_greenlet(self): 188 if self._is_auto_flushing: 189 log.info('Autoflushing already started.') 190 return 191 self._is_auto_flushing = True 192 import gevent 193 194 def flush(): 195 while True: 196 try: 197 log.debug('Flushing metrics in greenlet') 198 self.flush() 199 gevent.sleep(self.flush_interval) 200 except: 201 try: 202 log.exception('Error flushing in greenlet') 203 except: 204 pass 205 206 log.info('Starting flush greenlet with interval %s.' % self.flush_interval) 207 gevent.spawn(flush)