entropy/services/client-updates-daemon

#!/usr/bin/python2 -O
# -*- coding: utf-8 -*-
"""
    # DESCRIPTION:
    # Entropy Client Updates daemon

    Copyright (C) 2007-2009 Fabio Erculiani

    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 2 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, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
"""
from __future__ import with_statement
import os
import sys
# this makes the daemon to not write the entropy pid file
# avoiding to lock other instances
sys.argv.append('--no-pid-handling')
import time
import gobject
import dbus
import dbus.service
import dbus.mainloop.glib
import signal
from threading import Lock

DAEMON_DEBUG = False
if "--debug" in sys.argv:
    DAEMON_DEBUG = True

# Entropy imports
sys.path.insert(0,'/usr/lib/entropy/libraries')
sys.path.insert(0,'/usr/lib/entropy/server')
sys.path.insert(0,'/usr/lib/entropy/client')
sys.path.insert(0,'../libraries')
sys.path.insert(0,'../server')
sys.path.insert(0,'../client')
if DAEMON_DEBUG:
    sys.path.insert(0,'/home/fabio/entropy/libraries')
from entropy.misc import LogFile
from entropy.i18n import _
from entropy.exceptions import PermissionDenied, RepositoryError, \
    MissingParameter
import entropy.tools as entropyTools
from entropy.client.interfaces import Client
from entropy.transceivers import urlFetcher
from entropy.const import etpConst
from entropy.core import SystemSettings as SysSet
from entropy.output import TextInterface, nocolor
nocolor()


SYS_SETTINGS = SysSet()
TEXT = TextInterface()
DAEMON_LOGFILE = os.path.join(etpConst['syslogdir'],"client-updater.log")
DAEMON_LOG = LogFile(SYS_SETTINGS['system']['log_level']+1,
    DAEMON_LOGFILE, header = "[client-updater]")
PREVIOUS_PROGRESS = ''
CHECK_DELAY_SECS = 3600*4
for xopt in sys.argv:
    if xopt.startswith("--secs=") and (len(xopt.split("=")) > 1):
        try:
            delay_secs = int(xopt.split("=")[1])
            CHECK_DELAY_SECS = delay_secs
        except ValueError:
            continue

def write_output(*args, **kwargs):
    message = args[0]
    global PREVIOUS_PROGRESS
    if PREVIOUS_PROGRESS == message:
        return
    message = time.strftime('[%H:%M:%S %d/%m/%Y %Z]') + " " + message
    PREVIOUS_PROGRESS = message
    DAEMON_LOG.write(message)
    DAEMON_LOG.flush()
    if DAEMON_DEBUG:
        TEXT.updateProgress(*args, **kwargs)

class DaemonUrlFetcher(urlFetcher):

    daemon_last_avg = 100
    __average = 0
    __downloadedsize = 0
    __remotesize = 0
    __datatransfer = 0

    def handle_statistics(self, th_id, downloaded_size, total_size,
            average, old_average, update_step, show_speed, data_transfer,
            time_remaining, time_remaining_secs):
        self.__average = average
        self.__downloadedsize = downloaded_size
        self.__remotesize = total_size
        self.__datatransfer = data_transfer

    def updateProgress(self):
        myavg = abs(int(round(float(self.__average), 1)))
        if abs((myavg - self.daemon_last_avg)) < 1:
            return
        if int(myavg) % 10 == 0:
            DAEMON_LOG.write("fetch @ %s%s" % (myavg, "%",))
            self.daemon_last_avg = myavg



class Entropy(Client):

    def init_singleton(self):
        Client.init_singleton(self, load_ugc = False,
            url_fetcher = DaemonUrlFetcher, repo_validation = False,
            xcache = False)
        self.updateProgress(
            "Loading Entropy Updates daemon: check every %ss, logfile: %s" % (
                CHECK_DELAY_SECS, DAEMON_LOGFILE,)
            )

    def updateProgress(self, *args, **kwargs):
        return write_output(*args, **kwargs)


class UpdatesDaemon(dbus.service.Object):

    def __init__(self):

        if not entropyTools.is_user_in_entropy_group():
            raise PermissionDenied('insufficient permissions')
        self.__alive = False
        self.__is_working_mutex = Lock()
        self.__updater = None
        self.__oncall_updater = None
        self.__quit_service_wd = None
        self.__quit_service_trigger = False
        self.__trigger_oncall_updater = False
        self.__fetch_mutex = Lock()
        self.__updates = []
        self.__updates_atoms = []
        self.__system_db_hash = None

        # start dbus service
        object_path = "/notifier"
        dbus_loop = dbus.mainloop.glib.DBusGMainLoop(set_as_default = True)
        system_bus = dbus.SystemBus(mainloop = dbus_loop)
        name = dbus.service.BusName("org.entropy.Client", bus = system_bus)
        dbus.service.Object.__init__ (self, system_bus, object_path)
        write_output("__init__: dbus service loaded")

        # this seems to avoid race conditions
        # with dbus service not being available
        time.sleep(2)


    def start(self):
        self.stop()
        self.__alive = True
        self.__updater = gobject.timeout_add(
            CHECK_DELAY_SECS*1000, self.run_fetcher)
        self.__oncall_updater = gobject.timeout_add(
            1000, self.run_oncall_fetcher)
        self.__quit_service_wd = gobject.timeout_add(
            2000, self.quit_service_watchdog)

    def stop(self):
        if self.__alive:
            self.__alive = False
            if self.__updater != None:
                gobject.source_remove(self.__updater)
            if self.__oncall_updater != None:
                gobject.source_remove(self.__oncall_updater)
            if self.__quit_service_wd != None:
                gobject.source_remove(self.__quit_service_wd)

    def do_alert(self, string, msg, urgency = "critical"):
        write_output('alert: %s, %s, urgency: %s' % (string, msg, urgency,))

    def run_oncall_fetcher(self):
        if not self.__trigger_oncall_updater:
            return self.__alive
        with self.__fetch_mutex:
            self.__trigger_oncall_updater = False
            self.__run_fetcher()
        return self.__alive

    def quit_service_watchdog(self):
        if self.__quit_service_trigger:
            self.__alive = False
            with self.__is_working_mutex:
                raise SystemExit(0)
        return self.__alive

    def run_fetcher(self):
        with self.__fetch_mutex:
            self.__run_fetcher()
        return self.__alive

    def __run_sync(self, repos, entropy):

        try:
            repo_conn = entropy.Repositories(
                repos, fetchSecurity = False, noEquoCheck = True)
            if DAEMON_DEBUG:
                write_output("__run_fetcher: repository interface loaded")
        except MissingParameter, err:
            if DAEMON_DEBUG:
                write_output(
                    "__run_fetcher: MissingParameter exception, error: %s" % (
                        err,))
        except Exception, err:
            if DAEMON_DEBUG:
                write_output(
                    "__run_fetcher: Unhandled exception, error: %s" % (err,))
        else:

            # 128: sync error, something bad happened
            # 2: repositories not available (all)
            # 1: not able to update all the repositories
            if DAEMON_DEBUG:
                write_output("__run_fetcher: preparing to run sync")
            rc_res = repo_conn.sync()
            del repo_conn
            if DAEMON_DEBUG:
                write_output("__run_fetcher: sync done")

            if rc_res == 1:
                err = _("Not all the repositories have been fetched")
                self.do_alert( _("Updates: repository issues"), err )
                entropy.destroy()
                del entropy
                return rc_res
            elif rc_res == 2:
                err = _("No repositories found online")
                self.do_alert( _("Updates: repository issues"), err )
                entropy.destroy()
                del entropy
                return rc_res
            elif rc_res == 128:
                err = _("Synchronization errors. Cannot update repositories.")
                self.do_alert( _("Updates: sync issues"), err )
                entropy.destroy()
                del entropy
                return rc_res
            elif isinstance(rc_res, basestring):
                self.do_alert( _("Updates: unhandled error"), rc_res )
                entropy.destroy()
                del entropy
                return rc_res

        return 0

    def __run_fetcher(self):

        if self.__updater == None:
            return 0

        with self.__is_working_mutex:

            rc_fetch = 0
            entropy = Entropy()
            entropy.SystemSettings.clear()
            # entropy resources locked?
            locked = entropy._resources_run_check_lock()
            if locked:
                if DAEMON_DEBUG:
                    write_output("__run_fetcher: resources locked, skipping")
                return rc_fetch

            if DAEMON_DEBUG:
                write_output("__run_fetcher: called %s" % (time.time(),))

            repos_to_up = self.get_repo_status(entropy)

            if repos_to_up:

                self.signal_updating()
                repos = repos_to_up.keys()
                rc_fetch = self.__run_sync(repos, entropy)
                if rc_fetch != 0:
                    return rc_fetch
                if DAEMON_DEBUG:
                    write_output("__run_fetcher: sync closed, rc: %s" % (
                        rc_fetch,))

            try:
                update, remove, fine = entropy.calculate_world_updates()
                del fine, remove
            except Exception, err:
                entropyTools.print_traceback(f = DAEMON_LOG)
                msg = "%s: %s" % (_("Updates: error"), err,)
                self.do_alert(_("Updates: error"), msg)
                return 1

            if update:
                self.do_alert(
                    _("Updates available"),
                    "%s %d %s" % (
                        _("There are"), len(update),
                        _("updates available."),),
                    urgency = 'critical'
                )
                self.__system_db_hash = entropy.clientDbconn.database_checksum()
                self.__updates = update[:]
                del self.__updates_atoms[:]
                self.signal_updates()

            return 0

    # compare repos status for updates
    @dbus.service.method ( "org.entropy.Client", in_signature = '',
        out_signature = 'a{si}')
    def get_repo_status(self, entropy):

        repos = {}
        try:
            repo_conn = entropy.Repositories(
                noEquoCheck = True, fetchSecurity = False)
        except MissingParameter:
            return repos
        except Exception, e:
            return repos

        # now get remote
        for repoid in entropy.SystemSettings['repositories']['available']:
            #entropy.update_repository_revision(repoid)
            if repo_conn.is_repository_updatable(repoid):
                if DAEMON_DEBUG:
                    write_output(
                        "get_repo_status: repo needs to be updated: %s" % (
                            repoid,))
                entropy.repository_move_clear_cache(repoid)
                repo_rev = entropy.get_repository_revision(repoid)
                online_rev = repo_conn.get_online_repository_revision(repoid)
                repos[repoid] = {
                    'local': repo_rev,
                    'remote': online_rev,
                }

        del repo_conn
        return repos

    @dbus.service.method ( "org.entropy.Client", in_signature = '',
        out_signature = '')
    def trigger_check(self):
        self.__trigger_oncall_updater = True

    @dbus.service.method ( "org.entropy.Client", in_signature = '',
        out_signature = 'av')
    def get_updates(self):

        entropy = Entropy()
        curr_hash = entropy.clientDbconn.database_checksum()
        if curr_hash == self.__system_db_hash:
            return self.__updates

        try:
            update, remove, fine = entropy.calculate_world_updates()
        except Exception, err:
            entropyTools.print_traceback(f = DAEMON_LOG)
            msg = "get_updates: %s: %s" % (_("Updates: error"), err,)
            self.do_alert(_("Updates: error"), msg)
            return self.__updates

        self.__updates = update[:]
        self.__system_db_hash = curr_hash
        return self.__updates

    @dbus.service.method ( "org.entropy.Client", in_signature = '',
        out_signature = 'av')
    def get_updates_atoms(self):

        if self.__updates_atoms:
            return self.__updates_atoms

        with self.__is_working_mutex:
            atoms = []
            entropy = Entropy()

            for idpackage, repoid in self.__updates:
                try:
                    dbc = entropy.open_repository(repoid)
                    atoms.append(dbc.retrieveAtom(idpackage))
                except RepositoryError:
                    continue

            self.__updates_atoms.extend(atoms)
            entropy.destroy()
            del entropy
            return self.__updates_atoms

    @dbus.service.method ( "org.entropy.Client", in_signature = '',
        out_signature = '')
    def close_connection(self):
        self.__quit_service_trigger = True

    # signal sent when updates are available for retrive
    @dbus.service.signal(dbus_interface = 'org.entropy.Client',
        signature = '')
    def signal_updates(self):
        if DAEMON_DEBUG:
            write_output("signal_updates: updates available!")

    # signal sent when daemon is updating the repositories
    @dbus.service.signal(dbus_interface = 'org.entropy.Client',
        signature = '')
    def signal_updating(self):
        if DAEMON_DEBUG:
            write_output("signal_updating: updating repos!")

if __name__ == "__main__":
    signal.signal(signal.SIGINT, signal.SIG_DFL)
    try:
        ud_i = UpdatesDaemon()
    except dbus.exceptions.DBusException:
        raise SystemExit(1)
    ud_i.start()
    main_loop = gobject.MainLoop()
    main_loop.run()
    ud_i.stop()
    raise SystemExit(0)