path: root/qpid/cpp/management/python/bin/qpid-ha

#!/usr/bin/env python

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

import optparse, sys, time, os, re, math
from qpid.messaging import Connection
from qpid.messaging import Message as QpidMessage
from qpid.util import URL
from qpidtoollibs.broker import BrokerAgent
from qpidtoollibs.config import parse_qpidd_conf
try:
    from uuid import uuid4
except ImportError:
    from qpid.datatypes import uuid4

# QMF address for the HA broker object.
HA_BROKER = "org.apache.qpid.ha:habroker:ha-broker"
# Define these defaults here rather than in add_option because we want
# to use qpidd.conf for defaults if --config is specified and
# these defaults otherwise:
DEFAULTS = { "broker":"0.0.0.0", "timeout":10.0}

class ExitStatus(Exception):
    """Raised if a command want's a non-0 exit status from the script"""
    def __init__(self, status): self.status = status

def find_qpidd_conf():
    """Return the path to the local qpid.conf file or None if it is not found"""
    p = os.path
    prefix, bin = p.split(p.dirname(__file__))
    if bin == "bin":            # Installed in a standard place.
        conf = p.join(prefix, "etc", "qpid", "qpidd.conf")
        if p.isfile(conf): return conf
    return None

class Command(object):
    """
    Common options and logic for all commands. Subclasses provide additional
    options and execution logic.
    """

    commands = []

    def __init__(self, name, help, arg_names=[], connect_agent=True):
        """@param connect_agent true if we should establish a QMF agent connection"""
        Command.commands.append(self)
        self.name = name
        self.connect_agent = connect_agent
        self.arg_names = arg_names
        usage="%s [options] %s\n\n%s"%(name, " ".join(arg_names), help)
        self.help = help
        self.op=optparse.OptionParser(usage)
        common = optparse.OptionGroup(self.op, "Broker connection options")
        def help_default(what): return " (Default %s)"%DEFAULTS[what]
        common.add_option("-b", "--broker", metavar="<address>", help="Address of qpidd broker with syntax: [username/password@] hostname | ip-address [:<port>]"+help_default("broker"))
        common.add_option("--timeout", type="float", metavar="<seconds>", help="Give up if the broker does not respond within the timeout. 0 means wait forever"+help_default("timeout"))
        common.add_option("--sasl-mechanism", metavar="<mech>", help="SASL mechanism for authentication (e.g. EXTERNAL, ANONYMOUS, PLAIN, CRAM-MD5, DIGEST-MD5, GSSAPI). SASL automatically picks the most secure available mechanism - use this option to override")
        common.add_option("--sasl-service-name", action="store", type="string", help="SASL service name to use")
        common.add_option("--ssl-certificate", metavar="<cert>", help="Client SSL certificate (PEM Format)")
        common.add_option("--ssl-key", metavar="<key>", help="Client SSL private key (PEM Format)")
        common.add_option("--config", metavar="<path/to/qpidd.conf>", help="Read default connection configuration from the qpidd.conf broker configuration file. Defaults are overridden by command-line options.)")
        self.op.add_option_group(common)

    def connect(self, opts):
        conn_options = {}
        if not opts.broker:
            opts.broker = DEFAULTS["broker"]
            # If we are connecting locally, use local qpidd.conf by default
            if not opts.config: opts.config = find_qpidd_conf()
        url = URL(opts.broker)
        if opts.config:         # Use broker config file for defaults
            config = parse_qpidd_conf(opts.config)
            if not url.user: url.user = config.get("ha-username")
            if not url.password: url.password = config.get("ha-password")
            if not url.port: url.port = config.get("port")
            opts.broker = str(url)
            if not opts.sasl_mechanism: opts.sasl_mechanism = config.get("ha-mechanism")
            if  not opts.timeout:
                timeout = config.get("ha-heartbeat-interval") or config.get("link-heartbeat-interval")
                if timeout: opts.timeout = float(timeout)
        else:                   # Use DEFAULTS
            if not opts.timeout: opts.timeout = DEFAULTS["timeout"]
        if opts.sasl_mechanism: conn_options['sasl_mechanisms'] = opts.sasl_mechanism
        if opts.sasl_service_name:
            conn_options['sasl_service'] = opts.sasl_service_name
        if opts.ssl_certificate: conn_options['ssl_certfile'] = opts.ssl_certificate
        if opts.ssl_key:
            if not opts.ssl_certificate:
                self.op.error("missing '--ssl-certificate' (required by '--ssl-key')")
            conn_options['ssl_keyfile'] = opts.ssl_key
        conn_options['client_properties'] = {'qpid.ha-admin' : 1}
        if opts.timeout:
            conn_options['timeout'] = opts.timeout
            conn_options['heartbeat'] = int(math.ceil(opts.timeout/2))
        connection = Connection.establish(opts.broker, **conn_options)
        qmf_broker = self.connect_agent and BrokerAgent(connection)
        ha_broker = self.connect_agent and qmf_broker.getHaBroker()
        return (connection, qmf_broker, ha_broker)

    def all_brokers(self, ha_broker, opts, func):
        """@return: List of (broker_addr, ha_broker) for all brokers in the cluster.
        Returns (broker_addr, Exception) if an exception is raised accessing a broker.
        """
        # The brokersUrl setting is not in python URL format, simpler parsing here.
        result = []
        brokers = filter(None, re.sub(r'(^amqps?:)|(tcp:)', "", ha_broker.brokersUrl).split(","))
        if brokers and opts.all:
            if "@" in opts.broker: userpass = opts.broker.split("@")[0]
            else: userpass = None
            for b in brokers:
                if userpass and not "@" in b: opts.broker = userpass+"@"+b
                else: opts.broker = b
                try:
                    connection, qmf_broker, ha_broker = self.connect(opts)
                    func(ha_broker, b)
                except Exception,e:
                    func(ha_broker, b, e)
        else:
            func(ha_broker)

    def execute(self, args):
        opts, args = self.op.parse_args(args)
        if len(args) != len(self.arg_names)+1:
            self.op.print_help()
            raise Exception("Wrong number of arguments")
        self.connection, qmf_broker, ha_broker = self.connect(opts)
        if self.connect_agent and not ha_broker:
            raise Exception("HA module is not loaded on broker at %s" % opts.broker)
        try: self.do_execute(qmf_broker, ha_broker, opts, args)
        finally: self.connection.close()

    def do_execute(self, qmf_broker, opts, args):
        raise Exception("Command '%s' is not yet implemented"%self.name)

class ManagerCommand(Command):
    """
    Base for commands that should only be used by a cluster manager tool that ensures
    cluster consistency.
    """

    manager_commands = []           # Cluster manager commands

    def __init__(self, name, help, arg_names=[], connect_agent=True):
        """@param connect_agent true if we should establish a QMF agent connection"""
        super(ManagerCommand, self).__init__(name, "[Cluster manager only] "+help, arg_names, connect_agent)
        self.commands.remove(self)   # Not a user command
        self.manager_commands.append(self)


class PingCmd(Command):
    def __init__(self):
        Command.__init__(self, "ping","Check if the broker is alive and responding", connect_agent=False)
    def do_execute(self, qmf_broker, ha_broker, opts, args):
        self.connection.session() # Make sure we can establish a session.
PingCmd()

class PromoteCmd(ManagerCommand):
    def __init__(self):
        super(PromoteCmd, self).__init__("promote", "Promote a backup broker to primary. This command should *only* be used by a cluster manager (such as rgmanager) that ensures only one broker is primary at a time. Promoting more than one broker to primary at the same time will make the cluster inconsistent and will cause data loss and unexpected behavior.")

    def do_execute(self, qmf_broker, ha_broker, opts, args):
        qmf_broker._method("promote", {}, HA_BROKER, timeout=opts.timeout)

PromoteCmd()


class StatusCmd(Command):
    def __init__(self):
        Command.__init__(self, "status", "Print HA status")
        self.op.add_option(
            "--expect", metavar="<status>",
            help="Don't print status. Return 0 if it matches <status>, 1 otherwise")
        self.op.add_option(
            "--is-primary", action="store_true", default=False,
            help="Don't print status. Return 0 if the broker is primary, 1 otherwise")
        self.op.add_option(
            "--all", action="store_true", default=False,
            help="Print status for all brokers in the cluster")

    def do_execute(self, qmf_broker, ha_broker, opts, args):
        if opts.is_primary:
            if not ha_broker.status in ["active", "recovering"]: raise ExitStatus(1)
            return
        if opts.expect:
            if opts.expect != ha_broker.status: raise ExitStatus(1)
            return

        def status(hb, b=None, ex=None):
            if ex: print b, ex
            elif b: print b, hb.status
            else: print hb.status
        self.all_brokers(ha_broker, opts, status)

StatusCmd()

class ReplicateCmd(Command):
    def __init__(self):
        Command.__init__(self, "replicate", "Set up replication from <queue> on <remote-broker> to <queue> on the current broker.", ["<queue>", "<remote-broker>"])
    def do_execute(self, qmf_broker, ha_broker, opts, args):
        qmf_broker._method("replicate", {"broker":args[1], "queue":args[2]}, HA_BROKER, timeout=opts.timeout)
ReplicateCmd()

class QueryCmd(Command):
    def __init__(self):
        Command.__init__(self, "query", "Print HA configuration and status")
        self.op.add_option(
            "--all", action="store_true", default=False,
            help="Print configuration and status for all brokers in the cluster")

    def do_execute(self, qmf_broker, ha_broker, opts, args):
        def query(hb, b=None, ex=None):
            if ex:
                print "%s %s\n" % (b, ex)
            else:
                if b:
                    print "%-20s %s"%("Address:", b)
                for x in [("Status:", hb.status),
                          ("Broker ID:", hb.systemId),
                          ("Brokers URL:", hb.brokersUrl),
                          ("Public URL:", hb.publicUrl),
                          ("Replicate: ", hb.replicateDefault)
                          ]:
                    print "%-20s %s"%x
                if b: print
        self.all_brokers(ha_broker, opts, query)


QueryCmd()

def print_usage(prog):
    print "usage: %s <command> [<arguments>]\n\nCommands are:\n"%prog
    for cmd in Command.commands:
        print "  %-12s %s."%(cmd.name, cmd.help.split(".")[0])
    print "\nFor help with a command type: %s <command> --help\n"%prog

def find_command(args, commands):
    """Find a command among the arguments and options"""
    for arg in args:
        cmds = [cmd for cmd in commands if cmd.name == arg]
        if cmds: return cmds[0]
    return None

def main_except(argv):
    """This version of main raises exceptions"""
    args = argv[1:]
    commands = Command.commands
    if "--cluster-manager" in args:
        commands += ManagerCommand.manager_commands
        args.remove("--cluster-manager")
    if len(args) and args[0] in ['help', '--help', '-help', '-h', 'help-all', '--help-all']:
        if 'help-all' in args[0]:
            for c in commands: c.op.print_help(); print
        else:
            print_usage(os.path.basename(argv[0]));
    else:
        command = find_command(args, commands)
        if command:
            command.execute(args)
        else:
            # Check for attempt to use a manager command without --cluster-manager
            command = find_command(args, ManagerCommand.manager_commands)
            if command:
                message="""'%s' should only be called by the cluster manager.
Incorrect use of '%s' will cause cluster malfunction.
To call from a cluster manager use '%s --cluster-manager'. """
                raise Exception(message%((command.name,)*3))
            else:
                print_usage(os.path.basename(argv[0]));
                raise Exception("No valid command")

def main(argv):
    try:
        main_except(argv)
        return 0
    except ExitStatus, e:
        return e.status
    except Exception, e:
        print "%s: %s"%(type(e).__name__, e)
        return 1

if __name__ == "__main__":
    sys.exit(main(sys.argv))