| #!/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 |
| |
| 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 |
| |
| # 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 |
| |
| 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 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.ducc_properties.get('ducc.broker.hostname') |
| 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 |
| |
| 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 |
| os.system(CMD) |
| |
| |
| def db_configure(self): |
| dbhost = self.ducc_properties.get('ducc.database.host') |
| if ( dbhost == self.db_disabled ): |
| self.db_bypass = True |
| return; |
| else: |
| self.db_bypass = False |
| |
| dbprops = Properties() |
| dbprops.load(self.DUCC_HOME + '/resources.private/ducc.private.properties') |
| 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' |
| |
| # does the database process exist? |
| def db_process_alive(self): |
| pidfile = self.DUCC_HOME + '/state/cassandra.pid' |
| |
| if ( not os.path.exists(pidfile) ): |
| return False |
| |
| f = open(self.DUCC_HOME + '/state/cassandra.pid') |
| 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 |
| |
| # 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): |
| dbnode = self.ducc_properties.get('ducc.database.host') |
| if ( dbnode == None ): |
| if(verbose): |
| print 'No database location defined.' |
| return False |
| |
| pidfile = self.DUCC_HOME + '/state/cassandra.pid' |
| if ( not os.path.exists(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' |
| |
| CMD = [self.java(), 'org.apache.uima.ducc.database.DbAlive', dbnode, 'ducc', self.db_password, 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): |
| |
| # bypass all of this for the initial delivery |
| if ( self.db_bypass == True) : |
| print ' (Bypass database start because ducc.database.host =', self.db_disabled + ')' |
| return True |
| |
| print 'Starting database' |
| dbnode = self.ducc_properties.get('ducc.database.host') |
| 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): |
| |
| if ( self.db_bypass == True) : |
| print ' (Bypass database stop because ducc.database.host =', self.db_disabled + ')' |
| return True |
| |
| pidfile = self.DUCC_HOME + '/state/cassandra.pid' |
| if ( os.path.exists(pidfile) ): |
| # for cassandra, just send it a terminate signal. a pidfile is written on startup |
| CMD = ['kill', '-TERM', '`cat ' + pidfile + '`'] |
| CMD = ' '.join(CMD) |
| os.system(CMD) |
| |
| 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 |
| |
| def stop_broker(self): |
| |
| broker_host = self.ducc_properties.get('ducc.broker.hostname') |
| 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 |
| print '--------------------', 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 ssh_operational(self, node): |
| is_operational = False |
| req = node.split('.')[0] |
| cmd = '/bin/hostname' |
| ssh_cmd = 'ssh -q -o BatchMode=yes -o ConnectTimeout=10'+' '+node+" "+cmd |
| resp = self.popen(ssh_cmd) |
| lines = resp.readlines() |
| if(len(lines)== 1): |
| line = lines[0] |
| line = line.strip(); |
| rsp = line.split('.')[0] |
| if(req == rsp): |
| is_operational = True; |
| if(not is_operational): |
| print 'ssh not operational - unexpected results' |
| print ssh_cmd |
| for line in lines: |
| print line |
| return is_operational |
| |
| # 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 BatchMode=yes -o ConnectTimeout=10', host, cmd) |
| else: |
| if 'false' == self.ssh_enabled: |
| return self.spawn(cmd) |
| return self.spawn('ssh -q -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 |
| |
| # inspect ducc.head.failover |
| def verify_head_failover(self, head): |
| key = "ducc.head.failover" |
| failover = self.ducc_properties.get(key) |
| # check for no failover |
| if(failover == None): |
| logger.debug(key+" not specified") |
| else: |
| # insure ducc.head listed in ducc.head.failover |
| if(not head in failover): |
| text = head+" not found in "+key |
| logger.error(text) |
| sys.exit(1); |
| # test viability fo failover nodes |
| nodes = failover.replace(',',' ').split() |
| for node in nodes: |
| if(self.ssh_operational(node)): |
| text = "ssh is operational to "+node |
| logger.debug(text) |
| else: |
| text = "ssh to specified failover node unsuccessful or otherwise problematic: "+node |
| logger.warn(text) |
| |
| # Exit if this is not the head node. Ignore the domain as uname sometimes drops it. |
| # Also check that ssh to this node works |
| # Also restrict operations to the userid that installed ducc |
| def verify_head(self): |
| head = self.ducc_properties.get("ducc.head").split('.')[0] |
| local = self.localhost.split('.')[0] |
| if local != head: |
| print ">>> ERROR - this script must be run from the head node" |
| sys.exit(1); |
| node = head |
| 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); |
| self.verify_head_failover(head) |
| |
| |
| # |
| # 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) |
| |
| 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_properties.get('ducc.head') |
| self.spawn(self.java(), DUCC_JVM_OPTS, 'org.apache.uima.ducc.common.main.DuccAdmin', '--killAll') |
| |
| 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. |
| # |
| def read_nodefile(self, nodefile, ret): |
| #print 'READ_NODEFILE:', nodefile, ret |
| n_nodes = 0 |
| if ( 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: |
| 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 verify_head_failover_configuration(self): |
| rc = 0 |
| failover_nodes = self.ducc_properties.get('ducc.head.failover') |
| message = "OK: Head node failover not configured." |
| if(failover_nodes != None): |
| failover_nodes = failover_nodes.strip() |
| if(len(failover_nodes) >= 0): |
| nodes = failover_nodes.split() |
| head_node = self.ducc_properties.get('ducc.head') |
| head_pool = self.get_nodepool(head_node,'<None>') |
| for node in nodes: |
| node_pool = self.get_nodepool(node,'<None>') |
| #print 'head:'+head_pool+' '+'node:'+node_pool |
| if( head_pool != node_pool): |
| if(rc == 0): |
| message = 'OK: Head failover node '+head_node+' in node pool '+head_pool |
| print message |
| message = 'NOTOK: Head failover node '+node+' in node pool '+node_pool |
| print message |
| rc = 1 |
| if (rc > 0): |
| message = "NOTOK: Head failover nodepools incorrectly configured." |
| else: |
| message = "OK: Head failover nodepools correctly configured." |
| print message |
| return (rc == 0) |
| |
| def disable_threading(self): |
| global use_threading |
| use_threading = False |
| |
| def installed(self): |
| head = self.ducc_properties.get('ducc.head') |
| if ( head == '<head-node>' ): |
| return False |
| return True |
| |
| def __init__(self, merge=False): |
| global use_threading |
| DuccBase.__init__(self, merge) |
| |
| self.db_disabled = '--disabled--' |
| 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.default_nodefiles = [self.DUCC_HOME + '/resources/ducc.nodes'] |
| |
| if ( self.localhost == self.ducc_properties.get("ducc.head")): |
| self.is_ducc_head = True |
| |
| os.environ['DUCC_NODENAME'] = self.localhost # to match java code's implicit propery so script and java match |
| |
| self.pid_file = self.DUCC_HOME + '/state/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 = False |
| if (manage_broker in ('t', 'true', 'T', 'True')) : |
| self.automanage = 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() |
| |
