@cryptotaxi247 / netdata-1 / commits / dc36d60a7

multithreading

paulfantom committed Jun 23, 2016 at 11:21 UTC dc36d60a7da55e6c47bf802e3a2df28f7043ad87

5 files changed +192 -200

plugins.d/python.d.plugin

+59 -172

@@ -7,6 +7,7 @@

7 import os

8 import sys

9 import time

10 +import threading

12 # -----------------------------------------------------------------------------

13 # globals & environment setup

@@ -17,7 +18,7 @@ BASE_CONFIG = {'update_every': os.getenv('NETDATA_UPDATE_EVERY', 1),

18 'retries': 10}

20 MODULES_DIR = os.path.abspath(os.getenv('NETDATA_PLUGINS_DIR',

20 - os.path.dirname(__file__)) + "/../python.d") + "/"

21 + os.path.dirname(__file__)) + "/../python.d") + "/"

22 CONFIG_DIR = os.getenv('NETDATA_CONFIG_DIR', "/etc/netdata/")

23 # directories should end with '/'

24 if CONFIG_DIR[-1] != "/":

@@ -28,60 +29,10 @@ PROGRAM = os.path.basename(__file__).replace(".plugin", "")

29 DEBUG_FLAG = False

30 OVERRIDE_UPDATE_EVERY = False

31 -

32 # -----------------------------------------------------------------------------

33 -# logging functions

34 -def log_msg(msg_type, *args):

35 - """

36 - Print message on stderr.

37 - :param msg_type: str

38 - """

39 - sys.stderr.write(PROGRAM)

40 - sys.stderr.write(" ")

41 - sys.stderr.write(msg_type)

42 - sys.stderr.write(": ")

43 - for i in args:

44 - sys.stderr.write(" ")

45 - sys.stderr.write(str(i))

46 - sys.stderr.write("\n")

47 - sys.stderr.flush()

48 -

49 -

50 -def debug(*args):

51 - """

52 - Print debug message on stderr.

53 - """

54 - if not DEBUG_FLAG:

55 - return

56 -

57 - log_msg("DEBUG", *args)

58 -

59 -

60 -def error(*args):

61 - """

62 - Print message on stderr.

63 - """

64 - log_msg("ERROR", *args)

65 -

66 -

67 -def info(*args):

68 - """

69 - Print message on stderr.

70 - """

71 - log_msg("INFO", *args)

72 -

73 -

74 -def fatal(*args):

75 - """

76 - Print message on stderr and exit.

77 - """

78 - log_msg("FATAL", *args)

79 - sys.stdout.write('DISABLE\n')

80 - sys.exit(1)

33 +# custom, third party and version specific python modules management

34 +import msg

82 -

83 -# -----------------------------------------------------------------------------

84 -# third party and version specific python modules management

36 try:

37 assert sys.version_info >= (3, 1)

38 import importlib.machinery

@@ -90,7 +41,7 @@ try:

41 # import builtins

42 # builtins.PY_VERSION = 3

43 PY_VERSION = 3

93 - info('Using python v3')

44 + msg.info('Using python v3')

45 except (AssertionError, ImportError):

46 try:

47 import imp

@@ -99,19 +50,20 @@ except (AssertionError, ImportError):

50 # import __builtin__

51 # __builtin__.PY_VERSION = 2

52 PY_VERSION = 2

102 - info('Using python v2')

53 + msg.info('Using python v2')

54 except ImportError:

104 - fatal('Cannot start. No importlib.machinery on python3 or lack of imp on python2')

55 + msg.fatal('Cannot start. No importlib.machinery on python3 or lack of imp on python2')

56 try:

57 import yaml

58 except ImportError:

108 - fatal('Cannot find yaml library')

59 + msg.fatal('Cannot find yaml library')

62 class PythonCharts(object):

63 """

64 Main class used to control every python module.

65 """

66 +

67 def __init__(self,

68 modules=None,

69 modules_path='../python.d/',

@@ -167,7 +119,7 @@ class PythonCharts(object):

119 else:

120 return imp.load_source(name, path)

121 except Exception as e:

170 - error("Problem loading", name, str(e))

122 + msg.error("Problem loading", name, str(e))

123 return None

124

125 def _load_modules(self, path, modules, disabled):

@@ -181,7 +133,7 @@ class PythonCharts(object):

133

134 # check if plugin directory exists

135 if not os.path.isdir(path):

184 - fatal("cannot find charts directory ", path)

136 + msg.fatal("cannot find charts directory ", path)

137

138 # load modules

139 loaded = []

@@ -193,17 +145,17 @@ class PythonCharts(object):

145 if mod is not None:

146 loaded.append(mod)

147 else: # exit if plugin is not found

196 - fatal('no modules found.')

148 + msg.fatal('no modules found.')

149 else:

150 # scan directory specified in path and load all modules from there

151 names = os.listdir(path)

152 for mod in names:

153 if mod.replace(MODULE_EXTENSION, "") in disabled:

202 - error(mod + ": disabled module ", mod.replace(MODULE_EXTENSION, ""))

154 + msg.error(mod + ": disabled module ", mod.replace(MODULE_EXTENSION, ""))

155 continue

156 m = self._import_module(path + mod)

157 if m is not None:

206 - debug(mod + ": loading module '" + path + mod + "'")

158 + msg.debug(mod + ": loading module '" + path + mod + "'")

159 loaded.append(m)

160 return loaded

161

@@ -218,7 +170,7 @@ class PythonCharts(object):

170 for mod in modules:

171 configfile = self.configs + mod.__name__ + ".conf"

172 if os.path.isfile(configfile):

221 - debug(mod.__name__ + ": loading module configuration: '" + configfile + "'")

173 + msg.debug(mod.__name__ + ": loading module configuration: '" + configfile + "'")

174 try:

175 if not hasattr(mod, 'config'):

176 mod.config = {}

@@ -226,12 +178,12 @@ class PythonCharts(object):

178 'config',

179 self._parse_config(mod, read_config(configfile)))

180 except Exception as e:

229 - error(mod.__name__ + ": cannot parse configuration file '" + configfile + "':", str(e))

181 + msg.error(mod.__name__ + ": cannot parse configuration file '" + configfile + "':", str(e))

182 else:

231 - error(mod.__name__ + ": configuration file '" + configfile + "' not found. Using defaults.")

183 + msg.error(mod.__name__ + ": configuration file '" + configfile + "' not found. Using defaults.")

184 # set config if not found

185 if not hasattr(mod, 'config'):

234 - debug(mod.__name__ + ": setting configuration for only one job")

186 + msg.debug(mod.__name__ + ": setting configuration for only one job")

187 mod.config = {None: {}}

188 for var in BASE_CONFIG:

189 try:

@@ -256,7 +208,7 @@ class PythonCharts(object):

208 """

209 # get default values

210 defaults = {}

259 - debug(module.__name__ + ": reading configuration")

211 + msg.debug(module.__name__ + ": reading configuration")

212 for key in BASE_CONFIG:

213 try:

214 # get defaults from module config

@@ -305,10 +257,10 @@ class PythonCharts(object):

257 try:

258 job = module.Service(configuration=conf, name=name)

259 except Exception as e:

308 - error(module.__name__ +

309 - ("/" + str(name) if name is not None else "") +

310 - ": cannot start job: '" +

311 - str(e))

260 + msg.error(module.__name__ +

261 + ("/" + str(name) if name is not None else "") +

262 + ": cannot start job: '" +

263 + str(e))

264 return None

265 else:

266 # set chart_name (needed to plot run time graphs)

@@ -316,7 +268,7 @@ class PythonCharts(object):

268 if name is not None:

269 job.chart_name += "_" + name

270 jobs.append(job)

319 - debug(module.__name__ + ("/" + str(name) if name is not None else "") + ": job added")

271 + msg.debug(module.__name__ + ("/" + str(name) if name is not None else "") + ": job added")

272

273 return [j for j in jobs if j is not None]

274

@@ -336,22 +288,22 @@ class PythonCharts(object):

288 if reason is None:

289 return

290 elif reason[:3] == "no ":

339 - error(prefix +

340 - "does not seem to have " +

341 - reason[3:] +

342 - "() function. Disabling it.")

291 + msg.error(prefix +

292 + "does not seem to have " +

293 + reason[3:] +

294 + "() function. Disabling it.")

295 elif reason[:7] == "failed ":

344 - error(prefix +

345 - reason[7:] +

346 - "() function reports failure.")

296 + msg.error(prefix +

297 + reason[7:] +

298 + "() function reports failure.")

299 elif reason[:13] == "configuration":

348 - error(prefix +

349 - "configuration file '" +

350 - self.configs +

351 - job.__module__ +

352 - ".conf' not found. Using defaults.")

300 + msg.error(prefix +

301 + "configuration file '" +

302 + self.configs +

303 + job.__module__ +

304 + ".conf' not found. Using defaults.")

305 elif reason[:11] == "misbehaving":

354 - error(prefix + "is " + reason)

306 + msg.error(prefix + "is " + reason)

307

308 def check(self):

309 """

@@ -366,7 +318,7 @@ class PythonCharts(object):

318 if not job.check():

319 self._stop(job, "failed check")

320 else:

369 - debug(job.chart_name, ": check succeeded")

321 + msg.debug(job.chart_name, ": check succeeded")

322 i += 1

323 except AttributeError:

324 self._stop(job, "no check")

@@ -397,7 +349,7 @@ class PythonCharts(object):

349 str(job.timetable['freq']) +

350 '\n')

351 sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")

400 - debug("created charts for", job.chart_name)

352 + msg.debug("created charts for", job.chart_name)

353 # sys.stdout.flush()

354 i += 1

355 except AttributeError:

@@ -405,82 +357,14 @@ class PythonCharts(object):

357 except (UnboundLocalError, Exception) as e:

358 self._stop(job, "misbehaving. Reason: " + str(e))

359

408 - def _update_job(self, job):

409 - """

410 - Tries to execute update() on specified job.

411 - This cannot fail thus it is catching every exception.

412 - If job.update() returns False, number of retries_left is decremented.

413 - If there are no more retries, job is stopped.

414 - Job is also stopped if it throws an exception.

415 - This is also updating job run time chart.

416 - Return False if job is stopped

417 - :param job: object

418 - :return: boolean

419 - """

420 - t_start = time.time()

421 - # check if it is time to execute job update() function

422 - if job.timetable['next'] > t_start:

423 - debug(job.chart_name + " will be run in " + str(int((job.timetable['next'] - t_start) * 1000)) + " ms")

424 - return True

425 - try:

426 - if self.first_run:

427 - since_last = 0

428 - else:

429 - since_last = int((t_start - job.timetable['last']) * 1000000)

430 - debug(job.chart_name +

431 - " ready to run, after " + str(int((t_start - job.timetable['last']) * 1000)) +

432 - " ms (update_every: " + str(job.timetable['freq'] * 1000) +

433 - " ms, latency: " + str(int((t_start - job.timetable['next']) * 1000)) + " ms)")

434 - if not job.update(since_last):

435 - if job.retries_left <= 0:

436 - self._stop(job, "update failed")

437 - return False

438 - job.retries_left -= 1

439 - job.timetable['next'] += job.timetable['freq']

440 - return True

441 - except AttributeError:

442 - self._stop(job, "no update")

443 - return False

444 - except (UnboundLocalError, Exception) as e:

445 - self._stop(job, "misbehaving. Reason: " + str(e))

446 - return False

447 - t_end = time.time()

448 - job.timetable['next'] = t_end - (t_end % job.timetable['freq']) + job.timetable['freq']

449 - # draw performance graph

450 - run_time = str(int((t_end - t_start) * 1000))

451 - debug(job.chart_name, "updated in", run_time, "ms")

452 - sys.stdout.write("BEGIN netdata.plugin_pythond_" + job.chart_name + " " + str(since_last) + '\n')

453 - sys.stdout.write("SET run_time = " + run_time + '\n')

454 - sys.stdout.write("END\n")

455 - # sys.stdout.flush()

456 - job.timetable['last'] = t_start

457 - job.retries_left = job.retries

458 - self.first_run = False

459 - return True

460 -

360 def update(self):

462 - """

463 - Tries to execute update() on every job by using _update_job()

464 - This will stay forever and ever and ever forever and ever it'll be the one...

465 - """

466 - self.first_run = True

361 + for job in self.jobs:

362 + job.start()

363 +

364 while True:

468 - next_runs = []

469 - i = 0

470 - while i < len(self.jobs):

471 - job = self.jobs[i]

472 - if self._update_job(job):

473 - try:

474 - next_runs.append(job.timetable['next'])

475 - except KeyError:

476 - pass

477 - i += 1

478 - if len(next_runs) == 0:

479 - fatal('no python.d modules loaded.')

480 - try:

481 - time.sleep(min(next_runs) - time.time())

482 - except IOError:

483 - pass

365 + if threading.active_count() <= 1:

366 + msg.fatal("no more jobs")

367 + time.sleep(1)

368

369

370 def read_config(path):

@@ -493,10 +377,10 @@ def read_config(path):

377 with open(path, 'r') as stream:

378 config = yaml.load(stream)

379 except (OSError, IOError):

496 - error(str(path), "is not a valid configuration file")

380 + msg.error(str(path), "is not a valid configuration file")

381 return None

382 except yaml.YAMLError as e:

499 - error(str(path), "is malformed:", e)

383 + msg.error(str(path), "is malformed:", e)

384 return None

385 return config

386

@@ -531,9 +415,9 @@ def parse_cmdline(directory, *commands):

415 pass

416 if changed_update and DEBUG_FLAG:

417 OVERRIDE_UPDATE_EVERY = True

534 - debug(PROGRAM, "overriding update interval to", str(BASE_CONFIG['update_every']))

418 + msg.debug(PROGRAM, "overriding update interval to", str(BASE_CONFIG['update_every']))

419

536 - debug("started from", commands[0], "with options:", *commands[1:])

420 + msg.debug("started from", commands[0], "with options:", *commands[1:])

421

422 return mods

423

@@ -548,14 +432,14 @@ def run():

432 # read configuration file

433 disabled = []

434 configfile = CONFIG_DIR + "python.d.conf"

551 - debug(PROGRAM, "reading configuration file:", configfile)

435 + msg.info(PROGRAM, "reading configuration file:", configfile)

436

437 conf = read_config(configfile)

438 if conf is not None:

439 try:

440 # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'

441 if conf['enabled'] is False:

558 - fatal('disabled in configuration file.\n')

442 + msg.fatal('disabled in configuration file.\n')

443 except (KeyError, TypeError):

444 pass

445 try:

@@ -573,19 +457,22 @@ def run():

457 if v is False:

458 disabled.append(k)

459

460 + DEBUG_FLAG = True

461 + msg.DEBUG_FLAG = DEBUG_FLAG

462 + msg.PROGRAM = PROGRAM

463 # parse passed command line arguments

464 modules = parse_cmdline(MODULES_DIR, *sys.argv)

578 - info("MODULES_DIR='" + MODULES_DIR +

579 - "', CONFIG_DIR='" + CONFIG_DIR +

580 - "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +

581 - ", ONLY_MODULES=" + str(modules))

465 + msg.info("MODULES_DIR='" + MODULES_DIR +

466 + "', CONFIG_DIR='" + CONFIG_DIR +

467 + "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +

468 + ", ONLY_MODULES=" + str(modules))

469

470 # run plugins

471 charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)

472 charts.check()

473 charts.create()

474 charts.update()

588 - fatal("finished")

475 + msg.fatal("finished")

476

477

478 if __name__ == '__main__':

python.d/apache.chart.py

+1 -2

@@ -112,8 +112,7 @@ class Service(UrlService):

112 if str(tmp[0]) in self.assignment:

113 try:

114 data[self.assignment[tmp[0]]] = int(float(tmp[1]))

115 - except (IndexError, ValueError) as a:

116 - print(a)

115 + except (IndexError, ValueError):

116 pass

117 if len(data) == 0:

118 return None

python.d/example.chart.py

+9 -8

@@ -13,6 +13,7 @@ update_every = 4

13 priority = 90000

14 retries = 7

16 +

17 class Service(BaseService):

18 def __init__(self, configuration=None, name=None):

19 super(self.__class__,self).__init__(configuration=configuration, name=name)

@@ -21,15 +22,15 @@ class Service(BaseService):

22 return True

24 def create(self):

24 - print("CHART example.python_random '' 'A random number' 'random number' random random line " +

25 - str(self.priority) +

26 - " " +

27 - str(self.update_every))

28 - print("DIMENSION random1 '' absolute 1 1")

25 + chart = "CHART example.python_random '' 'A random number' 'random number' random random line " + \

26 + str(self.priority) + " " + \

27 + str(self.update_every) + "\n" + \

28 + "DIMENSION random1 '' absolute 1 1"

29 + print(chart)

30 return True

32 def update(self, interval):

32 - print("BEGIN example.python_random "+str(interval))

33 - print("SET random1 = "+str(random.randint(0,100)))

34 - print("END")

33 + chart = "BEGIN example.python_random "+str(interval)+"\n"

34 + chart += "SET random1 = "+str(random.randint(0,100))+"\n"

35 + print(chart + "END")

36 return True

python.d/python_modules/base.py

+67 -18

@@ -2,25 +2,36 @@

2 # Description: prototypes for netdata python.d modules

3 # Author: Pawel Krupa (paulfantom)

5 -from time import time

5 +import time

6 import sys

7 try:

8 from urllib.request import urlopen

9 except ImportError:

10 from urllib2 import urlopen

12 +import threading

13 +import msg

13 -class BaseService(object):

15 +

16 +class BaseService(threading.Thread):

17 """

18 Prototype of Service class.

19 Implemented basic functionality to run jobs by `python.d.plugin`

20 """

21 + debugging = False

22 +

23 def __init__(self, configuration=None, name=None):

24 """

25 This needs to be initialized in child classes

26 :param configuration: dict

27 :param name: str

28 """

29 + threading.Thread.__init__(self)

30 + self.daemon = True

31 + self.retries = 0

32 + self.retries_left = 0

33 + self.priority = 140000

34 + self.update_every = 1

35 self.name = name

36 if configuration is None:

37 self.error("BaseService: no configuration parameters supplied. Cannot create Service.")

@@ -58,24 +69,64 @@ class BaseService(object):

69 """

70 if freq is None:

71 freq = self.update_every

61 - now = time()

72 + now = time.time()

73 self.timetable = {'last': now,

74 'next': now - (now % freq) + freq,

75 'freq': freq}

66 - @staticmethod

67 - def error(msg, exception=""):

68 - if exception != "":

69 - exception = " " + str(exception).replace("\n", " ")

70 - sys.stderr.write(str(msg)+exception+"\n")

71 - sys.stderr.flush()

77 + def _run_once(self):

78 + t_start = time.time()

79 + # check if it is time to execute job update() function

80 + if self.timetable['next'] > t_start:

81 + msg.debug(self.chart_name + " will be run in " +

82 + str(int((self.timetable['next'] - t_start) * 1000)) + " ms")

83 + return True

84 +

85 + since_last = int((t_start - self.timetable['last']) * 1000000)

86 + msg.debug(self.chart_name +

87 + " ready to run, after " + str(int((t_start - self.timetable['last']) * 1000)) +

88 + " ms (update_every: " + str(self.timetable['freq'] * 1000) +

89 + " ms, latency: " + str(int((t_start - self.timetable['next']) * 1000)) + " ms)")

90 + if not self.update(since_last):

91 + return False

92 + t_end = time.time()

93 + self.timetable['next'] = t_end - (t_end % self.timetable['freq']) + self.timetable['freq']

94 +

95 + # draw performance graph

96 + run_time = str(int((t_end - t_start) * 1000))

97 + run_time_chart = "BEGIN netdata.plugin_pythond_" + self.chart_name + " " + str(since_last) + '\n'

98 + run_time_chart += "SET run_time = " + run_time + '\n'

99 + run_time_chart += "END\n"

100 + sys.stdout.write(run_time_chart)

101 + msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")

102 + self.timetable['last'] = t_start

103 + return True

104 +

105 + def run(self):

106 + self.timetable['last'] = time.time()

107 + while True:

108 + try:

109 + status = self._run_once()

110 + except Exception as e:

111 + msg.error("Something wrong: " + str(e))

112 + return

113 + if status:

114 + time.sleep(self.timetable['next'] - time.time())

115 + self.retries_left = self.retries

116 + else:

117 + self.retries -= 1

118 + if self.retries_left <= 0:

119 + msg.error("no more retries. Exiting")

120 + return

121 + else:

122 + time.sleep(self.timetable['freq'])

123

124 def check(self):

125 """

126 check() prototype

127 :return: boolean

128 """

78 - self.error("Service " + str(self.__module__) + "doesn't implement check() function")

129 + msg.error("Service " + str(self.__module__) + "doesn't implement check() function")

130 return False

131

132 def create(self):

@@ -83,7 +134,7 @@ class BaseService(object):

134 create() prototype

135 :return: boolean

136 """

86 - self.error("Service " + str(self.__module__) + "doesn't implement create() function?")

137 + msg.error("Service " + str(self.__module__) + "doesn't implement create() function?")

138 return False

139

140 def update(self, interval):

@@ -92,7 +143,7 @@ class BaseService(object):

143 :param interval: int

144 :return: boolean

145 """

95 - self.error("Service " + str(self.__module__) + "doesn't implement update() function")

146 + msg.error("Service " + str(self.__module__) + "doesn't implement update() function")

147 return False

148

149

@@ -128,7 +179,7 @@ class UrlService(BaseService):

179 f = urlopen(self.url, timeout=self.update_every)

180 raw = f.read().decode('utf-8')

181 except Exception as e:

131 - self.error(self.__module__, str(e))

182 + msg.error(self.__module__, str(e))

183 finally:

184 try:

185 f.close()

@@ -148,7 +199,7 @@ class UrlService(BaseService):

199 Format configuration data and try to connect to server

200 :return: boolean

201 """

151 - if self.name is None:

202 + if self.name is None or self.name == str(None):

203 self.name = 'local'

204 else:

205 self.name = str(self.name)

@@ -193,8 +244,7 @@ class UrlService(BaseService):

244 content += "DIMENSION " + line['name'] + " " + line['options'] + "\n"

245

246 if len(content) > 0:

196 - print(header)

197 - print(content)

247 + print(header + "\n" + content)

248 idx += 1

249

250 if idx == 0:

@@ -220,7 +270,6 @@ class UrlService(BaseService):

270 except KeyError:

271 pass

272 if len(c) != 0:

223 - print(header + c)

224 - print("END")

273 + print(header + c + "\nEND")

274

275 return True

python.d/python_modules/msg.py new

+56

@@ -0,0 +1,56 @@

1 +# -*- coding: utf-8 -*-

2 +# Description: logging for netdata python.d modules

3 +

4 +import sys

5 +

6 +DEBUG_FLAG = False

7 +PROGRAM = ""

8 +

9 +

10 +def log_msg(msg_type, *args):

11 + """

12 + Print message on stderr.

13 + :param msg_type: str

14 + """

15 + sys.stderr.write(PROGRAM)

16 + sys.stderr.write(" ")

17 + sys.stderr.write(msg_type)

18 + sys.stderr.write(": ")

19 + for i in args:

20 + sys.stderr.write(" ")

21 + sys.stderr.write(str(i))

22 + sys.stderr.write("\n")

23 + sys.stderr.flush()

24 +

25 +

26 +def debug(*args):

27 + """

28 + Print debug message on stderr.

29 + """

30 + if not DEBUG_FLAG:

31 + return

32 +

33 + log_msg("DEBUG", *args)

34 +

35 +

36 +def error(*args):

37 + """

38 + Print message on stderr.

39 + """

40 + log_msg("ERROR", *args)

41 +

42 +

43 +def info(*args):

44 + """

45 + Print message on stderr.

46 + """

47 + log_msg("INFO", *args)

48 +

49 +

50 +def fatal(*args):

51 + """

52 + Print message on stderr and exit.

53 + """

54 + log_msg("FATAL", *args)

55 + sys.stdout.write('DISABLE\n')

56 + sys.exit(1)

\ No newline at end of file