~dmedia/dmedia/trunk

#!/usr/bin/python3

# dmedia: distributed media library
# Copyright (C) 2011 Novacut Inc
#
# This file is part of `dmedia`.
#
# `dmedia` is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# `dmedia` 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 Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License along
# with `dmedia`.  If not, see <http://www.gnu.org/licenses/>.
#
# Authors:
#   Jason Gerard DeRose <jderose@novacut.com>

"""
Dmedia DBus service on org.freedesktop.Dmedia.

A quick note about why we don't need (and don't want) a "graceful" shutdown
procedure:

In terms of data-safety, neither the Dmedia `FileStore` nor CouchDB need a
"clean" shutdown.  Both are guaranteed to be in a consistent and otherwise
recoverable state even after "pulling the plug" with no warning.

We've sometimes have had problems with the service not being un-published from
Avahi when we yank the plug like this, but that's an Avahi bug.

Dmedia *must* be able to survive an ungraceful shutdown because in the real
world that will sometimes happen (power failure, catastrophic hardware failure,
etc).  So we *always* shutdown Dmedia this way, as a way to force us to ensure
that Dmedia can survive such a scenario.
"""

import os
os.nice(10)
import time
start_time = time.monotonic()
import argparse

import dbus
import dbus.service
from gi.repository import GLib, Gio
from microfiber import dumps

import dmedia
from dmedia import schema
from dmedia.units import bytes10, minsec
from dmedia.util import isfilestore
from dmedia.parallel import start_thread
from dmedia.startup import DmediaCouch
from dmedia.core import Core, start_httpd
from dmedia.service.background import Snapshots, LazyAccess, Downloads
from dmedia.service.avahi import Avahi
from dmedia.service.peers import Browser, Publisher
from dmedia.drives import Devices


def start_delta():
    return time.monotonic() - start_time

BUS = dmedia.BUS
IFACE = BUS
mainloop = GLib.MainLoop()
VolumeMonitor = Gio.VolumeMonitor.get()
devices = Devices()


class Service(dbus.service.Object):
    core = None
    peer = None
    ui = None
    thread = None
    publisher = None
    env_s = '{}'

    def __init__(self, busname):
        super().__init__(busname, object_path='/')

        # Before we do anything else, aquire the UserCouch lock:
        self.couch = DmediaCouch(dmedia.get_dmedia_dir())

        # Now go for it:
        log.info('Started at monotonic clock time: %s', minsec(int(start_time)))
        self.first_snapshot = True
        self.update_thread = None
        self.pending_update = None

    @dbus.service.signal(IFACE, signature='sb')
    def SnapshotComplete(self, name, success):
        log.info('@Dmedia.SnapshotComplete(%r, %r)', name, success)
        if name == '__all__':
            self.SnapshotAllComplete(success)

    @dbus.service.signal(IFACE, signature='b')
    def SnapshotAllComplete(self, success):
        log.info('@Dmedia.SnapshotAllComplete(%r)', success)
        #if self.first_snapshot:
        #    self.first_snapshot = False
        #    GLib.idle_add(self.on_idle3)

    @dbus.service.method(IFACE, in_signature='s', out_signature='b')
    def Snapshot(self, name):
        name = str(name)
        log.info('Dmedia.Snapshot(%r)', name)
        return self.snapshots.run(name)

    @dbus.service.method(IFACE, in_signature='', out_signature='b')
    def SnapshotAll(self):
        log.info('Dmedia.SnapshotAll()')
        return self.Snapshot('__all__')

    def run(self):
        if self.couch.user is None:
            log.info('First run, not starting CouchDB.')
            self.couch.create_machine_if_needed()
        else:
            self.start_core()
        mainloop.run()

    def start_core(self):
        start = time.monotonic()
        env = self.couch.auto_bootstrap()
        log.info('%r', self.couch._welcome)
        log.info('Starting CouchDB took %.3f', time.monotonic() - start)
        self.core = Core(env,
            self.couch.machine,
            self.couch.user,
            self.couch.get_ssl_config()
        )
        self.env_s = dumps(self.core.env, pretty=True)
        self.snapshots = Snapshots(
                self.core.env,
                self.couch.paths.dump,
                self.SnapshotComplete,
        )
        self.lazy_access = LazyAccess(self.core.db)
        self.downloads = Downloads(self.core.env, self.couch.get_ssl_config())
        log.info('Finished core startup in %.3f', time.monotonic() - start)
        GLib.timeout_add(300, self.on_idle1)

    def on_idle1(self):
        """
        Log CouchDB stats for dmedia-1, connect FileStore.
        """
        log.info('---- idle1 at time %.3f ----', start_delta())
        d = self.core.db.get()
        log.info('doc_count: %s, update_seq: %s, data_size: %s, disk_size: %s',
            d['doc_count'],
            d['update_seq'],
            bytes10(d['data_size']),
            bytes10(d['disk_size']),
        )
        if self.core.get_skip_internal():
            log.info('Skipping internal FileStore in %r', self.couch.basedir)
        else:
            if isfilestore(self.couch.basedir):
                self.core.connect_filestore(self.couch.basedir)
            else:
                info = devices.get_parentdir_info(self.couch.basedir)
                self.core.create_filestore(self.couch.basedir, **info)
        VolumeMonitor.connect('mount-added', self.on_mount_added)
        VolumeMonitor.connect('mount-pre-unmount', self.on_mount_pre_unmount)
        VolumeMonitor.connect('mount-removed', self.on_mount_removed)
        for mount in VolumeMonitor.get_mounts():
            self.on_mount_added(VolumeMonitor, mount)
        GLib.timeout_add(1500, self.on_idle2)

    def on_idle2(self):
        """
        Init project DB views, start peering browser, start HTTPD.
        """
        log.info('---- idle2 at time %.3f ----', start_delta())
        start_thread(self.core.init_project_views)
        if self.couch.pki.user.key_file is not None:
            self.peer = Browser(self, self.couch)
        self.httpd = start_httpd(self.core.env, self.core.ssl_config)
        self.httpd_port = self.httpd.address[1]
        GLib.timeout_add(1500, self.on_idle3)

    def on_idle3(self):
        """
        Publish "_dmedia._tcp" over Avahi, start replication, add timers.

        3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67
        """
        log.info('---- idle3 at time %.3f ----', start_delta())
        self.avahi = Avahi(self.core, self.httpd_port)
        self.avahi.run()
        GLib.timeout_add(10 * 1000, self.on_idle4)
        GLib.timeout_add(30 * 1000, self.on_idle5)
        GLib.timeout_add(90 * 1000, self.on_idle6)
        GLib.timeout_add(7 * 60 * 1000, self.core.restart_replication_tasks)
        GLib.timeout_add(11 * 60 * 1000, self.core.restart_downgrade_task)
        GLib.timeout_add(13 * 60 * 1000, self.core.restart_vigilance)
        GLib.timeout_add(17 * 60 * 1000, self.core.restart_filestore_tasks)

    def on_idle4(self):
        """
        Start background tasks.
        """
        log.info('---- idle4 at time %.3f ----', start_delta())
        self.core.start_background_tasks()

    def on_idle5(self):
        """
        Start vigilance task.
        """
        log.info('---- idle5 at time %.3f ----', start_delta())
        self.core.task_master.add_vigilance_task()
 
    def on_idle6(self):
        """
        Start downgrade and reclaim tasks.
        """
        log.info('---- idle6 at time %.3f ----', start_delta())
        self.core.task_master.add_downgrade_task()
        GLib.timeout_add(60 * 1000, self.core.reclaim_if_possible)

    def on_mount_added(self, vm, mount):
        parentdir = mount.get_root().get_path()
        log.info('mount-added: %r', parentdir)
        if parentdir is None:
            log.warning('parentdir is None, skipping possible blank optical media')
            return
        GLib.idle_add(self._add_mount, parentdir)

    def _add_mount(self, parentdir):
        if not isfilestore(parentdir):
            log.warning('%r is not a filestore, skipping', parentdir)
            return
        try:
            self.core.connect_filestore(parentdir)
        except Exception:
            log.exception('Could not connect FileStore %r', parentdir)

    def on_mount_pre_unmount(self, vm, mount):
        parentdir = mount.get_root().get_path()
        log.info('mount-pre-unmount: %r', parentdir)
        if parentdir is None:
            log.warning('parentdir is None, skipping possible blank optical media')
            return
        GLib.idle_add(self._remove_mount, parentdir)

    def on_mount_removed(self, vm, mount):
        parentdir = mount.get_root().get_path()
        log.info('mount-remove: %r', parentdir)
        if parentdir is None:
            log.warning('parentdir is None, skipping possible blank optical media')
            return
        GLib.idle_add(self._remove_mount, parentdir)

    def _remove_mount(self, parentdir):
        try:
            self.core.disconnect_filestore(parentdir)
        except KeyError:
            log.warning('%r is not a connected filestore', parentdir)

    ###########################################
    # Signals and Methods for Peering Protocol:

    @dbus.service.signal(IFACE, signature='s')
    def Message(self, message):
        log.info('@Dmedia.Message(%r)', message)

    @dbus.service.signal(IFACE, signature='sb')
    def DisplaySecret(self, secret, typo):
        log.info('@Dmedia.DisplaySecret(<secret>, %r)', typo)

    @dbus.service.method(IFACE, in_signature='s', out_signature='b')
    def GetSecret(self, peer_id):
        log.info('Dmedia.GetSecret(%r)', peer_id)
        if self.peer is None:
            return False
        return self.peer.get_secret(peer_id)

    @dbus.service.method(IFACE, in_signature='s', out_signature='b')
    def Cancel(self, peer_id):
        log.info('Dmedia.Cancel(%r)', peer_id)
        if self.peer is None:
            return False
        GLib.idle_add(self.peer.cancel, peer_id)
        return True

    @dbus.service.signal(IFACE, signature='')
    def Accept(self):
        log.info('@Dmedia.Accept()')

    @dbus.service.signal(IFACE, signature='b')
    def Response(self, success):
        log.info('@Dmedia.Response(%r)', success)

    @dbus.service.signal(IFACE, signature='')
    def InitDone(self):
        log.info('@Dmedia.InitDone()')

    @dbus.service.signal(IFACE, signature='')
    def PeeringDone(self):
        log.info('@Dmedia.PeeringDone()')

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def Version(self):
        """
        Return dmedia version.
        """
        return dmedia.__version__

    @dbus.service.method(IFACE, in_signature='', out_signature='b')
    def NeedsInit(self):
        """
        Return True if we need to do the firstrun init.
        """
        return self.couch.user is None

    @dbus.service.method(IFACE, in_signature='', out_signature='b')
    def CreateUser(self):
        log.info('Dmedia.CreateUser()')
        if self.couch.user is not None:
            return False
        if self.thread is not None:
            return False
        self.thread = start_thread(self.create_user)
        return True

    def create_user(self):
        log.info('create_user()')
        self.couch.wait_for_machine()
        self.couch.create_user()
        self.start_core()
        GLib.idle_add(self.on_init_done)

    def on_init_done(self):
        log.info('on_init_done()')
        self.thread.join()
        self.thread = None
        if self.publisher is not None:
            self.publisher.free()
            self.publisher = None
        self.InitDone()

    @dbus.service.method(IFACE, in_signature='', out_signature='b')
    def PeerWithExisting(self):
        log.info('Dmedia.PeerWithExisting()')
        if self.couch.user is not None:
            return False
        if self.publisher is not None:
            return False
        if self.thread is not None:
            return False
        self.publisher = Publisher(self, self.couch)
        self.thread = start_thread(self.wait)
        return True

    def wait(self):
        self.couch.wait_for_machine()
        GLib.idle_add(self.on_wait_done)

    def on_wait_done(self):
        self.thread.join()
        self.thread = None
        self.publisher.run()

    @dbus.service.method(IFACE, in_signature='s', out_signature='b')
    def SetSecret(self, secret):
        log.info('Dmedia.SetSecret()')
        return self.publisher.set_secret(secret)

    def set_user(self, user_id):
        log.info('set_user(%r)', user_id)
        assert self.couch.user is None
        assert self.thread is None
        self.thread = start_thread(self._set_user, user_id)

    def _set_user(self, user_id):
        self.couch.set_user(user_id)
        self.start_core()
        GLib.idle_add(self.on_init_done)

    @dbus.service.method(IFACE, in_signature='', out_signature='i')
    def Kill(self):
        """
        Kill the `dmedia-service` process.
        """
        log.info('Service.Kill()')
        mainloop.quit()
        return int(start_delta())

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def GetEnv(self):
        """
        Return dmedia env as JSON string.
        """
        log.info('Service.GetEnv()')
        return self.env_s

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def Tasks(self):
        """
        Info about currently running background tasks.
        """
        if self.core is None:
            return ''
        lines = [
            ('PID', 'AGE', 'TASK')
        ]
        curtime = time.time()
        for key in sorted(self.core.task_master.pool.active_tasks):
            task = self.core.task_master.pool.active_tasks[key]
            pid = str(task.process.pid)
            delta = max(0, curtime - task.start_time)
            uptime = minsec(int(delta))
            desc = str(key)
            cols = [pid, uptime, desc]
            lines.append(cols)
        widths = tuple(
            max(len(line[i]) for line in lines)
            for i in range(3)
        )
        lines.insert(1, tuple(
            ('=') * w for w in widths
        ))
        return '\n'.join(
            '  '.join(col.ljust(widths[i]) for (i, col) in enumerate(line))
            for line in lines
        )

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def Stores(self):
        """
        Return currently connected filestores
        """
        return dumps(self.core.machine['stores'], pretty=True)

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def Peers(self):
        """
        Return peers currently known on local network.
        """
        return dumps(self.core.machine['peers'], pretty=True)

    @dbus.service.method(IFACE, in_signature='s', out_signature='s')
    def CreateFileStore(self, parentdir):
        parentdir = str(parentdir)
        log.info('Dmedia.CreateFileStore(%r)', parentdir)
        self.core.create_filestore(parentdir)
        return self.Stores()

    @dbus.service.method(IFACE, in_signature='s', out_signature='')
    def DowngradeStore(self, store_id):
        store_id = str(store_id)
        log.info('Dmedia.DowngradeStore(%r)', store_id)
        start_thread(self.core.ms.downgrade_store, store_id)

    @dbus.service.method(IFACE, in_signature='', out_signature='')
    def DowngradeAll(self):
        log.info('Dmedia.DowngradeAll()')
        start_thread(self.core.ms.downgrade_all)

    @dbus.service.method(IFACE, in_signature='s', out_signature='')
    def PurgeStore(self, store_id):
        store_id = str(store_id)
        log.info('Dmedia.PurgeStore(%r)', store_id)
        start_thread(self.core.ms.purge_store, store_id)

    @dbus.service.method(IFACE, in_signature='', out_signature='')
    def PurgeAll(self):
        log.info('Dmedia.PurgeAll()')
        start_thread(self.core.ms.purge_all)

    @dbus.service.method(IFACE, in_signature='s', out_signature='(sys)')
    def Resolve(self, file_id):
        file_id = str(file_id)
        (file_id, status, filename) = self.core.resolve(file_id)
        if status in (0, 1):
            self.lazy_access.access(file_id)
            if status == 1:
                self.downloads.download(file_id)
        log.info('Dmedia.Resolve(%r) --> %r', file_id, filename)
        return (file_id, status, filename)

    @dbus.service.method(IFACE, in_signature='as', out_signature='a(sys)')
    def ResolveMany(self, ids):
        result = self.core.resolve_many([str(_id) for _id in ids])
        for (_id, status, filename) in result:
            if status in (0, 1):
                self.lazy_access.access(_id)
        return result

    @dbus.service.method(IFACE, in_signature='', out_signature='s')
    def AllocateTmp(self):
        return self.core.allocate_tmp()

    @dbus.service.method(IFACE, in_signature='ss', out_signature='a{ss}')
    def HashAndMove(self, tmp, origin):
        return self.core.hash_and_move(tmp, origin)

    @dbus.service.method(IFACE, in_signature='s', out_signature='b')
    def UpdateProject(self, project_id):
        if self.update_thread is not None:
            return False
        project_id = str(project_id)
        log.info('Dmedia.UpdateProject(%r)', project_id)
        assert self.update_thread is None
        self.update_thread = start_thread(self.do_update, project_id)
        return True

    def do_update(self, project_id):
        self.core.update_project(project_id)
        GLib.idle_add(self.on_update_complete)

    def on_update_complete(self):
        self.update_thread.join()
        self.update_thread = None
        if self.pending_update is not None:
            log.info('flushing pending project update')
            self.UpdateProject(self.pending_update)
            self.pending_update = None  

    @dbus.service.method(IFACE, in_signature='s', out_signature='')
    def SnapshotProject(self, project_id):
        project_id = str(project_id)
        log.info('Dmedia.SnapshotProject(%r)', project_id)
        if not self.UpdateProject(project_id):
            self.pending_update = project_id
        self.Snapshot(schema.project_db_name(project_id))
        #self.Snapshot(schema.DB_NAME)

    @dbus.service.method(IFACE, in_signature='s', out_signature='s')
    def AutoFormat(self, value):
        value = str(value)
        log.info('Dmedia.AutoFormat(%r)', value)
        if value == '':
            return dumps(self.core.get_auto_format())
        flag = {'true': True, 'false': False}.get(value)
        if flag is None:
            return "Error: value must be 'true' or 'false'"
        self.core.set_auto_format(flag)
        return dumps(flag)

    @dbus.service.method(IFACE, in_signature='s', out_signature='s')
    def SkipInternal(self, value):
        value = str(value)
        log.info('Dmedia.SkipInternal(%r)', value)
        if value == '':
            return dumps(self.core.get_skip_internal())
        flag = {'true': True, 'false': False}.get(value)
        if flag is None:
            return "Error: value must be 'true' or 'false'"
        self.core.set_skip_internal(flag)
        return dumps(flag)   


parser = argparse.ArgumentParser()
parser.add_argument('--version', action='version', version=dmedia.__version__)
parser.add_argument('--bus', default=BUS,
    help='DBus bus name; default is {!r}'.format(BUS),
)
parser.add_argument('--modules', action='store_true', default=False,
    help='print loaded modules and exit',
)
args = parser.parse_args()
if args.modules:
    import sys
    names = sorted(sys.modules)
    for name in names:
        print(name)
    print('[{:d}]'.format(len(names)))
    sys.exit()


log = dmedia.configure_logging2()
busname = dbus.service.BusName(args.bus, dbus.SessionBus())
service = Service(busname)
service.run()