blob: 4bbcda5d0936d6f8ce78a7c0841411320dda7e93 [file] [log] [blame]
#!/usr/bin/env python
# -----------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# -----------------------------------------------------------------------
import sys
version_min = [2, 7]
version_info = sys.version_info
version_error = False
if(version_info[0] < version_min[0]):
version_error = True
elif(version_info[0] == version_min[0]):
if(version_info[1] < version_min[1]):
version_error = True
if(version_error):
print('Python minimum requirement is version '+str(version_min[0])+'.'+str(version_min[1]))
sys.exit(1)
import argparse
import datetime
import logging
import os
import shlex
import socket
import subprocess
import time
import traceback
from ducc_util import DuccUtil
# produce a time stamp
def get_timestamp():
tod = time.time()
timestamp = datetime.datetime.fromtimestamp(tod).strftime('%Y-%m-%d %H:%M:%S')
return timestamp
# embedded class to log to file
# default level is info level,
# controlled by environment variable LOGLEVEL = { info, debug, trace }
class Logger():
flag_info = True
flag_debug = False
flag_trace = False
flag_error = True
flag_warn = True
def __init__(self,filename,level='info'):
self.filename = filename
if(level == 'debug'):
self.flag_debug = True
elif(level == 'trace'):
self.flag_debug = True
self.flag_trace = True
# write to file
def output(self,type,mn,text):
message = get_timestamp()+' '+type+' '+mn+' '+text
with open(self.filename, 'a') as logfile:
logfile.write(message+'\n')
# record info message
def info(self,mn,text):
if(self.flag_info):
self.output('I',mn,text)
# record debug message
def debug(self,mn,text):
if(self.flag_debug):
self.output('D',mn,text)
# record trace message
def trace(self,mn,text):
if(self.flag_trace):
self.output('T',mn,text)
# record error message
def error(self,mn,text):
if(self.flag_error):
self.output('E',mn,text)
# record warn message
def warn(self,mn,text):
if(self.flag_warn):
self.output('W',mn,text)
# class to automagically start DUCC daemons
class AutoStart(DuccUtil):
components = [ 'agent', 'broker', 'orchestrator', 'pm', 'rm', 'sm', 'ws' ]
map = { 'ag':'agent',
'br':'broker',
'or':'or',
'pm':'pm',
'rm':'rm',
'sm':'sm',
'ws':'ws',
}
def __init__(self):
DuccUtil.__init__(self)
self.ssh_enabled = False
# return file name
def _fn(self):
fpath = __file__.split('/')
flen = len(fpath)
return fpath[flen-1]
# return class name
def _cn(self):
return self.__class__.__name__
# return method name
def _mn(self):
return traceback.extract_stack(None,2)[0][2]
description = 'Start daemon(s) on the present node when listed in the autostart database table but not already running.'
def get_args(self):
parser = argparse.ArgumentParser(description=self.description)
self.args = parser.parse_args()
# setup logging to file for autostart script
def setup_logging(self,NAME):
LOGDIR = self.DUCC_HOME+'/logs/'+NAME
self.makedirs(LOGDIR)
self.LOCAL_HOST = socket.getfqdn()
FN = self.LOCAL_HOST.split('.')[0]+'.'+NAME+'.log'
LOGFILE = LOGDIR+'/'+FN
LOGLEVEL = os.environ.get('LOGLEVEL','info')
self.logger = Logger(LOGFILE,LOGLEVEL)
# check if host names with domain match
def is_host_match_with_domain(self,h1,h2):
retVal = False
if(h1 == h2):
retVal = True
text = str(h1)+' '+str(h2)+' '+str(retVal)
self.logger.debug(self._mn(),text)
return retVal
# check if host names without domain match
def is_host_match_without_domain(self,h1,h2):
retVal = False
h10 = h1.split('.')[0]
h20 = h2.split('.')[0]
if(h10 == h20):
retVal = True
text = str(h10)+' '+str(h20)+' '+str(retVal)
self.logger.debug(self._mn(),text)
return retVal
#check if host names match with/without domain
def is_host_match(self,h1,h2):
retVal = False
if(h1 != None):
if(h2 != None):
if(self.is_host_match_with_domain(h1,h2)):
retVal = True
elif(self.is_host_match_without_domain(h1,h2)):
retVal = True
text = str(h1)+' '+str(h2)+' '+str(retVal)
self.logger.debug(self._mn(),text)
return retVal
def parse_line(self,line):
retVal = '', '', ''
try:
if(line != None):
text = line
self.logger.debug(self._mn(),text)
tokens = line.split()
if(len(tokens) == 1):
host = line.split('.')[0]
remainder = line.split('.')[1]
name = remainder.split('=')[0]
state = remainder.split('=')[1]
retVal = host, name, state
except:
pass
return retVal
# get daemons started/all in DB for host
def get_daemons_host(self):
db = []
started = []
jclass = 'org.apache.uima.ducc.database.lifetime.DbDaemonLifetimeUI'
option = '--query'
cmd = [self.jvm, '-DDUCC_HOME='+self.DUCC_HOME, jclass, option]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate()
lines = out.split('\n')
for line in lines:
host, daemon, state = self.parse_line(line)
if(self.is_host_match(self.LOCAL_HOST, host)):
db.append(daemon)
if(state == 'Start'):
started.append(daemon)
text = 'add'+' '+host+' '+daemon
self.logger.debug(self._mn(),text)
else:
text = 'skip'+' '+host+' '+daemon
self.logger.debug(self._mn(),text)
else:
text = 'skip'+' '+host+' '+daemon
self.logger.debug(self._mn(),text)
text = 'started'+' '+str(started)+' '+'db'+' '+str(db)
self.logger.debug(self._mn(),text)
return db, started
def normalize_component(self,component):
daemon = component[:2]
return daemon
# get daemons running (from system)
def get_components_running(self):
daemons = []
result = self.find_ducc_process(self.LOCAL_HOST)
find_status = result[0]
text = 'find_status:'+str(find_status)
self.logger.debug(self._mn(),text)
tuples = result[1]
text = 'tuples:'+str(tuples)
self.logger.debug(self._mn(),text)
for tuple in tuples:
component = tuple[0]
pid = tuple[1]
user = tuple[2]
if(user == self.ducc_uid):
if(component in self.components):
text = 'keep:'+str(tuple)
self.logger.debug(self._mn(),text)
daemon = self.normalize_component(component)
daemons.append(daemon)
else:
text = 'skip:'+str(tuple)
self.logger.debug(self._mn(),text)
else:
text = 'skip:'+str(tuple)+' '+str(self.ducc_uid)
self.logger.debug(self._mn(),text)
text = 'daemons:'+str(daemons)
self.logger.debug(self._mn(),text)
return daemons
def start(self,daemon):
component = self.map[daemon]
text = str(component)
self.logger.warn(self._mn(),text)
if(daemon == 'ag'):
python_script = os.path.join(self.DUCC_HOME,'admin','ducc.py')
cmd = [ python_script, '-c', component, '-b', '-d', str(time.time()), '--nodup' ]
else:
python_script = os.path.join(self.DUCC_HOME,'admin','start_ducc',)
cmd = [ python_script, '-c', component ]
text = str(cmd)
self.logger.info(self._mn(),text)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate()
text = str(out)
self.logger.info(self._mn(),text)
def insert(self):
python_script = os.path.join(self.DUCC_HOME,'admin','db_autostart_insert.py')
node = self.get_node_name()
component = 'ag'
cmd = [ python_script, '--host', node, '--name', component]
text = str(cmd)
self.logger.info(self._mn(),text)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate()
text = str(out)
self.logger.info(self._mn(),text)
# autostart: start head or agent daemons, as required
def main(self, argv):
NAME = 'autostart'
self.setup_logging(NAME)
self.get_args()
try:
daemons_db, daemons_started = self.get_daemons_host()
text = 'daemons_db '+str(len(daemons_db))
self.logger.debug(self._mn(),text)
text = 'daemons_started '+str(len(daemons_started))
self.logger.debug(self._mn(),text)
# if agent node is not in db, then insert it!
if(len(daemons_db) == 0):
if(self.is_head_node()):
pass
else:
self.insert()
daemons_db, daemons_started = self.get_daemons_host()
components_running = self.get_components_running()
text = 'components_running '+str(components_running)
self.logger.debug(self._mn(),text)
for daemon in daemons_started:
component = self.normalize_component(daemon)
text = 'component '+str(component)
self.logger.debug(self._mn(),text)
if(component in components_running):
pass
else:
text = 'start daemon '+str(daemon)
self.logger.debug(self._mn(),text)
self.start(daemon)
except Exception,e:
lines = traceback.format_exc().splitlines()
for line in lines:
text = line
self.logger.debug(self._mn(),text)
if __name__ == '__main__':
instance = AutoStart()
instance.main(sys.argv[1:])