091bc7394 · cryptotaxi247/netdata-1

@@ -1,9 +1,24 @@

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

2 -# Description: prototypes for netdata python.d modules

2 +# Description: netdata python modules framework

3 # Author: Pawel Krupa (paulfantom)

4

5 +# Remember:

6 +# ALL CODE NEEDS TO BE COMPATIBLE WITH Python > 2.7 and Python > 3.1

7 +# Follow PEP8 as much as it is possible

8 +# "check" and "create" CANNOT be blocking.

9 +# "update" CAN be blocking

10 +# "update" function needs to be fast, so follow:

11 +# https://wiki.python.org/moin/PythonSpeed/PerformanceTips

12 +# basically:

13 +# - use local variables wherever it is possible

14 +# - avoid dots in expressions that are executed many times

15 +# - use "join()" instead of "+"

16 +# - use "import" only at the beginning

17 +#

18 +# using ".encode()" in one thread can block other threads as well (only in python2)

19 +

20 import time

6 -import sys

21 +# import sys

22 import os

23 import socket

24 import select

@@ -18,7 +33,8 @@ import threading

33 import msg

34

35

21 -class BaseService(threading.Thread):

36 +# class BaseService(threading.Thread):

37 +class SimpleService(threading.Thread):

38 """

39 Prototype of Service class.

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

@@ -42,6 +58,9 @@ class BaseService(threading.Thread):

58 self._dimensions = []

59 self._charts = []

60 self.__chart_set = False

61 + self.__first_run = True

62 + self.order = []

63 + self.definitions = {}

64 if configuration is None:

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

66 raise RuntimeError

@@ -50,6 +69,8 @@ class BaseService(threading.Thread):

69 self.timetable = {}

70 self.create_timetable()

71

72 + # --- BASIC SERVICE CONFIGURATION ---

73 +

74 def _extract_base_config(self, config):

75 """

76 Get basic parameters to run service

@@ -59,13 +80,14 @@ class BaseService(threading.Thread):

80 'retries':0}

81 :param config: dict

82 """

83 + pop = config.pop

84 try:

63 - self.override_name = config.pop('name')

85 + self.override_name = pop('name')

86 except KeyError:

87 pass

66 - self.update_every = int(config.pop('update_every'))

67 - self.priority = int(config.pop('priority'))

68 - self.retries = int(config.pop('retries'))

88 + self.update_every = int(pop('update_every'))

89 + self.priority = int(pop('priority'))

90 + self.retries = int(pop('retries'))

91 self.retries_left = self.retries

92 self.configuration = config

93

@@ -86,6 +108,8 @@ class BaseService(threading.Thread):

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

109 'freq': freq}

110

111 + # --- THREAD CONFIGURATION ---

112 +

113 def _run_once(self):

114 """

115 Executes self.update(interval) and draws run time chart.

@@ -93,39 +117,36 @@ class BaseService(threading.Thread):

117 :return: boolean

118 """

119 t_start = time.time()

120 + timetable = self.timetable

121 + chart_name = self.chart_name

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

97 - if self.timetable['next'] > t_start:

98 - #msg.debug(self.chart_name + " will be run in " +

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

100 - msg.debug(self.chart_name, "will be run in", str(int((self.timetable['next'] - t_start) * 1000)), "ms")

123 + if timetable['next'] > t_start:

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

125 return True

126

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

104 - #msg.debug(self.chart_name +

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

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

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

108 - msg.debug(self.chart_name,

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

110 - "ms (update_every:", str(self.timetable['freq'] * 1000),

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

127 + since_last = int((t_start - timetable['last']) * 1000000)

128 + self.debug(chart_name,

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

130 + "ms (update_every:", str(timetable['freq'] * 1000),

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

132 + if self.__first_run:

133 + since_last = 0

134 if not self.update(since_last):

135 self.error("update function failed.")

136 return False

137 t_end = time.time()

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

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

139 # draw performance graph

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

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

120 - #run_time_chart += "SET run_time = " + run_time + '\n'

121 - #run_time_chart += "END\n"

122 - #sys.stdout.write(run_time_chart)

123 - sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" % \

124 - (self.chart_name, str(since_last), run_time))

125 -

126 - #msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")

127 - msg.debug(self.chart_name, "updated in", str(run_time), "ms")

141 + # noinspection SqlNoDataSourceInspection

142 + print("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" %

143 + (self.chart_name, str(since_last), run_time))

144 + # sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" %

145 + # (self.chart_name, str(since_last), run_time))

146 +

147 + self.debug(chart_name, "updated in", str(run_time), "ms")

148 self.timetable['last'] = t_start

149 + self.__first_run = False

150 return True

151

152 def run(self):

@@ -135,25 +156,32 @@ class BaseService(threading.Thread):

156 :return: None

157 """

158 self.timetable['last'] = time.time()

138 - while True:

159 + while True: # run forever, unless something is wrong

160 try:

161 status = self._run_once()

162 except Exception as e:

142 - msg.error("Something wrong: ", str(e))

163 + self.error("Something wrong: ", str(e))

164 return

144 - if status:

165 + if status: # handle retries if update failed

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

167 self.retries_left = self.retries

168 else:

169 self.retries_left -= 1

170 if self.retries_left <= 0:

150 - msg.error("no more retries. Exiting")

171 + self.error("no more retries. Exiting")

172 return

173 else:

174 time.sleep(self.timetable['freq'])

175

176 + # --- CHART ---

177 +

178 @staticmethod

179 def _format(*args):

180 + """

181 + Escape and convert passed arguments.

182 + :param args: anything

183 + :return: list

184 + """

185 params = []

186 append = params.append

187 for p in args:

@@ -170,28 +198,14 @@ class BaseService(threading.Thread):

198 def _line(self, instruction, *params):

199 """

200 Converts *params to string and joins them with one space between every one.

201 + Result is appended to self._data_stream

202 :param params: str/int/float

203 """

175 - #self._data_stream += instruction

204 tmp = list(map((lambda x: "''" if x is None or len(x) == 0 else x), params))

177 -

205 self._data_stream += "%s %s\n" % (instruction, str(" ".join(tmp)))

206

180 - # self.error(str(" ".join(tmp)))

181 - # for p in params:

182 - # if p is None:

183 - # p = ""

184 - # else:

185 - # p = str(p)

186 - # if len(p) == 0:

187 - # p = "''"

188 - # if ' ' in p:

189 - # p = "'" + p + "'"

190 - # self._data_stream += " " + p

191 - #self._data_stream += "\n"

192 -

207 def chart(self, type_id, name="", title="", units="", family="",

194 - category="", charttype="line", priority="", update_every=""):

208 + category="", chart_type="line", priority="", update_every=""):

209 """

210 Defines a new chart.

211 :param type_id: str

@@ -200,14 +214,13 @@ class BaseService(threading.Thread):

214 :param units: str

215 :param family: str

216 :param category: str

203 - :param charttype: str

217 + :param chart_type: str

218 :param priority: int/str

219 :param update_every: int/str

220 """

221 self._charts.append(type_id)

208 - #self._line("CHART", type_id, name, title, units, family, category, charttype, priority, update_every)

222

210 - p = self._format(type_id, name, title, units, family, category, charttype, priority, update_every)

223 + p = self._format(type_id, name, title, units, family, category, chart_type, priority, update_every)

224 self._line("CHART", *p)

225

226 def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):

@@ -239,10 +252,8 @@ class BaseService(threading.Thread):

252 self._dimensions.append(str(id))

253 if hidden:

254 p = self._format(id, name, algorithm, multiplier, divisor, "hidden")

242 - #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor), "hidden")

255 else:

256 p = self._format(id, name, algorithm, multiplier, divisor)

245 - #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor))

257

258 self._line("DIMENSION", *p)

259

@@ -294,11 +305,13 @@ class BaseService(threading.Thread):

305

306 def commit(self):

307 """

297 - Upload new data to netdata

308 + Upload new data to netdata.

309 """

310 print(self._data_stream)

311 self._data_stream = ""

312

313 + # --- ERROR HANDLING ---

314 +

315 def error(self, *params):

316 """

317 Show error message on stderr

@@ -317,37 +330,7 @@ class BaseService(threading.Thread):

330 """

331 msg.info(self.chart_name, *params)

332

320 - def check(self):

321 - """

322 - check() prototype

323 - :return: boolean

324 - """

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

326 - return False

327 -

328 - def create(self):

329 - """

330 - create() prototype

331 - :return: boolean

332 - """

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

334 - return False

335 -

336 - def update(self, interval):

337 - """

338 - update() prototype

339 - :param interval: int

340 - :return: boolean

341 - """

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

343 - return False

344 -

345 -

346 -class SimpleService(BaseService):

347 - def __init__(self, configuration=None, name=None):

348 - self.order = []

349 - self.definitions = {}

350 - BaseService.__init__(self, configuration=configuration, name=name)

333 + # --- MAIN METHODS ---

334

335 def _get_data(self):

336 """

@@ -358,9 +341,14 @@ class SimpleService(BaseService):

341

342 def check(self):

343 """

361 - :return:

344 + check() prototype

345 + :return: boolean

346 """

363 - return True

347 + self.debug("Module", str(self.__module__), "doesn't implement check() function. Using default.")

348 + if self._get_data() is None or len(self._get_data()) == 0:

349 + return False

350 + else:

351 + return True

352

353 def create(self):

354 """

@@ -414,17 +402,45 @@ class SimpleService(BaseService):

402

403

404 class UrlService(SimpleService):

405 + # TODO add support for https connections

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

407 self.url = ""

408 self.user = None

409 self.password = None

410 + self.proxies = {}

411 SimpleService.__init__(self, configuration=configuration, name=name)

412

423 - def __add_auth(self):

424 - passman = urllib2.HTTPPasswordMgrWithDefaultRealm()

425 - passman.add_password(None, self.url, self.user, self.password)

426 - authhandler = urllib2.HTTPBasicAuthHandler(passman)

427 - opener = urllib2.build_opener(authhandler)

413 + def __add_openers(self):

414 + # TODO add error handling

415 + opener = urllib2.build_opener()

416 +

417 + # Proxy handling

418 + # TODO currently self.proxies isn't parsed from configuration file

419 + if len(self.proxies) > 0:

420 + for proxy in self.proxies:

421 + url = proxy['url']

422 + # TODO test this:

423 + if "user" in proxy and "pass" in proxy:

424 + if url.lower().startswith('https://'):

425 + url = 'https://' + proxy['user'] + ':' + proxy['pass'] + '@' + url[8:]

426 + else:

427 + url = 'http://' + proxy['user'] + ':' + proxy['pass'] + '@' + url[7:]

428 + # FIXME move proxy auth to sth like this:

429 + # passman = urllib2.HTTPPasswordMgrWithDefaultRealm()

430 + # passman.add_password(None, url, proxy['user'], proxy['password'])

431 + # opener.add_handler(urllib2.HTTPBasicAuthHandler(passman))

432 +

433 + if url.lower().startswith('https://'):

434 + opener.add_handler(urllib2.ProxyHandler({'https': url}))

435 + else:

436 + opener.add_handler(urllib2.ProxyHandler({'https': url}))

437 +

438 + # HTTP Basic Auth

439 + if self.user is not None and self.password is not None:

440 + passman = urllib2.HTTPPasswordMgrWithDefaultRealm()

441 + passman.add_password(None, self.url, self.user, self.password)

442 + opener.add_handler(urllib2.HTTPBasicAuthHandler(passman))

443 +

444 urllib2.install_opener(opener)

445

446 def _get_raw_data(self):

@@ -434,7 +450,7 @@ class UrlService(SimpleService):

450 """

451 raw = None

452 try:

437 - f = urllib2.urlopen(self.url, timeout=self.update_every)

453 + f = urllib2.urlopen(self.url, timeout=self.update_every * 2)

454 except Exception as e:

455 self.error(str(e))

456 return None

@@ -470,13 +486,12 @@ class UrlService(SimpleService):

486 except (KeyError, TypeError):

487 pass

488

473 - if self.user is not None and self.password is not None:

474 - self.__add_auth()

489 + self.__add_openers()

490

476 - if self._get_data() is not None:

477 - return True

478 - else:

491 + if self._get_data() is None or len(self._get_data()) == 0:

492 return False

493 + else:

494 + return True

495

496

497 class SocketService(SimpleService):

@@ -502,14 +517,17 @@ class SocketService(SimpleService):

517 # establish ipv6 or ipv4 connection.

518 for res in socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM):

519 try:

520 + # noinspection SpellCheckingInspection

521 af, socktype, proto, canonname, sa = res

522 self._sock = socket.socket(af, socktype, proto)

507 - except socket.error as msg:

523 + except socket.error as e:

524 + self.debug("Cannot create socket:", str(e))

525 self._sock = None

526 continue

527 try:

528 self._sock.connect(sa)

512 - except socket.error as msg:

529 + except socket.error as e:

530 + self.debug("Cannot connect to socket:", str(e))

531 self._disconnect()

532 continue

533 self.__socket_config = res

@@ -520,7 +538,8 @@ class SocketService(SimpleService):

538 af, socktype, proto, canonname, sa = self.__socket_config

539 self._sock = socket.socket(af, socktype, proto)

540 self._sock.connect(sa)

523 - except socket.error as msg:

541 + except socket.error as e:

542 + self.debug("Cannot create or connect to socket:", str(e))

543 self._disconnect()

544 else:

545 # connect to unix socket

@@ -665,7 +684,7 @@ class LogService(SimpleService):

684 self._last_position = 0

685 elif os.path.getsize(self.log_path) == self._last_position:

686 self.debug("Log file hasn't changed. No new data.")

668 - return None

687 + return False

688 with open(self.log_path, "r") as fp:

689 fp.seek(self._last_position)

690 for i, line in enumerate(fp):

@@ -701,13 +720,13 @@ class LogService(SimpleService):

720 return False

721

722 def create(self):

723 + self._last_position = os.path.getsize(self.log_path)

724 status = SimpleService.create(self)

705 - self._last_position = 0

725 + # self._last_position = 0

726 return status

727

728

729 class ExecutableService(SimpleService):

710 - #command_whitelist = ['exim', 'postqueue']

730 bad_substrings = ('&', '|', ';', '>', '<')

731

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

@@ -748,9 +767,6 @@ class ExecutableService(SimpleService):

767 except (KeyError, TypeError):

768 self.error("No command specified. Using: '" + self.command + "'")

769 self.command = self.command.split(' ')

751 - #if self.command[0] not in self.command_whitelist:

752 - # self.error("Command is not whitelisted.")

753 - # return False

770

771 for arg in self.command[1:]:

772 if any(st in arg for st in self.bad_substrings):

@@ -763,8 +779,6 @@ class ExecutableService(SimpleService):

779 self.command[0] = prefix + base

780 if os.path.isfile(self.command[0]):

781 break

766 - #if self._get_raw_data() is not None:

767 - # break

782

783 if self._get_data() is None or len(self._get_data()) == 0:

784 self.error("Command", self.command, "returned no data")

code cleanup + bug fixes