[collecty.git] / src / collecty / plugins / base.py

#!/usr/bin/python
###############################################################################
#                                                                             #
# collecty - A system statistics collection daemon for IPFire                 #
# Copyright (C) 2012 IPFire development team                                  #
#                                                                             #
# This program is free software: you can redistribute it and/or modify        #
# it under the terms of the GNU General Public License as published by        #
# the Free Software Foundation, either version 3 of the License, or           #
# (at your option) any later version.                                         #
#                                                                             #
# This program is distributed in the hope that it will be useful,             #
# but WITHOUT ANY WARRANTY; without even the implied warranty of              #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
# GNU General Public License for more details.                                #
#                                                                             #
# You should have received a copy of the GNU General Public License           #
# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

from __future__ import division

import datetime
import logging
import math
import os
import rrdtool
import threading
import time

from ..constants import *
from ..i18n import _

_plugins = {}

def get():
	"""
		Returns a list with all automatically registered plugins.
	"""
	return _plugins.values()

class Timer(object):
	def __init__(self, timeout, heartbeat=1):
		self.timeout = timeout
		self.heartbeat = heartbeat

		self.delay = 0

		self.reset()

	def reset(self, delay=0):
		# Save start time.
		self.start = time.time()

		self.delay = delay

		# Has this timer been killed?
		self.killed = False

	@property
	def elapsed(self):
		return time.time() - self.start - self.delay

	def cancel(self):
		self.killed = True

	def wait(self):
		while self.elapsed < self.timeout and not self.killed:
			time.sleep(self.heartbeat)

		return self.elapsed > self.timeout


class Plugin(threading.Thread):
	# The name of this plugin.
	name = None

	# A description for this plugin.
	description = None

	# Templates which can be used to generate a graph out of
	# the data from this data source.
	templates = []

	# The default interval for all plugins
	interval = 60

	# Automatically register all providers.
	class __metaclass__(type):
		def __init__(plugin, name, bases, dict):
			type.__init__(plugin, name, bases, dict)

			# The main class from which is inherited is not registered
			# as a plugin.
			if name == "Plugin":
				return

			if not all((plugin.name, plugin.description)):
				raise RuntimeError(_("Plugin is not properly configured: %s") \
					% plugin)

			_plugins[plugin.name] = plugin

	def __init__(self, collecty, **kwargs):
		threading.Thread.__init__(self, name=self.description)
		self.daemon = True

		self.collecty = collecty

		# Check if this plugin was configured correctly.
		assert self.name, "Name of the plugin is not set: %s" % self.name
		assert self.description, "Description of the plugin is not set: %s" % self.description

		# Initialize the logger.
		self.log = logging.getLogger("collecty.plugins.%s" % self.name)
		self.log.propagate = 1

		self.data = []

		# Run some custom initialization.
		self.init(**kwargs)

		# Keepalive options
		self.running = True
		self.timer = Timer(self.interval)

		self.log.info(_("Successfully initialized %s") % self.__class__.__name__)

	@property
	def path(self):
		"""
			Returns the name of the sub directory in which all RRD files
			for this plugin should be stored in.
		"""
		return self.name

	### Basic methods

	def init(self, **kwargs):
		"""
			Do some custom initialization stuff here.
		"""
		pass

	def collect(self):
		"""
			Gathers the statistical data, this plugin collects.
		"""
		time_start = time.time()

		# Run through all objects of this plugin and call the collect method.
		for o in self.objects:
			now = datetime.datetime.utcnow()
			try:
				result = o.collect()
			except:
				self.log.warning(_("Unhandled exception in %s.collect()") % o, exc_info=True)
				continue

			if not result:
				self.log.warning(_("Received empty result: %s") % o)
				continue

			self.log.debug(_("Collected %s: %s") % (o, result))

			# Add the object to the write queue so that the data is written
			# to the databases later.
			self.collecty.write_queue.add(o, now, result)

		# Returns the time this function took to complete.
		return (time.time() - time_start)

	def run(self):
		self.log.debug(_("%s plugin has started") % self.name)

		# Initially collect everything
		self.collect()

		while self.running:
			# Reset the timer.
			self.timer.reset()

			# Wait until the timer has successfully elapsed.
			if self.timer.wait():
				delay = self.collect()
				self.timer.reset(delay)

		self.log.debug(_("%s plugin has stopped") % self.name)

	def shutdown(self):
		self.log.debug(_("Received shutdown signal."))
		self.running = False

		# Kill any running timers.
		if self.timer:
			self.timer.cancel()


class Object(object):
	# The schema of the RRD database.
	rrd_schema = None

	# RRA properties.
	rra_types     = ["AVERAGE", "MIN", "MAX"]
	rra_timespans = [3600, 86400, 604800, 2678400, 31622400]
	rra_rows      = 2880

	def __init__(self, plugin, *args, **kwargs):
		self.plugin = plugin

		# Indicates if this object has collected its data
		self.collected = False

		# Initialise this object
		self.init(*args, **kwargs)

		# Create the database file.
		self.create()

	def __repr__(self):
		return "<%s>" % self.__class__.__name__

	@property
	def collecty(self):
		return self.plugin.collecty

	@property
	def log(self):
		return self.plugin.log

	@property
	def id(self):
		"""
			Returns a UNIQUE identifier for this object. As this is incorporated
			into the path of RRD file, it must only contain ASCII characters.
		"""
		raise NotImplementedError

	@property
	def file(self):
		"""
			The absolute path to the RRD file of this plugin.
		"""
		return os.path.join(DATABASE_DIR, self.plugin.path, "%s.rrd" % self.id)

	### Basic methods

	def init(self, *args, **kwargs):
		"""
			Do some custom initialization stuff here.
		"""
		pass

	def create(self):
		"""
			Creates an empty RRD file with the desired data structures.
		"""
		# Skip if the file does already exist.
		if os.path.exists(self.file):
			return

		dirname = os.path.dirname(self.file)
		if not os.path.exists(dirname):
			os.makedirs(dirname)

		# Create argument list.
		args = self.get_rrd_schema()

		rrdtool.create(self.file, *args)

		self.log.debug(_("Created RRD file %s.") % self.file)
		for arg in args:
			self.log.debug("  %s" % arg)

	def info(self):
		return rrdtool.info(self.file)

	@property
	def stepsize(self):
		return self.plugin.interval

	@property
	def heartbeat(self):
		return self.stepsize * 2

	def get_rrd_schema(self):
		schema = [
			"--step", "%s" % self.stepsize,
		]
		for line in self.rrd_schema:
			if line.startswith("DS:"):
				try:
					(prefix, name, type, lower_limit, upper_limit) = line.split(":")

					line = ":".join((
						prefix,
						name,
						type,
						"%s" % self.heartbeat,
						lower_limit,
						upper_limit
					))
				except ValueError:
					pass

			schema.append(line)

		xff = 0.1

		cdp_length = 0
		for rra_timespan in self.rra_timespans:
			if (rra_timespan / self.stepsize) < self.rra_rows:
				rra_timespan = self.stepsize * self.rra_rows

			if cdp_length == 0:
				cdp_length = 1
			else:
				cdp_length = rra_timespan // (self.rra_rows * self.stepsize)

			cdp_number = math.ceil(rra_timespan / (cdp_length * self.stepsize))

			for rra_type in self.rra_types:
				schema.append("RRA:%s:%.10f:%d:%d" % \
					(rra_type, xff, cdp_length, cdp_number))

		return schema

	def execute(self):
		if self.collected:
			raise RuntimeError("This object has already collected its data")

		self.collected = True
		self.now = datetime.datetime.utcnow()

		# Call the collect
		result = self.collect()

	def commit(self):
		"""
			Will commit the collected data to the database.
		"""
		# Make sure that the RRD database has been created
		self.create()


class GraphTemplate(object):
	# A unique name to identify this graph template.
	name = None

	# Instructions how to create the graph.
	rrd_graph = None

	# Extra arguments passed to rrdgraph.
	rrd_graph_args = []

	def __init__(self, ds):
		self.ds = ds

	@property
	def collecty(self):
		return self.ds.collecty

	def graph(self, file, interval=None,
			width=GRAPH_DEFAULT_WIDTH, height=GRAPH_DEFAULT_HEIGHT):
		args = [
			"--width", "%d" % width,
			"--height", "%d" % height,
		]
		args += self.collecty.graph_default_arguments
		args += self.rrd_graph_args

		intervals = {
			None   : "-3h",
			"hour" : "-1h",
			"day"  : "-25h",
			"week" : "-360h",
			"year" : "-365d",
		}

		args.append("--start")
		try:
			args.append(intervals[interval])
		except KeyError:
			args.append(interval)

		info = { "file" : self.ds.file }
		for item in self.rrd_graph:
			try:
				args.append(item % info)
			except TypeError:
				args.append(item)

		rrdtool.graph(file, *args)
Commit	Line	Data
eed405de MT	1	#!/usr/bin/python
	2	###############################################################################
	3	# #
	4	# collecty - A system statistics collection daemon for IPFire #
	5	# Copyright (C) 2012 IPFire development team #
	6	# #
	7	# This program is free software: you can redistribute it and/or modify #
	8	# it under the terms of the GNU General Public License as published by #
	9	# the Free Software Foundation, either version 3 of the License, or #
	10	# (at your option) any later version. #
	11	# #
	12	# This program is distributed in the hope that it will be useful, #
	13	# but WITHOUT ANY WARRANTY; without even the implied warranty of #
	14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
	15	# GNU General Public License for more details. #
	16	# #
	17	# You should have received a copy of the GNU General Public License #
	18	# along with this program. If not, see <http://www.gnu.org/licenses/>. #
	19	# #
	20	###############################################################################
	21
965a9c51 MT	22	from __future__ import division
965a9c51 MT	23
72364063	24	import datetime
4ac0cdf0	25	import logging
965a9c51	26	import math
4ac0cdf0 MT	27	import os
4ac0cdf0 MT	28	import rrdtool
49ce926e	29	import threading
4ac0cdf0 MT	30	import time
4ac0cdf0 MT	31
4ac0cdf0	32	from ..constants import *
eed405de MT	33	from ..i18n import _
eed405de MT	34
5d140577 MT	35	_plugins = {}
	36
	37	def get():
	38	"""
	39	Returns a list with all automatically registered plugins.
	40	"""
	41	return _plugins.values()
	42
4be39bf9 MT	43	class Timer(object):
	44	def __init__(self, timeout, heartbeat=1):
	45	self.timeout = timeout
	46	self.heartbeat = heartbeat
	47
e746a56e MT	48	self.delay = 0
e746a56e MT	49
4be39bf9 MT	50	self.reset()
4be39bf9 MT	51
e746a56e	52	def reset(self, delay=0):
4be39bf9 MT	53	# Save start time.
	54	self.start = time.time()
	55
e746a56e MT	56	self.delay = delay
e746a56e MT	57
4be39bf9 MT	58	# Has this timer been killed?
	59	self.killed = False
	60
	61	@property
	62	def elapsed(self):
e746a56e	63	return time.time() - self.start - self.delay
4be39bf9 MT	64
	65	def cancel(self):
	66	self.killed = True
	67
	68	def wait(self):
	69	while self.elapsed < self.timeout and not self.killed:
	70	time.sleep(self.heartbeat)
	71
	72	return self.elapsed > self.timeout
	73
	74
5d140577	75	class Plugin(threading.Thread):
4ac0cdf0 MT	76	# The name of this plugin.
	77	name = None
	78
	79	# A description for this plugin.
	80	description = None
	81
b1ea4956 MT	82	# Templates which can be used to generate a graph out of
	83	# the data from this data source.
	84	templates = []
	85
72364063 MT	86	# The default interval for all plugins
72364063 MT	87	interval = 60
4ac0cdf0	88
5d140577 MT	89	# Automatically register all providers.
	90	class __metaclass__(type):
	91	def __init__(plugin, name, bases, dict):
	92	type.__init__(plugin, name, bases, dict)
	93
	94	# The main class from which is inherited is not registered
	95	# as a plugin.
	96	if name == "Plugin":
	97	return
	98
	99	if not all((plugin.name, plugin.description)):
	100	raise RuntimeError(_("Plugin is not properly configured: %s") \
	101	% plugin)
	102
	103	_plugins[plugin.name] = plugin
	104
eed405de	105	def __init__(self, collecty, **kwargs):
49ce926e MT	106	threading.Thread.__init__(self, name=self.description)
	107	self.daemon = True
	108
eed405de MT	109	self.collecty = collecty
eed405de MT	110
4ac0cdf0 MT	111	# Check if this plugin was configured correctly.
	112	assert self.name, "Name of the plugin is not set: %s" % self.name
	113	assert self.description, "Description of the plugin is not set: %s" % self.description
4ac0cdf0 MT	114
	115	# Initialize the logger.
	116	self.log = logging.getLogger("collecty.plugins.%s" % self.name)
	117	self.log.propagate = 1
eed405de	118
eed405de MT	119	self.data = []
eed405de MT	120
269f74cd MT	121	# Run some custom initialization.
	122	self.init(**kwargs)
	123
4be39bf9 MT	124	# Keepalive options
	125	self.running = True
	126	self.timer = Timer(self.interval)
	127
72364063	128	self.log.info(_("Successfully initialized %s") % self.__class__.__name__)
4ac0cdf0	129
73db5226	130	@property
72364063	131	def path(self):
73db5226	132	"""
72364063 MT	133	Returns the name of the sub directory in which all RRD files
72364063 MT	134	for this plugin should be stored in.
73db5226 MT	135	"""
	136	return self.name
	137
72364063 MT	138	### Basic methods
	139
	140	def init(self, **kwargs):
4ac0cdf0	141	"""
72364063	142	Do some custom initialization stuff here.
4ac0cdf0	143	"""
72364063 MT	144	pass
	145
	146	def collect(self):
	147	"""
	148	Gathers the statistical data, this plugin collects.
	149	"""
	150	time_start = time.time()
	151
	152	# Run through all objects of this plugin and call the collect method.
	153	for o in self.objects:
	154	now = datetime.datetime.utcnow()
	155	try:
	156	result = o.collect()
	157	except:
	158	self.log.warning(_("Unhandled exception in %s.collect()") % o, exc_info=True)
	159	continue
	160
	161	if not result:
	162	self.log.warning(_("Received empty result: %s") % o)
	163	continue
	164
	165	self.log.debug(_("Collected %s: %s") % (o, result))
	166
	167	# Add the object to the write queue so that the data is written
	168	# to the databases later.
	169	self.collecty.write_queue.add(o, now, result)
	170
	171	# Returns the time this function took to complete.
	172	return (time.time() - time_start)
	173
	174	def run(self):
	175	self.log.debug(_("%s plugin has started") % self.name)
	176
	177	# Initially collect everything
	178	self.collect()
	179
	180	while self.running:
	181	# Reset the timer.
	182	self.timer.reset()
	183
	184	# Wait until the timer has successfully elapsed.
	185	if self.timer.wait():
	186	delay = self.collect()
	187	self.timer.reset(delay)
	188
	189	self.log.debug(_("%s plugin has stopped") % self.name)
	190
	191	def shutdown(self):
	192	self.log.debug(_("Received shutdown signal."))
	193	self.running = False
	194
	195	# Kill any running timers.
	196	if self.timer:
	197	self.timer.cancel()
4ac0cdf0	198
72364063 MT	199
	200	class Object(object):
	201	# The schema of the RRD database.
	202	rrd_schema = None
	203
	204	# RRA properties.
	205	rra_types = ["AVERAGE", "MIN", "MAX"]
	206	rra_timespans = [3600, 86400, 604800, 2678400, 31622400]
	207	rra_rows = 2880
	208
	209	def __init__(self, plugin, args, *kwargs):
	210	self.plugin = plugin
	211
	212	# Indicates if this object has collected its data
	213	self.collected = False
	214
	215	# Initialise this object
	216	self.init(args, *kwargs)
	217
	218	# Create the database file.
	219	self.create()
	220
	221	def __repr__(self):
	222	return "<%s>" % self.__class__.__name__
4ac0cdf0	223
965a9c51	224	@property
72364063 MT	225	def collecty(self):
72364063 MT	226	return self.plugin.collecty
965a9c51	227
881751ed	228	@property
72364063 MT	229	def log(self):
	230	return self.plugin.log
	231
	232	@property
	233	def id(self):
	234	"""
	235	Returns a UNIQUE identifier for this object. As this is incorporated
	236	into the path of RRD file, it must only contain ASCII characters.
	237	"""
	238	raise NotImplementedError
881751ed	239
4ac0cdf0 MT	240	@property
	241	def file(self):
	242	"""
	243	The absolute path to the RRD file of this plugin.
	244	"""
72364063 MT	245	return os.path.join(DATABASE_DIR, self.plugin.path, "%s.rrd" % self.id)
	246
	247	### Basic methods
	248
	249	def init(self, args, *kwargs):
	250	"""
	251	Do some custom initialization stuff here.
	252	"""
	253	pass
eed405de	254
4ac0cdf0 MT	255	def create(self):
	256	"""
	257	Creates an empty RRD file with the desired data structures.
	258	"""
	259	# Skip if the file does already exist.
	260	if os.path.exists(self.file):
	261	return
eed405de	262
4ac0cdf0 MT	263	dirname = os.path.dirname(self.file)
	264	if not os.path.exists(dirname):
	265	os.makedirs(dirname)
eed405de	266
965a9c51	267	# Create argument list.
ff0bbd88	268	args = self.get_rrd_schema()
965a9c51 MT	269
965a9c51 MT	270	rrdtool.create(self.file, *args)
eed405de	271
4ac0cdf0	272	self.log.debug(_("Created RRD file %s.") % self.file)
dadb8fb0 MT	273	for arg in args:
dadb8fb0 MT	274	self.log.debug(" %s" % arg)
eed405de	275
72364063 MT	276	def info(self):
	277	return rrdtool.info(self.file)
	278
	279	@property
	280	def stepsize(self):
	281	return self.plugin.interval
	282
	283	@property
	284	def heartbeat(self):
	285	return self.stepsize * 2
	286
965a9c51 MT	287	def get_rrd_schema(self):
	288	schema = [
	289	"--step", "%s" % self.stepsize,
	290	]
	291	for line in self.rrd_schema:
	292	if line.startswith("DS:"):
	293	try:
	294	(prefix, name, type, lower_limit, upper_limit) = line.split(":")
	295
	296	line = ":".join((
	297	prefix,
	298	name,
	299	type,
881751ed	300	"%s" % self.heartbeat,
965a9c51 MT	301	lower_limit,
	302	upper_limit
	303	))
	304	except ValueError:
	305	pass
	306
	307	schema.append(line)
	308
	309	xff = 0.1
	310
	311	cdp_length = 0
	312	for rra_timespan in self.rra_timespans:
	313	if (rra_timespan / self.stepsize) < self.rra_rows:
	314	rra_timespan = self.stepsize * self.rra_rows
	315
	316	if cdp_length == 0:
	317	cdp_length = 1
	318	else:
	319	cdp_length = rra_timespan // (self.rra_rows * self.stepsize)
	320
	321	cdp_number = math.ceil(rra_timespan / (cdp_length * self.stepsize))
	322
	323	for rra_type in self.rra_types:
	324	schema.append("RRA:%s:%.10f:%d:%d" % \
	325	(rra_type, xff, cdp_length, cdp_number))
	326
	327	return schema
	328
72364063 MT	329	def execute(self):
	330	if self.collected:
	331	raise RuntimeError("This object has already collected its data")
eed405de	332
72364063 MT	333	self.collected = True
72364063 MT	334	self.now = datetime.datetime.utcnow()
4ac0cdf0	335
72364063 MT	336	# Call the collect
72364063 MT	337	result = self.collect()
4ac0cdf0	338
72364063	339	def commit(self):
4ac0cdf0	340	"""
72364063	341	Will commit the collected data to the database.
4ac0cdf0	342	"""
72364063	343	# Make sure that the RRD database has been created
dadb8fb0 MT	344	self.create()
dadb8fb0 MT	345
b1ea4956 MT	346
	347	class GraphTemplate(object):
	348	# A unique name to identify this graph template.
	349	name = None
	350
	351	# Instructions how to create the graph.
	352	rrd_graph = None
	353
	354	# Extra arguments passed to rrdgraph.
	355	rrd_graph_args = []
	356
	357	def __init__(self, ds):
	358	self.ds = ds
	359
	360	@property
	361	def collecty(self):
	362	return self.ds.collecty
	363
73db5226 MT	364	def graph(self, file, interval=None,
73db5226 MT	365	width=GRAPH_DEFAULT_WIDTH, height=GRAPH_DEFAULT_HEIGHT):
73db5226 MT	366	args = [
	367	"--width", "%d" % width,
	368	"--height", "%d" % height,
	369	]
	370	args += self.collecty.graph_default_arguments
	371	args += self.rrd_graph_args
eed405de	372
73db5226 MT	373	intervals = {
	374	None : "-3h",
	375	"hour" : "-1h",
	376	"day" : "-25h",
	377	"week" : "-360h",
	378	"year" : "-365d",
	379	}
eed405de MT	380
eed405de MT	381	args.append("--start")
b1ea4956	382	try:
eed405de	383	args.append(intervals[interval])
b1ea4956	384	except KeyError:
eed405de MT	385	args.append(interval)
eed405de MT	386
b1ea4956	387	info = { "file" : self.ds.file }
73db5226	388	for item in self.rrd_graph:
eed405de MT	389	try:
	390	args.append(item % info)
	391	except TypeError:
	392	args.append(item)
	393
73db5226	394	rrdtool.graph(file, *args)