#!/usr/bin/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 os
import sys
import string
import subprocess
import re
import grp
import pwd
import resource
import time
import platform
import httplib
import socket

from threading import *
import traceback
import Queue

from  stat import *
from local_hooks import find_other_processes

# Catch the annoying problem when the current directory has been changed, e.g. by installing a new release
try:
    os.getcwd()
except:
    print "ERROR getting current directory ... may have been replaced .. try cd'ing to it again"
    sys.exit(1)

# simple bootstrap to establish DUCC_HOME and to set the python path so it can
# find the common code in DUCC_HOME/admin
# Infer DUCC_HOME from our location - no longer use a (possibly inaccurate) environment variable
me = os.path.abspath(__file__)    
ndx = me.rindex('/')
ndx = me.rindex('/', 0, ndx)
DUCC_HOME = me[:ndx]          # split from 0 to ndx
    
sys.path.append(DUCC_HOME + '/bin')
from ducc_base import DuccBase
from properties import Properties

from ducc_logger import DuccLogger
logger = DuccLogger()   

import db_util as dbu

global use_threading
use_threading = True

ducc_util_debug_flag = False

def debug(label,data):
    if(ducc_util_debug_flag):
        print label, data

# The "ducc" userid is the user that installed DUCC and created this file.
# If the admin dir's permissions were 700 then could assume the current user is the ducc user
def find_ducc_uid():
    my_file = os.path.abspath(__file__)    
    my_stat = os.stat(my_file)
    my_uid = my_stat.st_uid
    pwdinfo = pwd.getpwuid(my_uid)
    return pwdinfo.pw_name

def base_dir():
        fpath = __file__.rsplit('/',2)[0]
        return fpath

class ThreadWorker(Thread):
    def __init__(self, queue, outlock):
        Thread.__init__(self)
        self.queue = queue
        self.outlock = outlock
        
    def run(self):
        while True:
            (method, args) = self.queue.get()

            if ( args == 'quit' ):
                self.queue.task_done()
                return;

            try:
                response = method(args)
                if ( response != None and len(response) > 0):
                    self.outlock.acquire()
                    for l in response:
                        print ' '.join(l)
                    self.outlock.release()
            except:
                print "Exception executing", str(method), str(args)
                traceback.print_exc()

            self.queue.task_done()

class ThreadPool:
    def __init__(self, size):
        if ( use_threading ):
            self.size = size
            self.queue = Queue.Queue()
            outlock = Lock()

            MAX_NPSIZE = 100
            if ( self.size > MAX_NPSIZE ):
                self.size = MAX_NPSIZE

            for i in range(self.size):
                worker = ThreadWorker(self.queue, outlock)
                worker.start()

    def invoke(self, method, *args):
        if ( use_threading ):
            self.queue.put((method, args))
        else:
            response = method(args)
            if ( response != None and len(response) > 0):
                for l in response:
                    print ' '.join(l)
            
    def quit(self):
        if ( use_threading ):
            for i in range(self.size):
                self.queue.put((None, 'quit'))

            print "Waiting for Completion"
            self.queue.join()
            print "All threads returned"
        else:
            print 'All Work completed'

class DuccUtil(DuccBase):

    def makedirs(self,dir):
        try:
            os.makedirs(dir)
            print 'make: '+dir
        except:
            pass
        
    def update_properties(self):

        if ( self.ducc_properties == None ):
            DuccBase.read_properties(self)

        self.ssh_enabled       = self.ducc_properties.get('ducc.ssh')
        self.duccling          = self.ducc_properties.get('ducc.agent.launcher.ducc_spawn_path')

        self.ducc_uid          = find_ducc_uid()
        # self.broker_url      = self.ducc_properties.get('ducc.broker.url')
        self.broker_protocol   = self.ducc_properties.get('ducc.broker.protocol')
        self.broker_host       = self.localhost
        self.broker_port       = self.ducc_properties.get('ducc.broker.port')
        self.broker_jmx_port   = self.ducc_properties.get('ducc.broker.jmx.port')
        self.broker_decoration = self.ducc_properties.get('ducc.broker.url.decoration')
        self.broker_url        = self.broker_protocol + '://' + self.broker_host + ':' + self.broker_port
        self.agent_jvm_args    = self.ducc_properties.get('ducc.agent.jvm.args')
        self.ws_jvm_args       = self.ducc_properties.get('ducc.ws.jvm.args')
        self.pm_jvm_args       = self.ducc_properties.get('ducc.pm.jvm.args')
        self.rm_jvm_args       = self.ducc_properties.get('ducc.rm.jvm.args')
        self.sm_jvm_args       = self.ducc_properties.get('ducc.sm.jvm.args')
        self.or_jvm_args       = self.ducc_properties.get('ducc.orchestrator.jvm.args')


        if ( self.broker_decoration == '' ):
            self.broker_decoration = None

        if ( self.broker_decoration != None ):
            self.broker_url = self.broker_url + '?' + self.broker_decoration
        
        # Obsolete ??
        #if ( self.webserver_node == None ):
        #    self.webserver_node = self.localhost

    def merge_properties(self):
        # first task, always, merge the properties so subsequent code can depend on their validity.
        base_props = DUCC_HOME + '/resources/default.ducc.properties'
        site_props = DUCC_HOME + '/resources/site.ducc.properties'
        run_props = DUCC_HOME + '/resources/ducc.properties'
        merger = DUCC_HOME + '/admin/ducc_props_manager'
        CMD = [merger, '--merge', base_props, '--with', site_props, '--to', run_props]
        CMD = ' '.join(CMD)
        #print 'Merging', base_props, 'with', site_props, 'into', run_props
        print 'Merging site properties into ducc.properties'
        os.system(CMD)

    def check_properties(self):
        database_host_list = self.ducc_properties.get('ducc.database.host.list')
        database_host = self.ducc_properties.get('ducc.database.host')
        database_jmx_host = self.ducc_properties.get('ducc.database.jmx.host')
        if(database_host_list != None):
            if(database_host != None):
                text = 'ducc.database.host_list and ducc.database.host both specified.'
                print 'Error: '+text
                sys.exit(1)
            if(database_jmx_host != None):
                if(database_jmx_host == 'localhost'):
                    pass
                elif(database_jmx_host in database_host_list.split()):
                    pass
                else:
                    text = 'ducc.database.host_list and ducc.database.jmx.host both specified.'
                    print 'Error: '+text
                    sys.exit(1)
    
    def get_db_host_list(self):
        result = []
        slist = self.ducc_properties.get('ducc.database.host.list')
        if(slist == None):
            slist = self.ducc_properties.get('ducc.database.host')
        if(slist != None):
            result = slist.split()
        return result
    
    def get_db_host(self):
        result = None
        host_list = self.get_db_host_list()
        if(host_list != None):
            if(len(host_list) > 0):
                if(result == None):
                    this_host = self.get_hostname_long()
                    if(this_host in host_list):
                        result = this_host
                if(result == None):
                    this_host = self.get_hostname_short()
                    if(this_host in host_list):
                        result = this_host
                if(result == None):
                    result = host_list[0]
        return result
    
    def is_db_disabled(self):
        result = False
        dbhost = self.get_db_host()
        if(dbhost == self.db_disabled):
            result = True
        return result
    
    def db_configure(self):
        if(self.is_db_disabled()):
            self.db_bypass = True
            return;
        else:
            self.db_bypass = False

        fdir = 'resources.private'
        fname = 'ducc.private.properties'
        fpath = os.path.join(self.DUCC_HOME,fdir,fname)
        if(os.path.isfile(fpath)):
            dbprops = Properties()
            dbprops.load(fpath)
            self.db_password = dbprops.get('db_password')
            if ( self.db_password == None ):
                print 'bypassing database because no password is set.'
                self.db_bypass = True
    
            self.db_password_guest = dbprops.get('db_password_guest')
            if ( self.db_password_guest == None ):
                self.db_password_guest = 'guest'
        else:
            self.db_password == None
            self.db_password_guest == None
            print 'bypassing database because '+fname+' not found.'
            self.db_bypass = True
            
    def db_password(self):
        if(self.db_password == None):
            self.db_configure()
        return self.db_password
    
    def db_password_guest(self):
        if(self.db_password == None):
            self.db_configure()
        return self.db_password_guest
    
    # does the database process exist?  
    def db_process_alive(self):
        if (not self.automanage_database):
            return False
        if ( not os.path.exists(self.db_pidfile) ):
            return False

        f = open(self.db_pidfile)
        pid = f.read();
        f.close()
        answer = []
        if ( self.system == 'Darwin'):
            ps = 'ps -eo user,pid,comm,args ' + pid
        else:
            ps = 'ps -eo user:14,pid,comm,args ' + pid
        lines = self.popen(ps)
        
        for line in lines:
            line = line.strip()
            if (pid in line and 'cassandra' in line):
                return True
        return False

    def db_process_kill(self,code):
        if (not self.automanage_database):
            return False
        if ( not os.path.exists(self.db_pidfile) ):
            return False

        f = open(self.db_pidfile)
        pid = f.read();
        f.close()
        answer = []
        if ( self.system == 'Darwin'):
            ps = 'ps -eo user,pid,comm,args ' + pid
        else:
            ps = 'ps -eo user:14,pid,comm,args ' + pid
        lines = self.popen(ps)
        
        for line in lines:
            line = line.strip()
            if (pid in line and 'cassandra' in line):
                cmd = [ 'kill', '-'+str(code), pid ]
                cmd = ' '.join(cmd)
                rc = os.system(cmd)
                if ( rc == 0 ):
                    return True
                else:
                    return False
        return False

    # contact the database and see how useful it seems to be
    def db_alive(self, retry=10, verbose=True):
        if ( self.db_bypass == True ):
            return True
        else:
            return self.db_alive_check(retry,verbose)
        
    def db_alive_check(self, retry=10, verbose=True):
        if(self.is_db_disabled()):
            if(verbose):
                print 'No database location defined.'
            return False
        if (not self.automanage_database):
            return False
        
        if ( not os.path.exists(self.db_pidfile) ):
            if(verbose):
                print 'Database pid file does not exist.  Checking DB connectivity.'

        # get our log4j config into the path to shut up noisy logging
        os.environ['CLASSPATH'] = os.environ['CLASSPATH'] + ':' + self.DUCC_HOME + '/resources'
        
        dbnode = self.get_db_host()
        
        CMD = [self.java(), 'org.apache.uima.ducc.database.DbAlive', self.DUCC_HOME, dbnode, str(retry)]

        CMD = ' '.join(CMD)
        if(not verbose):
            CMD = CMD + " >/dev/null 2>&1" 
        rc = os.system(CMD)
        if ( rc == 0 ):
            return True
        else:
            return False

            
    def db_start(self):

        if(not self.automanage_database):
            print '   (Bypass database start - not automanaged)'
            return False

        # bypass all of this for the initial delivery
        if ( self.db_bypass == True) :
            print '   (Bypass database start)'
            return True
        
        dbnode = self.get_db_host()
        
        if(dbnode == None):
            print '   (Bypass database start - no database configured)'
            return False
        
        dblist = self.get_db_host_list()
        if(len(dblist) > 1):
            print '   (Bypass database start - database list not supported)'
            return True
        
        print 'Starting database'
        dbu.update_cassandra_config(self.DUCC_HOME, dbnode)

        max_attempts = 5
        attempt = 1
        while attempt <= max_attempts:
            lines = self.ssh(dbnode, True, "'", self.DUCC_HOME + '/admin/ducc.py', '-c', 'db', '--nodup', "'")
            # we'll capture anything that the python shell spews because it may be useful, and then drop the
            # pipe when we see a PID message

            while True:
                try:
                    line = lines.readline().strip()
                except:
                    break
                #print '[]', line
                
                if ( not line ):
                    break
                if ( line == '' ):
                    break
                if ( line == 'Database is already running.' ):
                    print 'Database process is running ... if connection fails stop the DB and try again'
                    break
                if ( line == 'OK' ):
                    print 'GOT OK from db start'
                    lines.close();

            print 'waiting for database to start'
            if ( self.db_alive() ):
                return True

            attempt = attempt + 1
            if attempt <= max_attempts:
                print 'Did not connect to database, retrying (', attempt, 'of', max_attempts, ')'

        return False

    def db_stop(self):
        try:
            if(not self.automanage_database):
                print '   (Bypass database start - not automanaged)'
                return False
            if ( self.db_bypass == True) :
                print '   (Bypass database stop)'
                return True
            dbnode = self.get_db_host()
            if(dbnode == None):
                print '   (Bypass database stop - no database configured)'
                return False
            dblist = self.get_db_host_list()
            if(len(dblist) > 1):
                print '   (Bypass database stop - database list not supported)'
                return True
            dbnode = dbnode.strip()
            pidfile = os.path.join(DUCC_HOME,'state','database',dbnode,'cassandra.pid')
            cmd = [ 'less', '-FX', pidfile ]
            cmd = ' '.join(cmd)
            #print cmd
            stdout = self.ssh(dbnode, True, cmd)
            result = stdout.read().strip()
            tokens = result.split()
            if(len(tokens) == 1):
                pid = tokens[0].strip()
                #print pid
                cmd = [ 'kill', '-15', pid ]
                cmd = ' '.join(cmd)
                #print cmd
                stdout = self.ssh(dbnode, True, cmd)
                result = stdout.read().strip()
                #print result
                print 'Database stopped.'
                return True
            else:
                #print result
                print 'Database not running.'
                return True
        except Exception,e:
            print e
            return False

    def find_netstat(self):
        # don't you wish people would get together on where stuff lives?
        if ( os.path.exists('/sbin/netstat') ):
            return '/sbin/netstat'
        if ( os.path.exists('/usr/sbin/netstat') ):
            return '/usr/sbin/netstat'
        if ( os.path.exists('/bin/netstat') ):
            return '/bin/netstat'
        if ( os.path.exists('/sbin/netstat') ):
            return '/usr/bin/netstat'
        print 'Cannot find netstat'
        return None

    def is_amq_active(self):
        netstat = self.find_netstat()
        if ( netstat == None ):
            print "Cannot determine if ActiveMq broker is alive."
            return false

        lines = self.ssh(self.broker_host, True, netstat, '-an')
        #
        # look for lines like this with the configured port in the 4th token, and
        # ending with LISTEN:
        #
        # tcp        0      0 :::61616                :::*                    LISTEN      
        for line in lines:
            toks = line.split()
            #print '[]', line
            if ( toks[-1] == 'LISTEN' ):
                port = toks[3]
                if (port.endswith(self.broker_port)):
                    return True
        return False        

    # Set ACTIVEMQ_DATA_DIR the same as in start_broker to put the logs & pid with the daemon logs
    def stop_broker(self):

        broker_host = self.localhost
        broker_home = self.ducc_properties.get('ducc.broker.home')
        broker_name = self.ducc_properties.get('ducc.broker.name')
        broker_jmx  = self.ducc_properties.get('ducc.broker.jmx.port')
        here = os.getcwd()
        CMD = broker_home + '/bin/activemq'
        CMD = CMD + ' stop --all'
        CMD = CMD + ' --jmxurl service:jmx:rmi:///jndi/rmi://' + broker_host + ':' + broker_jmx + '/jmxrmi' 
        CMD = CMD + ' ' + broker_name
        CMD = 'JAVA_HOME=' + self.java_home() + ' ' + CMD
        CMD = 'ACTIVEMQ_DATA_DIR=' + self.DUCC_HOME + '/logs/' + self.localhost + ' ' + CMD
        print 'CMD:', CMD
        lines = self.ssh(broker_host, True, CMD)
        for l in lines:
            pass       # throw away junk from ssh


    def nohup(self, cmd, showpid=True):
        # Skip use of ssh?
        if cmd[0] == "ssh" and 'false' == self.ssh_enabled:
            cmd = cmd[2:]
        cmd = ' '.join(cmd)
        # print '**** nohup', cmd, '****'
        devnw = open(os.devnull, 'w')
        devnr = open(os.devnull, 'r')
        ducc = subprocess.Popen(cmd, shell=True, stdin=devnr, stdout=devnw, stderr=devnw)
        devnr.close()
        devnw.close()
        if ( showpid ) :
            print 'PID', ducc.pid

    def get_hostname_long(self):
        hostname = '?'
        cmd = '/bin/hostname'
        resp = self.popen(cmd)
        lines = resp.readlines()
        if(len(lines)== 1):
            line = lines[0]
            hostname = line.strip();
        return hostname
    
    def get_hostname_short(self):
        hostname = '?'
        cmd = '/bin/hostname -s'
        resp = self.popen(cmd)
        lines = resp.readlines()
        if(len(lines)== 1):
            line = lines[0]
            hostname = line.strip();
        return hostname

    def get_hostname(self):
        hostname = '?'
        cmd = '/bin/hostname'
        resp = self.popen(cmd)
        lines = resp.readlines()
        if(len(lines)== 1):
            line = lines[0]
            line = line.strip();
            hostname = line.split('.')[0]
        return hostname
    
    def ssh_operational(self, node, verbosity=True):
        is_operational = False
        req = node.split('.')[0]
        cmd = '/bin/hostname'
        if(node == 'localhost'):
            req = self.get_hostname()
        ssh_cmd = 'ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes -o ConnectTimeout=10'+' '+node+" "+cmd
        resp = self.popen(ssh_cmd)
        lines = resp.readlines()
        # Response should be just the full nodename so check that the short name matches
        # Note that error msgs may also contain the node name
        for line in lines:
            if (line.startswith(node) and len(line.split())==1):
                return True
        if(verbosity):
            print 'ssh not operational - unexpected results from:', ssh_cmd
            for line in lines:
                print '>>>>>',line
            print '?? Has the entry in .ssh/id_rsa.pub been appended to .ssh/authorized_keys ??'
        return False

    # like popen, only it spawns via ssh
    # Skip use of ssh?
    # NOTE: Current callers always have do_wait True
    def ssh(self, host, do_wait, *CMD):
        cmd = ' '.join(CMD)
                # Some callers quote the string which is OK for ssh but not the direct call
        if cmd[0] == "'" and cmd[-1] == "'":
            cmd = cmd[1:len(cmd)-2]
        if ( do_wait ):
            if 'false' == self.ssh_enabled:
                return self.popen(cmd)
            return self.popen('ssh -q -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes -o ConnectTimeout=10', host, cmd)
        else:
            if 'false' == self.ssh_enabled:
                return self.spawn(cmd)
            return self.spawn('ssh -q -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes -o ConnectTimeout=10', host, cmd)


    def set_classpath(self):
        DH        = self.DUCC_HOME + '/'
        LIB       = DH + 'lib/'

        local_jars  = self.ducc_properties.get('ducc.local.jars')   #local mods
    
        CLASSPATH = ''
    
        if ( local_jars != None ):
            extra_jars = local_jars.split()
            for j in extra_jars:
                CLASSPATH = CLASSPATH + ':' + LIB + j

        CLASSPATH = CLASSPATH + ':' + DH  + 'apache-uima/lib/*'           
        CLASSPATH = CLASSPATH + ':' + DH  + 'apache-uima/apache-activemq/lib/*'           
        CLASSPATH = CLASSPATH + ':' + DH  + 'apache-uima/apache-activemq/lib/optional/*'           
        CLASSPATH = CLASSPATH + ':' + LIB + 'apache-commons/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'guava/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'google-gson/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'apache-log4j/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'apache-camel/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'joda-time/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'springframework/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'jna/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'libpam4j/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'uima-ducc/*'
        CLASSPATH = CLASSPATH + ':' + LIB + 'cassandra/*'

        # CLASSPATH = CLASSPATH + ':' + DH  + 'resources'  UIMA-4168 Use API, not classpath to configure log4j
    
        # more are added to some components in ducc.py, e.g.
        #    apache-activemq/lib/optional, jetty from ws lib, jsp, http- client
        #   
        os.environ['CLASSPATH'] = CLASSPATH

    def format_classpath(self, cp):
        strings = cp.split(':')
        for s in strings:
            print s

    def check_clock_skew(self, localdate):
        user = os.environ['LOGNAME']
        bypass = (user != self.ducc_uid)
        
        if bypass:
            tag = 'NOTE'
        else:
            tag = 'NOTOK'

        # Check clock skew
        ok = True
        acceptable_skew = 300
        skew = abs(long(localdate) - long(time.time()))
        if ( skew > (acceptable_skew) ):
            ok = False
            print tag, 'Clock skew[', skew, '] on', os.uname()[1], ". Remote time is", time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.localtime())
        return ok or bypass

    def set_duccling_version(self):
        CMD = self.duccling + ' -v >' + self.DUCC_HOME + '/state/duccling.version'
        os.system(CMD)
        print 'Set ducc_ling version from', self.localhost, ':', CMD

    def verify_limits(self):
        ret = True
        if ( self.system == 'Darwin' ):
            return ret                 # on mac, just use what you have

        proclimit = 20000
        filelimit = 8192

        (softnproc , hardnproc)  = resource.getrlimit(resource.RLIMIT_NPROC)
        (softnfiles, hardnfiles)  = resource.getrlimit(resource.RLIMIT_NOFILE)
        if ( softnproc < hardnproc ):
            try:
                resource.setrlimit(resource.RLIMIT_NPROC, (hardnproc, hardnproc))
            except:
                print 'NOTOK: could not set soft RLIMIT_NPROC up to the hard limit'
                ret = False

        if ( softnfiles < hardnfiles ):
            try:
                resource.setrlimit(resource.RLIMIT_NOFILE, (hardnfiles, hardnfiles))
            except:
                print 'NOTOK: could not set soft RLIMIT_NOFILES up to the hard limit'
                ret = False

        (softnproc , hardnproc)  = resource.getrlimit(resource.RLIMIT_NPROC)
        (softnfiles, hardnfiles)  = resource.getrlimit(resource.RLIMIT_NOFILE)
        
        if ( softnproc < proclimit ):
            print 'WARN: Soft limit RLIMIT_NPROC is too small at', softnproc, '(<', proclimit, ' ).  DUCC may be unable to create sufficient threads.'

        if ( softnfiles < filelimit ):
            print 'WARN: Soft limit RLIMIT_NOFILES is too small at', softnfiles, '(<', filelimit, ').  DUCC may be unable to open sufficient files or sockets.'

        return ret

    def verify_jvm(self):
        jvm = self.java()
        CMD = jvm + ' -version > /dev/null 2>&1'
        rc = os.system(CMD)
        if ( rc != 0 ):
            print 'NOTOK', CMD, 'returns', int(rc), '.  Must return rc 0.  Startup cannot continue.'
            return False
        return True
    
    # Check if the node's name or ip matches one of the head nodes
    def is_head_node(self):
        head_node_list = self.head_nodes
        node = self.localhost
        if (node in head_node_list):
            return True
        node_ip = socket.gethostbyname(node)
        for hnode in head_node_list:
            if socket.gethostbyname(hnode) == node_ip:
                return True
        return False
    
    # Exit if this is not a head node
    # Also check that ssh to this node works
    # Also restrict operations to the userid that installed ducc
    def verify_head(self):
        node = self.localhost
        if (not self.is_head_node()):
            print ">>> ERROR - " + node + " is not one of the configured head nodes " + str(self.head_nodes)
            sys.exit(1);
        if(self.ssh_operational(node)):
            text = "ssh is operational to "+node
            #print text
        else:
            print ">>> ERROR - this script cannot ssh to head node"
            sys.exit(1);
        # Ensure that root or another id doesn't start/stop ducc
        dir_stat = os.stat(DUCC_HOME + '/resources/site.ducc.properties')
        if dir_stat.st_uid != os.getuid():
            print ">>> ERROR - this script must be run by the userid that installed DUCC"
            sys.exit(1);

    #
    # Verify the viability of ducc_ling.
    # Returns a tuple (viable, elevated, safe)
    #   where 'viable' is a boolean indicating whether the ducc_ling is viable (exists and is correct version)
    #            If this is false, the other two values are meaningless
    #
    #         'elevated' indicates whether priveleges are at least partially elevated
    #         'safe' indicates whether ducc_ling is safe (elevated privileges and correct permissions)
    # The caller will evaluate these and take appriopriate action
    #
    def verify_duccling(self):
        viable = True
        elevated = False
        safe = False

        dl   = self.duccling
        path = os.path.dirname(os.path.abspath(dl))       

        
        if ( not (os.path.exists(dl) and os.access(dl, os.X_OK)) ):
            print dl, 'does not exist or is not executable.'
            viable = False

        path = os.path.dirname(os.path.abspath(dl))
        dl   = path + '/ducc_ling'
        
        dl_stat = os.stat(dl)          # dl_stat is stat for ducc_ling
        dl_mode = dl_stat.st_mode

        if ( (dl_mode & S_ISUID) != S_ISUID):
            if ( os.environ['LOGNAME'] == self.ducc_uid ):
                print 'ducc_ling module', dl, ': setuid bit is not set. Processes will run as user '+self.ducc_uid
            elevated = False
            file_safe = True
            dir_safe  = True
            own_safe  = True
        else:
            elevated = True
            file_safe = False
            dir_safe  = False
            own_safe  = False


        if ( elevated ):

            #
            # if setuid bit is set, all this MUST be true or we won't mark ducc_ling safe:
            #    file permissions are 750
            #    dir  permissions are 700
            #    owenership is root.ducc
            #
            dl_perm = oct(dl_mode & (S_IRWXU | S_IRWXG | S_IRWXO))
            expected = oct(0750)
            if ( dl_perm != expected ):
                 print dl, ': Invalid execution bits', dl_perm, 'should be', expected
            else:
                file_safe = True

            dir_stat = os.stat(path)          # dir_stat is stat for ducc_ling
            dir_mode = dir_stat.st_mode

            dir_perm = oct(dir_mode & (S_IRWXU | S_IRWXG | S_IRWXO))
            expected = oct(0700)
            if ( dir_perm != expected ):
                 print 'ducc_ling', path, ': Invalid directory permissions', dir_perm, 'should be', expected
            else:
                 dir_safe = True

            try:
                grpinfo = grp.getgrnam(self.ducc_uid)
                duccgid = grpinfo.gr_gid

                if ( (dl_stat.st_uid != 0) or (dl_stat.st_gid != duccgid) ):
                    print 'ducc_ling module', dl, ': Invalid ownership. Should be '+self.ducc_uid
                else:
                    own_safe = True
            except:
                print 'ducc_ling group "'+self.ducc_uid+'" cannot be found.'

        safe = file_safe and dir_safe and own_safe

        # A last viability check, do versions match? This also runs it, proving it
        # can execute in this environment.
        lines = self.popen(self.duccling + ' -v')
        version_from_head = lines.readline().strip();
        toks = version_from_head.split()
        version_from_head = ' '.join(toks[0:4])

        version_file = self.DUCC_HOME + '/state/duccling.version';
        if ( os.path.exists(version_file) ):
            verfile = open(version_file)
            for line in verfile:
                line = line.strip();
                toks = line.split();
                line = ' '.join(toks[0:4])
                if ( line != version_from_head ):
                    print "Mismatched ducc_ling versions:"
                    print "ALERT: Version on Agent Node:", version_from_head
                    print "ALERT: Version on Ducc  Head:", line
                    viable = False
            verfile.close()
        else:
            print "NOTE: ducc_ling version file missing, cannot verify version."

        # leave the decisions to the caller
        return (viable, elevated, safe)        

    # Apply these rules to determine if ducc_ling is installed ok
    #
    #    Caller            Elevated         Protected (safe)     Action
    #    --------          --------         ---------            ------
    #    ducc                 Y                 Y                OK
    #                         Y                 N                Fail
    #
    #                         N                 Y (by def)       OK
    #
    #    ~ducc                Y                 N (by def)       Fail
    #                         N                 Y (by def)       OK, Note    
    def duccling_ok(self, viable, elevated, safe):

        if ( not viable ):
            return False

        user = os.environ['LOGNAME']
        
        if ( user == self.ducc_uid ):
            if ( elevated ):
                return safe
        else:
            if ( elevated ):
                return False
            print 'Note: Running unprivileged ducc_ling. Process will run as user', user

        return True

    def ssh_ok(self, node, line):
        spacer = '   '
        messages = []
        if ( line.startswith("Permission denied") ):
            messages.append(' ')
            messages.append(spacer + "ALERT: Passwordless SSH is not configured correctly for node " + node)
            messages.append(spacer + "ALERT: SSH returns '" + line + "'")
            return messages

        if ( line.startswith("Host key verification failed") ):
            messages.append(' ')
            messages.append(spacer + "ALERT: Passwordless SSH is not configured correctly for node " + node)
            messages.append(spacer + "ALERT: SSH returns '" + line + "'")
            return messages

        if ( line.find("Connection refused") >= 0 ):
            messages.append(' ')
            messages.append(spacer + "ALERT: SSH is not not enabled on node " + node)
            messages.append(spacer + "ALERT: SSH returns '" + line + "'")
            return messages
        
        if ( line.find("Connection timed") >= 0 ):
            messages.append(' ')
            messages.append(spacer + "\nALERT: SSH did not respond with timeout of 10 secnds " + node)
            messages.append(spacer + "ALERT: SSH returns '" + line + "'")
            return messages
        
        if ( line.find("No route")  >= 0 ):
            messages.append(' ')
            messages.append(spacer + 'ALERT: SSH cannot connect to host.')
            messages.append(spacer + "ALERT: SSH returns '" + line + "'")
            return messages

        return None

        
    #
    # Input is array lines from ps command looking for ducc processes owned this user.
    # Output is list of dictionaries, where each dictionary describes a ducc process.
    #
    # If no ducc processes are found here the list is empty.
    #
    # The caller executes the 'ps' command and knows the node this is for.
    #
    def find_ducc_process(self, node):
    
        answer = []
        if ( self.system == 'Darwin'):
            ps = 'ps -eo user,pid,comm,args'
        else:
            ps = 'ps -eo user:14,pid,comm,args'
        resp = self.ssh(node, True, ps)
        ok = True

        while True:
            line = resp.readline().strip()           
            if ( line.startswith('PID')):
                continue

            ssh_errors = self.ssh_ok(line, node)
            if ( ssh_errors != None ):
                for m in ssh_errors:
                    print m
                ok = False
                continue

            # from here on, assume no error
            if ( not line ):
                break
            
            toks = line.split()
            if ( len(toks) < 4):
                continue

            user = toks[0]
            pid = toks[1]
            procname = toks[2]
            fullargs = toks[3:]

            if ( not ('java' in procname) and not ('JIT' in procname)):
                continue

            cont = False
            for tok in fullargs:
                if ( tok.startswith('-Dducc.deploy.components=') ):
                    cmp = tok.split('=')
                    dp = (cmp[1],  pid, user)
                    answer.append(dp)
                    cont = True
                    break
                if ( tok.startswith('-DDUCC_BROKER_CREDENTIALS_FILE=') ):
                    dp = ('broker',  pid, user)
                    answer.append(dp)
                    cont = True
                    break
            if ( cont ):             # stupid python only continues out of inner loop
                continue
            if fullargs[-1] == 'org.apache.cassandra.service.CassandraDaemon':
                dp = ('database',  pid, user)
                answer.append(dp)
                continue
            
            # Look for site-specific processes
            other_processes = find_other_processes(pid, user, line)
            if ( type(other_processes) is list ):
                if ( len(other_processes) > 0 ):
                    answer = answer + other_processes
            else:
                print 'Invalid response from \'find_other_processes\':', other_processes

        return (ok, answer)

    #
    # Given the name of a file containing ducc nodes, a ducc user (usually the 'ducc' user unless you're running
    #   as yourself for test), find all ducc processes owned by this user and print them to the console.
    #
    def find_ducc(self, nodefile, user):
        if ( nodefile == None ):
            nodefile = self.DUCC_HOME + '/resources/ducc.nodes'
    
        if ( not os.path.exists(nodefile) ):
            print 'Nodefile', nodefile, 'does not exist or cannot be read.'
            sys.exit(1)
    
        answer = {}
        nodes = []
        f = open(nodefile)
        for node in f:
            node = node.strip()
            if ( not node ):
                continue
            if ( node.startswith('#') ):
                continue
            nodes.append(node)

        # obsolete ??
        #if ( self.webserver_node != 'localhost' ):           # might be configured somewhere else
        #    nodes.append(self.webserver_node)

        for node in nodes:                
            data = self.find_ducc_process(node, user)
            answer[node] = data

        return answer

    def kill_process(self, node, proc, signal):
        lines = self.ssh(node, True, 'kill', signal, proc[1])
        for l in lines:
            pass # throw away the noise
                
    def clean_shutdown(self):
        DUCC_JVM_OPTS = ' -Dducc.deploy.configuration=' + self.DUCC_HOME + "/resources/ducc.properties "
        DUCC_JVM_OPTS = DUCC_JVM_OPTS + ' -DDUCC_HOME=' + self.DUCC_HOME
        DUCC_JVM_OPTS = DUCC_JVM_OPTS + ' -Dducc.head=' + self.ducc_head
        self.spawn(self.java(), DUCC_JVM_OPTS, 'org.apache.uima.ducc.common.main.DuccAdmin', '--killAll')

    def ducc_admin(self,option,target):
        DUCC_JVM_OPTS = ' -Dducc.deploy.configuration=' + self.DUCC_HOME + "/resources/ducc.properties "
        DUCC_JVM_OPTS = DUCC_JVM_OPTS + ' -DDUCC_HOME=' + self.DUCC_HOME
        DUCC_JVM_OPTS = DUCC_JVM_OPTS + ' -Dducc.head=' + self.ducc_head
        java_class = 'org.apache.uima.ducc.common.main.DuccAdmin'
        cmd = self.java()+' '+DUCC_JVM_OPTS+' '+java_class+' '+option+' '+target
        debug('ducc_admin', cmd)
        print 'ducc_admin: ', option, target
        self.spawn(self.java(), DUCC_JVM_OPTS, java_class, option, target)
        
    def get_os_pagesize(self):
        lines = self.popen('/usr/bin/getconf', 'PAGESIZE')
        return lines.readline().strip()

    def show_ducc_environment(self):
        global use_threading
        #
        # Print the java version
        #
        response = []
        jvm = self.ducc_properties.get('ducc.jvm')
        jvm = jvm.replace('//', '/')
        check_java = True
        if ( jvm == None ):
            response.append('WARNING: No jvm configured.  Default is used.')
            jvm = 'java'
        else:
            response.append('ENV: Java is configured as: ' + jvm)
            if ( not os.path.exists(jvm) ):
                print 'NOTOK: configured jvm cannot be found:', jvm
                check_java = False

        if ( check_java ):
            print 'JVM:', jvm
            lines = self.popen(jvm + ' -fullversion')
            for line in lines:
                response.append('ENV ' + line.strip())
                

        response.append('ENV: Threading enabled: ' + str(use_threading))
        #
        # Get the total memory for the node
        #
        if ( self.system != 'Darwin' ):
            meminfo = Properties()
            meminfo.load('/proc/meminfo')
            mem = meminfo.get('MemTotal')
            if ( mem.endswith('kB') ):
                toks = mem.split(' ')
                mem = str(int(toks[0]) / (1024*1024)) + ' gB'
                response.append('MEM: memory is ' + mem)

        #
        # Get the operating system information
        #
        response.append('ENV: system is ' + self.system)

        return response

    #
    # Resolve the 'path' relative to the path 'relative_to'
    #
    def resolve(self, path, relative_to):
        if ( not path.startswith('/') ):                        
            (head, tail) = os.path.split(os.path.abspath(relative_to))
            path = head + '/' + path
        return path

    #
    # Read the nodefile, recursing into 'imports' if needed, returning a
    # map.  The map is keyed on filename, with each entry a list of the nodes.
    # Skip file with suffix ".regex".
    #
    def read_nodefile(self, rnodefile, ret):
        #print 'READ_NODEFILE:', nodefile, ret
        n_nodes = 0
        if(rnodefile.startswith('/')):
            nodefile = rnodefile
        else:
            nodefile = os.path.join(base_dir(),'resources',rnodefile)
        #print nodefile
        if(nodefile.endswith('.regex')):
            pass
        elif ( os.path.exists(nodefile) ):
            nodes = []
            f = open(nodefile)
            for node in f:
                node = node.strip()
                if ( not node ):
                    continue
                if ( node.startswith('#') ):
                    continue
                if ( node.startswith('import ') ):
                    toks = node.split(' ')
                    newfile = toks[1]
                    newfile = self.resolve(newfile, nodefile)  # resolve newfile relative to nodefile
                    (count, ret) = self.read_nodefile(newfile, ret)
                    n_nodes = n_nodes + count
                    continue
                nodes.append(node)
                n_nodes = n_nodes + 1
            ret[nodefile] = nodes
        else:
            n_nodes = -1
            print 'Cannot read nodefile', nodefile
            ret[nodefile] = None

        #print 'RETURN', n_nodes, nodefile, ret
        return (n_nodes, ret)

    def compare_nodes(self, n1, n2):

        if ( n1 == n2 ):             # exact match - covers both short and both long
            return True

        if ( n1.find('.') >= 0 ):    # shortened n1 == n2?
            t1 = n1.split('.')
            n1A = t1[0]
            if ( n1A == n2 ):
                return True

        if ( n2.find('.') >= 0 ):    # n1 == shortened n2?
            t2 = n2.split('.')
            n2A = t2[0]
            if ( n1 == n2A ):
                return True
        return False

    def verify_class_configuration(self, allnodes, verbose):

        print 'allnodes', allnodes
        answer = True
        # first, find the class definition
        classfile = self.ducc_properties.get('ducc.rm.class.definitions')

        print 'Class definition file is', classfile
        CMD = self.jvm
        CMD = CMD + " -DDUCC_HOME=" + self.DUCC_HOME
        CMD = CMD + " org.apache.uima.ducc.common.NodeConfiguration "
        CMD = CMD + " -n " + allnodes
        CMD = CMD + " -c " + classfile
        if ( verbose ):
            CMD = CMD + " -p " + classfile
            print CMD
        else:
            CMD = CMD + " " + classfile
        rc = os.system(CMD)
        if ( rc == 0 ):
            print "OK: Class and node definitions validated."
        else:
            print "NOTOK: Cannot validate class and/or node definitions."

        return (rc == 0)

    def get_nodepool(self, node, default=''):
        classpath = '"'+self.DUCC_HOME+'/lib/uima-ducc/*'+'"'
        #print classpath
        classfile = self.ducc_properties.get('ducc.rm.class.definitions')
        #print 'classfile: '+classfile
        cmd = ''
        cmd = cmd+self.jvm
        cmd = cmd + ' '
        cmd = cmd+'-cp '+classpath
        cmd = cmd + ' '
        cmd = cmd+'-DDUCC_HOME='+self.DUCC_HOME
        cmd = cmd + ' '
        cmd = cmd+'org.apache.uima.ducc.common.NodeConfiguration'
        cmd = cmd + ' '
        cmd = cmd+'-c'+' '+classfile
        cmd = cmd + ' '
        cmd = cmd+'-m'+' '+node
        cmd = ''.join(cmd)
        #print 'cmd: '+cmd
        p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
        (out, err) = p.communicate()
        status = p.wait()
        result = out.strip()
        if(result == ''):
            result = default
        #print 'result: '+result
        #print 'status: '+str(status)
        return result

    def get_nodepool_file(self, node, default=''):
        classpath = '"'+self.DUCC_HOME+'/lib/uima-ducc/*'+'"'
        #print classpath
        classfile = self.ducc_properties.get('ducc.rm.class.definitions')
        #print 'classfile: '+classfile
        cmd = ''
        cmd = cmd+self.jvm
        cmd = cmd + ' '
        cmd = cmd+'-cp '+classpath
        cmd = cmd + ' '
        cmd = cmd+'-DDUCC_HOME='+self.DUCC_HOME
        cmd = cmd + ' '
        cmd = cmd+'org.apache.uima.ducc.common.NodeConfiguration'
        cmd = cmd + ' '
        cmd = cmd+'-c'+' '+classfile
        cmd = cmd + ' '
        cmd = cmd+'-m'+' '+node
        cmd = cmd + ' '
        cmd = cmd+'-f'+' '+node
        cmd = ''.join(cmd)
        #print 'cmd: '+cmd
        p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
        (out, err) = p.communicate()
        status = p.wait()
        result = out.strip()
        if(result == ''):
            result = default
        #print 'result: '+result
        #print 'status: '+str(status)
        return result
        
    def disable_threading(self):
        global use_threading
        use_threading = False

    def installed(self):
        head = self.ducc_head
        if ( head == '<head-node>' ):
            return False
        return True

    # Check if not reliable or if this is the currently active head node
    # If reliable test if the ducc.head's IP address is one of our addresses
    def is_active(self):
        label = 'is_active'
        if (not self.reliable):
            return True
        text = 'cmd: ', '/sbin/ip', 'addr', 'list'
        debug(label, text)
        p = subprocess.Popen(['/sbin/ip', 'addr', 'list'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        output, err = p.communicate()
        debug(label, "output: "+output)
        head_ip = socket.gethostbyname(self.ducc_head)
        return (head_ip in output)
    
    def db_normalize_component(self,component):
        com = component
        if(component != None):
            if('agent'.startswith(component.lower())):
                com = 'Ag'
            elif('orchestrator'.startswith(component.lower())):
                com = 'Or'
            elif('pm'.startswith(component.lower())):
                com = 'Pm'
            elif('rm'.startswith(component.lower())):
                com = 'Rm'
            elif('sm'.startswith(component.lower())):
                com = 'Sm'
            elif('ws'.startswith(component.lower())):
                com = 'Ws'
            elif('broker'.startswith(component.lower())):
                com = 'Br'
            elif('db'.startswith(component.lower())):
                com = 'Db'
            elif('database'.startswith(component.lower())):
                com = 'Db'
        #print 'db_normalize_component', component, '-->', com
        return com
    
    def db_acct_start(self,node,com):
        label = 'db_acct_start'
        component = self.db_normalize_component(com)
        CMD = [self.java(), '-DDUCC_HOME='+self.DUCC_HOME, 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI', '--start', node, component]
        text = ' '.join(CMD)
        debug(label,text)
        p = subprocess.Popen(CMD, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = p.communicate()
        text = out
        debug(label,text)
        
    def db_acct_stop(self,node=None,com=None):
        label = 'db_acct_stop'
        component = self.db_normalize_component(com)
        if((node == None) and (component == None)):
            CMD = [self.java(), '-DDUCC_HOME='+self.DUCC_HOME, 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI', '--stop']
        elif(component == None):
            CMD = [self.java(), '-DDUCC_HOME='+self.DUCC_HOME, 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI', '--stop', node]
        else:
            CMD = [self.java(), '-DDUCC_HOME='+self.DUCC_HOME, 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI', '--stop', node, component]
        text = ' '.join(CMD)
        debug(label,text)
        p = subprocess.Popen(CMD, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = p.communicate()
        text = out
        debug(label,text)
    
    # Line format is: <host>.<component>=<state>
    def _db_get_tuple(self,line):
        retVal = None
        try:
            if(len(line.split()) == 1):
                lhs = line.split('=')[0]
                rhs = line.split('=')[1]
                host = lhs.split('.')[0]
                daemon = lhs.split('.')[1]
                state = rhs
                tuple = [ host, daemon, state ]
                retVal = tuple
        except:
            pass
        return retVal
    
    def db_acct_query(self):
        list = []
        label = 'db_acct_query'
        CMD = [self.java(), '-DDUCC_HOME='+self.DUCC_HOME, 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI', '--query']
        text = ' '.join(CMD)
        debug(label,text)
        p = subprocess.Popen(CMD, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = p.communicate()
        text = out
        debug(label,text)
        lines = out.split('\n')
        for line in lines:
            tuple = self._db_get_tuple(line)
            if(tuple != None):
                list.append(tuple)
        return list
        
    def __init__(self, merge=False):
        global use_threading
        DuccBase.__init__(self, merge)

        self.db_disabled = '--disabled--'
        self.db_password = None
        self.duccling = None
        self.broker_url = 'tcp://localhost:61616'
        self.broker_protocol = 'tcp'
        self.broker_host = 'localhost'
        self.broker_port = '61616'
        self.default_components = ['rm', 'pm', 'sm', 'or', 'ws', 'db', 'broker']
        self.local_components = ['rm', 'pm', 'sm', 'or', 'ws', 'broker']
        self.default_nodefiles = [self.DUCC_HOME + '/resources/ducc.nodes']

        # Get the only head node, or if  a "reliable" installation the virtual head node & all head nodes
        self.ducc_head = self.ducc_properties.get('ducc.head')
        ducc_head_list = self.ducc_properties.get('ducc.head.reliable.list')
        if (ducc_head_list != None and len(string.strip(ducc_head_list)) != 0):
            self.reliable = True
            self.head_nodes = ducc_head_list.split()
            if (len(self.head_nodes) <= 1):
                print '>>> ERROR - "ducc.head.reliable.list" must be empty or have more than one entry'
                sys.exit(1);
        else:
            self.reliable = False
            self.head_nodes = [ self.ducc_head ]

        os.environ['DUCC_NODENAME'] = self.localhost    # to match java code's implicit property so script and java match

        manage_database = self.ducc_properties.get('ducc.database.automanage')
        self.automanage_database = False
        if (manage_database in ('t', 'true', 'T', 'True')) :
            self.automanage_database = True     
            dbhost = self.get_db_host()
            if ( dbhost == None ):
                dbhost = self.ducc_head
            dir_db_state = self.DUCC_HOME + '/state/database/'+dbhost
            self.makedirs(dir_db_state)
            self.db_pidfile = dir_db_state+ '/cassandra.pid'
            dir_db_logs = self.DUCC_HOME + '/logs'
            self.makedirs(dir_db_logs)
            self.db_logfile = dir_db_logs + '/' + dbhost + '.cassandra.console'
        
        #self.pid_file_agents  = self.DUCC_HOME + '/state/agents/ducc.pids'
        #self.pid_file_daemons  = self.DUCC_HOME + '/state/daemons/'+self.localhosty+'/ducc.pids'
        self.set_classpath()
        self.os_pagesize = self.get_os_pagesize()
        self.update_properties()

        self.db_configure()
        
        manage_broker = self.ducc_properties.get('ducc.broker.automanage')
        self.automanage_broker = False
        if (manage_broker in ('t', 'true', 'T', 'True')) :
            self.automanage_broker = True                    

        py_version = platform.python_version().split('.')
        if ( int(py_version[0]) > 2 ):
            print "Warning, only Python Version 2 is supported."
        if ( int(py_version[1]) < 4 ):
            print "Python must be at least at version 2.4."
            sys.exit(1)
        if ( int(py_version[1]) < 6 ):
            use_threading = False


if __name__ == "__main__":
    util = DuccUtil()