diff options
Diffstat (limited to 'cloudinit/cmd/status.py')
| -rw-r--r-- | cloudinit/cmd/status.py | 141 |
1 files changed, 81 insertions, 60 deletions
diff --git a/cloudinit/cmd/status.py b/cloudinit/cmd/status.py index ea79a85b..5176549d 100644 --- a/cloudinit/cmd/status.py +++ b/cloudinit/cmd/status.py @@ -5,22 +5,28 @@ """Define 'status' utility and handler as part of cloud-init commandline.""" import argparse +import enum import os import sys -from time import gmtime, strftime, sleep +from time import gmtime, sleep, strftime from cloudinit.distros import uses_systemd from cloudinit.stages import Init from cloudinit.util import get_cmdline, load_file, load_json -CLOUDINIT_DISABLED_FILE = '/etc/cloud/cloud-init.disabled' +CLOUDINIT_DISABLED_FILE = "/etc/cloud/cloud-init.disabled" + # customer visible status messages -STATUS_ENABLED_NOT_RUN = 'not run' -STATUS_RUNNING = 'running' -STATUS_DONE = 'done' -STATUS_ERROR = 'error' -STATUS_DISABLED = 'disabled' +@enum.unique +class UXAppStatus(enum.Enum): + """Enum representing user-visible cloud-init application status.""" + + NOT_RUN = "not run" + RUNNING = "running" + DONE = "done" + ERROR = "error" + DISABLED = "disabled" def get_parser(parser=None): @@ -34,15 +40,25 @@ def get_parser(parser=None): """ if not parser: parser = argparse.ArgumentParser( - prog='status', - description='Report run status of cloud init') + prog="status", description="Report run status of cloud init" + ) parser.add_argument( - '-l', '--long', action='store_true', default=False, - help=('Report long format of statuses including run stage name and' - ' error messages')) + "-l", + "--long", + action="store_true", + default=False, + help=( + "Report long format of statuses including run stage name and" + " error messages" + ), + ) parser.add_argument( - '-w', '--wait', action='store_true', default=False, - help='Block waiting on cloud-init to complete') + "-w", + "--wait", + action="store_true", + default=False, + help="Block waiting on cloud-init to complete", + ) return parser @@ -51,23 +67,20 @@ def handle_status_args(name, args): # Read configured paths init = Init(ds_deps=[]) init.read_cfg() - - status, status_detail, time = _get_status_details(init.paths) + status, status_detail, time = get_status_details(init.paths) if args.wait: - while status in (STATUS_ENABLED_NOT_RUN, STATUS_RUNNING): - sys.stdout.write('.') + while status in (UXAppStatus.NOT_RUN, UXAppStatus.RUNNING): + sys.stdout.write(".") sys.stdout.flush() - status, status_detail, time = _get_status_details(init.paths) + status, status_detail, time = get_status_details(init.paths) sleep(0.25) - sys.stdout.write('\n') + sys.stdout.write("\n") + print("status: {0}".format(status.value)) if args.long: - print('status: {0}'.format(status)) if time: - print('time: {0}'.format(time)) - print('detail:\n{0}'.format(status_detail)) - else: - print('status: {0}'.format(status)) - return 1 if status == STATUS_ERROR else 0 + print("time: {0}".format(time)) + print("detail:\n{0}".format(status_detail)) + return 1 if status == UXAppStatus.ERROR else 0 def _is_cloudinit_disabled(disable_file, paths): @@ -81,83 +94,91 @@ def _is_cloudinit_disabled(disable_file, paths): is_disabled = False cmdline_parts = get_cmdline().split() if not uses_systemd(): - reason = 'Cloud-init enabled on sysvinit' - elif 'cloud-init=enabled' in cmdline_parts: - reason = 'Cloud-init enabled by kernel command line cloud-init=enabled' + reason = "Cloud-init enabled on sysvinit" + elif "cloud-init=enabled" in cmdline_parts: + reason = "Cloud-init enabled by kernel command line cloud-init=enabled" elif os.path.exists(disable_file): is_disabled = True - reason = 'Cloud-init disabled by {0}'.format(disable_file) - elif 'cloud-init=disabled' in cmdline_parts: + reason = "Cloud-init disabled by {0}".format(disable_file) + elif "cloud-init=disabled" in cmdline_parts: is_disabled = True - reason = 'Cloud-init disabled by kernel parameter cloud-init=disabled' - elif not os.path.exists(os.path.join(paths.run_dir, 'enabled')): + reason = "Cloud-init disabled by kernel parameter cloud-init=disabled" + elif os.path.exists(os.path.join(paths.run_dir, "disabled")): is_disabled = True - reason = 'Cloud-init disabled by cloud-init-generator' + reason = "Cloud-init disabled by cloud-init-generator" + elif os.path.exists(os.path.join(paths.run_dir, "enabled")): + reason = "Cloud-init enabled by systemd cloud-init-generator" else: - reason = 'Cloud-init enabled by systemd cloud-init-generator' + reason = "Systemd generator may not have run yet." return (is_disabled, reason) -def _get_status_details(paths): +def get_status_details(paths=None): """Return a 3-tuple of status, status_details and time of last event. @param paths: An initialized cloudinit.helpers.paths object. Values are obtained from parsing paths.run_dir/status.json. """ - status = STATUS_ENABLED_NOT_RUN - status_detail = '' + if not paths: + init = Init(ds_deps=[]) + init.read_cfg() + paths = init.paths + + status = UXAppStatus.NOT_RUN + status_detail = "" status_v1 = {} - status_file = os.path.join(paths.run_dir, 'status.json') - result_file = os.path.join(paths.run_dir, 'result.json') + status_file = os.path.join(paths.run_dir, "status.json") + result_file = os.path.join(paths.run_dir, "result.json") (is_disabled, reason) = _is_cloudinit_disabled( - CLOUDINIT_DISABLED_FILE, paths) + CLOUDINIT_DISABLED_FILE, paths + ) if is_disabled: - status = STATUS_DISABLED + status = UXAppStatus.DISABLED status_detail = reason if os.path.exists(status_file): if not os.path.exists(result_file): - status = STATUS_RUNNING - status_v1 = load_json(load_file(status_file)).get('v1', {}) + status = UXAppStatus.RUNNING + status_v1 = load_json(load_file(status_file)).get("v1", {}) errors = [] latest_event = 0 for key, value in sorted(status_v1.items()): - if key == 'stage': + if key == "stage": if value: - status = STATUS_RUNNING - status_detail = 'Running in stage: {0}'.format(value) - elif key == 'datasource': + status = UXAppStatus.RUNNING + status_detail = "Running in stage: {0}".format(value) + elif key == "datasource": status_detail = value elif isinstance(value, dict): - errors.extend(value.get('errors', [])) - start = value.get('start') or 0 - finished = value.get('finished') or 0 + errors.extend(value.get("errors", [])) + start = value.get("start") or 0 + finished = value.get("finished") or 0 if finished == 0 and start != 0: - status = STATUS_RUNNING + status = UXAppStatus.RUNNING event_time = max(start, finished) if event_time > latest_event: latest_event = event_time if errors: - status = STATUS_ERROR - status_detail = '\n'.join(errors) - elif status == STATUS_ENABLED_NOT_RUN and latest_event > 0: - status = STATUS_DONE + status = UXAppStatus.ERROR + status_detail = "\n".join(errors) + elif status == UXAppStatus.NOT_RUN and latest_event > 0: + status = UXAppStatus.DONE if latest_event: - time = strftime('%a, %d %b %Y %H:%M:%S %z', gmtime(latest_event)) + time = strftime("%a, %d %b %Y %H:%M:%S %z", gmtime(latest_event)) else: - time = '' + time = "" return status, status_detail, time def main(): """Tool to report status of cloud-init.""" parser = get_parser() - sys.exit(handle_status_args('status', parser.parse_args())) + sys.exit(handle_status_args("status", parser.parse_args())) -if __name__ == '__main__': +if __name__ == "__main__": main() # vi: ts=4 expandtab |