From e7c51f2a80814b214ff8dccbd380b6227ec88fdb Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 09:25:23 -0400 Subject: [PATCH 01/82] add mysql lock for polling --- includes/dbFacile.mysql.php | 28 ++++++++++++++++++++++++++++ includes/dbFacile.mysqli.php | 27 +++++++++++++++++++++++++++ poller.php | 8 +++++--- 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/includes/dbFacile.mysql.php b/includes/dbFacile.mysql.php index 0d1d7150c..44dc5d494 100644 --- a/includes/dbFacile.mysql.php +++ b/includes/dbFacile.mysql.php @@ -59,6 +59,34 @@ function dbQuery($sql, $parameters=array()) { }//end dbQuery() +/* + * Aquire a lock on a string + * */ +function dbGetLock($data, $timeout = 0) { + $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; + echo "printing sql: "; + echo "$sql\n"; + $result = dbFetchCell($sql); + echo "printing result: "; + echo "$result\n"; + return $result; +} + + +/* + * Release a lock on a string + * */ +function dbReleaseLock($data, $timeout = 0) { + $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; + echo "printing sql: "; + echo "$sql\n"; + $result = dbFetchCell($sql); + echo "printing result: "; + echo "$result\n"; + return $result; +} + + /* * Passed an array and a table name, it attempts to insert the data into the table. * Check for boolean false to determine whether insert failed diff --git a/includes/dbFacile.mysqli.php b/includes/dbFacile.mysqli.php index 2408ca990..7b48ad6b9 100644 --- a/includes/dbFacile.mysqli.php +++ b/includes/dbFacile.mysqli.php @@ -59,6 +59,33 @@ function dbQuery($sql, $parameters=array()) { }//end dbQuery() +/* + * Aquire a lock on a string + * */ +function dbGetLock($data, $timeout = 0) { + $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; + echo "printing sql: "; + echo "$sql\n"; + $result = dbFetchCell($sql); + echo "printing result: "; + echo "$result\n"; + return $result; +} + + +/* + * Release a lock on a string + * */ +function dbReleaseLock($data, $timeout = 0) { + $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; + echo "printing sql: "; + echo "$sql\n"; + $result = dbFetchCell($sql); + echo "printing result: "; + echo "$result\n"; + return $result; +} + /* * Passed an array and a table name, it attempts to insert the data into the table. * Check for boolean false to determine whether insert failed diff --git a/poller.php b/poller.php index 6412c37be..8c2b32881 100755 --- a/poller.php +++ b/poller.php @@ -110,9 +110,11 @@ if (!isset($query)) { foreach (dbFetch($query) as $device) { $device = dbFetchRow("SELECT * FROM `devices` WHERE `device_id` = '".$device['device_id']."'"); - poll_device($device, $options); - RunRules($device['device_id']); - echo "\r\n"; + if (dbGetLock('polling.' . $device['device_id'])) { + poll_device($device, $options); + RunRules($device['device_id']); + echo "\r\n"; + } $polled_devices++; } From 8e34e83d9740bea782775077ff130c5e5960e758 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 09:59:24 -0400 Subject: [PATCH 02/82] add swp to gitignore for vim --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index d56e86250..5c73ae6be 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ nbproject .alerts.lock .ircbot.alert .metadata_never_index +*.swp From b34fcf872851e0ee530aeea929dc298364d5445a Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 10:19:05 -0400 Subject: [PATCH 03/82] remove debug echos --- includes/dbFacile.mysql.php | 11 +---------- includes/dbFacile.mysqli.php | 10 +--------- 2 files changed, 2 insertions(+), 19 deletions(-) diff --git a/includes/dbFacile.mysql.php b/includes/dbFacile.mysql.php index 44dc5d494..9fd82c698 100644 --- a/includes/dbFacile.mysql.php +++ b/includes/dbFacile.mysql.php @@ -64,11 +64,7 @@ function dbQuery($sql, $parameters=array()) { * */ function dbGetLock($data, $timeout = 0) { $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; - echo "printing sql: "; - echo "$sql\n"; $result = dbFetchCell($sql); - echo "printing result: "; - echo "$result\n"; return $result; } @@ -78,15 +74,10 @@ function dbGetLock($data, $timeout = 0) { * */ function dbReleaseLock($data, $timeout = 0) { $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; - echo "printing sql: "; - echo "$sql\n"; $result = dbFetchCell($sql); - echo "printing result: "; - echo "$result\n"; return $result; -} - + /* * Passed an array and a table name, it attempts to insert the data into the table. * Check for boolean false to determine whether insert failed diff --git a/includes/dbFacile.mysqli.php b/includes/dbFacile.mysqli.php index 7b48ad6b9..36b2208cb 100644 --- a/includes/dbFacile.mysqli.php +++ b/includes/dbFacile.mysqli.php @@ -64,11 +64,7 @@ function dbQuery($sql, $parameters=array()) { * */ function dbGetLock($data, $timeout = 0) { $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; - echo "printing sql: "; - echo "$sql\n"; $result = dbFetchCell($sql); - echo "printing result: "; - echo "$result\n"; return $result; } @@ -78,13 +74,9 @@ function dbGetLock($data, $timeout = 0) { * */ function dbReleaseLock($data, $timeout = 0) { $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; - echo "printing sql: "; - echo "$sql\n"; $result = dbFetchCell($sql); - echo "printing result: "; - echo "$result\n"; return $result; -} + /* * Passed an array and a table name, it attempts to insert the data into the table. From 7f262fd2cc3fcbd1b8daca8440d19fef493bab35 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 10:47:18 -0400 Subject: [PATCH 04/82] limit sql query to amount_of_workers --- poller-service.py | 202 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 poller-service.py diff --git a/poller-service.py b/poller-service.py new file mode 100644 index 000000000..291e95534 --- /dev/null +++ b/poller-service.py @@ -0,0 +1,202 @@ +""" + poller-service A service to wrap SNMP polling. It will poll up to $threads devices at a time, and will not re-poll + devices that have been polled within the last $frequency seconds. It will prioritize devices based on + the last time polled. If resources are sufficient, this service should poll every device every + $frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as + frequently as possible. This service is based on poller-wrapper.py. + + Author: Clint Armstrong + Date: July 2015 + + Usage: poller-service [threads] [frequency] + Default is 16 threads and 300 seconds. +""" + +import json +import os +import Queue +import subprocess +import sys +import threading +import time +import MySQLdb + +install_dir = os.path.dirname(os.path.realpath(__file__)) +config_file = ob_install_dir + '/config.php' + +def get_config_data(): + config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % ob_install_dir] + try: + proc = subprocess.Popen(config_cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE) + except: + print "ERROR: Could not execute: %s" % config_cmd + sys.exit(2) + return proc.communicate()[0] + +try: + with open(config_file) as f: + pass +except IOError as e: + print "ERROR: Oh dear... %s does not seem readable" % config_file + sys.exit(2) + +try: + config = json.loads(get_config_data()) +except: + print "ERROR: Could not load or parse configuration, are PATHs correct?" + sys.exit(2) + +poller_path = config['install_dir'] + '/poller.php' +db_username = config['db_user'] +db_password = config['db_pass'] + +if config['db_host'][:5].lower() == 'unix:': + db_server = config['db_host'] + db_port = 0 +elif ':' in config['db_host']: + db_server = config['db_host'].rsplit(':')[0] + db_port = int(config['db_host'].rsplit(':')[1]) +else: + db_server = config['db_host'] + db_port =0 + +db_dbname = config['db_name'] + +# (c) 2015, GPLv3, Daniel Preussker << 300: + print "WARNING: the process took more than 5 minutes to finish, you need faster hardware or more threads" + print "INFO: in sequential style polling the elapsed time would have been: %s seconds" % real_duration + for device in per_device_duration: + if per_device_duration[device] > 300: + print "WARNING: device %s is taking too long: %s seconds" % (device, per_device_duration[device]) + show_stopper = True + if show_stopper: + print "ERROR: Some devices are taking more than 300 seconds, the script cannot recommend you what to do." + else: + recommend = int(total_time / 300.0 * amount_of_workers + 1) + print "WARNING: Consider setting a minimum of %d threads. (This does not constitute professional advice!)" % recommend + + sys.exit(2) From 72844d112ea00975c29ec5efb2b98dd43ec21f8c Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 10:50:08 -0400 Subject: [PATCH 05/82] make poller-service executable --- poller-service.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 poller-service.py diff --git a/poller-service.py b/poller-service.py old mode 100644 new mode 100755 From 2ba1d9e0ece2a9c7a1446b179878cf61ddfa3e4e Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 10:52:32 -0400 Subject: [PATCH 06/82] add shebang --- poller-service.py | 1 + 1 file changed, 1 insertion(+) diff --git a/poller-service.py b/poller-service.py index 291e95534..0bdf2a08c 100755 --- a/poller-service.py +++ b/poller-service.py @@ -1,3 +1,4 @@ +#! /usr/bin/env python """ poller-service A service to wrap SNMP polling. It will poll up to $threads devices at a time, and will not re-poll devices that have been polled within the last $frequency seconds. It will prioritize devices based on From e479d8cf69fbacc48a70c1885ebea1059ceac7e3 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 10:52:59 -0400 Subject: [PATCH 07/82] change variables --- poller-service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index 0bdf2a08c..08d1a7754 100755 --- a/poller-service.py +++ b/poller-service.py @@ -23,10 +23,10 @@ import time import MySQLdb install_dir = os.path.dirname(os.path.realpath(__file__)) -config_file = ob_install_dir + '/config.php' +config_file = install_dir + '/config.php' def get_config_data(): - config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % ob_install_dir] + config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % install_dir] try: proc = subprocess.Popen(config_cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE) except: From bd054d7a81856796576ec406c0e921dc23700919 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 12:00:21 -0400 Subject: [PATCH 08/82] first attempt at looping logic --- poller-service.py | 86 ++++++++++++++++++++++++++++++----------------- 1 file changed, 55 insertions(+), 31 deletions(-) diff --git a/poller-service.py b/poller-service.py index 08d1a7754..82fc7ed4c 100755 --- a/poller-service.py +++ b/poller-service.py @@ -21,6 +21,7 @@ import sys import threading import time import MySQLdb +from datetime import datetime, timedelta install_dir = os.path.dirname(os.path.realpath(__file__)) config_file = install_dir + '/config.php' @@ -70,6 +71,11 @@ else: poller_group = False # EOC1 +s_time = time.time() +real_duration = 0 +per_device_duration = {} +polled_devices = 0 + # Take the amount of threads we want to run in parallel from the commandline # if None are given or the argument was garbage, fall back to default of 16 try: @@ -100,15 +106,6 @@ except: print "ERROR: Could not connect to MySQL database!" sys.exit(2) -# Query ordering based on last polled date, so that the device with the oldest data is polled first. -query = 'select device_id from devices {} disabled = 0 order by last_polled asc limit {}'.format( - 'where poller_group IN({}) and'.format(poller_group) if poller_group else '', - amount_of_workers) - -cursor.execute(query) -devices_list = cursor.fetchall() - - # A seperate queue and a single worker for printing information to the screen prevents # the good old joke: # @@ -118,6 +115,7 @@ devices_list = cursor.fetchall() def printworker(): nodeso = 0 while True: + worker_id, device_id, elapsed_time = print_queue.get() global real_duration global per_device_duration global polled_devices @@ -130,41 +128,67 @@ def printworker(): print "WARNING: worker %s finished device %s in %s seconds" % (worker_id, device_id, elapsed_time) print_queue.task_done() -def poll_worker(): - while True: - device_id = poll_queue.get() - try: - start_time = time.time() - command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (poller_path, device_id) - subprocess.check_call(command, shell=True) - elapsed_time = int(time.time() - start_time) - print_queue.put([threading.current_thread().name, device_id, elapsed_time]) - except (KeyboardInterrupt, SystemExit): - raise - except: - pass - poll_queue.task_done() +def poll_worker(device_id): + try: + start_time = time.time() + command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (poller_path, device_id) + subprocess.check_call(command, shell=True) + elapsed_time = int(time.time() - start_time) + print_queue.put([threading.current_thread().name, device_id, elapsed_time]) + except (KeyboardInterrupt, SystemExit): + raise + except: + pass -poll_queue = Queue.Queue() print_queue = Queue.Queue() print "INFO: starting the poller at %s with %s threads, slowest devices first" % (time.strftime("%Y-%m-%d %H:%M:%S"), amount_of_workers) -for device_id in devices_list: - poll_queue.put(device_id) -for i in range(amount_of_workers): - t = threading.Thread(target=poll_worker) - t.setDaemon(True) - t.start() +def queue_feeder(): + while True: + if threading.active_count() > amount_of_workers + 1: + time.sleep(1) + continue + + query = 'select device_id,last_polled from devices {} disabled = 0 order by last_polled asc'.format( + 'where poller_group IN({}) and'.format(poller_group) if poller_group else '') + cursor.execute(query) + for device_id, last_polled in cursor.fetchall(): + cursor.execute("SELECT IS_FREE_LOCK('polling.{}')".format(device_id)) + if cursor.fetchone()[0] != 1: + continue + cursor.execute("SELECT IS_FREE_LOCK('queued.{}')".format(device_id)) + if cursor.fetchone()[0] != 1: + continue + last_polled_date = strptime(last_polled, '%Y-%m-%d %H:%M:%S') + if last_polled_date < datetime.now() - timedelta(seconds=frequency): + t = threading.Thread(target=poll_worker(device_id)) + t.setDaemon(True) + t.start() + break + else: + cursor.execute("SELECT GET_LOCK('queued.{}')".format(device_id)) + if cursor.fetchone()[0] != 1: + break + sleeptime = ((last_polled_date + timedelta(seconds=300)) - datetime.now()).seconds + datetime.sleep(sleeptime) + t = threading.Thread(target=poll_worker(device_id)) + t.setDaemon(True) + t.start() + cursor.execute("SELECT RELEASE_LOCK('queued.{}')".format(device_id)) + break p = threading.Thread(target=printworker) p.setDaemon(True) p.start() +f = threading.Thread(target=queue_feeder) +f.setDaemon(True) +f.start() + try: - poll_queue.join() print_queue.join() except (KeyboardInterrupt, SystemExit): raise From 4d3b7a3c0fe75ff31af5569714506b635bd3c748 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 14:51:03 -0400 Subject: [PATCH 09/82] working --- poller-service.py | 140 ++++++++++++++++++++++++++++++++-------------- 1 file changed, 98 insertions(+), 42 deletions(-) diff --git a/poller-service.py b/poller-service.py index 82fc7ed4c..60588dde6 100755 --- a/poller-service.py +++ b/poller-service.py @@ -90,17 +90,28 @@ except: # if None are given or the argument was garbage, fall back to default of 300 try: frequency = int(sys.argv[2]) - if amount_of_workers == 0: + if frequency == 0: print "ERROR: 0 seconds is not a valid value" sys.exit(2) except: frequency = 300 +# Take the down_retry value from the commandline +# if None are given or the argument was garbage, fall back to default of 15 +try: + down_retry = int(sys.argv[3]) + if down_retry == 0: + print "ERROR: 0 seconds is not a valid value" + sys.exit(2) +except: + down_retry = 15 + try: if db_port == 0: db = MySQLdb.connect(host=db_server, user=db_username, passwd=db_password, db=db_dbname) else: db = MySQLdb.connect(host=db_server, port=db_port, user=db_username, passwd=db_password, db=db_dbname) + db.autocommit(True) cursor = db.cursor() except: print "ERROR: Could not connect to MySQL database!" @@ -137,56 +148,101 @@ def poll_worker(device_id): print_queue.put([threading.current_thread().name, device_id, elapsed_time]) except (KeyboardInterrupt, SystemExit): raise - except: - pass + #except: + # pass print_queue = Queue.Queue() -print "INFO: starting the poller at %s with %s threads, slowest devices first" % (time.strftime("%Y-%m-%d %H:%M:%S"), +print "INFO: starting the poller at %s with %s threads" % (time.strftime("%Y-%m-%d %H:%M:%S"), amount_of_workers) - -def queue_feeder(): - while True: - if threading.active_count() > amount_of_workers + 1: - time.sleep(1) - continue - - query = 'select device_id,last_polled from devices {} disabled = 0 order by last_polled asc'.format( - 'where poller_group IN({}) and'.format(poller_group) if poller_group else '') - cursor.execute(query) - for device_id, last_polled in cursor.fetchall(): - cursor.execute("SELECT IS_FREE_LOCK('polling.{}')".format(device_id)) - if cursor.fetchone()[0] != 1: - continue - cursor.execute("SELECT IS_FREE_LOCK('queued.{}')".format(device_id)) - if cursor.fetchone()[0] != 1: - continue - last_polled_date = strptime(last_polled, '%Y-%m-%d %H:%M:%S') - if last_polled_date < datetime.now() - timedelta(seconds=frequency): - t = threading.Thread(target=poll_worker(device_id)) - t.setDaemon(True) - t.start() - break - else: - cursor.execute("SELECT GET_LOCK('queued.{}')".format(device_id)) - if cursor.fetchone()[0] != 1: - break - sleeptime = ((last_polled_date + timedelta(seconds=300)) - datetime.now()).seconds - datetime.sleep(sleeptime) - t = threading.Thread(target=poll_worker(device_id)) - t.setDaemon(True) - t.start() - cursor.execute("SELECT RELEASE_LOCK('queued.{}')".format(device_id)) - break - p = threading.Thread(target=printworker) p.setDaemon(True) p.start() -f = threading.Thread(target=queue_feeder) -f.setDaemon(True) -f.start() +def lockFree(lock): + global cursor + query = "SELECT IS_FREE_LOCK('{}')".format(lock) + #print 'checking lock: {}'.format(query) + cursor.execute(query) + return cursor.fetchall()[0][0] == 1 + +def getLock(lock): + global cursor + query = "SELECT GET_LOCK('{}', 0)".format(lock) + #print 'getting lock: {}'.format(query) + cursor.execute(query) + return cursor.fetchall()[0][0] == 1 + +def releaseLock(lock): + global cursor + query = "SELECT RELEASE_LOCK('{}')".format(lock) + #print 'releasing lock: {}'.format(query) + cursor.execute(query) + return cursor.fetchall()[0][0] == 1 + +recently_scanned = {} + +while True: + print '{} threads currently active'.format(threading.active_count()) + while threading.active_count() >= amount_of_workers: + time.sleep(.5) + + #print 'querying for devices' + query = 'select device_id,last_polled from devices {} disabled = 0 order by last_polled asc'.format( + 'where poller_group IN({}) and'.format(poller_group) if poller_group else '') + cursor.execute(query) + devices = cursor.fetchall() + dead_retry_in = frequency + #print "first 5 devices: {}".format(devices[:5]) + for device_id, last_polled in devices: +# print 'trying device {}'.format(device_id) +# time.sleep(1) + if not lockFree('polling.{}'.format(device_id)): +# print 'polling lock is not free on {} continuing'.format(device_id) +# time.sleep(1) + continue + if not lockFree('queued.{}'.format(device_id)): +# print 'queued lock is not free on {} continuing'.format(device_id) +# time.sleep(1) + continue + try: + if ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds > 1: + dead_retry_in = ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds +# print 'device {} recently scanned already'.format(device_id) +# time.sleep(1) + continue + except KeyError: + pass + + # add queue lock, so if we sleep, we lock the next device against any other pollers, break + # if aquiring lock fails + if not getLock('queued.{}'.format(device_id)): +# print 'getting queue lock on {} failed'.format(device_id) +# time.sleep(1) + break + + if last_polled > datetime.now() - timedelta(seconds=frequency): + sleeptime = ((last_polled + timedelta(seconds=300)) - datetime.now()).seconds + if sleeptime > dead_retry_in: + print 'Sleeping {} seconds before retrying failed device'.format(dead_retry_in, device_id, last_polled) + time.sleep(dead_retry_in) + break + + print 'Sleeping {} seconds before polling {}, last polled {}'.format(sleeptime, device_id, last_polled) + time.sleep(sleeptime) + + print 'Starting poll of device {}, last polled {}'.format(device_id, last_polled) + recently_scanned[device_id] = datetime.now() + t = threading.Thread(target=poll_worker, args=[device_id]) + #t.setDaemon(True) + t.start() + #print 'thread launched' + + releaseLock('queued.{}'.format(device_id)) + + # If we made it this far, break out of the loop and query again. + break try: print_queue.join() From 99cf049645fc28ee5c14b200cad6f79e3aa0ebfc Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 6 Jul 2015 15:10:48 -0400 Subject: [PATCH 10/82] cleaned up --- poller-service.py | 85 ++++++++++------------------------------------- 1 file changed, 17 insertions(+), 68 deletions(-) diff --git a/poller-service.py b/poller-service.py index 60588dde6..7150e313b 100755 --- a/poller-service.py +++ b/poller-service.py @@ -26,6 +26,7 @@ from datetime import datetime, timedelta install_dir = os.path.dirname(os.path.realpath(__file__)) config_file = install_dir + '/config.php' + def get_config_data(): config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % install_dir] try: @@ -53,24 +54,17 @@ db_username = config['db_user'] db_password = config['db_pass'] if config['db_host'][:5].lower() == 'unix:': - db_server = config['db_host'] - db_port = 0 + db_server = config['db_host'] + db_port = 0 elif ':' in config['db_host']: - db_server = config['db_host'].rsplit(':')[0] - db_port = int(config['db_host'].rsplit(':')[1]) + db_server = config['db_host'].rsplit(':')[0] + db_port = int(config['db_host'].rsplit(':')[1]) else: - db_server = config['db_host'] - db_port =0 + db_server = config['db_host'] + db_port = 0 db_dbname = config['db_name'] -# (c) 2015, GPLv3, Daniel Preussker <<= amount_of_workers: time.sleep(.5) - #print 'querying for devices' query = 'select device_id,last_polled from devices {} disabled = 0 order by last_polled asc'.format( - 'where poller_group IN({}) and'.format(poller_group) if poller_group else '') + 'where poller_group IN({}) and'.format(str(config['distributed_poller_group'])) + if 'distributed_poller_group' in config else '') cursor.execute(query) devices = cursor.fetchall() dead_retry_in = frequency - #print "first 5 devices: {}".format(devices[:5]) for device_id, last_polled in devices: -# print 'trying device {}'.format(device_id) -# time.sleep(1) if not lockFree('polling.{}'.format(device_id)): -# print 'polling lock is not free on {} continuing'.format(device_id) -# time.sleep(1) continue if not lockFree('queued.{}'.format(device_id)): -# print 'queued lock is not free on {} continuing'.format(device_id) -# time.sleep(1) continue try: if ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds > 1: dead_retry_in = ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds -# print 'device {} recently scanned already'.format(device_id) -# time.sleep(1) continue except KeyError: pass @@ -218,8 +204,6 @@ while True: # add queue lock, so if we sleep, we lock the next device against any other pollers, break # if aquiring lock fails if not getLock('queued.{}'.format(device_id)): -# print 'getting queue lock on {} failed'.format(device_id) -# time.sleep(1) break if last_polled > datetime.now() - timedelta(seconds=frequency): @@ -235,9 +219,7 @@ while True: print 'Starting poll of device {}, last polled {}'.format(device_id, last_polled) recently_scanned[device_id] = datetime.now() t = threading.Thread(target=poll_worker, args=[device_id]) - #t.setDaemon(True) t.start() - #print 'thread launched' releaseLock('queued.{}'.format(device_id)) @@ -248,36 +230,3 @@ try: print_queue.join() except (KeyboardInterrupt, SystemExit): raise - -total_time = int(time.time() - s_time) - -print "INFO: poller-wrapper polled %s devices in %s seconds with %s workers" % (polled_devices, total_time, amount_of_workers) - -show_stopper = False - -query = "update pollers set last_polled=NOW(), devices='%d', time_taken='%d' where poller_name='%s'" % (polled_devices, - total_time, config['distributed_poller_name']) -response = cursor.execute(query) -if response == 1: - db.commit() -else: - query = "insert into pollers set poller_name='%s', last_polled=NOW(), devices='%d', time_taken='%d'" % ( - config['distributed_poller_name'], polled_devices, total_time) - cursor.execute(query) - db.commit() -db.close() - -if total_time > 300: - print "WARNING: the process took more than 5 minutes to finish, you need faster hardware or more threads" - print "INFO: in sequential style polling the elapsed time would have been: %s seconds" % real_duration - for device in per_device_duration: - if per_device_duration[device] > 300: - print "WARNING: device %s is taking too long: %s seconds" % (device, per_device_duration[device]) - show_stopper = True - if show_stopper: - print "ERROR: Some devices are taking more than 300 seconds, the script cannot recommend you what to do." - else: - recommend = int(total_time / 300.0 * amount_of_workers + 1) - print "WARNING: Consider setting a minimum of %d threads. (This does not constitute professional advice!)" % recommend - - sys.exit(2) From 1dcaa51011acb7f5d3c434746bc87d116704b009 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 10:09:55 -0400 Subject: [PATCH 11/82] more efficient sql queries, working well all night --- poller-service.py | 96 ++++++++++++++++++++++++++++++----------------- 1 file changed, 61 insertions(+), 35 deletions(-) diff --git a/poller-service.py b/poller-service.py index 7150e313b..13c3af851 100755 --- a/poller-service.py +++ b/poller-service.py @@ -176,53 +176,79 @@ def releaseLock(lock): cursor.execute(query) return cursor.fetchall()[0][0] == 1 -recently_scanned = {} +def sleep_until(timestamp): + now = datetime.now() + if timestamp > now: + sleeptime = (timestamp - now).seconds + else: + sleeptime = 0 + time.sleep(sleeptime) + +poller_group = ('and poller_group IN({}) ' + .format(str(config['distributed_poller_group'])) if 'distributed_poller_group' in config else '') + +# Add last_polled and last_polled_timetaken so we can sort by the time the last poll started, with the goal +# of having each device complete a poll within the given time range. +dev_query = ( 'SELECT device_id, ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_polled, ' + ' INTERVAL last_polled_timetaken SECOND ' + ' ), ' + ' INTERVAL {0} SECOND) AS next_poll ' + 'FROM devices WHERE ' + 'disabled = 0 ' + 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' + 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' + '{1} ' + 'ORDER BY next_poll asc ').format(frequency, poller_group) + +dont_retry = {} +threads = 0 while True: - print '{} threads currently active'.format(threading.active_count()) + cur_threads = threading.active_count() + if cur_threads != threads: + threads = cur_threads + print 'INFO: {} threads currently active'.format(threads) + + dont_retry = dict((dev, time) for dev, time in dont_retry.iteritems() if time > datetime.now()) + while threading.active_count() >= amount_of_workers: time.sleep(.5) - query = 'select device_id,last_polled from devices {} disabled = 0 order by last_polled asc'.format( - 'where poller_group IN({}) and'.format(str(config['distributed_poller_group'])) - if 'distributed_poller_group' in config else '') - cursor.execute(query) + cursor.execute(dev_query) devices = cursor.fetchall() - dead_retry_in = frequency - for device_id, last_polled in devices: - if not lockFree('polling.{}'.format(device_id)): - continue - if not lockFree('queued.{}'.format(device_id)): - continue - try: - if ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds > 1: - dead_retry_in = ((recently_scanned[device_id] + timedelta(seconds=down_retry)) - datetime.now()).seconds - continue - except KeyError: - pass - - # add queue lock, so if we sleep, we lock the next device against any other pollers, break - # if aquiring lock fails + for device_id, next_poll in devices: + # add queue lock, so we lock the next device against any other pollers + # if this fails, the device is locked by another poller already if not getLock('queued.{}'.format(device_id)): - break + continue + if not lockFree('polling.{}'.format(device_id)): + releaseLock('queued.{}'.format(device_id)) + continue + if device_id in dont_retry: + releaseLock('queued.{}'.format(device_id)) + continue - if last_polled > datetime.now() - timedelta(seconds=frequency): - sleeptime = ((last_polled + timedelta(seconds=300)) - datetime.now()).seconds - if sleeptime > dead_retry_in: - print 'Sleeping {} seconds before retrying failed device'.format(dead_retry_in, device_id, last_polled) - time.sleep(dead_retry_in) - break + if next_poll > datetime.now(): + next_retry_device = min(dont_retry, key=dont_retry.get) + next_retry_time = dont_retry[next_retry_device] + if next_retry_time < next_poll: + device_id = next_retry_device + if not getLock('queued.{}'.format(device_id)): + continue + print 'INFO: Sleeping until {} before retrying failed device {}'.format(next_retry_time, device_id) + sleep_until(next_retry_time) + else: + print 'INFO: Sleeping until {} before polling {}'.format(next_poll, device_id) + sleep_until(next_poll) - print 'Sleeping {} seconds before polling {}, last polled {}'.format(sleeptime, device_id, last_polled) - time.sleep(sleeptime) - - print 'Starting poll of device {}, last polled {}'.format(device_id, last_polled) - recently_scanned[device_id] = datetime.now() + print 'INFO: Starting poll of device {}'.format(device_id) + dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) t = threading.Thread(target=poll_worker, args=[device_id]) t.start() - releaseLock('queued.{}'.format(device_id)) - # If we made it this far, break out of the loop and query again. break From 4736544e7e691e125265bc6194a60ed496afb8e4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 10:27:40 -0400 Subject: [PATCH 12/82] fix bug on retrying failed devices --- poller-service.py | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/poller-service.py b/poller-service.py index 13c3af851..b07424d83 100755 --- a/poller-service.py +++ b/poller-service.py @@ -219,6 +219,7 @@ while True: cursor.execute(dev_query) devices = cursor.fetchall() + retry_devs_found = {} for device_id, next_poll in devices: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already @@ -228,21 +229,27 @@ while True: releaseLock('queued.{}'.format(device_id)) continue if device_id in dont_retry: + retry_devs_found[device_id] = dont_retry[device_id] releaseLock('queued.{}'.format(device_id)) continue if next_poll > datetime.now(): - next_retry_device = min(dont_retry, key=dont_retry.get) - next_retry_time = dont_retry[next_retry_device] - if next_retry_time < next_poll: - device_id = next_retry_device - if not getLock('queued.{}'.format(device_id)): - continue - print 'INFO: Sleeping until {} before retrying failed device {}'.format(next_retry_time, device_id) - sleep_until(next_retry_time) - else: - print 'INFO: Sleeping until {} before polling {}'.format(next_poll, device_id) - sleep_until(next_poll) + poll_action = 'polling' + sleep_until_time = next_poll + try: + next_retry_device = min(retry_devs_found, key=retry_devs_found.get) + next_retry_time = retry_devs_found[next_retry_device] + if next_retry_time < next_poll: + device_id = next_retry_device + if not getLock('queued.{}'.format(device_id)): + continue + poll_action = 'retrying failed device' + sleep_until_time = next_retry_time + except ValueError: + pass + + print 'INFO: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id) + sleep_until(sleep_until_time) print 'INFO: Starting poll of device {}'.format(device_id) dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) From e9ffc7a374398134cbc435863146b6a802a921ac Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 10:35:36 -0400 Subject: [PATCH 13/82] thread numbers are meaningless --- poller-service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index b07424d83..bdf9a3aea 100755 --- a/poller-service.py +++ b/poller-service.py @@ -129,9 +129,9 @@ def printworker(): per_device_duration[device_id] = elapsed_time polled_devices += 1 if elapsed_time < 300: - print "INFO: worker %s finished device %s in %s seconds" % (worker_id, device_id, elapsed_time) + print "INFO: worker finished device %s in %s seconds" % (device_id, elapsed_time) else: - print "WARNING: worker %s finished device %s in %s seconds" % (worker_id, device_id, elapsed_time) + print "WARNING: worker finished device %s in %s seconds" % (device_id, elapsed_time) print_queue.task_done() From 2b3d70e928a35c7bd2e4f80fa191ad97818db897 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 11:20:09 -0400 Subject: [PATCH 14/82] add proper logging --- poller-service.py | 78 +++++++++++++++++++---------------------------- 1 file changed, 32 insertions(+), 46 deletions(-) diff --git a/poller-service.py b/poller-service.py index bdf9a3aea..e3172e285 100755 --- a/poller-service.py +++ b/poller-service.py @@ -21,18 +21,29 @@ import sys import threading import time import MySQLdb +import logging +import logging.handlers from datetime import datetime, timedelta +log = logging.getLogger('poller-service') +log.setLevel(logging.DEBUG) + +formatter = logging.Formatter('poller-service: %(message)s') +handler = logging.handlers.SysLogHandler(address= '/dev/log') +handler.setFormatter(formatter) +log.addHandler(handler) + install_dir = os.path.dirname(os.path.realpath(__file__)) config_file = install_dir + '/config.php' +log.info('Starting poller-service') def get_config_data(): config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % install_dir] try: proc = subprocess.Popen(config_cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE) except: - print "ERROR: Could not execute: %s" % config_cmd + log.critical("ERROR: Could not execute: %s" % config_cmd) sys.exit(2) return proc.communicate()[0] @@ -40,13 +51,13 @@ try: with open(config_file) as f: pass except IOError as e: - print "ERROR: Oh dear... %s does not seem readable" % config_file + log.critical("ERROR: Oh dear... %s does not seem readable" % config_file) sys.exit(2) try: config = json.loads(get_config_data()) except: - print "ERROR: Could not load or parse configuration, are PATHs correct?" + log.critical("ERROR: Could not load or parse configuration, are PATHs correct?") sys.exit(2) poller_path = config['install_dir'] + '/poller.php' @@ -65,17 +76,13 @@ else: db_dbname = config['db_name'] -s_time = time.time() -real_duration = 0 -per_device_duration = {} -polled_devices = 0 # Take the amount of threads we want to run in parallel from the commandline # if None are given or the argument was garbage, fall back to default of 16 try: amount_of_workers = int(sys.argv[1]) if amount_of_workers == 0: - print "ERROR: 0 threads is not a valid value" + log.critical("ERROR: 0 threads is not a valid value") sys.exit(2) except: amount_of_workers = 16 @@ -85,7 +92,7 @@ except: try: frequency = int(sys.argv[2]) if frequency == 0: - print "ERROR: 0 seconds is not a valid value" + log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) except: frequency = 300 @@ -95,7 +102,7 @@ except: try: down_retry = int(sys.argv[3]) if down_retry == 0: - print "ERROR: 0 seconds is not a valid value" + log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) except: down_retry = 15 @@ -108,32 +115,9 @@ try: db.autocommit(True) cursor = db.cursor() except: - print "ERROR: Could not connect to MySQL database!" + log.critical("ERROR: Could not connect to MySQL database!") sys.exit(2) -# A seperate queue and a single worker for printing information to the screen prevents -# the good old joke: -# -# Some people, when confronted with a problem, think, -# "I know, I'll use threads," and then two they hav erpoblesms. - - -def printworker(): - nodeso = 0 - while True: - worker_id, device_id, elapsed_time = print_queue.get() - global real_duration - global per_device_duration - global polled_devices - real_duration += elapsed_time - per_device_duration[device_id] = elapsed_time - polled_devices += 1 - if elapsed_time < 300: - print "INFO: worker finished device %s in %s seconds" % (device_id, elapsed_time) - else: - print "WARNING: worker finished device %s in %s seconds" % (device_id, elapsed_time) - print_queue.task_done() - def poll_worker(device_id): try: @@ -141,20 +125,15 @@ def poll_worker(device_id): command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (poller_path, device_id) subprocess.check_call(command, shell=True) elapsed_time = int(time.time() - start_time) - print_queue.put([threading.current_thread().name, device_id, elapsed_time]) + if elapsed_time < 300: + log.debug("DEBUG: worker finished device %s in %s seconds" % (device_id, elapsed_time)) + else: + log.warning("WARNING: worker finished device %s in %s seconds" % (device_id, elapsed_time)) except (KeyboardInterrupt, SystemExit): raise except: pass -print_queue = Queue.Queue() - -print "INFO: starting the poller at %s with %s threads" % (time.strftime("%Y-%m-%d %H:%M:%S"), amount_of_workers) - -p = threading.Thread(target=printworker) -p.setDaemon(True) -p.start() - def lockFree(lock): global cursor @@ -205,12 +184,18 @@ dev_query = ( 'SELECT device_id, ' dont_retry = {} threads = 0 +next_update = datetime.now() + timedelta(minutes=5) while True: cur_threads = threading.active_count() if cur_threads != threads: threads = cur_threads - print 'INFO: {} threads currently active'.format(threads) + log.debug('DEBUG: {} threads currently active'.format(threads)) + + if next_update < datetime.now(): + log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) + devices_scanned = 0 + next_update = datetime.now() + timedelta(minutes=5) dont_retry = dict((dev, time) for dev, time in dont_retry.iteritems() if time > datetime.now()) @@ -248,10 +233,11 @@ while True: except ValueError: pass - print 'INFO: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id) + log.debug('DEBUG: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id)) sleep_until(sleep_until_time) - print 'INFO: Starting poll of device {}'.format(device_id) + log.debug('INFO: Starting poll of device {}'.format(device_id)) + devices_scanned += 1 dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) t = threading.Thread(target=poll_worker, args=[device_id]) t.start() From f7993a204c2f1113217ddefd97bccf1fa9465c5b Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 11:29:04 -0400 Subject: [PATCH 15/82] log level from config --- poller-service.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index e3172e285..1b84e69e0 100755 --- a/poller-service.py +++ b/poller-service.py @@ -60,6 +60,13 @@ except: log.critical("ERROR: Could not load or parse configuration, are PATHs correct?") sys.exit(2) +try: + loglevel = config['poller_service_loglevel'] +except KeyError: + loglevel = 'INFO' +numeric_level = getattr(logging, loglevel.upper(), None) +log.basicConfig(level=numeric_level) + poller_path = config['install_dir'] + '/poller.php' db_username = config['db_user'] db_password = config['db_pass'] @@ -185,6 +192,7 @@ dev_query = ( 'SELECT device_id, ' dont_retry = {} threads = 0 next_update = datetime.now() + timedelta(minutes=5) +devices_scanned = 0 while True: cur_threads = threading.active_count() @@ -236,7 +244,7 @@ while True: log.debug('DEBUG: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id)) sleep_until(sleep_until_time) - log.debug('INFO: Starting poll of device {}'.format(device_id)) + log.debug('DEBUG: Starting poll of device {}'.format(device_id)) devices_scanned += 1 dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) t = threading.Thread(target=poll_worker, args=[device_id]) From e30aea0168e02c521dacd3def5de9b32bdb88722 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 11:40:29 -0400 Subject: [PATCH 16/82] get all config from config.php --- poller-service.py | 30 +++++++++++++----------------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/poller-service.py b/poller-service.py index 1b84e69e0..6492ab0bc 100755 --- a/poller-service.py +++ b/poller-service.py @@ -36,7 +36,7 @@ log.addHandler(handler) install_dir = os.path.dirname(os.path.realpath(__file__)) config_file = install_dir + '/config.php' -log.info('Starting poller-service') +log.info('INFO: Starting poller-service') def get_config_data(): config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % install_dir] @@ -61,11 +61,13 @@ except: sys.exit(2) try: - loglevel = config['poller_service_loglevel'] + loglevel = logging.getLevelName(config['poller_service_loglevel'].upper()) except KeyError: - loglevel = 'INFO' -numeric_level = getattr(logging, loglevel.upper(), None) -log.basicConfig(level=numeric_level) + loglevel = logging.getLevelName('INFO') +if not isinstance(loglevel, int): + log.warning('ERROR: {} is not a valid log level'.format(str(loglevel))) + loglevel = logging.getLevelName('INFO') +log.setLevel(loglevel) poller_path = config['install_dir'] + '/poller.php' db_username = config['db_user'] @@ -84,34 +86,28 @@ else: db_dbname = config['db_name'] -# Take the amount of threads we want to run in parallel from the commandline -# if None are given or the argument was garbage, fall back to default of 16 try: - amount_of_workers = int(sys.argv[1]) + amount_of_workers = int(config['poller_service_workers']) if amount_of_workers == 0: log.critical("ERROR: 0 threads is not a valid value") sys.exit(2) -except: +except KeyError: amount_of_workers = 16 -# Take the frequency of scans we want from the commandline -# if None are given or the argument was garbage, fall back to default of 300 try: - frequency = int(sys.argv[2]) + frequency = int(config['poller_service_frequency']) if frequency == 0: log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) -except: +except KeyError: frequency = 300 -# Take the down_retry value from the commandline -# if None are given or the argument was garbage, fall back to default of 15 try: - down_retry = int(sys.argv[3]) + down_retry = int(config['poller_service_down_retry']) if down_retry == 0: log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) -except: +except KeyError: down_retry = 15 try: From 57bf214df3c5ce02127bb672fd3e08d24fb14c9d Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 11:41:42 -0400 Subject: [PATCH 17/82] first attempt at upstart conf --- poller-service.conf | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 poller-service.conf diff --git a/poller-service.conf b/poller-service.conf new file mode 100644 index 000000000..c5a9f26d2 --- /dev/null +++ b/poller-service.conf @@ -0,0 +1,28 @@ +# poller-service - SNMP polling service for LibreNMS + +description "SNMP polling service for LibreNMS" +author "Clint Armstrong " + +# When to start the service +start on runlevel [2345] + +# When to stop the service +stop on runlevel [016] + +# Automatically restart process if crashed +respawn + +# Essentially lets upstart know the process will detach itself to the background +expect fork + +# # Run before process +# pre-start script +# [ -d /var/run/myservice ] || mkdir -p /var/run/myservice +# echo "Put bash code here" +# end script + +# Set working directory +chdir /opt/librenms + +# Start the process +exec su -u librenms poller-service.py From cf8844742ef8b5088faba17abd12b4e1f4f45b36 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 12:04:36 -0400 Subject: [PATCH 18/82] poller service upstart conf working --- poller-service.conf | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/poller-service.conf b/poller-service.conf index c5a9f26d2..36677e7d2 100644 --- a/poller-service.conf +++ b/poller-service.conf @@ -13,7 +13,7 @@ stop on runlevel [016] respawn # Essentially lets upstart know the process will detach itself to the background -expect fork +#expect fork # # Run before process # pre-start script @@ -21,8 +21,9 @@ expect fork # echo "Put bash code here" # end script -# Set working directory chdir /opt/librenms +setuid librenms +setgid librenms # Start the process -exec su -u librenms poller-service.py +exec /opt/librenms/poller-service.py From c7423d2c4e1b1ad81a9dafd4cbc87a82587666ce Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 12:48:36 -0400 Subject: [PATCH 19/82] start docs --- doc/Extensions/Poller-Service.md | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 doc/Extensions/Poller-Service.md diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md new file mode 100644 index 000000000..9c515c378 --- /dev/null +++ b/doc/Extensions/Poller-Service.md @@ -0,0 +1,2 @@ +# Poller Service +The Poller service is an alternative to polling cron jobs. From f293e3d9679b4b93965f4156f3a777ccd8886636 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 13:02:13 -0400 Subject: [PATCH 20/82] add discovery support to service --- poller-service.py | 50 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 37 insertions(+), 13 deletions(-) diff --git a/poller-service.py b/poller-service.py index 6492ab0bc..f1530f1b5 100755 --- a/poller-service.py +++ b/poller-service.py @@ -1,15 +1,15 @@ #! /usr/bin/env python """ poller-service A service to wrap SNMP polling. It will poll up to $threads devices at a time, and will not re-poll - devices that have been polled within the last $frequency seconds. It will prioritize devices based on + devices that have been polled within the last $poll_frequency seconds. It will prioritize devices based on the last time polled. If resources are sufficient, this service should poll every device every - $frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as + $poll_frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as frequently as possible. This service is based on poller-wrapper.py. Author: Clint Armstrong Date: July 2015 - Usage: poller-service [threads] [frequency] + Usage: poller-service [threads] [poll_frequency] Default is 16 threads and 300 seconds. """ @@ -70,6 +70,7 @@ if not isinstance(loglevel, int): log.setLevel(loglevel) poller_path = config['install_dir'] + '/poller.php' +discover_path = config['install_dir'] + '/discovery.php' db_username = config['db_user'] db_password = config['db_pass'] @@ -95,12 +96,20 @@ except KeyError: amount_of_workers = 16 try: - frequency = int(config['poller_service_frequency']) - if frequency == 0: + poll_frequency = int(config['poller_service_poll_frequency']) + if poll_frequency == 0: log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) except KeyError: - frequency = 300 + poll_frequency = 300 + +try: + discover_frequency = int(config['poller_service_discover_frequency']) + if poll_frequency == 0: + log.critical("ERROR: 0 seconds is not a valid value") + sys.exit(2) +except KeyError: + poll_frequency = 21600 try: down_retry = int(config['poller_service_down_retry']) @@ -122,10 +131,13 @@ except: sys.exit(2) -def poll_worker(device_id): +def poll_worker(device_id, action): try: start_time = time.time() - command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (poller_path, device_id) + path = poller_path + if action = 'discovery' + path = discovery_path + command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (path, device_id) subprocess.check_call(command, shell=True) elapsed_time = int(time.time() - start_time) if elapsed_time < 300: @@ -178,12 +190,20 @@ dev_query = ( 'SELECT device_id, ' ' INTERVAL last_polled_timetaken SECOND ' ' ), ' ' INTERVAL {0} SECOND) AS next_poll ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_discovered, ' + ' INTERVAL last_discovered_timetaken SECOND ' + ' ), ' + ' INTERVAL {1} SECOND) AS next_discovery ' 'FROM devices WHERE ' 'disabled = 0 ' 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' - '{1} ' - 'ORDER BY next_poll asc ').format(frequency, poller_group) + '{2} ' + 'ORDER BY next_poll asc ').format( poll_frequency, + discover_frequency, + poller_group ) dont_retry = {} threads = 0 @@ -209,7 +229,7 @@ while True: cursor.execute(dev_query) devices = cursor.fetchall() retry_devs_found = {} - for device_id, next_poll in devices: + for device_id, next_poll, next_discovery in devices: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already if not getLock('queued.{}'.format(device_id)): @@ -240,10 +260,14 @@ while True: log.debug('DEBUG: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id)) sleep_until(sleep_until_time) - log.debug('DEBUG: Starting poll of device {}'.format(device_id)) + action = 'poll' + if not next_discovery or next_discovery < datetime.now(): + action = 'discovery' + + log.debug('DEBUG: Starting {} of device {}'.format(action, device_id)) devices_scanned += 1 dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) - t = threading.Thread(target=poll_worker, args=[device_id]) + t = threading.Thread(target=poll_worker, args=[device_id, action]) t.start() # If we made it this far, break out of the loop and query again. From 18d8914548af4d2a438ca63cd762c8a334d430fe Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 13:10:36 -0400 Subject: [PATCH 21/82] discover integration working --- poller-service.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/poller-service.py b/poller-service.py index f1530f1b5..e6656036a 100755 --- a/poller-service.py +++ b/poller-service.py @@ -105,11 +105,11 @@ except KeyError: try: discover_frequency = int(config['poller_service_discover_frequency']) - if poll_frequency == 0: + if discover_frequency == 0: log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) except KeyError: - poll_frequency = 21600 + discover_frequency = 21600 try: down_retry = int(config['poller_service_down_retry']) @@ -135,8 +135,8 @@ def poll_worker(device_id, action): try: start_time = time.time() path = poller_path - if action = 'discovery' - path = discovery_path + if action == 'discovery': + path = discover_path command = "/usr/bin/env php %s -h %s >> /dev/null 2>&1" % (path, device_id) subprocess.check_call(command, shell=True) elapsed_time = int(time.time() - start_time) @@ -189,7 +189,7 @@ dev_query = ( 'SELECT device_id, ' ' last_polled, ' ' INTERVAL last_polled_timetaken SECOND ' ' ), ' - ' INTERVAL {0} SECOND) AS next_poll ' + ' INTERVAL {0} SECOND) AS next_poll, ' 'DATE_ADD( ' ' DATE_SUB( ' ' last_discovered, ' @@ -242,7 +242,7 @@ while True: releaseLock('queued.{}'.format(device_id)) continue - if next_poll > datetime.now(): + if next_poll and next_poll > datetime.now(): poll_action = 'polling' sleep_until_time = next_poll try: From 9ae0bc3f8319cafa1387f7c7284c12c1b172569c Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 13:33:16 -0400 Subject: [PATCH 22/82] documentation written --- doc/Extensions/Poller-Service.md | 21 ++++++++++++++++++++- poller-service.py | 2 +- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 9c515c378..8406067c6 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -1,2 +1,21 @@ # Poller Service -The Poller service is an alternative to polling cron jobs. +The Poller service is an alternative to polling and discovery cron jobs and provides support for distributed polling without memcache. It is multi-threaded and runs continuously discovering and polling devices with the oldest data attempting to honor the polling frequency configured in `config.php`. + +Configure the maximum number of threads for the service in `$config['poller_service_workers']`. Configure the minimum desired polling frequency in `$config['poller_service_poll_frequency']` and the minimum desired discovery frequency in `$config['poller_service_discover_frequency']`. The service will not poll or discover devices which have data newer than this this configured age in seconds. Configure how frequently the service will attempt to poll devices which are down in `$config['poller_service_down_retry']`. + +The poller service is designed to gracefully degrade. If not all devices can be polled within the configured frequency, the service will continuously poll devices refreshing as frequently as possible using the configured number of threads. + +## Configuration +```php +// Poller-Service settings +$config['poller_service_workers'] = 16; +$config['poller_service_poll_frequency'] = 300; +$config['poller_service_discover_frequency'] = 21600; +$config['poller_service_down_retry'] = 60; +``` + +## Distributed Polling +Distributed polling is possible. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. + +## Service Installation +The service is tested on Ubuntu 14.04. An upstart configuration `poller-service.conf` is provided. To install copy this file to `/etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. diff --git a/poller-service.py b/poller-service.py index e6656036a..c65c99a40 100755 --- a/poller-service.py +++ b/poller-service.py @@ -117,7 +117,7 @@ try: log.critical("ERROR: 0 seconds is not a valid value") sys.exit(2) except KeyError: - down_retry = 15 + down_retry = 60 try: if db_port == 0: From 1f0799fc03b2088a57107e26f6f7673af3892918 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 13:34:41 -0400 Subject: [PATCH 23/82] more details about which cronjobs this replaces --- doc/Extensions/Poller-Service.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 8406067c6..5248a1019 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -1,5 +1,5 @@ # Poller Service -The Poller service is an alternative to polling and discovery cron jobs and provides support for distributed polling without memcache. It is multi-threaded and runs continuously discovering and polling devices with the oldest data attempting to honor the polling frequency configured in `config.php`. +The Poller service is an alternative to polling and discovery cron jobs and provides support for distributed polling without memcache. It is multi-threaded and runs continuously discovering and polling devices with the oldest data attempting to honor the polling frequency configured in `config.php`. This service replaces all the required cron jobs except for `/opt/librenms/daily.sh` and `/opt/librenms/alerts.php`. Configure the maximum number of threads for the service in `$config['poller_service_workers']`. Configure the minimum desired polling frequency in `$config['poller_service_poll_frequency']` and the minimum desired discovery frequency in `$config['poller_service_discover_frequency']`. The service will not poll or discover devices which have data newer than this this configured age in seconds. Configure how frequently the service will attempt to poll devices which are down in `$config['poller_service_down_retry']`. From f8756dc956c43590929208f1acb20fbe8a187a51 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 13:58:56 -0400 Subject: [PATCH 24/82] more details indocs --- doc/Extensions/Poller-Service.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 5248a1019..0b552d6c3 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -15,7 +15,7 @@ $config['poller_service_down_retry'] = 60; ``` ## Distributed Polling -Distributed polling is possible. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. +Distributed polling is possible, and uses the same configuration options as are described for traditional distributed polling, except that the memcached options are not necessary. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. ## Service Installation The service is tested on Ubuntu 14.04. An upstart configuration `poller-service.conf` is provided. To install copy this file to `/etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. From 4fed175935d7723f065ef5fff85cea8e07d75142 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 14:08:30 -0400 Subject: [PATCH 25/82] document loglevel --- doc/Extensions/Poller-Service.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 0b552d6c3..354d58c39 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -5,9 +5,12 @@ Configure the maximum number of threads for the service in `$config['poller_serv The poller service is designed to gracefully degrade. If not all devices can be polled within the configured frequency, the service will continuously poll devices refreshing as frequently as possible using the configured number of threads. +The service logs to syslog. A loglevel of INFO will print status updates every 5 minutes. Loglevel of DEBUG will print updates on every device as it is scanned. + ## Configuration ```php // Poller-Service settings +$config['poller_service_loglevel'] = "INFO"; $config['poller_service_workers'] = 16; $config['poller_service_poll_frequency'] = 300; $config['poller_service_discover_frequency'] = 21600; From b0c02a167ef18b7f8341b06150eba2e57b832b57 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 14:16:55 -0400 Subject: [PATCH 26/82] add license to docstring --- poller-service.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index c65c99a40..6ef51d5ae 100755 --- a/poller-service.py +++ b/poller-service.py @@ -9,8 +9,7 @@ Author: Clint Armstrong Date: July 2015 - Usage: poller-service [threads] [poll_frequency] - Default is 16 threads and 300 seconds. + License: BSD """ import json From 67b8f9f4733785b98498199b95e802ef55b2f490 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 14:33:33 -0400 Subject: [PATCH 27/82] pep8 --- poller-service.py | 58 ++++++++++++++++++++++------------------------- 1 file changed, 27 insertions(+), 31 deletions(-) diff --git a/poller-service.py b/poller-service.py index 6ef51d5ae..07e1ea3f2 100755 --- a/poller-service.py +++ b/poller-service.py @@ -1,8 +1,8 @@ #! /usr/bin/env python """ poller-service A service to wrap SNMP polling. It will poll up to $threads devices at a time, and will not re-poll - devices that have been polled within the last $poll_frequency seconds. It will prioritize devices based on - the last time polled. If resources are sufficient, this service should poll every device every + devices that have been polled within the last $poll_frequency seconds. It will prioritize devices based + on the last time polled. If resources are sufficient, this service should poll every device every $poll_frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as frequently as possible. This service is based on poller-wrapper.py. @@ -14,7 +14,6 @@ import json import os -import Queue import subprocess import sys import threading @@ -28,7 +27,7 @@ log = logging.getLogger('poller-service') log.setLevel(logging.DEBUG) formatter = logging.Formatter('poller-service: %(message)s') -handler = logging.handlers.SysLogHandler(address= '/dev/log') +handler = logging.handlers.SysLogHandler(address='/dev/log') handler.setFormatter(formatter) log.addHandler(handler) @@ -37,6 +36,7 @@ config_file = install_dir + '/config.php' log.info('INFO: Starting poller-service') + def get_config_data(): config_cmd = ['/usr/bin/env', 'php', '%s/config_to_json.php' % install_dir] try: @@ -169,6 +169,7 @@ def releaseLock(lock): cursor.execute(query) return cursor.fetchall()[0][0] == 1 + def sleep_until(timestamp): now = datetime.now() if timestamp > now: @@ -178,31 +179,31 @@ def sleep_until(timestamp): time.sleep(sleeptime) poller_group = ('and poller_group IN({}) ' - .format(str(config['distributed_poller_group'])) if 'distributed_poller_group' in config else '') + .format(str(config['distributed_poller_group'])) if 'distributed_poller_group' in config else '') # Add last_polled and last_polled_timetaken so we can sort by the time the last poll started, with the goal # of having each device complete a poll within the given time range. -dev_query = ( 'SELECT device_id, ' - 'DATE_ADD( ' - ' DATE_SUB( ' - ' last_polled, ' - ' INTERVAL last_polled_timetaken SECOND ' - ' ), ' - ' INTERVAL {0} SECOND) AS next_poll, ' - 'DATE_ADD( ' - ' DATE_SUB( ' - ' last_discovered, ' - ' INTERVAL last_discovered_timetaken SECOND ' - ' ), ' - ' INTERVAL {1} SECOND) AS next_discovery ' - 'FROM devices WHERE ' - 'disabled = 0 ' - 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' - 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' - '{2} ' - 'ORDER BY next_poll asc ').format( poll_frequency, - discover_frequency, - poller_group ) +dev_query = ('SELECT device_id, ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_polled, ' + ' INTERVAL last_polled_timetaken SECOND ' + ' ), ' + ' INTERVAL {0} SECOND) AS next_poll, ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_discovered, ' + ' INTERVAL last_discovered_timetaken SECOND ' + ' ), ' + ' INTERVAL {1} SECOND) AS next_discovery ' + 'FROM devices WHERE ' + 'disabled = 0 ' + 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' + 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' + '{2} ' + 'ORDER BY next_poll asc ').format(poll_frequency, + discover_frequency, + poller_group) dont_retry = {} threads = 0 @@ -271,8 +272,3 @@ while True: # If we made it this far, break out of the loop and query again. break - -try: - print_queue.join() -except (KeyboardInterrupt, SystemExit): - raise From 0588ff41c301a792150e3bba40a1c691e2da3606 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 7 Jul 2015 15:53:04 -0400 Subject: [PATCH 28/82] get a lock when doing discovery --- discovery.php | 2 ++ 1 file changed, 2 insertions(+) diff --git a/discovery.php b/discovery.php index 3006399ff..93ca69a09 100755 --- a/discovery.php +++ b/discovery.php @@ -106,7 +106,9 @@ if ($config['distributed_poller'] === true) { } foreach (dbFetch("SELECT * FROM `devices` WHERE status = 1 AND disabled = 0 $where ORDER BY device_id DESC") as $device) { + if (dbGetLock('polling.' . $device['device_id'])) { discover_device($device, $options); + } } $end = utime(); From c65a160c18a424058edc146f330bb8a23e4ce189 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 10 Jul 2015 16:59:16 -0400 Subject: [PATCH 29/82] log progress to db every 5 minutes --- poller-service.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/poller-service.py b/poller-service.py index 07e1ea3f2..6477ee93c 100755 --- a/poller-service.py +++ b/poller-service.py @@ -217,6 +217,15 @@ while True: log.debug('DEBUG: {} threads currently active'.format(threads)) if next_update < datetime.now(): + seconds_taken = (datetime.now() - (next_update - timedelta(minutes=5))).seconds + update_query = ("insert into pollers set " + "poller_name='%s', " + "last_polled=NOW(), " + "devices='%d', " + "time_taken='%d' ").format(config['distributed_poller_name'], + devices_scanned, + seconds_taken) + cursor.execute(update_query) log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 next_update = datetime.now() + timedelta(minutes=5) From 4529ba2e7da25bb2bbe5ca4c90ddc333ff10f7e2 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 10 Jul 2015 17:19:51 -0400 Subject: [PATCH 30/82] fix stupid query --- poller-service.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/poller-service.py b/poller-service.py index 6477ee93c..e2649586e 100755 --- a/poller-service.py +++ b/poller-service.py @@ -218,13 +218,18 @@ while True: if next_update < datetime.now(): seconds_taken = (datetime.now() - (next_update - timedelta(minutes=5))).seconds - update_query = ("insert into pollers set " - "poller_name='%s', " - "last_polled=NOW(), " - "devices='%d', " - "time_taken='%d' ").format(config['distributed_poller_name'], - devices_scanned, - seconds_taken) + update_query = ("INSERT INTO pollers(poller_name, " + " last_polled, " + " devices, " + " time_taken) " + " values({0}, NOW(), {1}, {2}) " + "ON DUPLICATE KEY UPDATE " + " poller_name=values(poller_name), " + " last_polled=values(last_polled), " + " devices=values(devices) " + " time_taken=values(time_taken) ").format(config['distributed_poller_name'], + devices_scanned, + seconds_taken) cursor.execute(update_query) log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 From b9c3b3cec8a7de50fa59904a70cc3071ca7d2ea7 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 10 Jul 2015 17:34:34 -0400 Subject: [PATCH 31/82] sql stuff --- poller-service.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/poller-service.py b/poller-service.py index e2649586e..c794704f1 100755 --- a/poller-service.py +++ b/poller-service.py @@ -218,16 +218,15 @@ while True: if next_update < datetime.now(): seconds_taken = (datetime.now() - (next_update - timedelta(minutes=5))).seconds - update_query = ("INSERT INTO pollers(poller_name, " - " last_polled, " - " devices, " - " time_taken) " - " values({0}, NOW(), {1}, {2}) " - "ON DUPLICATE KEY UPDATE " - " poller_name=values(poller_name), " - " last_polled=values(last_polled), " - " devices=values(devices) " - " time_taken=values(time_taken) ").format(config['distributed_poller_name'], + update_query = ('INSERT INTO pollers(poller_name, ' + ' last_polled, ' + ' devices, ' + ' time_taken) ' + ' values("{0}", NOW(), "{1}", "{2}") ' + 'ON DUPLICATE KEY UPDATE ' + ' last_polled=values(last_polled), ' + ' devices=values(devices) ' + ' time_taken=values(time_taken) ').format(config['distributed_poller_name'], devices_scanned, seconds_taken) cursor.execute(update_query) From b4c5f0a5ec38beea1711f444b28e64f3db319a74 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 10 Jul 2015 17:34:45 -0400 Subject: [PATCH 32/82] notes --- notes.txt | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 notes.txt diff --git a/notes.txt b/notes.txt new file mode 100644 index 000000000..e3a826237 --- /dev/null +++ b/notes.txt @@ -0,0 +1,6 @@ +Need to set poller_name as primary key for this to work: + +ALTER TABLE pollers + ADD PRIMARY KEY (poller_name) + +Check in beginning? From bedf425c1cabf3c0c6326ad221fd97b3893778d7 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 08:43:18 -0400 Subject: [PATCH 33/82] update sql schema to add primary key for pollers --- notes.txt | 6 ------ 1 file changed, 6 deletions(-) delete mode 100644 notes.txt diff --git a/notes.txt b/notes.txt deleted file mode 100644 index e3a826237..000000000 --- a/notes.txt +++ /dev/null @@ -1,6 +0,0 @@ -Need to set poller_name as primary key for this to work: - -ALTER TABLE pollers - ADD PRIMARY KEY (poller_name) - -Check in beginning? From 55ad58b4e98a2d94f1c5e413d9631e04735b7e7c Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 08:43:23 -0400 Subject: [PATCH 34/82] update sql schema to add primary key for pollers --- sql-schema/065.sql | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 sql-schema/065.sql diff --git a/sql-schema/065.sql b/sql-schema/065.sql new file mode 100644 index 000000000..b201eab9a --- /dev/null +++ b/sql-schema/065.sql @@ -0,0 +1,3 @@ +ALTER TABLE `device_perf` DROP INDEX `id` , ADD INDEX `id` ( `id` ), ADD INDEX ( `device_id` ); +DELETE n1 FROM pollers n1, pollers n2 WHERE n1.last_polled < n2.last_polled and n1.poller_name = n2.poller_name; +ALTER TABLE pollers ADD PRIMARY KEY (poller_name); From 01b7648136cd692cfccfbf798a99cce683a5b9d4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 09:22:53 -0400 Subject: [PATCH 35/82] add last_poll_attempted column to devices table --- sql-schema/065.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql-schema/065.sql b/sql-schema/065.sql index b201eab9a..ab64beca9 100644 --- a/sql-schema/065.sql +++ b/sql-schema/065.sql @@ -1,3 +1,3 @@ -ALTER TABLE `device_perf` DROP INDEX `id` , ADD INDEX `id` ( `id` ), ADD INDEX ( `device_id` ); DELETE n1 FROM pollers n1, pollers n2 WHERE n1.last_polled < n2.last_polled and n1.poller_name = n2.poller_name; ALTER TABLE pollers ADD PRIMARY KEY (poller_name); +ALTER TABLE `devices` ADD `last_poll_attempted` timestamp NULL DEFAULT NULL AFTER `last_polled`; From f12725fc814371868ad0a7e164fce7bf4cef215d Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 09:38:03 -0400 Subject: [PATCH 36/82] store last attempted in sql for simpler down_retry --- poller-service.py | 43 ++++++++++++++++--------------------------- 1 file changed, 16 insertions(+), 27 deletions(-) diff --git a/poller-service.py b/poller-service.py index c794704f1..43ee9875c 100755 --- a/poller-service.py +++ b/poller-service.py @@ -140,9 +140,9 @@ def poll_worker(device_id, action): subprocess.check_call(command, shell=True) elapsed_time = int(time.time() - start_time) if elapsed_time < 300: - log.debug("DEBUG: worker finished device %s in %s seconds" % (device_id, elapsed_time)) + log.debug("DEBUG: worker finished %s of device %s in %s seconds" % (action, device_id, elapsed_time)) else: - log.warning("WARNING: worker finished device %s in %s seconds" % (device_id, elapsed_time)) + log.warning("WARNING: worker finished %s of device %s in %s seconds" % (action, device_id, elapsed_time)) except (KeyboardInterrupt, SystemExit): raise except: @@ -200,12 +200,14 @@ dev_query = ('SELECT device_id, ' 'disabled = 0 ' 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' - '{2} ' + 'AND last_poll_attempted < DATE_SUB( ' + ' NOW(), INTERVAL {2} SECOND ) ' + '{3} ' 'ORDER BY next_poll asc ').format(poll_frequency, discover_frequency, + down_retry, poller_group) -dont_retry = {} threads = 0 next_update = datetime.now() + timedelta(minutes=5) devices_scanned = 0 @@ -220,28 +222,26 @@ while True: seconds_taken = (datetime.now() - (next_update - timedelta(minutes=5))).seconds update_query = ('INSERT INTO pollers(poller_name, ' ' last_polled, ' - ' devices, ' + ' devices, ' ' time_taken) ' ' values("{0}", NOW(), "{1}", "{2}") ' 'ON DUPLICATE KEY UPDATE ' ' last_polled=values(last_polled), ' ' devices=values(devices) ' ' time_taken=values(time_taken) ').format(config['distributed_poller_name'], - devices_scanned, - seconds_taken) + devices_scanned, + seconds_taken) cursor.execute(update_query) + cursor.fetchall() log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 next_update = datetime.now() + timedelta(minutes=5) - dont_retry = dict((dev, time) for dev, time in dont_retry.iteritems() if time > datetime.now()) - while threading.active_count() >= amount_of_workers: time.sleep(.5) cursor.execute(dev_query) devices = cursor.fetchall() - retry_devs_found = {} for device_id, next_poll, next_discovery in devices: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already @@ -256,22 +256,8 @@ while True: continue if next_poll and next_poll > datetime.now(): - poll_action = 'polling' - sleep_until_time = next_poll - try: - next_retry_device = min(retry_devs_found, key=retry_devs_found.get) - next_retry_time = retry_devs_found[next_retry_device] - if next_retry_time < next_poll: - device_id = next_retry_device - if not getLock('queued.{}'.format(device_id)): - continue - poll_action = 'retrying failed device' - sleep_until_time = next_retry_time - except ValueError: - pass - - log.debug('DEBUG: Sleeping until {0} before {1} {2}'.format(next_poll, poll_action, device_id)) - sleep_until(sleep_until_time) + log.debug('DEBUG: Sleeping until {0} before polling {2}'.format(next_poll, device_id)) + sleep_until(next_poll) action = 'poll' if not next_discovery or next_discovery < datetime.now(): @@ -279,7 +265,10 @@ while True: log.debug('DEBUG: Starting {} of device {}'.format(action, device_id)) devices_scanned += 1 - dont_retry[device_id] = datetime.now() + timedelta(seconds=down_retry) + + cursor.execute('UPDATE devices SET last_poll_attempted = NOW() WHERE device_id = {}'.format(device_id)) + cursor.fetchall() + t = threading.Thread(target=poll_worker, args=[device_id, action]) t.start() From d23bed0af7686b08fa4a28b85bd4f2a7e1a551e6 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 09:40:54 -0400 Subject: [PATCH 37/82] don't check dont-retry' --- poller-service.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/poller-service.py b/poller-service.py index 43ee9875c..8852a1a65 100755 --- a/poller-service.py +++ b/poller-service.py @@ -250,10 +250,6 @@ while True: if not lockFree('polling.{}'.format(device_id)): releaseLock('queued.{}'.format(device_id)) continue - if device_id in dont_retry: - retry_devs_found[device_id] = dont_retry[device_id] - releaseLock('queued.{}'.format(device_id)) - continue if next_poll and next_poll > datetime.now(): log.debug('DEBUG: Sleeping until {0} before polling {2}'.format(next_poll, device_id)) From c35a1ea750ca2dd456b285a956e109304fd4635e Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 09:48:18 -0400 Subject: [PATCH 38/82] don't loop too fast --- poller-service.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/poller-service.py b/poller-service.py index 8852a1a65..e57f8f7ae 100755 --- a/poller-service.py +++ b/poller-service.py @@ -270,3 +270,7 @@ while True: # If we made it this far, break out of the loop and query again. break + + # Looping with no break causes the service to be killed by init. + # This point is only reached if the query is empty, so sleep half a second before querying again. + time.sleep(.5) From 4a66720969596805a7cf39207c1032c98bd2dbf2 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 09:59:08 -0400 Subject: [PATCH 39/82] release lock for 1 device --- poller-service.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index e57f8f7ae..e64c9800b 100755 --- a/poller-service.py +++ b/poller-service.py @@ -201,7 +201,7 @@ dev_query = ('SELECT device_id, ' 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' 'AND last_poll_attempted < DATE_SUB( ' - ' NOW(), INTERVAL {2} SECOND ) ' + ' NOW(), INTERVAL {2} SECOND ) ' '{3} ' 'ORDER BY next_poll asc ').format(poll_frequency, discover_frequency, @@ -268,6 +268,10 @@ while True: t = threading.Thread(target=poll_worker, args=[device_id, action]) t.start() + # If there is only one device, release the queue lock, because it won't release automatically. + if len(devices) == 1: + releaseLock('queued.{}'.format(device_id)) + # If we made it this far, break out of the loop and query again. break From 6c5f71bbb583990f753b6bc3a34ee479214caf71 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:06:57 -0400 Subject: [PATCH 40/82] remove queue locks on empty query --- poller-service.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/poller-service.py b/poller-service.py index e64c9800b..db20994f0 100755 --- a/poller-service.py +++ b/poller-service.py @@ -268,13 +268,12 @@ while True: t = threading.Thread(target=poll_worker, args=[device_id, action]) t.start() - # If there is only one device, release the queue lock, because it won't release automatically. - if len(devices) == 1: - releaseLock('queued.{}'.format(device_id)) - # If we made it this far, break out of the loop and query again. break # Looping with no break causes the service to be killed by init. # This point is only reached if the query is empty, so sleep half a second before querying again. time.sleep(.5) + + # Make sure we're not holding any device queue locks in this connection before querying again. + getLock('unlock.{}'.format(config['distributed_poller_name'])) From e68bebbc625fdd765d98e82934a0d2d9f089d1d4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:10:49 -0400 Subject: [PATCH 41/82] bad index --- poller-service.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index db20994f0..f92eae490 100755 --- a/poller-service.py +++ b/poller-service.py @@ -246,13 +246,15 @@ while True: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already if not getLock('queued.{}'.format(device_id)): + time.sleep(.5) continue if not lockFree('polling.{}'.format(device_id)): releaseLock('queued.{}'.format(device_id)) + time.sleep(.5) continue if next_poll and next_poll > datetime.now(): - log.debug('DEBUG: Sleeping until {0} before polling {2}'.format(next_poll, device_id)) + log.debug('DEBUG: Sleeping until {0} before polling {1}'.format(next_poll, device_id)) sleep_until(next_poll) action = 'poll' From 5672afd0a8acf6659dc1117444235a85ee7eadbb Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:15:06 -0400 Subject: [PATCH 42/82] test for null last_attempted --- poller-service.py | 48 +++++++++++++++++++++++------------------------ 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/poller-service.py b/poller-service.py index f92eae490..32d4deb32 100755 --- a/poller-service.py +++ b/poller-service.py @@ -183,30 +183,30 @@ poller_group = ('and poller_group IN({}) ' # Add last_polled and last_polled_timetaken so we can sort by the time the last poll started, with the goal # of having each device complete a poll within the given time range. -dev_query = ('SELECT device_id, ' - 'DATE_ADD( ' - ' DATE_SUB( ' - ' last_polled, ' - ' INTERVAL last_polled_timetaken SECOND ' - ' ), ' - ' INTERVAL {0} SECOND) AS next_poll, ' - 'DATE_ADD( ' - ' DATE_SUB( ' - ' last_discovered, ' - ' INTERVAL last_discovered_timetaken SECOND ' - ' ), ' - ' INTERVAL {1} SECOND) AS next_discovery ' - 'FROM devices WHERE ' - 'disabled = 0 ' - 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' - 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' - 'AND last_poll_attempted < DATE_SUB( ' - ' NOW(), INTERVAL {2} SECOND ) ' - '{3} ' - 'ORDER BY next_poll asc ').format(poll_frequency, - discover_frequency, - down_retry, - poller_group) +dev_query = ('SELECT device_id, ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_polled, ' + ' INTERVAL last_polled_timetaken SECOND ' + ' ), ' + ' INTERVAL {0} SECOND) AS next_poll, ' + 'DATE_ADD( ' + ' DATE_SUB( ' + ' last_discovered, ' + ' INTERVAL last_discovered_timetaken SECOND ' + ' ), ' + ' INTERVAL {1} SECOND) AS next_discovery ' + 'FROM devices WHERE ' + 'disabled = 0 ' + 'AND IS_FREE_LOCK(CONCAT("polling.", device_id)) ' + 'AND IS_FREE_LOCK(CONCAT("queued.", device_id)) ' + 'AND ( last_poll_attempted < DATE_SUB(NOW(), INTERVAL {2} SECOND ) ' + ' OR last_poll_attempted IS NULL ) ' + '{3} ' + 'ORDER BY next_poll asc ').format(poll_frequency, + discover_frequency, + down_retry, + poller_group) threads = 0 next_update = datetime.now() + timedelta(minutes=5) From c3b6fee0eb2705ef1f0fac64c590b175159904c3 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:51:56 -0400 Subject: [PATCH 43/82] bad sql --- poller-service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index 32d4deb32..13841df20 100755 --- a/poller-service.py +++ b/poller-service.py @@ -227,8 +227,8 @@ while True: ' values("{0}", NOW(), "{1}", "{2}") ' 'ON DUPLICATE KEY UPDATE ' ' last_polled=values(last_polled), ' - ' devices=values(devices) ' - ' time_taken=values(time_taken) ').format(config['distributed_poller_name'], + ' devices=values(devices), ' + ' time_taken=values(time_taken) ').format(config['distributed_poller_name'].strip(), devices_scanned, seconds_taken) cursor.execute(update_query) From 4d832067a9ace2e921b43f1be3c600ec88e9d138 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:57:49 -0400 Subject: [PATCH 44/82] change to 2 minute updates --- poller-service.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/poller-service.py b/poller-service.py index 13841df20..3f9f96082 100755 --- a/poller-service.py +++ b/poller-service.py @@ -209,7 +209,7 @@ dev_query = ('SELECT device_id, poller_group) threads = 0 -next_update = datetime.now() + timedelta(minutes=5) +next_update = datetime.now() + timedelta(minutes=2) devices_scanned = 0 while True: @@ -219,7 +219,7 @@ while True: log.debug('DEBUG: {} threads currently active'.format(threads)) if next_update < datetime.now(): - seconds_taken = (datetime.now() - (next_update - timedelta(minutes=5))).seconds + seconds_taken = (datetime.now() - (next_update - timedelta(minutes=2))).seconds update_query = ('INSERT INTO pollers(poller_name, ' ' last_polled, ' ' devices, ' @@ -235,7 +235,7 @@ while True: cursor.fetchall() log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 - next_update = datetime.now() + timedelta(minutes=5) + next_update = datetime.now() + timedelta(minutes=2) while threading.active_count() >= amount_of_workers: time.sleep(.5) From 9f1c2232f02812cdcc380f9f253c13d3136e1dc0 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 10:59:26 -0400 Subject: [PATCH 45/82] update every minute --- poller-service.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/poller-service.py b/poller-service.py index 3f9f96082..8591cdfe0 100755 --- a/poller-service.py +++ b/poller-service.py @@ -209,7 +209,7 @@ dev_query = ('SELECT device_id, poller_group) threads = 0 -next_update = datetime.now() + timedelta(minutes=2) +next_update = datetime.now() + timedelta(minutes=1) devices_scanned = 0 while True: @@ -219,7 +219,7 @@ while True: log.debug('DEBUG: {} threads currently active'.format(threads)) if next_update < datetime.now(): - seconds_taken = (datetime.now() - (next_update - timedelta(minutes=2))).seconds + seconds_taken = (datetime.now() - (next_update - timedelta(minutes=1))).seconds update_query = ('INSERT INTO pollers(poller_name, ' ' last_polled, ' ' devices, ' @@ -235,7 +235,7 @@ while True: cursor.fetchall() log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 - next_update = datetime.now() + timedelta(minutes=2) + next_update = datetime.now() + timedelta(minutes=1) while threading.active_count() >= amount_of_workers: time.sleep(.5) From 3ff7a4808903a9d0ba2d0563e9d7719c0a816461 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 11:10:14 -0400 Subject: [PATCH 46/82] catch sql errors --- poller-service.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index 8591cdfe0..b1170ed58 100755 --- a/poller-service.py +++ b/poller-service.py @@ -231,7 +231,10 @@ while True: ' time_taken=values(time_taken) ').format(config['distributed_poller_name'].strip(), devices_scanned, seconds_taken) - cursor.execute(update_query) + try: + cursor.execute(update_query) + except: + log.critical('ERROR: MySQL query error. Is your schema up to date?') cursor.fetchall() log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 @@ -240,7 +243,11 @@ while True: while threading.active_count() >= amount_of_workers: time.sleep(.5) - cursor.execute(dev_query) + try: + cursor.execute(dev_query) + except: + log.critical('ERROR: MySQL query error. Is your schema up to date?') + devices = cursor.fetchall() for device_id, next_poll, next_discovery in devices: # add queue lock, so we lock the next device against any other pollers From 21e9dac646da714a213ce0ec7bbbe98140b0185c Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 11:16:51 -0400 Subject: [PATCH 47/82] exit on mysql error --- poller-service.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/poller-service.py b/poller-service.py index b1170ed58..7f94f1423 100755 --- a/poller-service.py +++ b/poller-service.py @@ -235,6 +235,7 @@ while True: cursor.execute(update_query) except: log.critical('ERROR: MySQL query error. Is your schema up to date?') + sys.exit(2) cursor.fetchall() log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 @@ -247,6 +248,7 @@ while True: cursor.execute(dev_query) except: log.critical('ERROR: MySQL query error. Is your schema up to date?') + sys.exit(2) devices = cursor.fetchall() for device_id, next_poll, next_discovery in devices: From 9931d9dc442d0f3d4a1f2a2a29504f40ceec0a0b Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Mon, 13 Jul 2015 14:32:00 -0400 Subject: [PATCH 48/82] don't try to discover down devices --- poller-service.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/poller-service.py b/poller-service.py index 7f94f1423..e9aa123bc 100755 --- a/poller-service.py +++ b/poller-service.py @@ -183,7 +183,7 @@ poller_group = ('and poller_group IN({}) ' # Add last_polled and last_polled_timetaken so we can sort by the time the last poll started, with the goal # of having each device complete a poll within the given time range. -dev_query = ('SELECT device_id, ' +dev_query = ('SELECT device_id, status, ' 'DATE_ADD( ' ' DATE_SUB( ' ' last_polled, ' @@ -251,7 +251,7 @@ while True: sys.exit(2) devices = cursor.fetchall() - for device_id, next_poll, next_discovery in devices: + for device_id, status, next_poll, next_discovery in devices: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already if not getLock('queued.{}'.format(device_id)): @@ -267,7 +267,7 @@ while True: sleep_until(next_poll) action = 'poll' - if not next_discovery or next_discovery < datetime.now(): + if (not next_discovery or next_discovery < datetime.now()) and status == 1: action = 'discovery' log.debug('DEBUG: Starting {} of device {}'.format(action, device_id)) From b73b49bfcb36c39676fa4fd1799402918d880648 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 16 Jul 2015 11:55:04 -0400 Subject: [PATCH 49/82] remove unnecessary exit conditions --- poller-service.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/poller-service.py b/poller-service.py index e9aa123bc..c9d39b8b5 100755 --- a/poller-service.py +++ b/poller-service.py @@ -89,32 +89,28 @@ db_dbname = config['db_name'] try: amount_of_workers = int(config['poller_service_workers']) if amount_of_workers == 0: - log.critical("ERROR: 0 threads is not a valid value") - sys.exit(2) + amount_of_workers = 16 except KeyError: amount_of_workers = 16 try: poll_frequency = int(config['poller_service_poll_frequency']) if poll_frequency == 0: - log.critical("ERROR: 0 seconds is not a valid value") - sys.exit(2) + poll_frequency = 300 except KeyError: poll_frequency = 300 try: discover_frequency = int(config['poller_service_discover_frequency']) if discover_frequency == 0: - log.critical("ERROR: 0 seconds is not a valid value") - sys.exit(2) + discover_frequency = 21600 except KeyError: discover_frequency = 21600 try: down_retry = int(config['poller_service_down_retry']) if down_retry == 0: - log.critical("ERROR: 0 seconds is not a valid value") - sys.exit(2) + down_retry = 60 except KeyError: down_retry = 60 From bb697a35bd81bd71428edd6593ac49bc7332e035 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 16 Jul 2015 12:11:45 -0400 Subject: [PATCH 50/82] attribution --- poller-service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/poller-service.py b/poller-service.py index c9d39b8b5..0ca3b5cd9 100755 --- a/poller-service.py +++ b/poller-service.py @@ -4,12 +4,12 @@ devices that have been polled within the last $poll_frequency seconds. It will prioritize devices based on the last time polled. If resources are sufficient, this service should poll every device every $poll_frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as - frequently as possible. This service is based on poller-wrapper.py. + frequently as possible. This service is based on Job Snijders' poller-wrapper.py. Author: Clint Armstrong Date: July 2015 - License: BSD + License: BSD 2-Clause """ import json From c9a56e248d0312568318fee0b758e8d59e0db215 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 17 Jul 2015 13:33:24 -0400 Subject: [PATCH 51/82] index and limit --- poller-service.py | 9 ++++----- sql-schema/065.sql | 2 ++ 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/poller-service.py b/poller-service.py index 0ca3b5cd9..bbcfa7617 100755 --- a/poller-service.py +++ b/poller-service.py @@ -199,7 +199,8 @@ dev_query = ('SELECT device_id, status, 'AND ( last_poll_attempted < DATE_SUB(NOW(), INTERVAL {2} SECOND ) ' ' OR last_poll_attempted IS NULL ) ' '{3} ' - 'ORDER BY next_poll asc ').format(poll_frequency, + 'ORDER BY next_poll asc ' + 'LIMIT 5 ').format(poll_frequency, discover_frequency, down_retry, poller_group) @@ -251,11 +252,9 @@ while True: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already if not getLock('queued.{}'.format(device_id)): - time.sleep(.5) continue if not lockFree('polling.{}'.format(device_id)): releaseLock('queued.{}'.format(device_id)) - time.sleep(.5) continue if next_poll and next_poll > datetime.now(): @@ -278,9 +277,9 @@ while True: # If we made it this far, break out of the loop and query again. break - # Looping with no break causes the service to be killed by init. # This point is only reached if the query is empty, so sleep half a second before querying again. time.sleep(.5) - # Make sure we're not holding any device queue locks in this connection before querying again. + # Make sure we're not holding any device queue locks in this connection before querying again + # by locking a different string. getLock('unlock.{}'.format(config['distributed_poller_name'])) diff --git a/sql-schema/065.sql b/sql-schema/065.sql index ab64beca9..4fb76d358 100644 --- a/sql-schema/065.sql +++ b/sql-schema/065.sql @@ -1,3 +1,5 @@ DELETE n1 FROM pollers n1, pollers n2 WHERE n1.last_polled < n2.last_polled and n1.poller_name = n2.poller_name; ALTER TABLE pollers ADD PRIMARY KEY (poller_name); ALTER TABLE `devices` ADD `last_poll_attempted` timestamp NULL DEFAULT NULL AFTER `last_polled`; +ALTER TABLE `devices` ADD INDEX `last_polled` (`last_polled`); +ALTER TABLE `devices` ADD INDEX `last_poll_attempted` (`last_poll_attempted`); From 71c1771d8492e6ea57f710c5d9e3bb28fdd47a7d Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 20:04:44 -0400 Subject: [PATCH 52/82] zero length field in format --- poller-service.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/poller-service.py b/poller-service.py index bbcfa7617..82b486316 100755 --- a/poller-service.py +++ b/poller-service.py @@ -64,7 +64,7 @@ try: except KeyError: loglevel = logging.getLevelName('INFO') if not isinstance(loglevel, int): - log.warning('ERROR: {} is not a valid log level'.format(str(loglevel))) + log.warning('ERROR: {0} is not a valid log level'.format(str(loglevel))) loglevel = logging.getLevelName('INFO') log.setLevel(loglevel) @@ -147,21 +147,21 @@ def poll_worker(device_id, action): def lockFree(lock): global cursor - query = "SELECT IS_FREE_LOCK('{}')".format(lock) + query = "SELECT IS_FREE_LOCK('{0}')".format(lock) cursor.execute(query) return cursor.fetchall()[0][0] == 1 def getLock(lock): global cursor - query = "SELECT GET_LOCK('{}', 0)".format(lock) + query = "SELECT GET_LOCK('{0}', 0)".format(lock) cursor.execute(query) return cursor.fetchall()[0][0] == 1 def releaseLock(lock): global cursor - query = "SELECT RELEASE_LOCK('{}')".format(lock) + query = "SELECT RELEASE_LOCK('{0}')".format(lock) cursor.execute(query) return cursor.fetchall()[0][0] == 1 @@ -174,7 +174,7 @@ def sleep_until(timestamp): sleeptime = 0 time.sleep(sleeptime) -poller_group = ('and poller_group IN({}) ' +poller_group = ('and poller_group IN({0}) ' .format(str(config['distributed_poller_group'])) if 'distributed_poller_group' in config else '') # Add last_polled and last_polled_timetaken so we can sort by the time the last poll started, with the goal @@ -213,7 +213,7 @@ while True: cur_threads = threading.active_count() if cur_threads != threads: threads = cur_threads - log.debug('DEBUG: {} threads currently active'.format(threads)) + log.debug('DEBUG: {0} threads currently active'.format(threads)) if next_update < datetime.now(): seconds_taken = (datetime.now() - (next_update - timedelta(minutes=1))).seconds @@ -234,7 +234,7 @@ while True: log.critical('ERROR: MySQL query error. Is your schema up to date?') sys.exit(2) cursor.fetchall() - log.info('INFO: {} devices scanned in the last 5 minutes'.format(devices_scanned)) + log.info('INFO: {0} devices scanned in the last 5 minutes'.format(devices_scanned)) devices_scanned = 0 next_update = datetime.now() + timedelta(minutes=1) @@ -251,10 +251,10 @@ while True: for device_id, status, next_poll, next_discovery in devices: # add queue lock, so we lock the next device against any other pollers # if this fails, the device is locked by another poller already - if not getLock('queued.{}'.format(device_id)): + if not getLock('queued.{0}'.format(device_id)): continue - if not lockFree('polling.{}'.format(device_id)): - releaseLock('queued.{}'.format(device_id)) + if not lockFree('polling.{0}'.format(device_id)): + releaseLock('queued.{0}'.format(device_id)) continue if next_poll and next_poll > datetime.now(): @@ -265,10 +265,10 @@ while True: if (not next_discovery or next_discovery < datetime.now()) and status == 1: action = 'discovery' - log.debug('DEBUG: Starting {} of device {}'.format(action, device_id)) + log.debug('DEBUG: Starting {0} of device {1}'.format(action, device_id)) devices_scanned += 1 - cursor.execute('UPDATE devices SET last_poll_attempted = NOW() WHERE device_id = {}'.format(device_id)) + cursor.execute('UPDATE devices SET last_poll_attempted = NOW() WHERE device_id = {0}'.format(device_id)) cursor.fetchall() t = threading.Thread(target=poll_worker, args=[device_id, action]) @@ -282,4 +282,4 @@ while True: # Make sure we're not holding any device queue locks in this connection before querying again # by locking a different string. - getLock('unlock.{}'.format(config['distributed_poller_name'])) + getLock('unlock.{0}'.format(config['distributed_poller_name'])) From 73ada34f306b3c8608ae544b8f19730401a4873e Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 20:37:34 -0400 Subject: [PATCH 53/82] make logging python3 compatible --- poller-service.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/poller-service.py b/poller-service.py index 82b486316..c0c861413 100755 --- a/poller-service.py +++ b/poller-service.py @@ -44,7 +44,7 @@ def get_config_data(): except: log.critical("ERROR: Could not execute: %s" % config_cmd) sys.exit(2) - return proc.communicate()[0] + return proc.communicate()[0].decode() try: with open(config_file) as f: @@ -60,13 +60,12 @@ except: sys.exit(2) try: - loglevel = logging.getLevelName(config['poller_service_loglevel'].upper()) + log.setLevel(config['poller_service_loglevel'].upper()) except KeyError: - loglevel = logging.getLevelName('INFO') -if not isinstance(loglevel, int): - log.warning('ERROR: {0} is not a valid log level'.format(str(loglevel))) - loglevel = logging.getLevelName('INFO') -log.setLevel(loglevel) + log.setLevel('INFO') +except ValueError: + log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'])) + log.setLevel('INFO') poller_path = config['install_dir'] + '/poller.php' discover_path = config['install_dir'] + '/discovery.php' From 9022c168367732d300b9f4552dc032e0839e71d1 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 20:57:25 -0400 Subject: [PATCH 54/82] upper case warning --- poller-service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index c0c861413..fd97860e2 100755 --- a/poller-service.py +++ b/poller-service.py @@ -64,7 +64,7 @@ try: except KeyError: log.setLevel('INFO') except ValueError: - log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'])) + log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'].upper())) log.setLevel('INFO') poller_path = config['install_dir'] + '/poller.php' From 0f34042b84aaff7c4a5f732f027d25dc738944e4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 21:09:10 -0400 Subject: [PATCH 55/82] Revert "upper case warning" This reverts commit c87d626105ad80d7856e2c15f98219c43ebe0c2e. --- poller-service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index fd97860e2..c0c861413 100755 --- a/poller-service.py +++ b/poller-service.py @@ -64,7 +64,7 @@ try: except KeyError: log.setLevel('INFO') except ValueError: - log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'].upper())) + log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'])) log.setLevel('INFO') poller_path = config['install_dir'] + '/poller.php' From 85aa9ccdb8b530aac72221bcc62260b5edd31f66 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 21:09:20 -0400 Subject: [PATCH 56/82] Revert "make logging python3 compatible" This reverts commit 74531c6e69c31c252a48d2bab90bc98e4f566111. --- poller-service.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/poller-service.py b/poller-service.py index c0c861413..82b486316 100755 --- a/poller-service.py +++ b/poller-service.py @@ -44,7 +44,7 @@ def get_config_data(): except: log.critical("ERROR: Could not execute: %s" % config_cmd) sys.exit(2) - return proc.communicate()[0].decode() + return proc.communicate()[0] try: with open(config_file) as f: @@ -60,12 +60,13 @@ except: sys.exit(2) try: - log.setLevel(config['poller_service_loglevel'].upper()) + loglevel = logging.getLevelName(config['poller_service_loglevel'].upper()) except KeyError: - log.setLevel('INFO') -except ValueError: - log.warning('ERROR: {0} is not a valid log level'.format(config['poller_service_loglevel'])) - log.setLevel('INFO') + loglevel = logging.getLevelName('INFO') +if not isinstance(loglevel, int): + log.warning('ERROR: {0} is not a valid log level'.format(str(loglevel))) + loglevel = logging.getLevelName('INFO') +log.setLevel(loglevel) poller_path = config['install_dir'] + '/poller.php' discover_path = config['install_dir'] + '/discovery.php' From 403d4c53646c11825d85cb4ac0acfb89bdf599a8 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 21:09:49 -0400 Subject: [PATCH 57/82] decode proc --- poller-service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index 82b486316..3a98a6ac2 100755 --- a/poller-service.py +++ b/poller-service.py @@ -44,7 +44,7 @@ def get_config_data(): except: log.critical("ERROR: Could not execute: %s" % config_cmd) sys.exit(2) - return proc.communicate()[0] + return proc.communicate()[0].decode() try: with open(config_file) as f: From 779862601762680a120c9370fc4b17dd1c614646 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 21:31:01 -0400 Subject: [PATCH 58/82] fix logging options --- poller-service.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/poller-service.py b/poller-service.py index 3a98a6ac2..4fcf37818 100755 --- a/poller-service.py +++ b/poller-service.py @@ -60,12 +60,14 @@ except: sys.exit(2) try: - loglevel = logging.getLevelName(config['poller_service_loglevel'].upper()) + loglevel = config['poller_service_loglevel'] except KeyError: - loglevel = logging.getLevelName('INFO') + loglevel = 20 if not isinstance(loglevel, int): - log.warning('ERROR: {0} is not a valid log level'.format(str(loglevel))) - loglevel = logging.getLevelName('INFO') + loglevel = logging.getLevelName(loglevel) +if not isinstance(loglevel, int): + log.warning('ERROR: {0} is not a valid log level. If using python 3.4.0-3.4.1 you must specify loglevel by number'.format(str(loglevel))) + loglevel = 20 log.setLevel(loglevel) poller_path = config['install_dir'] + '/poller.php' From 617e695b2012fb94303d555376b151fc83b64ff4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Sat, 18 Jul 2015 21:38:27 -0400 Subject: [PATCH 59/82] better logging error checking --- poller-service.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/poller-service.py b/poller-service.py index 4fcf37818..0989b27a2 100755 --- a/poller-service.py +++ b/poller-service.py @@ -60,15 +60,17 @@ except: sys.exit(2) try: - loglevel = config['poller_service_loglevel'] + loglevel = int(config['poller_service_loglevel']) except KeyError: loglevel = 20 -if not isinstance(loglevel, int): - loglevel = logging.getLevelName(loglevel) -if not isinstance(loglevel, int): +except ValueError: + loglevel = logging.getLevelName(config['poller_service_loglevel']) + +try: + log.setLevel(loglevel) +except ValueError: log.warning('ERROR: {0} is not a valid log level. If using python 3.4.0-3.4.1 you must specify loglevel by number'.format(str(loglevel))) - loglevel = 20 -log.setLevel(loglevel) + log.setLevel(20) poller_path = config['install_dir'] + '/poller.php' discover_path = config['install_dir'] + '/discovery.php' From 9039dbb425555c48d9ce4ac5927e65f97ae7c803 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 07:37:02 -0400 Subject: [PATCH 60/82] give discovery it's own lock --- discovery.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/discovery.php b/discovery.php index 93ca69a09..a8e48c649 100755 --- a/discovery.php +++ b/discovery.php @@ -106,8 +106,8 @@ if ($config['distributed_poller'] === true) { } foreach (dbFetch("SELECT * FROM `devices` WHERE status = 1 AND disabled = 0 $where ORDER BY device_id DESC") as $device) { - if (dbGetLock('polling.' . $device['device_id'])) { - discover_device($device, $options); + if (dbGetLock('discovering.' . $device['device_id'])) { + discover_device($device, $options); } } From a47ae33cfd0e096bbc980e4370eca7e79f571983 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 07:39:37 -0400 Subject: [PATCH 61/82] check for schema_update lock when poller-service runs --- poller-service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index 0989b27a2..178282026 100755 --- a/poller-service.py +++ b/poller-service.py @@ -242,7 +242,7 @@ while True: devices_scanned = 0 next_update = datetime.now() + timedelta(minutes=1) - while threading.active_count() >= amount_of_workers: + while threading.active_count() >= amount_of_workers or not lockFree('schema_update'): time.sleep(.5) try: From 86e0f42dc75a678d1a78b0f20a179b47c0215928 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 07:52:58 -0400 Subject: [PATCH 62/82] bail out if we can't get a lock on schema_update --- includes/sql-schema/update.php | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 4bb7df721..19b043ada 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -31,6 +31,11 @@ if (!isset($debug)) { } } +if (!dbGetLock('schema_update')) { + echo "Schema update already in progress. Exiting"; + exit(1); +} + $insert = 0; if ($db_rev = @dbFetchCell('SELECT version FROM `dbSchema` ORDER BY version DESC LIMIT 1')) { From 777d5d6c6f586dedaeadc2c607b172ed9c36e4a4 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 07:58:36 -0400 Subject: [PATCH 63/82] add function to check for a lock --- includes/dbFacile.mysql.php | 13 +++++++------ includes/dbFacile.mysqli.php | 11 ++++++----- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/includes/dbFacile.mysql.php b/includes/dbFacile.mysql.php index 9fd82c698..bac21b008 100644 --- a/includes/dbFacile.mysql.php +++ b/includes/dbFacile.mysql.php @@ -60,12 +60,12 @@ function dbQuery($sql, $parameters=array()) { /* - * Aquire a lock on a string + * Check a lock on a string * */ -function dbGetLock($data, $timeout = 0) { - $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; - $result = dbFetchCell($sql); - return $result; +function dbCheckLock($data, $timeout = 0) { + $sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')'; + $result = dbFetchCell($sql); + return $result; } @@ -76,8 +76,9 @@ function dbReleaseLock($data, $timeout = 0) { $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; $result = dbFetchCell($sql); return $result; - +} + /* * Passed an array and a table name, it attempts to insert the data into the table. * Check for boolean false to determine whether insert failed diff --git a/includes/dbFacile.mysqli.php b/includes/dbFacile.mysqli.php index 36b2208cb..158a1dae5 100644 --- a/includes/dbFacile.mysqli.php +++ b/includes/dbFacile.mysqli.php @@ -60,12 +60,12 @@ function dbQuery($sql, $parameters=array()) { /* - * Aquire a lock on a string + * Check a lock on a string * */ -function dbGetLock($data, $timeout = 0) { - $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; - $result = dbFetchCell($sql); - return $result; +function dbCheckLock($data, $timeout = 0) { + $sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')'; + $result = dbFetchCell($sql); + return $result; } @@ -76,6 +76,7 @@ function dbReleaseLock($data, $timeout = 0) { $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; $result = dbFetchCell($sql); return $result; +} /* From 1379a13c7686cfef66f52aa85625d609cf8b7400 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:09:19 -0400 Subject: [PATCH 64/82] wait for all locks to be free when updating schema --- includes/sql-schema/update.php | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 19b043ada..46233e8dd 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -34,7 +34,11 @@ if (!isset($debug)) { if (!dbGetLock('schema_update')) { echo "Schema update already in progress. Exiting"; exit(1); -} +} //end if + +do { + sleep(1); +} while (@dbFetchCell('SELECT COUNT(*) FROM `devices` WHERE NOT IS_FREE_LOCK(CONCAT("polling.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("queued.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("discovering.", device_id))') > 0); $insert = 0; From c5fcb87b51b7590195e3e5b27570ace0fc8d8da6 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:31:51 -0400 Subject: [PATCH 65/82] bail if schema is already up to date --- includes/sql-schema/update.php | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 46233e8dd..23038e5fc 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -32,7 +32,7 @@ if (!isset($debug)) { } if (!dbGetLock('schema_update')) { - echo "Schema update already in progress. Exiting"; + echo "Schema update already in progress. Exiting\n"; exit(1); } //end if @@ -93,6 +93,13 @@ if ($handle = opendir($config['install_dir'].'/sql-schema')) { asort($filelist); +if (explode('.', max($filelist), 2) <= $db_rev) { + if ($debug) { + echo "DB Schema already up to date.\n"; + } + exit(0); +} + foreach ($filelist as $file) { list($filename,$extension) = explode('.', $file, 2); if ($filename > $db_rev) { From e74a6db21650f1e3b08d4f190450ea0a712a1f2f Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:32:30 -0400 Subject: [PATCH 66/82] move lock checks after bail out --- includes/sql-schema/update.php | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 23038e5fc..c07909845 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -31,15 +31,6 @@ if (!isset($debug)) { } } -if (!dbGetLock('schema_update')) { - echo "Schema update already in progress. Exiting\n"; - exit(1); -} //end if - -do { - sleep(1); -} while (@dbFetchCell('SELECT COUNT(*) FROM `devices` WHERE NOT IS_FREE_LOCK(CONCAT("polling.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("queued.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("discovering.", device_id))') > 0); - $insert = 0; if ($db_rev = @dbFetchCell('SELECT version FROM `dbSchema` ORDER BY version DESC LIMIT 1')) { @@ -100,6 +91,15 @@ if (explode('.', max($filelist), 2) <= $db_rev) { exit(0); } +if (!dbGetLock('schema_update')) { + echo "Schema update already in progress. Exiting\n"; + exit(1); +} //end if + +do { + sleep(1); +} while (@dbFetchCell('SELECT COUNT(*) FROM `devices` WHERE NOT IS_FREE_LOCK(CONCAT("polling.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("queued.", device_id)) OR NOT IS_FREE_LOCK(CONCAT("discovering.", device_id))') > 0); + foreach ($filelist as $file) { list($filename,$extension) = explode('.', $file, 2); if ($filename > $db_rev) { From d9efadb441c243fb39a1dfe9eb26850ce1061013 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:44:55 -0400 Subject: [PATCH 67/82] fix bailout comparison --- includes/sql-schema/update.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index c07909845..141139a35 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -84,7 +84,7 @@ if ($handle = opendir($config['install_dir'].'/sql-schema')) { asort($filelist); -if (explode('.', max($filelist), 2) <= $db_rev) { +if (explode('.', max($filelist), 2)[0] <= $db_rev) { if ($debug) { echo "DB Schema already up to date.\n"; } From 38af8b4c4c6cf568487342b3469811091466b32b Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:48:33 -0400 Subject: [PATCH 68/82] return instead of exit --- includes/sql-schema/update.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 141139a35..1d0c70804 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -88,7 +88,7 @@ if (explode('.', max($filelist), 2)[0] <= $db_rev) { if ($debug) { echo "DB Schema already up to date.\n"; } - exit(0); + return; } if (!dbGetLock('schema_update')) { From 33baea6d6a9edcec35f1e01690aca86405feff46 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 08:50:35 -0400 Subject: [PATCH 69/82] release schema lock --- includes/sql-schema/update.php | 2 ++ 1 file changed, 2 insertions(+) diff --git a/includes/sql-schema/update.php b/includes/sql-schema/update.php index 1d0c70804..154ac007c 100644 --- a/includes/sql-schema/update.php +++ b/includes/sql-schema/update.php @@ -169,3 +169,5 @@ if ($updating) { echo "-- Done\n"; } + +dbReleaseLock('schema_update'); From 6f0969c1516b15c246606d3b16e12bbd5f6ccbf0 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Wed, 22 Jul 2015 09:04:12 -0400 Subject: [PATCH 70/82] typo --- poller-service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.py b/poller-service.py index 178282026..74e584e64 100755 --- a/poller-service.py +++ b/poller-service.py @@ -238,7 +238,7 @@ while True: log.critical('ERROR: MySQL query error. Is your schema up to date?') sys.exit(2) cursor.fetchall() - log.info('INFO: {0} devices scanned in the last 5 minutes'.format(devices_scanned)) + log.info('INFO: {0} devices scanned in the last minute'.format(devices_scanned)) devices_scanned = 0 next_update = datetime.now() + timedelta(minutes=1) From 507e4baa95daacec409e31077cbea65f5175ccba Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 14:48:08 -0400 Subject: [PATCH 71/82] simplify poller-service.conf --- poller-service.conf | 9 --------- 1 file changed, 9 deletions(-) diff --git a/poller-service.conf b/poller-service.conf index 36677e7d2..b41af1830 100644 --- a/poller-service.conf +++ b/poller-service.conf @@ -12,15 +12,6 @@ stop on runlevel [016] # Automatically restart process if crashed respawn -# Essentially lets upstart know the process will detach itself to the background -#expect fork - -# # Run before process -# pre-start script -# [ -d /var/run/myservice ] || mkdir -p /var/run/myservice -# echo "Put bash code here" -# end script - chdir /opt/librenms setuid librenms setgid librenms From bba88039338fdd7a22508a9b149164a0d226bc67 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 14:49:34 -0400 Subject: [PATCH 72/82] symlink instead of copy --- doc/Extensions/Poller-Service.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 354d58c39..d6efd97a2 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -21,4 +21,4 @@ $config['poller_service_down_retry'] = 60; Distributed polling is possible, and uses the same configuration options as are described for traditional distributed polling, except that the memcached options are not necessary. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. ## Service Installation -The service is tested on Ubuntu 14.04. An upstart configuration `poller-service.conf` is provided. To install copy this file to `/etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. +The service is tested on Ubuntu 14.04. An upstart configuration `poller-service.conf` is provided. To install symlink this file to `/etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. From ef91005f6c9b6a84a425f6612bf3efa33409e269 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 14:53:37 -0400 Subject: [PATCH 73/82] add LSB script --- poller-service.init | 88 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100755 poller-service.init diff --git a/poller-service.init b/poller-service.init new file mode 100755 index 000000000..ff5b4766c --- /dev/null +++ b/poller-service.init @@ -0,0 +1,88 @@ +### BEGIN INIT INFO +# Provides: poller-service +# Required-Start: networking +# Required-Stop: networking +# Default-Start: 2 3 4 5 +# Default-Stop: 0 1 6 +# Short-Description: The LibreNMS poller-service daemon +# Description: The LibreNMS poller-service daemon +# This polls devices monitored by LibreNMS +### END INIT INFO + +./lib/lsb/init-functions + +NAME=poller-service + +DAEMON=/opt/librenms/poller-service.py + +PIDFILE=/var/run/poller-service.pid + +test -x $DAEMON || exit 5 + +case $1 in + + start) + # Checked the PID file exists and check the actual status of process + if [ -e $PIDFILE ]; then + status_of_proc -p $PIDFILE $DAEMON "$NAME process" && status="0" || status="$?" + # If the status is SUCCESS then don't need to start again. + if [ $status = "0" ]; then + exit # Exit + fi + fi + # Start the daemon. + log_daemon_msg "Starting the process" "$NAME" + # Start the daemon with the help of start-stop-daemon + # Log the message appropriately + if start-stop-daemon --start --quiet --oknodo --pidfile $PIDFILE --exec $DAEMON ; then + log_end_msg 0 + else + log_end_msg 1 + fi + ;; + + stop) + # Stop the daemon. + if [ -e $PIDFILE ]; then + status_of_proc -p $PIDFILE $DAEMON "Stoppping the $NAME process" && status="0" || status="$?" + if [ "$status" = 0 ]; then + start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE + /bin/rm -rf $PIDFILE + fi + else + log_daemon_msg "$NAME process is not running" + log_end_msg 0 + fi + ;; + restart) + # Restart the daemon. + $0 stop && sleep 2 && $0 start + ;; + + status) + # Check the status of the process. + if [ -e $PIDFILE ]; then + status_of_proc -p $PIDFILE $DAEMON "$NAME process" && exit 0 || exit $? + else + log_daemon_msg "$NAME Process is not running" + log_end_msg 0 + fi + ;; + + reload) + # Reload the process. Basically sending some signal to a daemon to reload + # it configurations. + if [ -e $PIDFILE ]; then + start-stop-daemon --stop --signal USR1 --quiet --pidfile $PIDFILE --name $NAME + log_success_msg "$NAME process reloaded successfully" + else + log_failure_msg "$PIDFILE does not exists" + fi + ;; + + *) + # For invalid arguments, print the usage message. + echo "Usage: $0 {start|stop|restart|reload|status}" + exit 2 + ;; +esac From 4c5f7f9dd1ce34ac23849fda751fa976b5719367 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 14:56:42 -0400 Subject: [PATCH 74/82] add lsb docs --- doc/Extensions/Poller-Service.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index d6efd97a2..4d709ccd6 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -21,4 +21,6 @@ $config['poller_service_down_retry'] = 60; Distributed polling is possible, and uses the same configuration options as are described for traditional distributed polling, except that the memcached options are not necessary. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. ## Service Installation -The service is tested on Ubuntu 14.04. An upstart configuration `poller-service.conf` is provided. To install symlink this file to `/etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. +An upstart configuration `poller-service.conf` is provided. To install run `ln -s /opt/librenms/poller-service.conf /etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. + +An LSB init script `poller-service.init` is also provided. To install run `ln -s /opt/librenms/poller-service.init /etc/init.d/poller-service && update-rc.d poller-service defaults`. From 92177414ad0bb477b4470e73fa54f4dbb2c35cea Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 14:59:17 -0400 Subject: [PATCH 75/82] space --- poller-service.init | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.init b/poller-service.init index ff5b4766c..7e1e5620f 100755 --- a/poller-service.init +++ b/poller-service.init @@ -9,7 +9,7 @@ # This polls devices monitored by LibreNMS ### END INIT INFO -./lib/lsb/init-functions +. /lib/lsb/init-functions NAME=poller-service From 29a6f1b0579555d4badee0a16330bb9891ebfe29 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 15:03:01 -0400 Subject: [PATCH 76/82] run as librenms and background --- poller-service.init | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/poller-service.init b/poller-service.init index 7e1e5620f..25c4bd253 100755 --- a/poller-service.init +++ b/poller-service.init @@ -15,6 +15,8 @@ NAME=poller-service DAEMON=/opt/librenms/poller-service.py +USER=librenms + PIDFILE=/var/run/poller-service.pid test -x $DAEMON || exit 5 @@ -34,7 +36,7 @@ case $1 in log_daemon_msg "Starting the process" "$NAME" # Start the daemon with the help of start-stop-daemon # Log the message appropriately - if start-stop-daemon --start --quiet --oknodo --pidfile $PIDFILE --exec $DAEMON ; then + if start-stop-daemon --start --quiet --oknodo --pidfile $PIDFILE --exec $DAEMON --chuid $USER --background; then log_end_msg 0 else log_end_msg 1 @@ -69,17 +71,6 @@ case $1 in fi ;; - reload) - # Reload the process. Basically sending some signal to a daemon to reload - # it configurations. - if [ -e $PIDFILE ]; then - start-stop-daemon --stop --signal USR1 --quiet --pidfile $PIDFILE --name $NAME - log_success_msg "$NAME process reloaded successfully" - else - log_failure_msg "$PIDFILE does not exists" - fi - ;; - *) # For invalid arguments, print the usage message. echo "Usage: $0 {start|stop|restart|reload|status}" From 985ea21f6e2321d68626bfb8a842529f26b25807 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 15:09:06 -0400 Subject: [PATCH 77/82] make and remove pidfile --- poller-service.init | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/poller-service.init b/poller-service.init index 25c4bd253..15f2509d5 100755 --- a/poller-service.init +++ b/poller-service.init @@ -36,7 +36,7 @@ case $1 in log_daemon_msg "Starting the process" "$NAME" # Start the daemon with the help of start-stop-daemon # Log the message appropriately - if start-stop-daemon --start --quiet --oknodo --pidfile $PIDFILE --exec $DAEMON --chuid $USER --background; then + if start-stop-daemon --start --quiet --oknodo --make-pidfile --pidfile $PIDFILE --exec $DAEMON --chuid $USER --background; then log_end_msg 0 else log_end_msg 1 @@ -48,7 +48,7 @@ case $1 in if [ -e $PIDFILE ]; then status_of_proc -p $PIDFILE $DAEMON "Stoppping the $NAME process" && status="0" || status="$?" if [ "$status" = 0 ]; then - start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE + start-stop-daemon --stop --quiet --oknodo --remove-pidfile --pidfile $PIDFILE /bin/rm -rf $PIDFILE fi else From 2fdea68e493e82ca5e77480a9d526115451d9286 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 15:10:59 -0400 Subject: [PATCH 78/82] no removepidfile --- poller-service.init | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/poller-service.init b/poller-service.init index 15f2509d5..6e870a6db 100755 --- a/poller-service.init +++ b/poller-service.init @@ -48,7 +48,7 @@ case $1 in if [ -e $PIDFILE ]; then status_of_proc -p $PIDFILE $DAEMON "Stoppping the $NAME process" && status="0" || status="$?" if [ "$status" = 0 ]; then - start-stop-daemon --stop --quiet --oknodo --remove-pidfile --pidfile $PIDFILE + start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE /bin/rm -rf $PIDFILE fi else From 2caba3ee589c6d5480c2bbc6781c96e9440aca93 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 23 Jul 2015 15:35:32 -0400 Subject: [PATCH 79/82] add note to reload initctl after linking upstart job --- doc/Extensions/Poller-Service.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 4d709ccd6..26bf898c1 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -21,6 +21,6 @@ $config['poller_service_down_retry'] = 60; Distributed polling is possible, and uses the same configuration options as are described for traditional distributed polling, except that the memcached options are not necessary. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. ## Service Installation -An upstart configuration `poller-service.conf` is provided. To install run `ln -s /opt/librenms/poller-service.conf /etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. The service is configured to run as the user `librenms` and will fail if that user does not exist. +An upstart configuration `poller-service.conf` is provided. To install run `ln -s /opt/librenms/poller-service.conf /etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. If you recieve an error that the service does not exist, run `initctl reload-configuration`. The service is configured to run as the user `librenms` and will fail if that user does not exist. An LSB init script `poller-service.init` is also provided. To install run `ln -s /opt/librenms/poller-service.init /etc/init.d/poller-service && update-rc.d poller-service defaults`. From a71172969f88ed4d2898e5a5e38722a051fb5003 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Tue, 28 Jul 2015 07:31:46 -0400 Subject: [PATCH 80/82] fix unused code --- includes/dbFacile.mysql.php | 22 ++++++++++++++++------ includes/dbFacile.mysqli.php | 21 +++++++++++++++------ 2 files changed, 31 insertions(+), 12 deletions(-) diff --git a/includes/dbFacile.mysql.php b/includes/dbFacile.mysql.php index bac21b008..7b75aa9fb 100644 --- a/includes/dbFacile.mysql.php +++ b/includes/dbFacile.mysql.php @@ -59,10 +59,20 @@ function dbQuery($sql, $parameters=array()) { }//end dbQuery() +/* + * Aquire a lock on a string + * */ +function dbGetLock($data, $timeout = 0) { + $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; + $result = dbFetchCell($sql); + return $result; +} + + /* * Check a lock on a string * */ -function dbCheckLock($data, $timeout = 0) { +function dbCheckLock($data) { $sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')'; $result = dbFetchCell($sql); return $result; @@ -72,12 +82,12 @@ function dbCheckLock($data, $timeout = 0) { /* * Release a lock on a string * */ -function dbReleaseLock($data, $timeout = 0) { - $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; - $result = dbFetchCell($sql); - return $result; +function dbReleaseLock($data) { + $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; + $result = dbFetchCell($sql); + return $result; } - + /* * Passed an array and a table name, it attempts to insert the data into the table. diff --git a/includes/dbFacile.mysqli.php b/includes/dbFacile.mysqli.php index 158a1dae5..d7016368d 100644 --- a/includes/dbFacile.mysqli.php +++ b/includes/dbFacile.mysqli.php @@ -59,10 +59,20 @@ function dbQuery($sql, $parameters=array()) { }//end dbQuery() +/* + * Aquire a lock on a string + * */ +function dbGetLock($data, $timeout = 0) { + $sql = 'SELECT GET_LOCK(\'' . $data . '\',' . $timeout . ')'; + $result = dbFetchCell($sql); + return $result; +} + + /* * Check a lock on a string * */ -function dbCheckLock($data, $timeout = 0) { +function dbCheckLock($data) { $sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')'; $result = dbFetchCell($sql); return $result; @@ -72,13 +82,12 @@ function dbCheckLock($data, $timeout = 0) { /* * Release a lock on a string * */ -function dbReleaseLock($data, $timeout = 0) { - $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; - $result = dbFetchCell($sql); - return $result; +function dbReleaseLock($data) { + $sql = 'SELECT RELEASE_LOCK(\'' . $data . '\')'; + $result = dbFetchCell($sql); + return $result; } - /* * Passed an array and a table name, it attempts to insert the data into the table. * Check for boolean false to determine whether insert failed From f28c0cceca229b17fc1177aa8cef5e6618af3513 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Thu, 30 Jul 2015 09:39:08 -0400 Subject: [PATCH 81/82] multi-master mysql docs --- doc/Extensions/Poller-Service.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/doc/Extensions/Poller-Service.md b/doc/Extensions/Poller-Service.md index 26bf898c1..ab5461ca6 100644 --- a/doc/Extensions/Poller-Service.md +++ b/doc/Extensions/Poller-Service.md @@ -20,6 +20,9 @@ $config['poller_service_down_retry'] = 60; ## Distributed Polling Distributed polling is possible, and uses the same configuration options as are described for traditional distributed polling, except that the memcached options are not necessary. The database must be acessable from the distributed pollers, and properly configured. Remote access to the RRD directory must also be configured as described in the Distributed Poller documentation. Memcache is not required. Concurrency is managed using mysql GET_LOCK to ensure that devices are only being polled by one device at at time. The poller service is compatible with poller groups. +## Multi-Master MySQL considerations +Because locks are not replicated in Multi-Master MySQL configurations, if you are using such a configuration, you will need to make sure that all pollers are using the same MySQL server. + ## Service Installation An upstart configuration `poller-service.conf` is provided. To install run `ln -s /opt/librenms/poller-service.conf /etc/init/poller-service.conf`. The service will start on boot and can be started manually by running `start poller-service`. If you recieve an error that the service does not exist, run `initctl reload-configuration`. The service is configured to run as the user `librenms` and will fail if that user does not exist. From 588ea4083eee6d3dbf9f93f4416347f8204ffad2 Mon Sep 17 00:00:00 2001 From: Clint Armstrong Date: Fri, 21 Aug 2015 07:44:05 -0400 Subject: [PATCH 82/82] add license text --- poller-service.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/poller-service.py b/poller-service.py index 74e584e64..d8b48f455 100755 --- a/poller-service.py +++ b/poller-service.py @@ -10,6 +10,17 @@ Date: July 2015 License: BSD 2-Clause + +Copyright (c) 2015, Clint Armstrong +All rights reserved. + +Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """ import json