| 1 | # -*- coding: utf-8 -*- |
| 2 | # Description: |
| 3 | # SPDX-License-Identifier: GPL-3.0-or-later |
| 4 | |
| 5 | import os |
| 6 | |
| 7 | from bases.charts import Charts, ChartError, create_runtime_chart |
| 8 | from bases.collection import safe_print |
| 9 | from bases.loggers import PythonDLogger |
| 10 | from third_party.monotonic import monotonic |
| 11 | from time import sleep, time |
| 12 | |
| 13 | RUNTIME_CHART_UPDATE = 'BEGIN netdata.runtime_{job_name} {since_last}\n' \ |
| 14 | 'SET run_time = {elapsed}\n' \ |
| 15 | 'END\n' |
| 16 | |
| 17 | PENALTY_EVERY = 5 |
| 18 | MAX_PENALTY = 10 * 60 # 10 minutes |
| 19 | |
| 20 | ND_INTERNAL_MONITORING_DISABLED = os.getenv("NETDATA_INTERNALS_MONITORING") == "NO" |
| 21 | |
| 22 | |
| 23 | class RuntimeCounters: |
| 24 | def __init__(self, configuration): |
| 25 | """ |
| 26 | :param configuration: <dict> |
| 27 | """ |
| 28 | self.update_every = int(configuration.pop('update_every')) |
| 29 | self.do_penalty = configuration.pop('penalty') |
| 30 | |
| 31 | self.start_mono = 0 |
| 32 | self.start_real = 0 |
| 33 | self.retries = 0 |
| 34 | self.penalty = 0 |
| 35 | self.elapsed = 0 |
| 36 | self.prev_update = 0 |
| 37 | |
| 38 | self.runs = 1 |
| 39 | |
| 40 | def calc_next(self): |
| 41 | self.start_mono = monotonic() |
| 42 | return self.start_mono - (self.start_mono % self.update_every) + self.update_every + self.penalty |
| 43 | |
| 44 | def sleep_until_next(self): |
| 45 | next_time = self.calc_next() |
| 46 | while self.start_mono < next_time: |
| 47 | sleep(next_time - self.start_mono) |
| 48 | self.start_mono = monotonic() |
| 49 | self.start_real = time() |
| 50 | |
| 51 | def handle_retries(self): |
| 52 | self.retries += 1 |
| 53 | if self.do_penalty and self.retries % PENALTY_EVERY == 0: |
| 54 | self.penalty = round(min(self.retries * self.update_every / 2, MAX_PENALTY)) |
| 55 | |
| 56 | |
| 57 | def clean_module_name(name): |
| 58 | if name.startswith('pythond_'): |
| 59 | return name[8:] |
| 60 | return name |
| 61 | |
| 62 | |
| 63 | class SimpleService(PythonDLogger, object): |
| 64 | """ |
| 65 | Prototype of Service class. |
| 66 | Implemented basic functionality to run jobs by `python.d.plugin` |
| 67 | """ |
| 68 | |
| 69 | def __init__(self, configuration, name=''): |
| 70 | """ |
| 71 | :param configuration: <dict> |
| 72 | :param name: <str> |
| 73 | """ |
| 74 | PythonDLogger.__init__(self) |
| 75 | self.configuration = configuration |
| 76 | self.order = list() |
| 77 | self.definitions = dict() |
| 78 | |
| 79 | self.module_name = clean_module_name(self.__module__) |
| 80 | self.job_name = configuration.pop('job_name') |
| 81 | self.actual_job_name = self.job_name or self.module_name |
| 82 | self.override_name = configuration.pop('override_name') |
| 83 | self.fake_name = None |
| 84 | |
| 85 | self._runtime_counters = RuntimeCounters(configuration=configuration) |
| 86 | self.charts = Charts(job_name=self.actual_name, |
| 87 | actual_job_name=self.actual_job_name, |
| 88 | priority=configuration.pop('priority'), |
| 89 | cleanup=configuration.pop('chart_cleanup'), |
| 90 | get_update_every=self.get_update_every, |
| 91 | module_name=self.module_name) |
| 92 | |
| 93 | def __repr__(self): |
| 94 | return '<{cls_bases}: {name}>'.format(cls_bases=', '.join(c.__name__ for c in self.__class__.__bases__), |
| 95 | name=self.name) |
| 96 | |
| 97 | @property |
| 98 | def name(self): |
| 99 | name = self.override_name or self.job_name |
| 100 | if name and name != self.module_name: |
| 101 | return '_'.join([self.module_name, name]) |
| 102 | return self.module_name |
| 103 | |
| 104 | def actual_name(self): |
| 105 | return self.fake_name or self.name |
| 106 | |
| 107 | @property |
| 108 | def runs_counter(self): |
| 109 | return self._runtime_counters.runs |
| 110 | |
| 111 | @property |
| 112 | def update_every(self): |
| 113 | return self._runtime_counters.update_every |
| 114 | |
| 115 | @update_every.setter |
| 116 | def update_every(self, value): |
| 117 | """ |
| 118 | :param value: <int> |
| 119 | :return: |
| 120 | """ |
| 121 | self._runtime_counters.update_every = value |
| 122 | |
| 123 | def get_update_every(self): |
| 124 | return self.update_every |
| 125 | |
| 126 | def check(self): |
| 127 | """ |
| 128 | check() prototype |
| 129 | :return: boolean |
| 130 | """ |
| 131 | self.debug("job doesn't implement check() method. Using default which simply invokes get_data().") |
| 132 | data = self.get_data() |
| 133 | if data and isinstance(data, dict): |
| 134 | return True |
| 135 | self.debug('returned value is wrong: {0}'.format(data)) |
| 136 | return False |
| 137 | |
| 138 | @create_runtime_chart |
| 139 | def create(self): |
| 140 | for chart_name in self.order: |
| 141 | chart_config = self.definitions.get(chart_name) |
| 142 | |
| 143 | if not chart_config: |
| 144 | self.debug("create() => [NOT ADDED] chart '{chart_name}' not in definitions. " |
| 145 | "Skipping it.".format(chart_name=chart_name)) |
| 146 | continue |
| 147 | |
| 148 | # create chart |
| 149 | chart_params = [chart_name] + chart_config['options'] |
| 150 | try: |
| 151 | self.charts.add_chart(params=chart_params) |
| 152 | except ChartError as error: |
| 153 | self.error("create() => [NOT ADDED] (chart '{chart}': {error})".format(chart=chart_name, |
| 154 | error=error)) |
| 155 | continue |
| 156 | |
| 157 | # add dimensions to chart |
| 158 | for dimension in chart_config['lines']: |
| 159 | try: |
| 160 | self.charts[chart_name].add_dimension(dimension) |
| 161 | except ChartError as error: |
| 162 | self.error("create() => [NOT ADDED] (dimension '{dimension}': {error})".format(dimension=dimension, |
| 163 | error=error)) |
| 164 | continue |
| 165 | |
| 166 | # add variables to chart |
| 167 | if 'variables' in chart_config: |
| 168 | for variable in chart_config['variables']: |
| 169 | try: |
| 170 | self.charts[chart_name].add_variable(variable) |
| 171 | except ChartError as error: |
| 172 | self.error("create() => [NOT ADDED] (variable '{var}': {error})".format(var=variable, |
| 173 | error=error)) |
| 174 | continue |
| 175 | |
| 176 | del self.order |
| 177 | del self.definitions |
| 178 | |
| 179 | # True if job has at least 1 chart else False |
| 180 | return bool(self.charts) |
| 181 | |
| 182 | def run(self): |
| 183 | """ |
| 184 | Runs job in thread. Handles retries. |
| 185 | Exits when job failed or timed out. |
| 186 | :return: None |
| 187 | """ |
| 188 | job = self._runtime_counters |
| 189 | self.debug('started, update frequency: {freq}'.format(freq=job.update_every)) |
| 190 | |
| 191 | while True: |
| 192 | job.sleep_until_next() |
| 193 | |
| 194 | since = 0 |
| 195 | if job.prev_update: |
| 196 | since = int((job.start_real - job.prev_update) * 1e6) |
| 197 | |
| 198 | try: |
| 199 | updated = self.update(interval=since) |
| 200 | except Exception as error: |
| 201 | self.error('update() unhandled exception: {error}'.format(error=error)) |
| 202 | updated = False |
| 203 | |
| 204 | job.runs += 1 |
| 205 | |
| 206 | if not updated: |
| 207 | job.handle_retries() |
| 208 | else: |
| 209 | job.elapsed = int((monotonic() - job.start_mono) * 1e3) |
| 210 | job.prev_update = job.start_real |
| 211 | job.retries, job.penalty = 0, 0 |
| 212 | if not ND_INTERNAL_MONITORING_DISABLED: |
| 213 | safe_print(RUNTIME_CHART_UPDATE.format(job_name=self.name, |
| 214 | since_last=since, |
| 215 | elapsed=job.elapsed)) |
| 216 | self.debug('update => [{status}] (elapsed time: {elapsed}, failed retries in a row: {retries})'.format( |
| 217 | status='OK' if updated else 'FAILED', |
| 218 | elapsed=job.elapsed if updated else '-', |
| 219 | retries=job.retries)) |
| 220 | |
| 221 | def update(self, interval): |
| 222 | """ |
| 223 | :return: |
| 224 | """ |
| 225 | data = self.get_data() |
| 226 | if not data: |
| 227 | self.debug('get_data() returned no data') |
| 228 | return False |
| 229 | elif not isinstance(data, dict): |
| 230 | self.debug('get_data() returned incorrect type data') |
| 231 | return False |
| 232 | |
| 233 | updated = False |
| 234 | |
| 235 | for chart in self.charts: |
| 236 | if chart.flags.obsoleted: |
| 237 | if chart.can_be_updated(data): |
| 238 | chart.refresh() |
| 239 | else: |
| 240 | continue |
| 241 | elif self.charts.cleanup and chart.penalty >= self.charts.cleanup: |
| 242 | chart.obsolete() |
| 243 | self.info("chart '{0}' was suppressed due to non updating".format(chart.name)) |
| 244 | continue |
| 245 | |
| 246 | ok = chart.update(data, interval) |
| 247 | if ok: |
| 248 | updated = True |
| 249 | |
| 250 | if not updated: |
| 251 | self.debug('none of the charts has been updated') |
| 252 | |
| 253 | return updated |
| 254 | |
| 255 | def get_data(self): |
| 256 | return self._get_data() |
| 257 | |
| 258 | def _get_data(self): |
| 259 | raise NotImplementedError |