/usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxutil
Edit: /usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxutil/cli.py (10702B)
#!/opt/imh-python/bin/python3.9
# vim: set ts=4 sw=4 expandtab syntax=python:
"""
ngxutil.cli
Command-line functions & CLI entry-point
@author J. Hipps
"""
import os
import sys
import json
import logging
import logging.handlers
from urllib.parse import urlparse
from argparse import ArgumentParser, Action, SUPPRESS
import signal
import fcntl
import yaml
from ngxutil.util import format_size, excepthook
from ngxutil import report, cache, sendmail, api, info, profile
from ngxutil import __version__, __date__, email_from_addr
logger = logging.getLogger('ngxutil')
STATS_LOCK_FILE = '/var/run/ngxutil_stats.lock'
class SafeStoreAction(Action):
def __init__(self, option_strings, dest, nargs=None, **kwargs):
super(SafeStoreAction, self).__init__(option_strings, dest, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):
is_safe = False
if 'SUDO_UID' in os.environ:
if os.environ['SUDO_UID'] == '0':
is_safe = True
else:
is_safe = False
else:
is_safe = True
if is_safe:
setattr(namespace, self.dest, values)
else:
print("ERROR: Attempted to set dangerous option as a non-root sudoer! Aborting.")
sys.exit(249)
class SafeConstAction(Action):
def __init__(self, option_strings, dest, const, default=None,
required=False, help=None, metavar=None):
super(SafeConstAction, self).__init__(option_strings=option_strings,
dest=dest, nargs=0, const=const, default=default, required=required,
help=help)
def __call__(self, parser, namespace, values, option_string=None):
is_safe = False
if 'SUDO_UID' in os.environ:
if os.environ['SUDO_UID'] == '0':
is_safe = True
else:
is_safe = False
else:
is_safe = True
if is_safe:
setattr(namespace, self.dest, self.const)
else:
print("ERROR: Attempted to set dangerous option as a non-root sudoer! Aborting.")
sys.exit(249)
class OperationTimeout(Exception):
"""Raised when an operation times out"""
pass
def timeout_handler(signum, frame):
"""Signal handler for SIGALRM timeout"""
raise OperationTimeout("Operation timed out")
def setup_logging(clevel=logging.INFO, flevel=logging.DEBUG, logfile='/var/log/ngxutil.log'):
"""
Setup logging
"""
logger.setLevel(logging.DEBUG)
# Console
con = logging.StreamHandler()
con.setLevel(clevel)
con_format = logging.Formatter("%(levelname)s: %(message)s")
con.setFormatter(con_format)
logger.addHandler(con)
# File
try:
flog = logging.handlers.WatchedFileHandler(logfile)
flog.setLevel(flevel)
flog_format = logging.Formatter("[%(asctime)s] %(name)s: %(levelname)s: %(message)s")
flog.setFormatter(flog_format)
logger.addHandler(flog)
except Exception as e:
logger.warning("Failed to open logfile: %s", str(e))
def parse_cli(show_help=False):
"""
Parse CLI arguments
"""
parser = ArgumentParser(description="Nginx status reporting and cache manipulation utility")
parser.set_defaults(user=None, domain=None, loglevel=logging.INFO, logfile="/var/log/ngxutil.log",
email=None, action='stats', outfmt='ansi', logdata=False, resolve=True,
span=None, lines=None, url=None, profile=None, timeout=60)
parser.add_argument('--user', '-u', action='store', metavar="USER",
help="Specify user")
parser.add_argument('--domain', '-d', action='store', metavar="DOMAIN",
help="Specify domain")
parser.add_argument('--email', '-E', action='store', metavar="ADDR",
help="Email report to specified address")
parser.add_argument('--profile', '-P', action='store', metavar="PROFILE",
help="Reset domain configuration to defaults using specified profile. Must also specify user (-u) and domain (-d).")
parser.add_argument('--purge', '-X', action='store_const', dest='action', const='purge',
help="Mode: Purge cache for specified user or page")
parser.add_argument('--purgeall', '-Z', action=SafeConstAction, dest='action', const='purgeall',
help="Mode: Purge ALL cached data, for all users")
parser.add_argument('--stats', '-S', action='store_const', dest='action', const='stats',
help="Mode: Show cache stats [default]")
parser.add_argument('--info', '-I', action='store_const', dest='action', const='info',
help="Mode: Show info for URL")
parser.add_argument('--span', '-t', action='store', type=int, help="Parse specified span from logs (in hours)")
parser.add_argument('--lines', '-z', action='store', type=int, metavar="X",
help="Parse X number of lines from log tail")
parser.add_argument('--logfile', '-l', action=SafeStoreAction, metavar="PATH",
help="Log output file")
# Deprecated no-op. Kept so existing callers (notably the cpanel-cache-manager
# plugin's `/usr/local/cpanel/bin/admin/nginx/Nginx`) don't fail argparse on
# upgrade. The InfluxDB code path was removed; logparse is now the only
# data source. Safe to drop once all packaged callers stop passing it.
parser.add_argument('--noflux', '-x', action='store_true',
help=SUPPRESS)
parser.add_argument('--debug', '-D', dest='loglevel', action='store_const', const=logging.DEBUG,
help="Enable debug output")
parser.add_argument('--logdata', action='store_true', help="Return logdata in JSON or YAML output")
parser.add_argument('--timeout', '-T', action='store', type=int, default=60,
metavar="SECONDS",
help="Timeout for stats operation in seconds (default: 60, 0 to disable)")
parser.add_argument('--json', '-J', dest='outfmt', action='store_const', const='json',
help="Output format: JSON")
parser.add_argument('--yaml', '-Y', dest='outfmt', action='store_const', const='yaml',
help="Output format: YAML")
parser.add_argument('--html', '-H', dest='outfmt', action='store_const', const='html',
help="Output format: HTML")
parser.add_argument('--version', '-v', action='version', version="%s (%s)" % (__version__, __date__))
parser.add_argument('url', nargs='?', metavar="URL_OR_LOGFILE",
help="Target URL when using --info or --purge modes; logfile when using report modes")
if show_help:
parser.print_help()
sys.exit(1)
return parser.parse_args()
def _main():
"""Entry point"""
sys.excepthook = excepthook
args = parse_cli()
setup_logging(clevel=args.loglevel, flevel=logging.DEBUG, logfile=args.logfile)
if args.profile and args.user and args.domain:
profile.set_domain_profile(args.user, args.domain, args.profile)
elif args.action == 'purge':
if args.user and not args.url:
cache.purge_cache_zone(args.user)
else:
if args.url:
if urlparse(args.url).scheme == 'all':
if args.user:
cache.purge_url_all(args.url, args.user)
else:
logger.error("Must specify --user when using all://")
parse_cli(show_help=True)
else:
cache.purge_url(args.url)
else:
logger.error("Must specify --user or url with --purge action")
parse_cli(show_help=True)
elif args.action == 'purgeall':
cache.purge_full_cache()
elif args.action == 'info':
if args.url:
info.check_url(args.url)
else:
logger.error("Must specify URL when using --info")
parse_cli(show_help=True)
elif args.action == 'stats':
infile = os.path.expanduser(args.url) if args.url else '/var/log/nginx/access.log'
lock_fd = None
# Acquire exclusive lock for stats operation
try:
lock_fd = open(STATS_LOCK_FILE, 'w')
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except (IOError, OSError):
logger.error("Another ngxutil stats operation is already running")
sys.exit(125)
# Set up timeout if specified
if args.timeout > 0:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(args.timeout)
try:
if args.outfmt in ('ansi', 'html'):
if args.domain:
vstat = report.repgen_domain(args.domain, span=args.span, tail=args.lines, infile=infile)
elif args.user:
logger.error("Must specify --domain with --stats action")
parse_cli(show_help=True)
else:
vstat = report.repgen_server(span=args.span, tail=args.lines, infile=infile)
if vstat:
if args.outfmt == 'html':
print(vstat[1])
else:
print(vstat[0])
if args.email:
sendmail.email_report(args.email, email_from_addr, "Cache Status Report", *vstat)
else:
if args.domain:
vstat = api.get_domain_info(args.domain, include_logdata=args.logdata, span=args.span, tail=args.lines, infile=infile)
else:
vstat = api.get_server_info(include_logdata=args.logdata, span=args.span, tail=args.lines, infile=infile)
if args.outfmt == 'json':
outstr = json.dumps(vstat, sort_keys=True, indent=4, separators=(',', ': '))
else:
outstr = yaml.dump(vstat, Dumper=yaml.SafeDumper, default_flow_style=False)
sys.stdout.write(outstr)
sys.stdout.flush()
except OperationTimeout:
logger.error("Operation timed out after %d seconds", args.timeout)
sys.exit(124)
finally:
if args.timeout > 0:
signal.alarm(0)
if lock_fd:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
else:
parse_cli(show_help=True)
if __name__ == '__main__':
_main()