Distribution EOLE » Modules » Zéphir » zephir-client

# -*- coding: UTF-8 -*-

###########################################################################

# Eole NG - 2007

# Copyright Pole de Competence Eole  (Ministere Education - Academie Dijon)

# Licence CeCill  cf /root/LicenceEole.txt

# eole@ac-dijon.fr

###########################################################################

"""

Services Twisted de collection et de publication de données.

"""

import locale, gettext, os, pwd, shutil, random

from pathlib2 import Path

from glob import glob

import cjson

import traceback

# install locales early

from zephir.monitor.agentmanager import ZEPHIRAGENTS_DATADIR

APP = 'zephir-agents'

DIR = os.path.join(ZEPHIRAGENTS_DATADIR, 'i18n')

gettext.install(APP, DIR, unicode=False)

from creole.client import CreoleClient

from twisted.application import internet, service

from twisted.internet import utils, reactor

from twisted.web import resource, server, static, util, xmlrpc

from twisted.python import syslog

from zephir.monitor.agentmanager import config as cfg

from zephir.monitor.agentmanager.util import ensure_dirs, md5file, get_md5files, log

from zephir.monitor.agentmanager.web_resources import ZephirServerResource

from zephir.monitor.agentmanager.clientmanager import ClientManager

try:

    import zephir.zephir_conf.zephir_conf as conf_zeph

    from zephir.lib_zephir import zephir_proxy, convert, zephir_dir, update_sudoers, charset

    from zephir.lib_zephir import log as zeph_log

    registered = 1

except:

    # serveur non enregistré sur zephir

    registered = 0

creole_client = None

def container_aware_path(path):

    global creole_client

    path_elements = path.split('::')

    path = Path(path_elements[0])

    name = path.name

    if len(path_elements) > 1 and path_elements[1]:

        container = path_elements[1]

        if not creole_client:

            creole_client = CreoleClient()

        container_path = creole_client.get_creole('container_path_{}'.format(container), None)

        if container_path:

            path = Path(container_path).joinpath(path)

        name = name + "::{}".format(container)

    return path, name

class ZephirService(service.MultiService):

    """Main Twisted service for Zephir apps"""

    def __init__(self, config, root_resource=None, serve_static=False):

        """config will be completed by default values"""

        service.MultiService.__init__(self)

        self.config = cfg.DEFAULT_CONFIG.copy()

        self.config.update(config)

        self.updater = self.publisher = None

        # mise à jour des scripts clients dans sudoers

        if registered:

            update_sudoers()

        # parent web server

        if root_resource is None:

            self.root_resource = resource.Resource()

            webserver = internet.TCPServer(self.config['webserver_port'],

                                                server.Site(self.root_resource))

            webserver.setServiceParent(service.IServiceCollection(self))

        else:

            self.root_resource = root_resource

        # serve global static files

        if serve_static:

            self.root_resource.putChild('static',

                                        static.File(self.config['static_web_dir']))

    # subservices factory methods

    def with_updater(self):

        assert self.updater is None

        self.updater = UpdaterService(self.config, self, self.root_resource)

        return self

    def with_publisher(self):

        assert self.publisher is None

        self.publisher = PublisherService(self.config, self, self.root_resource)

        return self

    def with_updater_and_publisher(self):

        assert self.updater is None

        assert self.publisher is None

        self.updater = UpdaterService(self.config, self, self.root_resource)

        self.publisher = PublisherService(self.config, self, self.root_resource,

                                          show_clients_page = False,

                                          live_agents={self.config['host_ref']: self.updater.agents})

        return self

class UpdaterService(service.MultiService, xmlrpc.XMLRPC):

    """Schedules measures, data serialisation and upload."""

    def __init__(self, config, parent, root_resource):

        """config should be complete"""

        service.MultiService.__init__(self)

        xmlrpc.XMLRPC.__init__(self)

        self.old_obs = None

        self.config = config

        # updates site.cfg file

        self.update_static_data()

        # start subservices

        loc, enc = locale.getdefaultlocale()

        log.msg(_('default locale: %s encoding: %s') % (loc, enc))

        if enc == 'utf':

            log.msg(_('Warning: locale encoding %s broken in RRD graphs, set e.g: LC_ALL=fr_FR') % enc)

        self.agents = self.load_agents()

        # attach to parent service

        self.setServiceParent(service.IServiceCollection(parent))

        root_resource.putChild('xmlrpc', self)

    def startService(self):

        """initialize zephir services"""

        service.MultiService.startService(self)

        reactor.callLater(2,self.schedule_all)

        # mise à jour du préfixe de log (twisted par défaut)

        # FIX : on conserve la référence à l'ancien observer pour

        # éviter les pb à la fermeture du service

        self.old_obs = None

        if len(log.theLogPublisher.observers) >= 1:

            self.old_obs = log.theLogPublisher.observers[0]

        try:

            from zephir.backend import config as conf_zeph

            log_prefix = 'zephir_backend'

        except:

            log_prefix = 'zephiragents'

        new_obs = syslog.SyslogObserver(log_prefix, options=syslog.DEFAULT_OPTIONS, facility=syslog.DEFAULT_FACILITY)

        log.addObserver(new_obs.emit)

        log.removeObserver(self.old_obs)

        if registered != 0:

            # on est enregistré sur zephir => initiation de

            # la création et l'envoi d'archives

            self.setup_uucp()

            # dans le cas ou un reboot a été demandé, on indique que le redémarrage est bon

            if os.path.isfile(os.path.join(zephir_dir,'reboot.lck')):

                try:

                    zeph_log('REBOOT',0,'redémarrage du serveur terminé')

                    os.unlink(os.path.join(zephir_dir,'reboot.lck'))

                except:

                    pass

    def stopService(self):

        """stops zephir services"""

        if self.old_obs:

            log.removeObserver(log.theLogPublisher.observers[0])

            log.addObserver(self.old_obs)

        service.MultiService.stopService(self)

    def load_agents(self):

        """Charge tous les agents du répertoire de configurations."""

        log.msg(_("Loading agents from %s...") % self.config['config_dir'])

        loaded_agents = {}

        list_agents = glob(os.path.join(self.config['config_dir'], "*.agent"))

        for f in list_agents:

            log.msg(_("  from %s:") % os.path.basename(f))

            h = { 'AGENTS': None }

            execfile(f, globals(), h)

            assert h.has_key('AGENTS')

            for a in h['AGENTS']:

                assert not loaded_agents.has_key(a.name)

                # init agent data and do a first archive

                a.init_data(os.path.join(self.config['state_dir'],

                                         self.config['host_ref'],

                                         a.name))

                a.manager = self

                a.archive()

                loaded_agents[a.name] = a # /!\ écrasement des clés

                log.msg(_("    %s, period %d") % (a.name, a.period))

        log.msg(_("Loaded."))

        return loaded_agents

    # scheduling measures

    def schedule(self, agent_name):

        """Planifie les mesures périodiques d'un agent."""

        assert self.agents.has_key(agent_name)

        if self.agents[agent_name].period > 0:

            timer = internet.TimerService(self.agents[agent_name].period,

                                          self.wakeup_for_measure, agent_name)

            timer.setName(agent_name)

            timer.setServiceParent(service.IServiceCollection(self))

    def wakeup_for_measure(self, agent_name):

        """Callback pour les mesures planifiées."""

        assert self.agents.has_key(agent_name)

        # log.debug("Doing scheduled measure on " + agent_name)

        self.agents[agent_name].scheduled_measure()

    def schedule_all(self):

        """Planifie tous les agents chargés.

        Démarre le cycle de mesures périodiques de chaque agent

        chargé. La première mesure est prise immédiatement.

        """

        for agent_name in self.agents.keys():

            # charge les actions disponibles (standard en premier, puis les actions locales)

            # les actions locales écrasent les actions standard si les 2 existent

            for action_dir in (os.path.join(self.config['action_dir'],'eole'), self.config['action_dir']):

                f_actions = os.path.join(action_dir, "%s.actions" % agent_name)

                if os.path.isfile(f_actions):

                    actions = {}

                    execfile(f_actions, globals(), actions)

                    for item in actions.keys():

                        if item.startswith('action_'):

                            setattr(self.agents[agent_name], item, actions[item])

            # self.wakeup_for_measure(agent_name) # first measure at launch

            self.schedule(agent_name)

    def timer_for_agent_named(self, agent_name):

        assert self.agents.has_key(agent_name)

        return self.getServiceNamed(agent_name)

    # data upload to zephir server

    def setup_uucp(self):

        ensure_dirs(self.config['uucp_dir'])

        self.update_static_data()

        # récupération du délai de connexion à zephir

        try:

            reload(conf_zeph)

            # supression des éventuels répertoires de stats invalides

            # par ex, en cas de désinscription zephir 'manuelle'.

            # sur zephir : on garde toujours 0 pour éviter les conflits avec les serveurs enregistrés

            if not os.path.isdir('/var/lib/zephir'):

                for st_dir in os.listdir(self.config['state_dir']):

                    if st_dir != str(conf_zeph.id_serveur):

                        shutil.rmtree(os.path.join(self.config['state_dir'],st_dir))

            # vérification sur zephir du délai de connexion

            period = convert(zephir_proxy.serveurs.get_timeout(conf_zeph.id_serveur)[1])

        except:

            period = 0

        if period < 30:

            period = self.config['upload_period']

            log.msg(_('Using default period : %s seconds') % period)

        # on ajoute un décalage aléatoire (entre 30 secondes et period) au premier démarrage

        # (pour éviter trop de connexions simultanées si le service est relancé par crontab)

        delay = random.randrange(30,period)

        reactor.callLater(delay,self.wakeup_for_upload)

    def update_static_data(self):

        original = os.path.join(self.config['config_dir'], 'site.cfg')

        if os.path.isfile(original):

            destination = cfg.client_data_dir(self.config, self.config['host_ref'])

            ensure_dirs(destination)

            need_copy = False

            try:

                org_mtime = os.path.getmtime(original)

                dest_mtime = os.path.getmtime(os.path.join(destination, 'site.cfg'))

            except OSError:

                need_copy = True

            if need_copy or (org_mtime > dest_mtime):

                shutil.copy(original, destination)

    def wakeup_for_upload(self, recall=True):

        # relecture du délai de connexion sur zephir

        try:

            reload(conf_zeph)

            period = convert(zephir_proxy.serveurs.get_timeout(conf_zeph.id_serveur)[1])

        except:

            period = 0

        # on relance la fonction dans le délai demandé

        if period < 30:

            period = self.config['upload_period']

            log.msg(_('Using default period : %s seconds') % period)

        # on ajoute un décalage au premier démarrage

        # (pour éviter trop de connexions simultanées si le service est relancé par crontab)

        if recall:

            reactor.callLater(period,self.wakeup_for_upload)

        # virer l'ancienne archive du rép. uucp

        for agent in self.agents.values():

            agent.archive()

            # agent.reset_max_status()

        self.update_static_data()

        # archiver dans rép. uucp, donner les droits en lecture sur l'archive

        try:

            assert conf_zeph.id_serveur != 0

            client_dir = os.path.join(self.config['tmp_data_dir'],str(conf_zeph.id_serveur))

        except:

            client_dir = os.path.join(self.config['tmp_data_dir'],self.config['host_ref'])

        try:

            # purge du répertoire temporaire

            if os.path.isdir(client_dir):

                shutil.rmtree(client_dir)

            os.makedirs(client_dir)

        except: # non existant

            pass

        args = ['-Rf',os.path.abspath(os.path.join(cfg.client_data_dir(self.config, self.config['host_ref']),'site.cfg'))]

        ignore_file = os.path.abspath(os.path.join(self.config['state_dir'],'ignore_list'))

        if os.path.exists(ignore_file):

            args.append(ignore_file)

        # on ne copie que les données des agents instanciés

        # cela évite de remonter par exemple les stats rvp si le service a été désactivé

        for agent_name in self.agents.keys():

            args.append(os.path.abspath(cfg.agent_data_dir(self.config, self.config['host_ref'],agent_name)))

        args.append(os.path.abspath(client_dir))

        res = utils.getProcessOutput('/bin/cp', args = args)

        res.addCallbacks(self._make_archive,

                         lambda x: log.msg(_("/!\ copy failed (%s)\n"

                                             "data: %s")

                                           % (x, self.config['state_dir'])))

    def _check_md5(self):

        def to_bytes(objet):

            """Transforme les objets unicode contenus dans un objet en bytes

            """

            if isinstance(objet, tuple):

                l = []

                for item in objet:

                    l.append(to_bytes(item))

                return '({})'.format(', '.join(l))

            if isinstance(objet, list):

                l = []

                for item in objet:

                    l.append(to_bytes(item))

                return '[{}]'.format(', '.join(l))

            if isinstance(objet, dict):

                dico={}

                for cle in objet:

                    dico[to_bytes(cle)] = to_bytes(objet[cle])

                return '{{{}}}'.format(', '.join(['{}: {}'.format(el[0], el[1]) for el in sorted(dico.items())]))

            if isinstance(objet, unicode):

                string =  objet.encode(charset)

                return "'{}'".format(string)

            if isinstance(objet, int):

                return str(objet)

            if isinstance(objet, float):

                return str(objet)

            if objet == None:

                return 'None'

            return objet

        # calcul de sommes md5 pour config.eol et les patchs

        rep_src = "/usr/share/eole/creole"

        rep_conf = "/etc/eole"

        data = []

        try:

            for src, dst, pattern in get_md5files(cfg.distrib_version):

                if src == 'variables.eol':

                    # cas particulier : variables.eol, on génère le fichier à chaque fois

                    orig_eol = os.path.join(rep_conf, 'config.eol')

                    if os.path.isfile(orig_eol):

                        var_eol = os.path.join(rep_src, 'variables.eol')

                        # on crée un fichier avec variable:valeur ordonné par nom de variable

                        conf = cjson.decode(file(orig_eol).read(), all_unicode=True)

                        var_names = conf.keys()

                        var_names.sort()

                        f_var = file(var_eol, 'w')

                        with open(var_eol, 'w') as f_var:

                            for key, value in sorted(conf.items()):

                                if key not in ['mode_zephir', '___version___'] and isinstance(value, dict) and 'val' in value:

                                    if type(value['val']) in [list, dict, tuple]:

                                        converted_value = to_bytes(value['val'])

                                    else:

                                        converted_value = value['val']

                                    f_var.write('{}:{}\n'.format(key, converted_value))

                if os.path.isdir(os.path.join(rep_src,src)):

                    fics = os.listdir(os.path.join(rep_src,src))

                    fics = [(os.path.join(src,fic),os.path.join(dst,fic)) for fic in fics]

                else:

                    fics = [(src,dst)]

                for fic, fic_dst in fics:

                    if os.path.isfile(os.path.join(rep_src,fic)):

                        if (pattern is None) or fic.endswith(pattern):

                            md5res = md5file(os.path.join(rep_src,fic))

                            data.append("%s  %s\n" % (md5res, fic_dst))

            if Path('/usr/share/zephir/zephir_conf/fichiers_zephir').is_file():

                with open('/usr/share/zephir/zephir_conf/fichiers_zephir', 'r') as fz_fh:

                    fichiers_zephir = []

                    for l in fz_fh.readlines():

                        if l.startswith('#'):

                            continue

                        if l.startswith('%%'):

                            break

                        fichier_serveur, name = container_aware_path(l.strip())

                        if fichier_serveur.is_file():

                            md5res = md5file(fichier_serveur.as_posix())

                            data.append("{}  {}\n".format(md5res, Path('fichiers_zephir').joinpath(name)))

                        elif fichier_serveur.is_dir():

                            for step in os.walk(fichier_serveur.as_posix()):

                                for sub_fichier_serveur in step[2]:

                                    sub_fichier_serveur_full_path = Path(step[0]).joinpath(sub_fichier_serveur)

                                    md5res = md5file(sub_fichier_serveur_full_path.as_posix())

                                    sub_fichier_serveur = sub_fichier_serveur_full_path.relative_to(fichier_serveur)

                                    data.append("{}  {}\n".format(md5res,

                                                                Path('fichiers_zephir').joinpath(name, sub_fichier_serveur)))

            if Path('/usr/share/zephir/zephir_conf/fichiers_variante').is_file():

                with open('/usr/share/zephir/zephir_conf/fichiers_variante', 'r') as fz_fh:

                    fichiers_zephir = []

                    for l in fz_fh.readlines():

                        if l.startswith('#'):

                            continue

                        if l.startswith('%%'):

                            break

                        fichier_serveur, name = container_aware_path(l.strip())

                        if fichier_serveur.is_file():

                            md5res = md5file(fichier_serveur.as_posix())

                            data.append("{}  {}\n".format(md5res, Path('fichiers_zephir').joinpath('variante', name)))

        except:

            # on n'empêche pas de continuer les opérations si le calcul du md5 n'est pas bon

            log.msg('!! Erreur rencontrée lors du calcul du md5 de config.eol !!')

            traceback.print_exc()

        try:

            assert conf_zeph.id_serveur != 0

            outf = file(os.path.join(self.config['tmp_data_dir'],"config%s.md5" % str(conf_zeph.id_serveur)), "w")

        except:

            outf = file(os.path.join(self.config['tmp_data_dir'],"config%s.md5" % self.config['host_ref']), "w")

        outf.writelines(data)

        outf.close()

    def _get_packages(self, *args):

        """génère une liste des paquets installés

        """

        try:

            assert conf_zeph.id_serveur != 0

            cmd_pkg = ("/usr/bin/dpkg-query -W >" + os.path.join(self.config['tmp_data_dir'],"packages%s.list" % str(conf_zeph.id_serveur)))

        except:

            cmd_pkg = ("/usr/bin/dpkg-query -W >" + os.path.join(self.config['tmp_data_dir'],"packages%s.list" % self.config['host_ref']))

        os.system(cmd_pkg)

    def _make_archive(self,*args):

        self._check_md5()

        self._get_packages()

        # compression des données à envoyer

        try:

            assert conf_zeph.id_serveur != 0

            tarball = os.path.join(self.config['uucp_dir'],'site%s.tar' % str(conf_zeph.id_serveur))

        except:

            tarball = os.path.join(self.config['uucp_dir'],'site%s.tar' % self.config['host_ref'])

        tar_cwd = os.path.dirname(os.path.abspath(self.config['tmp_data_dir']))

        tar_dir = os.path.basename(os.path.abspath(self.config['tmp_data_dir']))

        res = utils.getProcessOutput('/bin/tar',

                                     args = ('czf', tarball,

                                             '--exclude', 'private',

                                             '-C', tar_cwd,

                                             tar_dir))

        res.addCallbacks(self._try_chown,

                         lambda x: log.msg(_("/!\ archiving failed (%s)\n"

                                             "data: %s\narchive: %s")

                                           % (str(x), self.config['state_dir'], tarball)),

                         callbackArgs = [tarball])

    def _try_chown(self, tar_output, tarball):

        try:

            uucp_uid, uucp_gid = pwd.getpwnam('uucp')[2:4]

            uid = os.getuid()

            os.chown(tarball, uucp_uid, uucp_gid) # only change group id so that uucp can read while we can still write

        except OSError, e:

            log.msg("/!\ chown error, check authorizations (%s)" % e)

        # upload uucp

        # on fait également un chown sur le fichier deffered_logs au cas ou il serait en root

        try:

            uucp_uid, uucp_gid = pwd.getpwnam('uucp')[2:4]

            os.chown('/usr/share/zephir/deffered_logs', uucp_uid, uucp_gid)

        except:

            log.msg("/!\ chown error on deffered_logs")

        os.system('/usr/share/zephir/scripts/zephir_client call &> /dev/null')

    # xmlrpc methods

    def xmlrpc_list_agents(self):

        """@return: Liste des agents chargés"""

        return self.agents.keys()

    xmlrpc_list_agents.signature = [['array']]

    def xmlrpc_agents_menu(self):

        """@return: Liste des agents chargés et structure d'affichage"""

        try:

            menu = {}

            for name, agent in self.agents.items():

                if agent.section != None:

                    if not menu.has_key(agent.section):

                        menu[agent.section] = []

                    menu[agent.section].append((name, agent.description))

            return menu

        except Exception, e:

            log.msg(e)

    xmlrpc_agents_menu.signature = [['struct']]

    def xmlrpc_status_for_agents(self, agent_name_list = []):

        """

        @return: Les statuts des agents listés dans un dictionnaire

        C{{nom:status}}. Le status est lui-même un dictionnaire avec

        pour clés C{'level'} et C{'message'}. Seuls les noms d'agents

        effectivement chargés apparaîtront parmi les clés du

        dictionnaire.

        """

        result = {}

        if len(agent_name_list) == 0:

            agent_name_list = self.agents.keys()

        for agent_name in agent_name_list:

            if self.agents.has_key(agent_name):

                result[agent_name] = self.agents[agent_name].check_status().to_dict()

        return result

    xmlrpc_status_for_agents.signature = [['string', 'struct']]

    def xmlrpc_reset_max_status_for_agents(self, agent_name_list=[]):

            if len(agent_name_list) == 0:

                agent_name_list = self.agents.keys()

            for agent_name in agent_name_list:

                if self.agents.has_key(agent_name):

                    self.agents[agent_name].reset_max_status()

            return "ok"

    def xmlrpc_archive_for_upload(self):

        self.wakeup_for_upload(False)

        return "ok"

class PublisherService(service.MultiService):

    """Serves the web interface for current agent data"""

    def __init__(self, config, parent, root_resource,

                 live_agents=None,

                 show_clients_page=True):

        """config should be complete"""

        service.MultiService.__init__(self)

        self.config = config

        self.show_clients_page = show_clients_page

        self.manager = ClientManager(self.config, live_agents)

        # attach to parent service

        self.setServiceParent(service.IServiceCollection(parent))

        # run webserver

        rsrc = ZephirServerResource(self.config, self.manager)

        root_resource.putChild('agents', rsrc)

        default_page = './agents/'

        if not self.show_clients_page:

            default_page += self.config['host_ref'] + '/'

        root_resource.putChild('', util.Redirect(default_page))

#TODO

# update resources: loading host structures, manager -> agent dict

# connect publisher and updater to zephir service (web server, config...)

# client manager: liste des host_ref, {host_ref => agent_manager}

# agent manager: structure, {nom => agent_data}