blob: 0126f829164ab836307195f475255dd5583a798d [file] [log] [blame]
#!/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
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
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 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
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'
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 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 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 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
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
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 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()
for line in lines:
if(node in line):
return True
print 'not found: ', node
for line in lines:
print line
if(verbosity):
print 'ssh not operational - unexpected results from:', ssh_cmd
for line in lines:
print '>>>>>',line
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
# determine if string represent an integer
def is_int(self,string):
result = True
try:
number = int(string)
except:
result = False
return result
# transform hostname into ip address
def get_ip_address(self,hostname):
label = 'get_ip_address'
result = None
try:
# get virtual ip address from keepalived.conf
result = self.get_virtual_ipaddress()
if(result == None):
# get virtual ip address from nameserver
p = subprocess.Popen(['/usr/bin/nslookup', hostname], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate()
#print hostname, output, err
name = None
for line in output.splitlines():
tokens = line.split()
if(len(tokens) == 2):
t0 = tokens[0]
t1 = tokens[1]
if(t0 == 'Address:'):
if(name != None):
result = t1
break
elif(t0 == 'Name:'):
name = t1
except Exception as e:
print e
debug(label, str(result))
return result
# get ducc.head.reliable.list
def get_head_node_list(self):
head_node_list = []
# add ducc.head.reliable.list node(s)
ducc_head_list = self.ducc_properties.get("ducc.head.reliable.list")
if(ducc_head_list != None):
ducc_head_nodes = ducc_head_list.split()
if(len(ducc_head_nodes)== 0):
pass
elif(len(ducc_head_nodes)== 1):
print '>>> ERROR - "ducc.head.reliable.list" missing or invalid.'
sys.exit(1);
else:
head_node_list = ducc_head_nodes
return head_node_list
# get all possible hostnames & ip addresses for a head node
def get_head_node_list_variations(self):
# start with ducc.head.reliable.list node(s)
head_node_list = self.get_head_node_list()
# add ducc.head node
ducc_head = self.ducc_properties.get("ducc.head")
if(ducc_head == None):
print '>>> ERROR - "ducc.head" missing or invalid.'
sys.exit(1);
ducc_head_nodes = ducc_head.split()
if(len(ducc_head_nodes) != 1):
print '>>> ERROR - "ducc.head" missing or invalid.'
sys.exit(1);
head_node = ducc_head_nodes[0]
if(not head_node in head_node_list):
head_node_list.append(head_node)
# add short names
list = head_node_list
for node in list:
short_name = node.split('.')[0]
if(not self.is_int(short_name)):
if(not short_name in head_node_list):
head_node_list.append(short_name)
# add ip addresses
list = head_node_list
for node in list:
ip = self.get_ip_address(node)
if(ip != None):
if(not ip in head_node_list):
head_node_list.append(ip)
#
debug('head_node_list: ', head_node_list)
return head_node_list
# drop domain and whitespace
def normalize(self,name):
result = name
if(name != None):
result = name
result = result.strip()
result = result.split('.')[0]
return result
# get current host's name
def get_node_name(self):
node_name = 'unknown'
cmd = '/bin/hostname'
resp = self.popen(cmd)
lines = resp.readlines()
if(len(lines)== 1):
name = lines[0]
node_name = self.normalize(name)
debug('node_name: ', node_name)
return node_name
def is_head_node(self):
retVal = False
head_node_list = self.get_head_node_list_variations()
node = self.get_node_name()
if(node in head_node_list):
retVal = True
return retVal
# 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_node_list = self.get_head_node_list_variations()
node = self.get_node_name()
if(node in head_node_list):
pass
else:
ip = self.get_ip_address(node)
if(ip in head_node_list):
pass
else:
print ">>> ERROR - "+node+" not configured as head node."
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)
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 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_properties.get('ducc.head')
java_class = 'org.apache.uima.ducc.common.main.DuccAdmin'
cmd = self.java()+' '+DUCC_JVM_OPTS+' '+java_class+' '+option+' '+target
print cmd
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_properties.get('ducc.head')
if ( head == '<head-node>' ):
return False
return True
keepalived_conf = '/etc/keepalived/keepalived.conf'
def get_virtual_ipaddress(self):
state = 0
vip = None
if ( os.path.exists(self.keepalived_conf) ):
with open(self.keepalived_conf) as f:
for line in f:
tokens = line.split(' ')
if(tokens == None):
pass
elif(len(tokens) == 0):
pass
elif(tokens[0] == '!'):
pass
else:
for token in tokens:
token = token.strip()
if(len(token) == 0):
continue
elif(token == '#'):
break
if(state == 0):
if(token == 'virtual_ipaddress'):
state = 1
elif(state == 1):
if(token == '{'):
state = 2
elif(state == 2):
vip = token
state = 3
elif(state == 3):
if(token == '}'):
state = 4
else:
pass
return vip
# eligible when keepalived config comprises the ip
def is_reliable_eligible(self, ip):
retVal = False
if ( os.path.exists(self.keepalived_conf) ):
with open(self.keepalived_conf) as f:
for line in f:
if ip in line:
retVal = True
break
return retVal
def is_valid_ip_address(self,address):
retVal = False
try:
list = address.split('.')
for item in list:
int(item)
retVal = True
except Exception as e:
pass
return retVal
# master when current node keepalived answers for head node ip
# backup when current node keepalived does not answer for head ip, but is capable in config
# unspecified otherwise
def get_reliable_state(self):
label = 'get_reliable_state'
result = 'unspecified'
try:
ducc_head = self.ducc_properties.get('ducc.head')
if(self.is_valid_ip_address(ducc_head)):
head_ip = ducc_head
else:
head_ip = self.get_ip_address(ducc_head)
# Check if "reliable" ... i.e. ducc.head is the virtual ip in the keepalived conf file
# If so check if this node is connected to the virtual ip
if(self.is_reliable_eligible(head_ip)):
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()
text = "output: "+output
debug(label, text)
if(head_ip in output):
result = 'master'
else:
result = 'backup'
except Exception as e:
print e
return result
def is_reliable_backup(self):
return self.get_reliable_state() == 'backup'
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)
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']
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 property so script and java match
dbhost = self.get_db_host()
if ( dbhost == None ):
dbhost = self.ducc_properties.get('ducc.head')
if ( dbhost == None ):
dbhost = 'localhost'
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
if(manage_database):
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.get_node_name()+'/ducc.pids'
self.set_classpath()
self.os_pagesize = self.get_os_pagesize()
self.update_properties()
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()