mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2025-02-25 18:55:31 -06:00
Re-engineer the background process executor, to avoid using sqlite as some builds of components it relies on do not support working in forked children.
This commit is contained in:
Ashesh Vashi
Dave Page
@@ -23,66 +23,108 @@ This script will:
|
||||
database.
|
||||
|
||||
Args:
|
||||
process_id -- Process id
|
||||
db_file -- Database file which holds list of processes to be executed
|
||||
output_directory -- output directory
|
||||
list of program and arguments passed to it.
|
||||
|
||||
It also depends on the following environment variable for proper execution.
|
||||
PROCID - Process-id
|
||||
OUTDIR - Output directory
|
||||
"""
|
||||
from __future__ import print_function, unicode_literals
|
||||
|
||||
# To make print function compatible with python2 & python3
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta, tzinfo
|
||||
from subprocess import Popen, PIPE
|
||||
from threading import Thread
|
||||
import csv
|
||||
import pytz
|
||||
import codecs
|
||||
import signal
|
||||
|
||||
# SQLite3 needs all string as UTF-8
|
||||
# We need to make string for Python2/3 compatible
|
||||
if sys.version_info < (3,):
|
||||
from cStringIO import StringIO
|
||||
def log(msg):
|
||||
if 'OUTDIR' not in os.environ:
|
||||
return
|
||||
|
||||
with open(
|
||||
os.path.join(os.environ['OUTDIR'], ('log_%s' % os.getpid())), 'a'
|
||||
) as fp:
|
||||
fp.write(('INFO:: %s\n' % str(msg)))
|
||||
|
||||
|
||||
def u(x):
|
||||
return x
|
||||
else:
|
||||
from io import StringIO
|
||||
def log_exception():
|
||||
if 'OUTDIR' not in os.environ:
|
||||
return
|
||||
type_, value_, traceback_ = info=sys.exc_info()
|
||||
|
||||
with open(
|
||||
os.path.join(os.environ['OUTDIR'], ('log_%s' % os.getpid())), 'a'
|
||||
) as fp:
|
||||
from traceback import format_exception
|
||||
res = ''.join(
|
||||
format_exception(type_, value_, traceback_)
|
||||
)
|
||||
|
||||
fp.write('EXCEPTION::\n{0}'.format(res))
|
||||
return res
|
||||
|
||||
|
||||
def u(x):
|
||||
if hasattr(x, 'decode'):
|
||||
return x.decode()
|
||||
return x
|
||||
IS_WIN = (os.name == 'nt')
|
||||
|
||||
ZERO = timedelta(0)
|
||||
default_encoding = sys.getdefaultencoding() or "utf-8"
|
||||
|
||||
def usage():
|
||||
# Copied the 'UTC' class from the 'pytz' package to allow to run this script
|
||||
# without any external dependent library, and can be used with any python
|
||||
# version.
|
||||
class UTC(tzinfo):
|
||||
"""UTC
|
||||
|
||||
Optimized UTC implementation. It unpickles using the single module global
|
||||
instance defined beneath this class declaration.
|
||||
"""
|
||||
This function will display usage message.
|
||||
zone = "UTC"
|
||||
|
||||
Args:
|
||||
None
|
||||
_utcoffset = ZERO
|
||||
_dst = ZERO
|
||||
_tzname = zone
|
||||
|
||||
Returns:
|
||||
Displays help message
|
||||
"""
|
||||
def fromutc(self, dt):
|
||||
if dt.tzinfo is None:
|
||||
return self.localize(dt)
|
||||
return super(UTC.__class__, self).fromutc(dt)
|
||||
|
||||
help_msg = """
|
||||
Usage:
|
||||
def utcoffset(self, dt):
|
||||
return ZERO
|
||||
|
||||
executer.py [-h|--help]
|
||||
[-p|--process] Process ID
|
||||
[-d|--db_file] SQLite3 database file path
|
||||
"""
|
||||
print(help_msg)
|
||||
def tzname(self, dt):
|
||||
return "UTC"
|
||||
|
||||
def dst(self, dt):
|
||||
return ZERO
|
||||
|
||||
def localize(self, dt, is_dst=False):
|
||||
'''Convert naive time to local time'''
|
||||
if dt.tzinfo is not None:
|
||||
raise ValueError('Not naive datetime (tzinfo is already set)')
|
||||
return dt.replace(tzinfo=self)
|
||||
|
||||
def normalize(self, dt, is_dst=False):
|
||||
'''Correct the timezone information on the given datetime'''
|
||||
if dt.tzinfo is self:
|
||||
return dt
|
||||
if dt.tzinfo is None:
|
||||
raise ValueError('Naive time - no tzinfo set')
|
||||
return dt.astimezone(self)
|
||||
|
||||
def __repr__(self):
|
||||
return "<UTC>"
|
||||
|
||||
def __str__(self):
|
||||
return "UTC"
|
||||
|
||||
|
||||
def get_current_time(format='%Y-%m-%d %H:%M:%S.%f %z'):
|
||||
return datetime.utcnow().replace(
|
||||
tzinfo=pytz.utc
|
||||
tzinfo=UTC()
|
||||
).strftime(format)
|
||||
|
||||
|
||||
@@ -93,35 +135,34 @@ class ProcessLogger(Thread):
|
||||
|
||||
Methods:
|
||||
--------
|
||||
* __init__(stream_type, configs)
|
||||
* __init__(stream_type)
|
||||
- This method is use to initlize the ProcessLogger class object
|
||||
|
||||
* logging(msg)
|
||||
- This method is use to log messages in sqlite3 database
|
||||
* log(msg)
|
||||
- Log message in the orderly manner.
|
||||
|
||||
* run()
|
||||
- Reads the stdout/stderr for messages and sent them to logger
|
||||
"""
|
||||
|
||||
def __init__(self, stream_type, configs):
|
||||
def __init__(self, stream_type):
|
||||
"""
|
||||
This method is use to initialize the ProcessLogger class object
|
||||
|
||||
Args:
|
||||
stream_type: Type of STD (std)
|
||||
configs: Process details dict
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
Thread.__init__(self)
|
||||
self.configs = configs
|
||||
self.process = None
|
||||
self.stream = None
|
||||
self.encoding = default_encoding
|
||||
self.logger = codecs.open(
|
||||
os.path.join(
|
||||
configs['output_directory'], stream_type
|
||||
), 'w', "utf-8"
|
||||
os.environ['OUTDIR'], stream_type
|
||||
), 'w', self.encoding, "ignore"
|
||||
)
|
||||
|
||||
def attach_process_stream(self, process, stream):
|
||||
@@ -153,7 +194,8 @@ class ProcessLogger(Thread):
|
||||
if msg:
|
||||
self.logger.write(
|
||||
str('{0},{1}').format(
|
||||
get_current_time(format='%Y%m%d%H%M%S%f'), u(msg)
|
||||
get_current_time(format='%y%m%d%H%M%S%f'),
|
||||
msg.decode(self.encoding, 'replace')
|
||||
)
|
||||
)
|
||||
return True
|
||||
@@ -176,44 +218,7 @@ class ProcessLogger(Thread):
|
||||
self.logger = None
|
||||
|
||||
|
||||
def read_configs(data):
|
||||
"""
|
||||
This reads SQLite3 database and fetches process details
|
||||
|
||||
Args:
|
||||
data - configuration details
|
||||
|
||||
Returns:
|
||||
Process details fetched from database as a dict
|
||||
"""
|
||||
if data.db_file is not None and data.process_id is not None:
|
||||
conn = sqlite3.connect(data.db_file)
|
||||
c = conn.cursor()
|
||||
t = (data.process_id,)
|
||||
|
||||
c.execute('SELECT command, arguments FROM process WHERE \
|
||||
exit_code is NULL \
|
||||
AND pid=?', t)
|
||||
|
||||
row = c.fetchone()
|
||||
conn.close()
|
||||
|
||||
if row and len(row) > 1:
|
||||
configs = {
|
||||
'pid': data.process_id,
|
||||
'cmd': row[0],
|
||||
'args': row[1],
|
||||
'output_directory': data.output_directory,
|
||||
'db_file': data.db_file
|
||||
}
|
||||
return configs
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
raise ValueError("Please verify pid and db_file arguments.")
|
||||
|
||||
|
||||
def update_configs(kwargs):
|
||||
def update_status(**kw):
|
||||
"""
|
||||
This function will updates process stats
|
||||
|
||||
@@ -223,166 +228,268 @@ def update_configs(kwargs):
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
if 'db_file' in kwargs and 'pid' in kwargs:
|
||||
conn = sqlite3.connect(kwargs['db_file'])
|
||||
sql = 'UPDATE process SET '
|
||||
params = list()
|
||||
import json
|
||||
|
||||
for param in ['start_time', 'end_time', 'exit_code']:
|
||||
if param in kwargs:
|
||||
sql += (',' if len(params) else '') + param + '=? '
|
||||
params.append(kwargs[param])
|
||||
|
||||
if len(params) == 0:
|
||||
return
|
||||
|
||||
sql += 'WHERE pid=?'
|
||||
params.append(kwargs['pid'])
|
||||
|
||||
with conn:
|
||||
c = conn.cursor()
|
||||
c.execute(sql, params)
|
||||
conn.commit()
|
||||
|
||||
# Commit & close cursor
|
||||
conn.close()
|
||||
if os.environ['OUTDIR']:
|
||||
status = {
|
||||
k: v for k, v in kw.items() if k in [
|
||||
'start_time', 'end_time', 'exit_code', 'pid'
|
||||
]
|
||||
}
|
||||
log('Updating the status:\n{0}'.format(json.dumps(status)))
|
||||
with open(os.path.join(os.environ['OUTDIR'], 'status'), 'w') as fp:
|
||||
json.dump(status, fp)
|
||||
else:
|
||||
raise ValueError("Please verify pid and db_file arguments.")
|
||||
|
||||
|
||||
def execute(configs):
|
||||
def execute():
|
||||
"""
|
||||
This function will execute the background process
|
||||
|
||||
Args:
|
||||
configs: Process configuration details
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
if configs is not None:
|
||||
command = [configs['cmd']]
|
||||
args_csv = StringIO(configs['args'])
|
||||
args_reader = csv.reader(args_csv, delimiter=str(','))
|
||||
for args in args_reader:
|
||||
command = command + args
|
||||
args = {
|
||||
'pid': configs['pid'],
|
||||
'db_file': configs['db_file']
|
||||
}
|
||||
command = sys.argv[1:]
|
||||
args = dict()
|
||||
log('Initialize the process execution: {0}'.format(command))
|
||||
|
||||
# Create seprate thread for stdout and stderr
|
||||
process_stdout = ProcessLogger('out')
|
||||
process_stderr = ProcessLogger('err')
|
||||
process = None
|
||||
|
||||
try:
|
||||
# update start_time
|
||||
args.update({
|
||||
'start_time': get_current_time(),
|
||||
'stdout': process_stdout.log,
|
||||
'stderr': process_stderr.log,
|
||||
'pid': os.getpid()
|
||||
})
|
||||
|
||||
# Update start time
|
||||
update_status(**args)
|
||||
log('Status updated...')
|
||||
|
||||
if 'PROCID' in os.environ and os.environ['PROCID'] in os.environ:
|
||||
os.environ['PGPASSWORD'] = os.environ[os.environ['PROCID']]
|
||||
|
||||
kwargs = dict()
|
||||
kwargs['close_fds'] = False
|
||||
kwargs['shell'] = True if IS_WIN else False
|
||||
|
||||
# We need environment variables & values in string
|
||||
log('Converting the environment variable in the bytes format...')
|
||||
kwargs['env'] = convert_environment_variables(os.environ.copy())
|
||||
|
||||
log('Starting the command execution...')
|
||||
process = Popen(
|
||||
command, stdout=PIPE, stderr=PIPE, stdin=None, **kwargs
|
||||
)
|
||||
|
||||
log('Attaching the loggers to stdout, and stderr...')
|
||||
# Attach the stream to the process logger, and start logging.
|
||||
process_stdout.attach_process_stream(process, process.stdout)
|
||||
process_stdout.start()
|
||||
process_stderr.attach_process_stream(process, process.stderr)
|
||||
process_stderr.start()
|
||||
|
||||
# Join both threads together
|
||||
process_stdout.join()
|
||||
process_stderr.join()
|
||||
|
||||
log('Waiting for the process to finish...')
|
||||
# Child process return code
|
||||
exitCode = process.wait()
|
||||
|
||||
if exitCode is None:
|
||||
exitCode = process.poll()
|
||||
|
||||
log('Process exited with code: {0}'.format(exitCode))
|
||||
args.update({'exit_code': exitCode})
|
||||
|
||||
# Add end_time
|
||||
args.update({'end_time': get_current_time()})
|
||||
|
||||
# Fetch last output, and error from process if it has missed.
|
||||
data = process.communicate()
|
||||
if data:
|
||||
if data[0]:
|
||||
process_stdout.log(data[0])
|
||||
if data[1]:
|
||||
process_stderr.log(data[1])
|
||||
|
||||
# If executable not found or invalid arguments passed
|
||||
except OSError:
|
||||
info = log_exception()
|
||||
args.update({'exit_code': 500})
|
||||
if process_stderr:
|
||||
process_stderr.log(info)
|
||||
else:
|
||||
print("WARNING: ", e.strerror, file=sys.stderr)
|
||||
args.update({'end_time': get_current_time()})
|
||||
args.update({'exit_code': e.errno})
|
||||
|
||||
# Unknown errors
|
||||
except Exception:
|
||||
info = log_exception()
|
||||
args.update({'exit_code': 501})
|
||||
if process_stderr:
|
||||
process_stderr.log(info)
|
||||
else:
|
||||
print("WARNING: ", str(e), file=sys.stderr)
|
||||
args.update({'end_time': get_current_time()})
|
||||
args.update({'exit_code': -1})
|
||||
finally:
|
||||
# Update the execution end_time, and exit-code.
|
||||
update_status(**args)
|
||||
log('Exiting the process executor...')
|
||||
if process_stderr:
|
||||
process_stderr.release()
|
||||
if process_stdout:
|
||||
process_stdout.release()
|
||||
log('Bye!')
|
||||
|
||||
|
||||
# Let's ignore all the signal comming to us.
|
||||
def signal_handler(signal, msg):
|
||||
pass
|
||||
|
||||
|
||||
def convert_environment_variables(env):
|
||||
"""
|
||||
This function is use to convert environment variable to string
|
||||
because environment variable must be string in popen
|
||||
:param env: Dict of environment variable
|
||||
:return: Encoded environment variable as string
|
||||
"""
|
||||
temp_env = dict()
|
||||
for key, value in env.items():
|
||||
try:
|
||||
reload(sys)
|
||||
sys.setdefaultencoding('utf8')
|
||||
except:
|
||||
pass
|
||||
|
||||
# Create seprate thread for stdout and stderr
|
||||
process_stdout = ProcessLogger('out', configs)
|
||||
process_stderr = ProcessLogger('err', configs)
|
||||
|
||||
try:
|
||||
# update start_time
|
||||
args.update({
|
||||
'start_time': get_current_time(),
|
||||
'stdout': process_stdout.log,
|
||||
'stderr': process_stderr.log
|
||||
})
|
||||
|
||||
# Update start time
|
||||
update_configs(args)
|
||||
|
||||
if args['pid'] in os.environ:
|
||||
os.environ['PGPASSWORD'] = os.environ[args['pid']]
|
||||
|
||||
process = Popen(
|
||||
command, stdout=PIPE, stderr=PIPE, stdin=PIPE,
|
||||
shell=(os.name == 'nt'), close_fds=(os.name != 'nt')
|
||||
)
|
||||
try:
|
||||
del (os.environ['PGPASSWORD'])
|
||||
except:
|
||||
pass
|
||||
|
||||
# Attach the stream to the process logger, and start logging.
|
||||
process_stdout.attach_process_stream(process, process.stdout)
|
||||
process_stdout.start()
|
||||
process_stderr.attach_process_stream(process, process.stderr)
|
||||
process_stderr.start()
|
||||
|
||||
# Join both threads together
|
||||
process_stdout.join()
|
||||
process_stderr.join()
|
||||
|
||||
# Child process return code
|
||||
exitCode = process.wait()
|
||||
|
||||
if exitCode is None:
|
||||
exitCode = process.poll()
|
||||
|
||||
args.update({'exit_code': exitCode})
|
||||
|
||||
# Add end_time
|
||||
args.update({'end_time': get_current_time()})
|
||||
|
||||
# Fetch last output, and error from process if it has missed.
|
||||
data = process.communicate()
|
||||
if data:
|
||||
if data[0]:
|
||||
process_stdout.log(data[0])
|
||||
if data[1]:
|
||||
process_stderr.log(data[1])
|
||||
|
||||
# If executable not found or invalid arguments passed
|
||||
except OSError as e:
|
||||
if process_stderr:
|
||||
process_stderr.log(e.strerror)
|
||||
else:
|
||||
print("WARNING: ", e.strerror, file=sys.stderr)
|
||||
args.update({'end_time': get_current_time()})
|
||||
args.update({'exit_code': e.errno})
|
||||
|
||||
# Unknown errors
|
||||
if not isinstance(key, str):
|
||||
key = key.encode(default_encoding)
|
||||
if not isinstance(value, str):
|
||||
value = value.encode(default_encoding)
|
||||
temp_env[key] = value
|
||||
except Exception as e:
|
||||
if process_stderr:
|
||||
process_stderr.log(str(e))
|
||||
else:
|
||||
print("WARNING: ", str(e), file=sys.stderr)
|
||||
args.update({'end_time': get_current_time()})
|
||||
args.update({'exit_code': -1})
|
||||
finally:
|
||||
# Update the execution end_time, and exit-code.
|
||||
update_configs(args)
|
||||
if process_stderr:
|
||||
process_stderr.release()
|
||||
process_stderr = None
|
||||
if process_stdout:
|
||||
process_stdout.release()
|
||||
process_stdout = None
|
||||
|
||||
else:
|
||||
raise ValueError("Please verify process configs.")
|
||||
log_exception()
|
||||
return temp_env
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Read command line arguments
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Process executor for pgAdmin 4'
|
||||
)
|
||||
parser.add_argument(
|
||||
'-p', '--process_id', help='Process ID', required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'-d', '--db_file', help='Configuration Database', required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'-o', '--output_directory',
|
||||
help='Location where the logs will be created', required=True
|
||||
)
|
||||
args = parser.parse_args()
|
||||
log('Starting the process executor...')
|
||||
|
||||
# Fetch bakcground process details from SQLite3 database file
|
||||
configs = read_configs(args)
|
||||
# Ignore any signals
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
log('Disabled the SIGINT, SIGTERM signals...')
|
||||
|
||||
# Execute the background process
|
||||
execute(configs)
|
||||
if IS_WIN:
|
||||
log('Disable the SIGBREAKM signal (windows)...')
|
||||
signal.signal(signal.SIGBREAK, signal_handler)
|
||||
log('Disabled the SIGBREAKM signal (windows)...')
|
||||
|
||||
# For windows:
|
||||
# We would run the process_executor in the detached mode again to make
|
||||
# the child process to run as a daemon. And, it would run without
|
||||
# depending on the status of the web-server.
|
||||
if 'PGA_BGP_FOREGROUND' in os.environ and \
|
||||
os.environ['PGA_BGP_FOREGROUND'] == "1":
|
||||
log('[CHILD] Start process execution...')
|
||||
log('Executing the command now from the detached child...')
|
||||
# This is a child process running as the daemon process.
|
||||
# Let's do the job assing to it.
|
||||
execute()
|
||||
else:
|
||||
from subprocess import CREATE_NEW_PROCESS_GROUP
|
||||
DETACHED_PROCESS = 0x00000008
|
||||
|
||||
# Forward the standard input, output, and error stream to the
|
||||
# 'devnull'.
|
||||
stdin = open(os.devnull, "r")
|
||||
stdout = open(os.devnull, "a")
|
||||
stderr = open(os.devnull, "a")
|
||||
env = os.environ.copy()
|
||||
env['PGA_BGP_FOREGROUND'] = "1"
|
||||
|
||||
# We need environment variables & values in string
|
||||
log('[PARENT] Converting the environment variable in the bytes format...')
|
||||
try:
|
||||
env = convert_environment_variables(env)
|
||||
except Exception as e:
|
||||
log_exception()
|
||||
|
||||
kwargs = {
|
||||
'stdin': stdin.fileno(),
|
||||
'stdout': stdout.fileno(),
|
||||
'stderr': stderr.fileno(),
|
||||
'creationflags': CREATE_NEW_PROCESS_GROUP | DETACHED_PROCESS,
|
||||
'close_fds': False,
|
||||
'cwd': os.environ['OUTDIR'],
|
||||
'env': env
|
||||
}
|
||||
|
||||
cmd = [sys.executable]
|
||||
cmd.extend(sys.argv)
|
||||
|
||||
log('[PARENT] Command executings: {0}'.format(cmd))
|
||||
|
||||
p = Popen(cmd, **kwargs)
|
||||
|
||||
exitCode = p.poll()
|
||||
|
||||
if exitCode is not None:
|
||||
log(
|
||||
'[PARENT] Child exited with exit-code#{0}...'.format(
|
||||
exitCode
|
||||
)
|
||||
)
|
||||
else:
|
||||
log('[PARENT] Started the child with PID#{0}'.format(p.pid))
|
||||
|
||||
# Question: Should we wait for sometime?
|
||||
# Answer: Looks the case...
|
||||
from time import sleep
|
||||
sleep(2)
|
||||
log('[PARENT] Exiting...')
|
||||
sys.exit(0)
|
||||
else:
|
||||
r, w = os.pipe()
|
||||
|
||||
# For POSIX:
|
||||
# We will fork the process, and run the child process as daemon, and
|
||||
# let it do the job.
|
||||
if os.fork() == 0:
|
||||
log('[CHILD] Forked the child process...')
|
||||
# Hmm... So - I need to do the job now...
|
||||
try:
|
||||
os.close(r)
|
||||
|
||||
log('[CHILD] Make the child process leader...')
|
||||
# Let me be the process leader first.
|
||||
os.setsid()
|
||||
os.umask(0)
|
||||
|
||||
log('[CHILD] Make the child process leader...')
|
||||
w = os.fdopen(w, 'w')
|
||||
# Let me inform my parent - I will do the job, do not worry
|
||||
# now, and die peacefully.
|
||||
log('[CHILD] Inform parent about successful child forking...')
|
||||
w.write('1')
|
||||
w.close()
|
||||
|
||||
log('[CHILD] Start executing the background process...')
|
||||
execute()
|
||||
except Exception:
|
||||
sys.exit(1)
|
||||
else:
|
||||
os.close(w)
|
||||
r = os.fdopen(r)
|
||||
# I do not care, what the child send.
|
||||
r.read()
|
||||
log('[PARENT] Got message from the child...')
|
||||
r.close()
|
||||
|
||||
log('[PARENT] Exiting...')
|
||||
sys.exit(0)
|
||||
|
||||
Reference in New Issue
Block a user