initial import of new codebase (7e731b86) · Commits · Mirror / cc-node

bin/cc-node

+24 −206

Original line number	Diff line number	Diff line
		#!/usr/bin/env python

		import logging
		import threading
		import logging.handlers
		import sys
		import os
		import atexit
		import ConfigParser
		from math import exp
		from time import sleep
		from optparse import OptionParser

		from daemon import DaemonContext
		from sjrpc.core import RpcError

		from ccnode.ccnode import CCNode
		from ccnode import __version__

		MAX_AUTH_TIMEOUT = 30
		DEFAULT_CONFIG_FILE = '/etc/cc-node.conf'
		DEFAULT_PID_FILE = '/var/run/cc-node.pid'
		DEFAULT_CONFIGURATION = {
		'address': '',
		'login': '',
		'password': '',
		'port': '1984',
		'verbosity': '0',
		'detect_hypervisor': 'yes',
		#'ssl_cert': '',
		'command_execution' : 'yes',
		'force_xen' : 'no',
		}

		def authentication(node, suicide, login, password):
		'''
		Node authentication thread
		'''
		timeout = 1
		while not suicide.is_set() and node.get_rpc():
		logging.debug('Sending authentication request')
		if node.authentify(login, password):
		logging.error('Authentication suscessfull')
		return
		else:
		logging.error('Authentication failure')
		timeout += 0.1
		if timeout >= MAX_AUTH_TIMEOUT:
		timeout = MAX_AUTH_TIMEOUT
		sleep(exp(timeout))

		def setup_logging(options, instance_id=None):
		'''
		'''
		level = logging.ERROR
		verb = int(options['verbosity'])
		if verb:
		if verb == 1:
		level = logging.WARNING
		elif verb == 2:
		level = logging.INFO
		else:
		level = logging.DEBUG
		logger = logging.getLogger()
		logger.setLevel(level)
		if options['stdout']:
		handler = logging.StreamHandler()
		else:
		facility = logging.handlers.SysLogHandler.LOG_DAEMON
		handler = logging.handlers.SysLogHandler(address='/dev/log',
		facility=facility)
		fmt = 'cc-node: %(levelname)s '
		if options['verbosity'] == 0:
		fmt += '%(message)s'
		else:
		if isinstance(instance_id, int):
		fmt += 'id=%i ' % instance_id
		if options['stdout']:
		fmt += ("\x1B[30;47m%(process)d\x1B[0m:"
		"\x1B[30;42m%(threadName)s\x1B[0m:"
		"\x1B[30;43m%(funcName)s\x1B[0m:"
		"\x1B[30;44m%(lineno)d\x1B[0m@"
		"\x1B[30;45m%(msecs)d"
		"\x1B[0m\t\t "
		"%(message)s")
		else:
		fmt += ("%(process)d:"
		"%(threadName)s:"
		"%(funcName)s:"
		"%(lineno)d@"
		"%(msecs)d "
		"%(message)s")
		handler.setFormatter(logging.Formatter(fmt))
		logger.handlers = []
		logger.addHandler(handler)

		def run_node(options):
		'''
		'''
		# instance death event
		suicide_event = threading.Event()

		try:
		# setup logging infrastructure
		setup_logging(options)

		# create client
		logging.debug('Initializing client')
		try:
		node = CCNode(options['address'], int(options['port']),
		options['detect_hypervisor'] == 'yes',
		options['command_execution'] == 'yes',
		force_xen=(options['force_xen'] == 'yes'))
		except Exception as err:
		logging.error('Client initialization failure: `%s`:`%s`',
		repr(err), err)
		raise err

		# reconfigure logging with instance ID
		setup_logging(options, instance_id=id(node))

		# auth thread
		logging.debug('Starting authentication thread')
		auth_thread = threading.Thread(target=authentication, name='Auth',
		args=(node, suicide_event, options['login'], options['password']))
		auth_thread.daemon = True
		auth_thread.start()

		# main loop
		logging.debug('Starting main loop')
		node.run()
		except Exception as err:
		logging.error('run_node: `%s` -> `%s`', repr(err), err)
		try:
		node.get_rpc().shutdown()
		except:
		pass
		finally:
		# ensure everything is killed properly
		suicide_event.set()
		# -- coding: utf-8 --
		#
		# Cloud-Control Node
		# Copyright (C) 2011 SmartJog [http://www.smartjog.com]
		#
		# 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/>.
		#
		# Authors:
		# * 2011-06: Thibault VINCENT <thibault.vincent@smartjog.com>
		#

		from ccnode.launcher import Launcher

		if __name__ == '__main__':

		# configure argument parser
		op = OptionParser(version='%%prog %s' % __version__)
		op.add_option('-c', '--config', default=DEFAULT_CONFIG_FILE,
		help='configuration file (default: %default)')
		op.add_option('-d', '--daemonize', default=False, action='store_true',
		help='run as daemon and write pid file')
		op.add_option('-p', '--pidfile', default=DEFAULT_PID_FILE,
		help='pid file (default: %default)')
		op.add_option('-s', '--stdout', action='store_true', default=False,
		help='log on standard output instead of syslog')
		cliopts, args = op.parse_args()

		# read the config file
		config = ConfigParser.SafeConfigParser()
		config.read(cliopts.config)
		try:
		options = dict(config.items('node'))
		except ConfigParser.NoSectionError:
		sys.stderr.write('Configuration error: `node` section must exist '
		'in `%s`\n' % cliopts.config)
		sys.exit(1)

		# applying default config file options
		for opt, default in DEFAULT_CONFIGURATION.iteritems():
		if opt not in options or not options[opt]:
		if default is None:
		sys.stderr.write('Configuration error: you must specify `%s` '
		'option in `%s` !\n' % (opt, cliopts.config))
		sys.exit(1)
		else:
		options[opt] = default

		# Merge command options and .conf file options
		for opt in ('daemonize', 'pidfile', 'stdout'):
		if getattr(cliopts, opt) is not None:
		options[opt] = getattr(cliopts, opt)

		# Create option set for the daemonization process
		daemon_opts = {}
		daemonize = options['daemonize']
		if isinstance(daemonize, str):
		daemonize = daemonize in ('yes', 'true')
		daemon_opts['detach_process'] = daemonize
		if not daemonize:
		daemon_opts['stderr'] = sys.stderr
		daemon_opts['stdout'] = sys.stderr
		if options['pidfile']:
		#daemon_opts['pidfile'] = lockfile.FileLock(options['pidfile'])
		pidfile = open(cliopts.pidfile, 'w')
		daemon_opts['files_preserve'] = [pidfile]

		with DaemonContext(**daemon_opts):
		# write pid in pidfile
		if pidfile is not None:
		pidfile.write('%s' % os.getpid())
		pidfile.flush()
		# register pidfile cleaning
		@atexit.register
		def clean_pidfile():
		pidfile.seek(0)
		pidfile.truncate()
		pidfile.flush()
		# run node
		while True:
		run_node(options)
		logging.critical('Critical error, restarting node !')
		sleep(2)
		Launcher().main()

ccnode/init.py

+46 −5

Original line number	Diff line number	Diff line
		#!/usr/bin/env python
		# -- coding: utf-8 --
		#
		# Cloud-Control Node
		# Copyright (C) 2011 SmartJog [http://www.smartjog.com]
		#
		# 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/>.
		#
		# Authors:
		# * 2011-06: Thibault VINCENT <thibault.vincent@smartjog.com>
		#

		"""
		__product__ = 'Cloud-Control Node'
		__version__ = '0~dev'
		__canonical__ = 'cc-node'
		__develmode__ = False

		TODO : rewrite based on older doc

		"""
		def git_version():
		global __version__
		import os
		import sys
		from subprocess import Popen, PIPE, CalledProcessError
		cwd = os.getcwd()
		try:
		os.chdir(os.path.dirname(sys.argv[0]))
		p = Popen(["git", "log", "--pretty=format:%H" ], stdout=PIPE,
		stderr=open("/dev/null", "wb"))
		p.wait()
		if p.returncode == 0:
		githash = p.stdout.readline().strip()
		if len(githash) > 0:
		__version__ += "-git-%s" % githash
		except OSError:
		pass
		finally:
		os.chdir(cwd)

		__version__ = '17'
		if __version__.find("dev") != -1:
		__develmode__ = True
		git_version()

ccnode/ccnode.py

deleted100644 → 0

+0 −102

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --

		import logging
		from threading import Timer, Lock, Event
		from time import sleep

		from sjrpc.utils import ConnectionProxy
		from sjrpc.core import RpcError, RpcConnection
		# FIXME relative import
		import handlers


		class CCNode(object):
		'''
		Handle node initialization, connection to server, and authentication
		'''
		def __init__(self, server, port, hypervisor, exec_cmd, force_xen=False,
		cert=None):
		'''
		'''
		self._scheduler_timer = None
		self._scheduler_mutex = Lock()
		self._scheduler_stopped = Event()
		self._is_hv = hypervisor # hugly
		self._is_xen = force_xen # hugly
		self._exec_cmd = exec_cmd # hugly
		self._handler = None
		self._rpc = RpcConnection.from_addr_ssl(server, port,
		handler=self._handler)
		self._server = ConnectionProxy(self._rpc)

		def run(self):
		'''
		'''
		try:
		self._rpc.run()
		except:
		raise
		finally:
		self._scheduler_stopped.set()

		def authentify(self, login, password):
		'''
		'''
		logging.debug('Authenticating user `%s` on connection `%i`' % (login,
		id(self._server)))
		try:
		role = self._server.authentify(login, password)
		except RpcError as err:
		if err.exception == 'AuthenticationError':
		logging.warning('Authentication error')
		else:
		logging.warning('Unknown error while authenticating: %s' % err)
		return False
		except Exception as err:
		logging.debug('Unhandled exception: `%s`' % err)
		else:
		if role == 'hv':
		self._handler = handlers.NodeHandler(self,
		self._is_hv,
		self._exec_cmd,
		self._is_xen)
		elif role == 'host':
		self._handler = handlers.NodeHandler(self,
		False,
		self._exec_cmd,
		self._is_xen)
		else:
		logging.warning('Bad role affected by server: %s' % role)
		raise Exception()

		self._rpc.get_protocol(0).set_handler(self._handler)
		self._scheduler_rearm()
		return True

		def _scheduler_rearm(self):
		'''
		'''
		self._scheduler_timer = Timer(5, self._scheduler_run)
		self._scheduler_timer.start()

		def _scheduler_run(self):
		'''
		'''
		with self._scheduler_mutex:
		if not self._scheduler_stopped.is_set():
		self._handler.scheduler_run()
		sleep(0.1)
		self._scheduler_rearm()

		def get_server(self):
		'''
		'''
		return self._server

		def get_rpc(self):
		return self._rpc

		def get_handler(self):
		'''
		'''
		return self._handler

ccnode/common.py

deleted100644 → 0

+0 −359

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --

		import os
		import re
		import psutil
		from subprocess import Popen, PIPE, STDOUT
		from multiprocessing import cpu_count
		from platform import platform, machine, system
		from socket import gethostbyaddr, gethostname
		from jobs import JobManager
		from lvm import LVM
		from drbd import DRBDPool


		class Host(object):
		'''
		Root class for all physical or virtual machines that CloudControl Node may
		run on, or manage. It is not intended for direct usage by NodeHandler.
		'''


		class LocalHost(Host):
		'''
		Regular host with no hypervisor support, all methods defined here are
		expected to provide information to the NodeHandler.
		'''

		ARCH = {
		'i386' : 'x86',
		'i486' : 'x86',
		'i586' : 'x86',
		'i686' : 'x86',
		'x86_64' : 'x64',
		}

		def __init__(self):
		'''
		'''
		super(LocalHost, self).__init__()
		self.jobmgr = JobManager()
		self.drbdpool = DRBDPool()
		self.lvm = LVM()

		def scheduler_run(self):
		'''
		'''
		pass

		def get_hw_serial(self):
		'''
		'''
		serial = None
		data = open('/sys/class/dmi/id/product_serial').read().strip()
		if data:
		serial = data
		return serial

		def get_hw_vendor(self):
		'''
		'''
		vendor = None
		data = open('/sys/class/dmi/id/sys_vendor').read().strip()
		if data:
		vendor = data
		return vendor

		def get_hw_product(self):
		'''
		'''
		product = None
		data = open('/sys/class/dmi/id/product_name').read().strip()
		if data:
		product = data
		return product

		def get_hw_bios(self):
		'''
		'''
		bios = ''
		bios_ver = open('/sys/class/dmi/id/bios_version').read().strip()
		bios_date = open('/sys/class/dmi/id/bios_date').read().strip()
		if bios_ver:
		bios += bios_ver
		if bios_date:
		bios += ' (%s)' % bios_date
		if not bios:
		bios = None
		return bios

		def get_chassis_asset(self):
		'''
		'''
		asset = None
		data = open('/sys/class/dmi/id/chassis_asset_tag').read().strip()
		if data:
		asset = data
		return asset

		def get_chassis_serial(self):
		'''
		'''
		serial = None
		data = open('/sys/class/dmi/id/chassis_serial').read().strip()
		if data:
		serial = data
		return serial

		def get_name(self):
		'''
		'''
		result = None
		hostname = gethostname()
		fqdn = gethostbyaddr(hostname)[0]
		result = fqdn if fqdn else hostname
		return result

		def get_uname(self):
		'''
		'''
		uname = None
		data = ' '.join(os.uname())
		if data:
		uname = data
		return uname

		def get_platform(self):
		'''
		'''
		result = None
		try:
		p = platform()
		if p:
		result = p
		except:
		pass
		return result

		def get_system(self):
		'''
		'''
		result = None
		try:
		p = system()
		if p:
		result = p.lower()
		except:
		pass
		return result

		def get_uptime(self):
		'''
		'''
		uptime = None
		try:
		data = open("/proc/uptime").read().split()
		if data:
		uptime = int(float(data[0]))
		except:
		pass
		return uptime

		def get_loadavg(self):
		'''
		'''
		load = None
		try:
		data = ' '.join('%.2f' % load for load in os.getloadavg())
		if data:
		load = data
		except:
		pass
		return load

		def get_arch(self):
		'''
		'''
		arch = None
		try:
		a = machine()
		if a in self.ARCH:
		arch = self.ARCH[a]
		except:
		pass
		return arch

		def get_cpu(self):
		'''
		'''
		cpucount = None
		try:
		data = cpu_count()
		if data:
		cpucount = data
		except:
		pass
		return cpucount

		def get_cpu_usage(self):
		'''
		'''
		usage = None
		try:
		data = '%.1f' % psutil.cpu_percent()
		if data:
		usage = data
		except:
		pass
		return usage

		def get_mem(self):
		'''
		'''
		mem = None
		try:
		data = psutil.avail_phymem() + psutil.used_phymem()
		if data:
		mem = data
		except:
		pass
		return mem

		def get_mem_free(self):
		'''
		'''
		free = None
		try:
		data = psutil.avail_phymem()
		if data:
		free = data
		except:
		pass
		return free

		def get_mem_used(self):
		'''
		'''
		used = None
		try:
		data = psutil.used_phymem()
		if data:
		used = data
		except:
		pass
		return used

		def get_disks(self):
		'''
		'''
		disks = {}
		try:
		re_pattern = re.compile(r'([sh]d[a-z]+)')
		found = [bd for bd in os.listdir('/sys/block/')
		if re_pattern.match(bd)]
		for disk in found:
		fullname = os.path.join('/sys/block', disk, 'size')
		size = int(open(fullname).read())
		if size > 0:
		disks[disk] = size * 512
		except:
		pass
		return disks

		def power_shutdown(self):
		'''
		'''
		return self.execute('/sbin/shutdown -h -P 0')

		def power_off(self):
		'''
		'''
		return self.execute('/sbin/shutdown -h -P -n 0')

		def power_reboot(self):
		'''
		'''
		return self.execute('/sbin/shutdown -r -f 0')

		def power_force_reboot(self):
		'''
		'''
		return self.execute('/sbin/shutdown -r -n 0')

		def execute(self, command):
		'''
		'''
		output = None
		try:
		#FIXME: stop using shell=true and parse arguments with shlex.split()
		data = Popen(command,
		shell=True,
		bufsize=-1,
		stdin=PIPE,
		stdout=PIPE,
		stderr=STDOUT).communicate()[0]
		if data:
		output = data
		except:
		pass
		return output


		class Hypervisor(LocalHost):
		'''
		'''

		def __init__(self):
		'''
		'''
		super(Hypervisor, self).__init__()

		def storage(self):
		'''
		'''
		raise NotImplementedError

		def vm_list(self):
		'''
		'''
		raise NotImplementedError

		def vm_list_running(self):
		'''
		'''
		raise NotImplementedError

		def vm_list_stopped(self):
		'''
		'''
		raise NotImplementedError

		def vm_list_paused(self):
		'''
		'''
		raise NotImplementedError

		def vm_get(self, name):
		'''
		'''
		raise NotImplementedError


		class VM(Host):
		'''
		'''


		class Storage(object):
		'''
		'''


		class StoragePool(object):
		'''
		'''


		class StorageVolume(object):
		'''
		'''

ccnode/config.py

0 → 100644

+193 −0

File added.

Preview size limit exceeded, changes collapsed.