Synnefo

#!/usr/bin/env python

# -*- coding: utf-8 -*-

#

# Copyright 2011 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

#

"""Utility to monitor the progress of image deployment

A small utility to monitor the progress of image deployment

by watching the contents of /proc/<pid>/io and producing

notifications of type 'ganeti-create-progress' to the rest

of the Synnefo infrastructure over AMQP.

"""

import os

import sys

import time

import json

import prctl

import signal

import socket

from amqplib import client_0_8 as amqp

from synnefo import settings

class AMQPClient(object):

    def __init__(self, routekey):

        self.conn = None

        self.chan = None

        self.routekey = routekey

    def open_channel(self):

        if not self.conn:

            try:

                sys.stderr.write("Attempting to connect to %s\n" %

                                 settings.RABBIT_HOST)

                self.conn = amqp.Connection(host=settings.RABBIT_HOST,

                                            userid=settings.RABBIT_USERNAME,

                                            password=settings.RABBIT_PASSWORD,

                                            virtual_host=settings.RABBIT_VHOST)

            except socket.error:

                sys.stderr.write("Connection failed, will retry in 1s\n")

                time.sleep(1)

        if self.conn:

            sys.stderr.write("Connection succesful, opening channel\n")

            self.chan = self.conn.channel()

    def send_message(self, msg):

        sys.stderr.write("Delivering msg with key=%s:\n%s\n" %

                         (self.routekey, json.dumps(msg)))

        msg = amqp.Message(json.dumps(msg))

        msg.properties["delivery_mode"] = 2  # Persistent

        if not self.chan:

            self.open_channel()

        if not self.chan:

            return

        try:

            self.chan.basic_publish(msg,

                                    exchange=settings.EXCHANGE_GANETI,

                                    routing_key=self.routekey)

        except socket.error:

            sys.stderr.write("Server went away, reconnecting...\n")

            self.conn = None

            self.chan = None

def parse_arguments(args):

    from optparse import OptionParser

    kw = {}

    kw['usage'] = "%prog [options] command [args...]"

    kw['description'] = \

        "%prog runs 'command' with the specified arguments, monitoring the " \

        "number of bytes read and written by it. 'command' is assumed to be " \

        "A program used to install the OS for a Ganeti instance. %prog " \

        "periodically issues notifications of type 'ganeti-create-progress' " \

        "to the rest of the Synnefo infrastructure over AMQP."

    parser = OptionParser(**kw)

    parser.disable_interspersed_args()

    parser.add_option("-r", "--read-bytes",

                      action="store", type="int", dest="read_bytes",

                      metavar="BYTES_TO_READ",

                      help="The expected number of bytes to be read, " \

                           "used to compute input progress",

                      default=0)

    parser.add_option("-w", "--write-bytes",

                      action="store", type="int", dest="write_bytes",

                      metavar="BYTES_TO_WRITE",

                      help="The expected number of bytes to be written, " \

                           "used to compute output progress",

                      default=0)

    parser.add_option("-i", "--instance-name",

                      dest="instance_name",

                      metavar="GANETI_INSTANCE",

                      help="The Ganeti instance name to be used in AMQP " \

                           "notifications")

    (opts, args) = parser.parse_args(args)

    if opts.instance_name is None or (opts.read_bytes == 0 and

                                      opts.write_bytes == 0):

        sys.stderr.write("Fatal: Options '-i' and at least one of '-r' " \

                         "or '-w' are mandatory.\n")

        parser.print_help()

        sys.exit(1)

    if len(args) == 0:

        sys.stderr.write("Fatal: You need to specify the command to run.\n")

        parser.print_help()

        sys.exit(1)

    return (opts, args)

def report_wait_status(pid, status):

    if os.WIFEXITED(status):

        sys.stderr.write("Child PID = %d exited, status = %d\n" %

                         (pid, os.WEXITSTATUS(status)))

    elif os.WIFSIGNALED(status):

        sys.stderr.write("Child PID = %d died by signal, signal = %d\n" %

                         (pid, os.WTERMSIG(status)))

    elif os.WIFSTOPPED(status):

        sys.stderr.write("Child PID = %d stopped by signal, signal = %d\n" %

                         (pid, os.WSTOPSIG(status)))

    else:

        sys.stderr.write("Internal error: Unhandled case, " \

                         "PID = %d, status = %d\n" % (pid, status))

        sys.exit(1)

    sys.stderr.flush()

def main():

    (opts, args) = parse_arguments(sys.argv[1:])

    # WARNING: This assumes that instance names

    # are of the form prefix-id, and uses prefix to

    # determine the routekey for AMPQ

    prefix = opts.instance_name.split('-')[0]

    routekey = "ganeti.%s.event.progress" % prefix

    amqp = AMQPClient(routekey)

    pid = os.fork()

    if pid == 0:

        # In child process:

        # Make sure we die with the parent and are not left behind

        # WARNING: This uses the prctl(2) call and is Linux-specific.

        prctl.set_pdeathsig(signal.SIGHUP)

        # exec command specified in arguments,

        # searching the $PATH, keeping all environment

        os.execvpe(args[0], args, os.environ)

        sys.stderr.write("execvpe failed, exiting with non-zero status")

        os.exit(1)

    # In parent process:

    iofname = "/proc/%d/io" % pid

    iof = open(iofname, "r", 0)   # 0: unbuffered open

    sys.stderr.write("%s: created child PID = %d, monitoring file %s\n" %

                     (sys.argv[0], pid, iofname))

    while True:

        # check if the child process is still alive

        (wpid, status) = os.waitpid(pid, os.WNOHANG)

        if wpid == pid:

            report_wait_status(pid, status)

            if (os.WIFEXITED(status) or os.WIFSIGNALED(status)):

                if not (os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0):

                    return 1

                else:

                    # send a final notification

                    final_msg = dict(type="ganeti-create-progress",

                                     instance=opts.instance_name)

                    if opts.read_bytes:

                        final_msg['rprogress'] = float(100)

                    if opts.write_bytes:

                        final_msg['wprogress'] = float(100)

                    amqp.send_message(final_msg)

                    return 0

        # retrieve the current values of the read/write byte counters

        iof.seek(0)

        for l in iof.readlines():

            if l.startswith("rchar:"):

                rchar = int(l.split(': ')[1])

            if l.startswith("wchar:"):

                wchar = int(l.split(': ')[1])

        # Construct notification of type 'ganeti-create-progress'

        msg = dict(type="ganeti-create-progress",

                   instance=opts.instance_name)

        if opts.read_bytes:

            msg['rprogress'] = float("%2.2f" %

                                     (rchar * 100.0 / opts.read_bytes))

        if opts.write_bytes:

            msg['wprogress'] = float("%2.2f" %

                                     (wchar * 100.0 / opts.write_bytes))

        # and send it over AMQP

        amqp.send_message(msg)

        # Sleep for a while

        time.sleep(3)

if __name__ == "__main__":

    sys.exit(main())