mirror of
https://github.com/stylersnico/librenms.git
synced 2026-07-26 00:18:03 +02:00
Merge pull request #1584 from clinta/poller-service-rebase
Poller service rebase
This commit is contained in:
Daniel Preussker
@@ -10,3 +10,4 @@ nbproject
|
||||
.alerts.lock
|
||||
.ircbot.alert
|
||||
.metadata_never_index
|
||||
*.swp
|
||||
|
||||
+3
-1
@@ -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) {
|
||||
discover_device($device, $options);
|
||||
if (dbGetLock('discovering.' . $device['device_id'])) {
|
||||
discover_device($device, $options);
|
||||
}
|
||||
}
|
||||
|
||||
$end = utime();
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
# 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`. 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']`.
|
||||
|
||||
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;
|
||||
$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.
|
||||
|
||||
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`.
|
||||
@@ -59,6 +59,36 @@ 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) {
|
||||
$sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')';
|
||||
$result = dbFetchCell($sql);
|
||||
return $result;
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
* Release a lock on a string
|
||||
* */
|
||||
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
|
||||
|
||||
@@ -59,6 +59,35 @@ 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) {
|
||||
$sql = 'SELECT IS_FREE_LOCK(\'' . $data . '\')';
|
||||
$result = dbFetchCell($sql);
|
||||
return $result;
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
* Release a lock on a string
|
||||
* */
|
||||
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
|
||||
|
||||
@@ -84,6 +84,22 @@ if ($handle = opendir($config['install_dir'].'/sql-schema')) {
|
||||
|
||||
asort($filelist);
|
||||
|
||||
if (explode('.', max($filelist), 2)[0] <= $db_rev) {
|
||||
if ($debug) {
|
||||
echo "DB Schema already up to date.\n";
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
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) {
|
||||
@@ -153,3 +169,5 @@ if ($updating) {
|
||||
|
||||
echo "-- Done\n";
|
||||
}
|
||||
|
||||
dbReleaseLock('schema_update');
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
# poller-service - SNMP polling service for LibreNMS
|
||||
|
||||
description "SNMP polling service for LibreNMS"
|
||||
author "Clint Armstrong <clint@clintarmstrong.net>"
|
||||
|
||||
# When to start the service
|
||||
start on runlevel [2345]
|
||||
|
||||
# When to stop the service
|
||||
stop on runlevel [016]
|
||||
|
||||
# Automatically restart process if crashed
|
||||
respawn
|
||||
|
||||
chdir /opt/librenms
|
||||
setuid librenms
|
||||
setgid librenms
|
||||
|
||||
# Start the process
|
||||
exec /opt/librenms/poller-service.py
|
||||
Executable
+79
@@ -0,0 +1,79 @@
|
||||
### 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
|
||||
|
||||
USER=librenms
|
||||
|
||||
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 --make-pidfile --pidfile $PIDFILE --exec $DAEMON --chuid $USER --background; 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
|
||||
;;
|
||||
|
||||
*)
|
||||
# For invalid arguments, print the usage message.
|
||||
echo "Usage: $0 {start|stop|restart|reload|status}"
|
||||
exit 2
|
||||
;;
|
||||
esac
|
||||
Executable
+300
@@ -0,0 +1,300 @@
|
||||
#! /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
|
||||
$poll_frequency seconds, but should gracefully degrade if resources are inefficient, polling devices as
|
||||
frequently as possible. This service is based on Job Snijders' poller-wrapper.py.
|
||||
|
||||
Author: Clint Armstrong <clint@clintarmstrong.net>
|
||||
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
|
||||
import os
|
||||
import subprocess
|
||||
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('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:
|
||||
log.critical("ERROR: Could not execute: %s" % config_cmd)
|
||||
sys.exit(2)
|
||||
return proc.communicate()[0].decode()
|
||||
|
||||
try:
|
||||
with open(config_file) as f:
|
||||
pass
|
||||
except IOError as e:
|
||||
log.critical("ERROR: Oh dear... %s does not seem readable" % config_file)
|
||||
sys.exit(2)
|
||||
|
||||
try:
|
||||
config = json.loads(get_config_data())
|
||||
except:
|
||||
log.critical("ERROR: Could not load or parse configuration, are PATHs correct?")
|
||||
sys.exit(2)
|
||||
|
||||
try:
|
||||
loglevel = int(config['poller_service_loglevel'])
|
||||
except KeyError:
|
||||
loglevel = 20
|
||||
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)))
|
||||
log.setLevel(20)
|
||||
|
||||
poller_path = config['install_dir'] + '/poller.php'
|
||||
discover_path = config['install_dir'] + '/discovery.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']
|
||||
|
||||
|
||||
try:
|
||||
amount_of_workers = int(config['poller_service_workers'])
|
||||
if amount_of_workers == 0:
|
||||
amount_of_workers = 16
|
||||
except KeyError:
|
||||
amount_of_workers = 16
|
||||
|
||||
try:
|
||||
poll_frequency = int(config['poller_service_poll_frequency'])
|
||||
if poll_frequency == 0:
|
||||
poll_frequency = 300
|
||||
except KeyError:
|
||||
poll_frequency = 300
|
||||
|
||||
try:
|
||||
discover_frequency = int(config['poller_service_discover_frequency'])
|
||||
if discover_frequency == 0:
|
||||
discover_frequency = 21600
|
||||
except KeyError:
|
||||
discover_frequency = 21600
|
||||
|
||||
try:
|
||||
down_retry = int(config['poller_service_down_retry'])
|
||||
if down_retry == 0:
|
||||
down_retry = 60
|
||||
except KeyError:
|
||||
down_retry = 60
|
||||
|
||||
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:
|
||||
log.critical("ERROR: Could not connect to MySQL database!")
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def poll_worker(device_id, action):
|
||||
try:
|
||||
start_time = time.time()
|
||||
path = poller_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)
|
||||
if elapsed_time < 300:
|
||||
log.debug("DEBUG: worker finished %s of device %s in %s seconds" % (action, device_id, elapsed_time))
|
||||
else:
|
||||
log.warning("WARNING: worker finished %s of device %s in %s seconds" % (action, device_id, elapsed_time))
|
||||
except (KeyboardInterrupt, SystemExit):
|
||||
raise
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
def lockFree(lock):
|
||||
global cursor
|
||||
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}', 0)".format(lock)
|
||||
cursor.execute(query)
|
||||
return cursor.fetchall()[0][0] == 1
|
||||
|
||||
|
||||
def releaseLock(lock):
|
||||
global cursor
|
||||
query = "SELECT RELEASE_LOCK('{0}')".format(lock)
|
||||
cursor.execute(query)
|
||||
return cursor.fetchall()[0][0] == 1
|
||||
|
||||
|
||||
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({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
|
||||
# of having each device complete a poll within the given time range.
|
||||
dev_query = ('SELECT device_id, status, '
|
||||
'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 '
|
||||
'LIMIT 5 ').format(poll_frequency,
|
||||
discover_frequency,
|
||||
down_retry,
|
||||
poller_group)
|
||||
|
||||
threads = 0
|
||||
next_update = datetime.now() + timedelta(minutes=1)
|
||||
devices_scanned = 0
|
||||
|
||||
while True:
|
||||
cur_threads = threading.active_count()
|
||||
if cur_threads != threads:
|
||||
threads = cur_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
|
||||
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'].strip(),
|
||||
devices_scanned,
|
||||
seconds_taken)
|
||||
try:
|
||||
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: {0} devices scanned in the last minute'.format(devices_scanned))
|
||||
devices_scanned = 0
|
||||
next_update = datetime.now() + timedelta(minutes=1)
|
||||
|
||||
while threading.active_count() >= amount_of_workers or not lockFree('schema_update'):
|
||||
time.sleep(.5)
|
||||
|
||||
try:
|
||||
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, 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.{0}'.format(device_id)):
|
||||
continue
|
||||
if not lockFree('polling.{0}'.format(device_id)):
|
||||
releaseLock('queued.{0}'.format(device_id))
|
||||
continue
|
||||
|
||||
if next_poll and next_poll > datetime.now():
|
||||
log.debug('DEBUG: Sleeping until {0} before polling {1}'.format(next_poll, device_id))
|
||||
sleep_until(next_poll)
|
||||
|
||||
action = 'poll'
|
||||
if (not next_discovery or next_discovery < datetime.now()) and status == 1:
|
||||
action = 'discovery'
|
||||
|
||||
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 = {0}'.format(device_id))
|
||||
cursor.fetchall()
|
||||
|
||||
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.
|
||||
break
|
||||
|
||||
# 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
|
||||
# by locking a different string.
|
||||
getLock('unlock.{0}'.format(config['distributed_poller_name']))
|
||||
+5
-3
@@ -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++;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +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`);
|
||||
Reference in New Issue
Block a user