923 lines
35 KiB
Python
923 lines
35 KiB
Python
import logging
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import pymysql # pylint: disable=import-error
|
|
|
|
import LibreNMS
|
|
from LibreNMS.config import DBConfig
|
|
|
|
try:
|
|
import psutil
|
|
except ImportError:
|
|
pass
|
|
|
|
from datetime import timedelta
|
|
from datetime import datetime
|
|
from platform import python_version
|
|
from time import sleep
|
|
from socket import gethostname
|
|
from signal import signal, SIGTERM, SIGQUIT, SIGINT, SIGHUP, SIGCHLD
|
|
from uuid import uuid1
|
|
|
|
try:
|
|
from systemd.daemon import notify
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
from redis.exceptions import ConnectionError as RedisConnectionError
|
|
except ImportError:
|
|
|
|
class RedisConnectionError(Exception):
|
|
pass
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ServiceConfig(DBConfig):
|
|
def __init__(self):
|
|
"""
|
|
Stores all of the configuration variables for the LibreNMS service in a common object
|
|
Starts with defaults, but can be populated with variables from config.php by calling populate()
|
|
"""
|
|
self._uuid = str(uuid1())
|
|
self.set_name(gethostname())
|
|
|
|
def set_name(self, name):
|
|
if name:
|
|
self.name = name.strip()
|
|
self.unique_name = "{}-{}".format(self.name, self._uuid)
|
|
|
|
class PollerConfig:
|
|
def __init__(self, workers, frequency, calculate=None):
|
|
self.enabled = True
|
|
self.workers = workers
|
|
self.frequency = frequency
|
|
self.calculate = calculate
|
|
|
|
# config variables with defaults
|
|
BASE_DIR = os.path.abspath(
|
|
os.path.join(os.path.dirname(os.path.realpath(__file__)), os.pardir)
|
|
)
|
|
|
|
node_id = None
|
|
name = None
|
|
unique_name = None
|
|
single_instance = True
|
|
distributed = False
|
|
group = 0
|
|
|
|
debug = False
|
|
log_level = 20
|
|
max_db_failures = 5
|
|
|
|
alerting = PollerConfig(1, 60)
|
|
poller = PollerConfig(24, 300)
|
|
services = PollerConfig(8, 300)
|
|
discovery = PollerConfig(16, 21600)
|
|
billing = PollerConfig(2, 300, 60)
|
|
ping = PollerConfig(1, 60)
|
|
down_retry = 60
|
|
update_enabled = True
|
|
update_frequency = 86400
|
|
|
|
master_resolution = 1
|
|
master_timeout = 10
|
|
|
|
redis_host = "localhost"
|
|
redis_port = 6379
|
|
redis_db = 0
|
|
redis_user = None
|
|
redis_pass = None
|
|
redis_socket = None
|
|
redis_sentinel = None
|
|
redis_sentinel_user = None
|
|
redis_sentinel_pass = None
|
|
redis_sentinel_service = None
|
|
redis_timeout = 60
|
|
|
|
log_output = False
|
|
logdir = "logs"
|
|
|
|
watchdog_enabled = False
|
|
watchdog_logfile = "logs/librenms.log"
|
|
|
|
def populate(self):
|
|
config = LibreNMS.get_config_data(self.BASE_DIR)
|
|
|
|
# populate config variables
|
|
self.node_id = os.getenv("NODE_ID")
|
|
self.set_name(config.get("distributed_poller_name", None))
|
|
self.distributed = config.get("distributed_poller", ServiceConfig.distributed)
|
|
self.group = ServiceConfig.parse_group(
|
|
config.get("distributed_poller_group", ServiceConfig.group)
|
|
)
|
|
|
|
# backward compatible options
|
|
self.master_timeout = config.get(
|
|
"service_master_timeout", ServiceConfig.master_timeout
|
|
)
|
|
self.poller.workers = config.get(
|
|
"poller_service_workers", ServiceConfig.poller.workers
|
|
)
|
|
self.poller.frequency = config.get(
|
|
"poller_service_poll_frequency", ServiceConfig.poller.frequency
|
|
)
|
|
self.discovery.frequency = config.get(
|
|
"poller_service_discover_frequency", ServiceConfig.discovery.frequency
|
|
)
|
|
self.down_retry = config.get(
|
|
"poller_service_down_retry", ServiceConfig.down_retry
|
|
)
|
|
self.log_level = config.get("poller_service_loglevel", ServiceConfig.log_level)
|
|
|
|
# new options
|
|
self.poller.enabled = config.get("service_poller_enabled", True) # unused
|
|
self.poller.workers = config.get(
|
|
"service_poller_workers", ServiceConfig.poller.workers
|
|
)
|
|
self.poller.frequency = config.get(
|
|
"service_poller_frequency", ServiceConfig.poller.frequency
|
|
)
|
|
self.discovery.enabled = config.get("service_discovery_enabled", True) # unused
|
|
self.discovery.workers = config.get(
|
|
"service_discovery_workers", ServiceConfig.discovery.workers
|
|
)
|
|
self.discovery.frequency = config.get(
|
|
"service_discovery_frequency", ServiceConfig.discovery.frequency
|
|
)
|
|
self.services.enabled = config.get("service_services_enabled", True)
|
|
self.services.workers = config.get(
|
|
"service_services_workers", ServiceConfig.services.workers
|
|
)
|
|
self.services.frequency = config.get(
|
|
"service_services_frequency", ServiceConfig.services.frequency
|
|
)
|
|
self.billing.enabled = config.get("service_billing_enabled", True)
|
|
self.billing.frequency = config.get(
|
|
"service_billing_frequency", ServiceConfig.billing.frequency
|
|
)
|
|
self.billing.calculate = config.get(
|
|
"service_billing_calculate_frequency", ServiceConfig.billing.calculate
|
|
)
|
|
self.alerting.enabled = config.get("service_alerting_enabled", True)
|
|
self.alerting.frequency = config.get(
|
|
"service_alerting_frequency", ServiceConfig.alerting.frequency
|
|
)
|
|
self.ping.enabled = config.get("service_ping_enabled", False)
|
|
self.ping.frequency = config.get("ping_rrd_step", ServiceConfig.ping.frequency)
|
|
self.down_retry = config.get(
|
|
"service_poller_down_retry", ServiceConfig.down_retry
|
|
)
|
|
self.log_level = config.get("service_loglevel", ServiceConfig.log_level)
|
|
self.update_enabled = config.get(
|
|
"service_update_enabled", ServiceConfig.update_enabled
|
|
)
|
|
self.update_frequency = config.get(
|
|
"service_update_frequency", ServiceConfig.update_frequency
|
|
)
|
|
|
|
self.redis_host = os.getenv(
|
|
"REDIS_HOST", config.get("redis_host", ServiceConfig.redis_host)
|
|
)
|
|
self.redis_db = os.getenv(
|
|
"REDIS_DB", config.get("redis_db", ServiceConfig.redis_db)
|
|
)
|
|
self.redis_user = os.getenv(
|
|
"REDIS_USERNAME", config.get("redis_user", ServiceConfig.redis_user)
|
|
)
|
|
self.redis_pass = os.getenv(
|
|
"REDIS_PASSWORD", config.get("redis_pass", ServiceConfig.redis_pass)
|
|
)
|
|
self.redis_port = int(
|
|
os.getenv("REDIS_PORT", config.get("redis_port", ServiceConfig.redis_port))
|
|
)
|
|
self.redis_socket = os.getenv(
|
|
"REDIS_SOCKET", config.get("redis_socket", ServiceConfig.redis_socket)
|
|
)
|
|
self.redis_sentinel = os.getenv(
|
|
"REDIS_SENTINEL", config.get("redis_sentinel", ServiceConfig.redis_sentinel)
|
|
)
|
|
self.redis_sentinel_user = os.getenv(
|
|
"REDIS_SENTINEL_USERNAME",
|
|
config.get("redis_sentinel_user", ServiceConfig.redis_sentinel_user),
|
|
)
|
|
self.redis_sentinel_pass = os.getenv(
|
|
"REDIS_SENTINEL_PASSWORD",
|
|
config.get("redis_sentinel_pass", ServiceConfig.redis_sentinel_pass),
|
|
)
|
|
self.redis_sentinel_service = os.getenv(
|
|
"REDIS_SENTINEL_SERVICE",
|
|
config.get("redis_sentinel_service", ServiceConfig.redis_sentinel_service),
|
|
)
|
|
self.redis_timeout = int(
|
|
os.getenv(
|
|
"REDIS_TIMEOUT",
|
|
self.alerting.frequency
|
|
if self.alerting.frequency != 0
|
|
else self.redis_timeout,
|
|
)
|
|
)
|
|
|
|
self.db_host = os.getenv(
|
|
"DB_HOST", config.get("db_host", ServiceConfig.db_host)
|
|
)
|
|
self.db_name = os.getenv(
|
|
"DB_DATABASE", config.get("db_name", ServiceConfig.db_name)
|
|
)
|
|
self.db_pass = os.getenv(
|
|
"DB_PASSWORD", config.get("db_pass", ServiceConfig.db_pass)
|
|
)
|
|
self.db_port = int(
|
|
os.getenv("DB_PORT", config.get("db_port", ServiceConfig.db_port))
|
|
)
|
|
self.db_socket = os.getenv(
|
|
"DB_SOCKET", config.get("db_socket", ServiceConfig.db_socket)
|
|
)
|
|
self.db_user = os.getenv(
|
|
"DB_USERNAME", config.get("db_user", ServiceConfig.db_user)
|
|
)
|
|
self.db_sslmode = os.getenv(
|
|
"DB_SSLMODE", config.get("db_sslmode", ServiceConfig.db_sslmode)
|
|
)
|
|
self.db_ssl_ca = os.getenv(
|
|
"MYSQL_ATTR_SSL_CA", config.get("db_ssl_ca", ServiceConfig.db_ssl_ca)
|
|
)
|
|
|
|
self.watchdog_enabled = config.get(
|
|
"service_watchdog_enabled", ServiceConfig.watchdog_enabled
|
|
)
|
|
self.logdir = config.get("log_dir", ServiceConfig.BASE_DIR + "/logs")
|
|
self.watchdog_logfile = config.get("log_file", self.logdir + "/librenms.log")
|
|
|
|
# set convenient debug variable
|
|
self.debug = logging.getLogger().isEnabledFor(logging.DEBUG)
|
|
|
|
if not self.debug and self.log_level:
|
|
try:
|
|
logging.getLogger().setLevel(self.log_level)
|
|
except ValueError:
|
|
logger.error(
|
|
"Unknown log level {}, must be one of 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'".format(
|
|
self.log_level
|
|
)
|
|
)
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
|
|
def load_poller_config(self, db):
|
|
try:
|
|
settings = {}
|
|
cursor = db.query(
|
|
"SELECT * FROM `poller_cluster` WHERE `node_id`=%s", self.node_id
|
|
)
|
|
if cursor.rowcount == 0:
|
|
return
|
|
|
|
for index, setting in enumerate(cursor.fetchone()):
|
|
name = cursor.description[index][0]
|
|
settings[name] = setting
|
|
|
|
if settings["poller_name"] is not None:
|
|
self.set_name(settings["poller_name"])
|
|
if settings["poller_groups"] is not None:
|
|
self.group = ServiceConfig.parse_group(settings["poller_groups"])
|
|
if settings["poller_enabled"] is not None:
|
|
self.poller.enabled = settings["poller_enabled"]
|
|
if settings["poller_frequency"] is not None:
|
|
self.poller.frequency = settings["poller_frequency"]
|
|
if settings["poller_workers"] is not None:
|
|
self.poller.workers = settings["poller_workers"]
|
|
if settings["poller_down_retry"] is not None:
|
|
self.down_retry = settings["poller_down_retry"]
|
|
if settings["discovery_enabled"] is not None:
|
|
self.discovery.enabled = settings["discovery_enabled"]
|
|
if settings["discovery_frequency"] is not None:
|
|
self.discovery.frequency = settings["discovery_frequency"]
|
|
if settings["discovery_workers"] is not None:
|
|
self.discovery.workers = settings["discovery_workers"]
|
|
if settings["services_enabled"] is not None:
|
|
self.services.enabled = settings["services_enabled"]
|
|
if settings["services_frequency"] is not None:
|
|
self.services.frequency = settings["services_frequency"]
|
|
if settings["services_workers"] is not None:
|
|
self.services.workers = settings["services_workers"]
|
|
if settings["billing_enabled"] is not None:
|
|
self.billing.enabled = settings["billing_enabled"]
|
|
if settings["billing_frequency"] is not None:
|
|
self.billing.frequency = settings["billing_frequency"]
|
|
if settings["billing_calculate_frequency"] is not None:
|
|
self.billing.calculate = settings["billing_calculate_frequency"]
|
|
if settings["alerting_enabled"] is not None:
|
|
self.alerting.enabled = settings["alerting_enabled"]
|
|
if settings["alerting_frequency"] is not None:
|
|
self.alerting.frequency = settings["alerting_frequency"]
|
|
if settings["ping_enabled"] is not None:
|
|
self.ping.enabled = settings["ping_enabled"]
|
|
if settings["ping_frequency"] is not None:
|
|
self.ping.frequency = settings["ping_frequency"]
|
|
if settings["update_enabled"] is not None:
|
|
self.update_enabled = settings["update_enabled"]
|
|
if settings["update_frequency"] is not None:
|
|
self.update_frequency = settings["update_frequency"]
|
|
if settings["loglevel"] is not None:
|
|
self.log_level = settings["loglevel"]
|
|
if settings["watchdog_enabled"] is not None:
|
|
self.watchdog_enabled = settings["watchdog_enabled"]
|
|
if settings["watchdog_log"] is not None:
|
|
self.watchdog_logfile = settings["watchdog_log"]
|
|
except pymysql.err.Error:
|
|
logger.warning("Unable to load poller (%s) config", self.node_id)
|
|
|
|
@staticmethod
|
|
def parse_group(g):
|
|
if g is None:
|
|
return [0]
|
|
elif type(g) is int:
|
|
return [g]
|
|
elif type(g) is str:
|
|
try:
|
|
return [int(x) for x in set(g.split(","))]
|
|
except ValueError:
|
|
pass
|
|
|
|
logger.error("Could not parse group string, defaulting to 0")
|
|
return [0]
|
|
|
|
|
|
class Service:
|
|
config = ServiceConfig()
|
|
_fp = False
|
|
_started = False
|
|
start_time = 0
|
|
queue_managers = {}
|
|
poller_manager = None
|
|
discovery_manager = None
|
|
last_poll = {}
|
|
reap_flag = False
|
|
terminate_flag = False
|
|
reload_flag = False
|
|
db_failures = 0
|
|
|
|
def __init__(self):
|
|
self.start_time = time.time()
|
|
self.config.populate()
|
|
self._db = LibreNMS.DB(self.config)
|
|
self.config.load_poller_config(self._db)
|
|
|
|
threading.current_thread().name = self.config.name # rename main thread
|
|
self.attach_signals()
|
|
|
|
self._lm = self.create_lock_manager()
|
|
self.daily_timer = LibreNMS.RecurringTimer(
|
|
self.config.update_frequency, self.run_maintenance, "maintenance"
|
|
)
|
|
self.stats_timer = LibreNMS.RecurringTimer(
|
|
self.config.poller.frequency, self.log_performance_stats, "performance"
|
|
)
|
|
if self.config.watchdog_enabled:
|
|
logger.info(
|
|
"Starting watchdog timer for log file: {}".format(
|
|
self.config.watchdog_logfile
|
|
)
|
|
)
|
|
self.watchdog_timer = LibreNMS.RecurringTimer(
|
|
self.config.poller.frequency, self.logfile_watchdog, "watchdog"
|
|
)
|
|
else:
|
|
logger.info("Watchdog is disabled.")
|
|
self.systemd_watchdog_timer = LibreNMS.RecurringTimer(
|
|
10, self.systemd_watchdog, "systemd-watchdog"
|
|
)
|
|
self.is_master = False
|
|
|
|
def service_age(self):
|
|
return time.time() - self.start_time
|
|
|
|
def attach_signals(self):
|
|
logger.debug(
|
|
"Attaching signal handlers on thread %s", threading.current_thread().name
|
|
)
|
|
signal(SIGTERM, self.terminate) # capture sigterm and exit gracefully
|
|
signal(SIGQUIT, self.terminate) # capture sigquit and exit gracefully
|
|
signal(SIGINT, self.terminate) # capture sigint and exit gracefully
|
|
signal(SIGHUP, self.reload) # capture sighup and restart gracefully
|
|
|
|
if "psutil" not in sys.modules:
|
|
logger.warning("psutil is not available, polling gap possible")
|
|
else:
|
|
signal(SIGCHLD, self.reap) # capture sigchld and reap the process
|
|
|
|
def reap_psutil(self):
|
|
"""
|
|
A process from a previous invocation is trying to report its status
|
|
"""
|
|
# Speed things up by only looking at direct zombie children
|
|
for p in psutil.Process().children(recursive=False):
|
|
try:
|
|
cmd = (
|
|
p.cmdline()
|
|
) # cmdline is uncached, so needs to go here to avoid NoSuchProcess
|
|
status = p.status()
|
|
|
|
if status == psutil.STATUS_ZOMBIE:
|
|
pid = p.pid
|
|
r = os.waitpid(p.pid, os.WNOHANG)
|
|
logger.warning(
|
|
'Reaped long running job "%s" in state %s with PID %d - job returned %d',
|
|
cmd,
|
|
status,
|
|
r[0],
|
|
r[1],
|
|
)
|
|
except (OSError, psutil.NoSuchProcess):
|
|
# process was already reaped
|
|
continue
|
|
|
|
def start(self):
|
|
logger.debug("Performing startup checks...")
|
|
|
|
if self.config.single_instance:
|
|
self.check_single_instance() # don't allow more than one service at a time
|
|
|
|
if self._started:
|
|
raise RuntimeWarning("Not allowed to start Poller twice")
|
|
self._started = True
|
|
|
|
logger.debug("Starting up queue managers...")
|
|
|
|
# initialize and start the worker pools
|
|
self.poller_manager = LibreNMS.PollerQueueManager(self.config, self._lm)
|
|
self.queue_managers["poller"] = self.poller_manager
|
|
self.discovery_manager = LibreNMS.DiscoveryQueueManager(self.config, self._lm)
|
|
self.queue_managers["discovery"] = self.discovery_manager
|
|
self.queue_managers["alerting"] = LibreNMS.AlertQueueManager(
|
|
self.config, self._lm
|
|
)
|
|
self.queue_managers["services"] = LibreNMS.ServicesQueueManager(
|
|
self.config, self._lm
|
|
)
|
|
self.queue_managers["billing"] = LibreNMS.BillingQueueManager(
|
|
self.config, self._lm
|
|
)
|
|
self.queue_managers["ping"] = LibreNMS.PingQueueManager(self.config, self._lm)
|
|
|
|
if self.config.update_enabled:
|
|
self.daily_timer.start()
|
|
self.stats_timer.start()
|
|
self.systemd_watchdog_timer.start()
|
|
if self.config.watchdog_enabled:
|
|
self.watchdog_timer.start()
|
|
|
|
logger.info("LibreNMS Service: {} started!".format(self.config.unique_name))
|
|
logger.info(
|
|
"Poller group {}. Using Python {} and {} locks and queues".format(
|
|
"0 (default)" if self.config.group == [0] else self.config.group,
|
|
python_version(),
|
|
"redis" if isinstance(self._lm, LibreNMS.RedisLock) else "internal",
|
|
)
|
|
)
|
|
logger.info(
|
|
"Queue Workers: Discovery={} Poller={} Services={} Alerting={} Billing={} Ping={}".format(
|
|
self.config.discovery.workers
|
|
if self.config.discovery.enabled
|
|
else "disabled",
|
|
self.config.poller.workers
|
|
if self.config.poller.enabled
|
|
else "disabled",
|
|
self.config.services.workers
|
|
if self.config.services.enabled
|
|
else "disabled",
|
|
"enabled" if self.config.alerting.enabled else "disabled",
|
|
"enabled" if self.config.billing.enabled else "disabled",
|
|
"enabled" if self.config.ping.enabled else "disabled",
|
|
)
|
|
)
|
|
|
|
if self.config.update_enabled:
|
|
logger.info(
|
|
"Maintenance tasks will be run every {}".format(
|
|
timedelta(seconds=self.config.update_frequency)
|
|
)
|
|
)
|
|
else:
|
|
logger.warning("Maintenance tasks are disabled.")
|
|
|
|
# Main dispatcher loop
|
|
try:
|
|
while not self.terminate_flag:
|
|
if self.reload_flag:
|
|
logger.info("Picked up reload flag, calling the reload process")
|
|
self.restart()
|
|
|
|
if self.reap_flag:
|
|
self.reap_flag = False
|
|
self.reap_psutil()
|
|
|
|
master_lock = self._acquire_master()
|
|
if master_lock:
|
|
if not self.is_master:
|
|
logger.info(
|
|
"{} is now the master dispatcher".format(self.config.name)
|
|
)
|
|
self.is_master = True
|
|
self.start_dispatch_timers()
|
|
|
|
devices = self.fetch_immediate_device_list()
|
|
for device in devices:
|
|
device_id = device[0]
|
|
group = device[1]
|
|
|
|
if device[2]: # polling
|
|
self.dispatch_immediate_polling(device_id, group)
|
|
|
|
if device[3]: # discovery
|
|
self.dispatch_immediate_discovery(device_id, group)
|
|
else:
|
|
if self.is_master:
|
|
logger.info(
|
|
"{} is no longer the master dispatcher".format(
|
|
self.config.name
|
|
)
|
|
)
|
|
self.stop_dispatch_timers()
|
|
self.is_master = False # no longer master
|
|
sleep(self.config.master_resolution)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
logger.info("Dispatch loop terminated")
|
|
self.shutdown()
|
|
|
|
def _acquire_master(self):
|
|
return self._lm.lock(
|
|
"dispatch.master", self.config.unique_name, self.config.master_timeout, True
|
|
)
|
|
|
|
def _release_master(self):
|
|
self._lm.unlock("dispatch.master", self.config.unique_name)
|
|
|
|
# ------------ Discovery ------------
|
|
def dispatch_immediate_discovery(self, device_id, group):
|
|
if not self.discovery_manager.is_locked(device_id):
|
|
self.discovery_manager.post_work(device_id, group)
|
|
|
|
# ------------ Polling ------------
|
|
def dispatch_immediate_polling(self, device_id, group):
|
|
if not self.poller_manager.is_locked(device_id):
|
|
self.poller_manager.post_work(device_id, group)
|
|
|
|
if self.config.debug:
|
|
cur_time = time.time()
|
|
elapsed = cur_time - self.last_poll.get(device_id, cur_time)
|
|
self.last_poll[device_id] = cur_time
|
|
# arbitrary limit to reduce spam
|
|
if elapsed > (
|
|
self.config.poller.frequency - self.config.master_resolution
|
|
):
|
|
logger.debug(
|
|
"Dispatching polling for device {}, time since last poll {:.2f}s".format(
|
|
device_id, elapsed
|
|
)
|
|
)
|
|
|
|
def fetch_immediate_device_list(self):
|
|
try:
|
|
poller_find_time = self.config.poller.frequency - 1
|
|
discovery_find_time = self.config.discovery.frequency - 1
|
|
|
|
result = self._db.query(
|
|
"""SELECT `device_id`,
|
|
`poller_group`,
|
|
COALESCE(`last_polled` <= DATE_ADD(DATE_ADD(NOW(), INTERVAL -%s SECOND), INTERVAL COALESCE(`last_polled_timetaken`, 0) SECOND), 1) AS `poll`,
|
|
IF(status=0, 0, IF (%s < `last_discovered_timetaken` * 1.25, 0, COALESCE(`last_discovered` <= DATE_ADD(DATE_ADD(NOW(), INTERVAL -%s SECOND), INTERVAL COALESCE(`last_discovered_timetaken`, 0) SECOND), 1))) AS `discover`
|
|
FROM `devices`
|
|
WHERE `disabled` = 0 AND (
|
|
`last_polled` IS NULL OR
|
|
`last_discovered` IS NULL OR
|
|
`last_polled` <= DATE_ADD(DATE_ADD(NOW(), INTERVAL -%s SECOND), INTERVAL COALESCE(`last_polled_timetaken`, 0) SECOND) OR
|
|
`last_discovered` <= DATE_ADD(DATE_ADD(NOW(), INTERVAL -%s SECOND), INTERVAL COALESCE(`last_discovered_timetaken`, 0) SECOND)
|
|
)
|
|
ORDER BY `last_polled_timetaken` DESC""",
|
|
(
|
|
poller_find_time,
|
|
self.service_age(),
|
|
discovery_find_time,
|
|
poller_find_time,
|
|
discovery_find_time,
|
|
),
|
|
)
|
|
self.db_failures = 0
|
|
return result
|
|
except pymysql.err.Error:
|
|
self.db_failures += 1
|
|
if self.db_failures > self.config.max_db_failures:
|
|
logger.warning(
|
|
"Too many DB failures ({}), attempting to release master".format(
|
|
self.db_failures
|
|
)
|
|
)
|
|
self._release_master()
|
|
sleep(
|
|
self.config.master_resolution
|
|
) # sleep to give another node a chance to acquire
|
|
return []
|
|
|
|
def run_maintenance(self):
|
|
"""
|
|
Runs update and cleanup tasks by calling daily.sh. Reloads the python script after the update.
|
|
Sets a schema-update lock so no distributed pollers will update until the schema has been updated.
|
|
"""
|
|
attempt = 0
|
|
wait = 5
|
|
max_runtime = 86100
|
|
max_tries = int(max_runtime / wait)
|
|
logger.info("Waiting for schema lock")
|
|
while not self._lm.lock("schema-update", self.config.unique_name, max_runtime):
|
|
attempt += 1
|
|
if attempt >= max_tries: # don't get stuck indefinitely
|
|
logger.warning(
|
|
"Reached max wait for other pollers to update, updating now"
|
|
)
|
|
break
|
|
sleep(wait)
|
|
|
|
logger.info("Running maintenance tasks")
|
|
exit_code, output = LibreNMS.call_script("daily.sh")
|
|
if exit_code == 0:
|
|
logger.info("Maintenance tasks complete\n{}".format(output))
|
|
else:
|
|
logger.error("Error {} in daily.sh:\n{}".format(exit_code, output))
|
|
|
|
self._lm.unlock("schema-update", self.config.unique_name)
|
|
|
|
self.restart()
|
|
|
|
def create_lock_manager(self):
|
|
"""
|
|
Create a new LockManager. Tries to create a Redis LockManager, but falls
|
|
back to python's internal threading lock implementation.
|
|
Exits if distributing poller is enabled and a Redis LockManager cannot be created.
|
|
:return: Instance of LockManager
|
|
"""
|
|
try:
|
|
return LibreNMS.RedisLock(
|
|
sentinel_kwargs={
|
|
"username": self.config.redis_sentinel_user,
|
|
"password": self.config.redis_sentinel_pass,
|
|
"socket_timeout": self.config.redis_timeout,
|
|
"unix_socket_path": self.config.redis_socket,
|
|
},
|
|
namespace="librenms.lock",
|
|
host=self.config.redis_host,
|
|
port=self.config.redis_port,
|
|
db=self.config.redis_db,
|
|
username=self.config.redis_user,
|
|
password=self.config.redis_pass,
|
|
unix_socket_path=self.config.redis_socket,
|
|
sentinel=self.config.redis_sentinel,
|
|
sentinel_service=self.config.redis_sentinel_service,
|
|
socket_timeout=self.config.redis_timeout,
|
|
)
|
|
except ImportError:
|
|
if self.config.distributed:
|
|
logger.critical(
|
|
"ERROR: Redis connection required for distributed polling"
|
|
)
|
|
logger.critical(
|
|
"Please install redis-py, either through your os software repository or from PyPI"
|
|
)
|
|
self.exit(2)
|
|
except Exception as e:
|
|
if self.config.distributed:
|
|
logger.critical(
|
|
"ERROR: Redis connection required for distributed polling"
|
|
)
|
|
logger.critical(
|
|
"Lock manager could not connect to Redis. {}: {}".format(
|
|
type(e).__name__, e
|
|
)
|
|
)
|
|
self.exit(2)
|
|
|
|
return LibreNMS.ThreadingLock()
|
|
|
|
def restart(self):
|
|
"""
|
|
Stop then recreate this entire process by re-calling the original script.
|
|
Has the effect of reloading the python files from disk.
|
|
"""
|
|
if sys.version_info < (3, 4, 0):
|
|
logger.warning(
|
|
"Skipping restart as running under an incompatible interpreter"
|
|
)
|
|
logger.warning("Please restart manually")
|
|
return
|
|
|
|
logger.info("Restarting service... ")
|
|
|
|
if "psutil" not in sys.modules:
|
|
logger.warning("psutil is not available, polling gap possible")
|
|
self._stop_managers_and_wait()
|
|
else:
|
|
self._stop_managers()
|
|
self._release_master()
|
|
|
|
python = sys.executable
|
|
sys.stdout.flush()
|
|
os.execl(python, python, *sys.argv)
|
|
|
|
def reap(self, signalnum=None, flag=None):
|
|
"""
|
|
Handle a set the reload flag to begin a clean restart
|
|
:param signalnum: UNIX signal number
|
|
:param flag: Flags accompanying signal
|
|
"""
|
|
self.reap_flag = True
|
|
|
|
def reload(self, signalnum=None, flag=None):
|
|
"""
|
|
Handle a set the reload flag to begin a clean restart
|
|
:param signalnum: UNIX signal number
|
|
:param flag: Flags accompanying signal
|
|
"""
|
|
logger.info(
|
|
"Received signal on thread %s, handling", threading.current_thread().name
|
|
)
|
|
self.reload_flag = True
|
|
|
|
def terminate(self, signalnum=None, flag=None):
|
|
"""
|
|
Handle a set the terminate flag to begin a clean shutdown
|
|
:param signalnum: UNIX signal number
|
|
:param flag: Flags accompanying signal
|
|
"""
|
|
logger.info(
|
|
"Received signal on thread %s, handling", threading.current_thread().name
|
|
)
|
|
self.terminate_flag = True
|
|
|
|
def shutdown(self, signalnum=None, flag=None):
|
|
"""
|
|
Stop and exit, waiting for all child processes to exit.
|
|
:param signalnum: UNIX signal number
|
|
:param flag: Flags accompanying signal
|
|
"""
|
|
logger.info("Shutting down, waiting for running jobs to complete...")
|
|
|
|
self.stop_dispatch_timers()
|
|
self._release_master()
|
|
|
|
self.daily_timer.stop()
|
|
self.stats_timer.stop()
|
|
self.systemd_watchdog_timer.stop()
|
|
if self.config.watchdog_enabled:
|
|
self.watchdog_timer.stop()
|
|
|
|
self._stop_managers_and_wait()
|
|
|
|
# try to release master lock
|
|
logger.info(
|
|
"Shutdown of %s/%s complete", os.getpid(), threading.current_thread().name
|
|
)
|
|
self.exit(0)
|
|
|
|
def start_dispatch_timers(self):
|
|
"""
|
|
Start all dispatch timers and begin pushing events into queues.
|
|
This should only be started when we are the master dispatcher.
|
|
"""
|
|
for manager in self.queue_managers.values():
|
|
try:
|
|
manager.start_dispatch()
|
|
except AttributeError:
|
|
pass
|
|
|
|
def stop_dispatch_timers(self):
|
|
"""
|
|
Stop all dispatch timers, this should be called when we are no longer the master dispatcher.
|
|
"""
|
|
for manager in self.queue_managers.values():
|
|
try:
|
|
manager.stop_dispatch()
|
|
except AttributeError:
|
|
pass
|
|
|
|
def _stop_managers(self):
|
|
for manager in self.queue_managers.values():
|
|
manager.stop()
|
|
|
|
def _stop_managers_and_wait(self):
|
|
"""
|
|
Stop all QueueManagers, and wait for their processing threads to complete.
|
|
We send the stop signal to all QueueManagers first, then wait for them to finish.
|
|
"""
|
|
self._stop_managers()
|
|
|
|
for manager in self.queue_managers.values():
|
|
manager.stop_and_wait()
|
|
|
|
def check_single_instance(self):
|
|
"""
|
|
Check that there is only one instance of the service running on this computer.
|
|
We do this be creating a file in the base directory (.lock.service) if it doesn't exist and
|
|
obtaining an exclusive lock on that file.
|
|
"""
|
|
lock_file = "{}/{}".format(self.config.BASE_DIR, ".lock.service")
|
|
|
|
import fcntl
|
|
|
|
self._fp = open(
|
|
lock_file, "w"
|
|
) # keep a reference so the file handle isn't garbage collected
|
|
self._fp.flush()
|
|
try:
|
|
fcntl.lockf(self._fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except IOError:
|
|
logger.warning("Another instance is already running, quitting.")
|
|
self.exit(2)
|
|
|
|
def log_performance_stats(self):
|
|
logger.info("Counting up time spent polling")
|
|
|
|
try:
|
|
# Report on the poller instance as a whole
|
|
self._db.query(
|
|
"INSERT INTO poller_cluster(node_id, poller_name, poller_version, poller_groups, last_report, master) "
|
|
'values("{0}", "{1}", "{2}", "{3}", NOW(), {4}) '
|
|
'ON DUPLICATE KEY UPDATE poller_version="{2}", poller_groups="{3}", last_report=NOW(), master={4}; '.format(
|
|
self.config.node_id,
|
|
self.config.name,
|
|
"librenms-service",
|
|
",".join(str(i) for i in self.config.group),
|
|
1 if self.is_master else 0,
|
|
)
|
|
)
|
|
|
|
# Find our ID
|
|
self._db.query(
|
|
'SELECT id INTO @parent_poller_id FROM poller_cluster WHERE node_id="{0}"; '.format(
|
|
self.config.node_id
|
|
)
|
|
)
|
|
|
|
for worker_type, manager in self.queue_managers.items():
|
|
worker_seconds, devices = manager.performance.reset()
|
|
|
|
# Record the queue state
|
|
self._db.query(
|
|
"INSERT INTO poller_cluster_stats(parent_poller, poller_type, depth, devices, worker_seconds, workers, frequency) "
|
|
'values(@parent_poller_id, "{0}", {1}, {2}, {3}, {4}, {5}) '
|
|
"ON DUPLICATE KEY UPDATE depth={1}, devices={2}, worker_seconds={3}, workers={4}, frequency={5}; ".format(
|
|
worker_type,
|
|
sum(
|
|
[
|
|
manager.get_queue(group).qsize()
|
|
for group in self.config.group
|
|
]
|
|
),
|
|
devices,
|
|
worker_seconds,
|
|
getattr(self.config, worker_type).workers,
|
|
getattr(self.config, worker_type).frequency,
|
|
)
|
|
)
|
|
except (pymysql.err.Error, ConnectionResetError, RedisConnectionError):
|
|
logger.critical(
|
|
"Unable to log performance statistics - is the database still online?",
|
|
exc_info=True,
|
|
)
|
|
|
|
def systemd_watchdog(self):
|
|
if "systemd.daemon" in sys.modules:
|
|
notify("WATCHDOG=1")
|
|
|
|
def logfile_watchdog(self):
|
|
|
|
try:
|
|
# check that lofgile has been written to within last poll period
|
|
logfile_mdiff = datetime.now().timestamp() - os.path.getmtime(
|
|
self.config.watchdog_logfile
|
|
)
|
|
except FileNotFoundError as e:
|
|
logger.error("Log file not found! {}".format(e))
|
|
return
|
|
|
|
if logfile_mdiff > self.config.poller.frequency:
|
|
logger.critical(
|
|
"BARK! Log file older than {}s, restarting service!".format(
|
|
self.config.poller.frequency
|
|
),
|
|
exc_info=True,
|
|
)
|
|
self.restart()
|
|
else:
|
|
logger.info("Log file updated {}s ago".format(int(logfile_mdiff)))
|
|
|
|
def exit(self, code=0):
|
|
sys.stdout.flush()
|
|
sys.exit(code)
|