src/ceph.in

   1 #!@Python3_EXECUTABLE@
   2 # -*- mode:python -*-
   3 # vim: ts=4 sw=4 smarttab expandtab
   4 #
   5 # Processed in Makefile to add python #! line and version variable
   6 #
   7 #
   8
   9
  10 """
  11 ceph.in becomes ceph, the command-line management tool for Ceph clusters.
  12 This is a replacement for tools/ceph.cc and tools/common.cc.
  13
  14 Copyright (C) 2013 Inktank Storage, Inc.
  15
  16 This is free software; you can redistribute it and/or
  17 modify it under the terms of the GNU General Public
  18 License version 2, as published by the Free Software
  19 Foundation.  See file COPYING.
  20 """
  21
  22 from time import sleep
  23 import grp
  24 import os
  25 import pwd
  26 import re
  27 import shutil
  28 import stat
  29 import sys
  30 import time
  31 import platform
  32
  33 from typing import Dict, List, Sequence, Tuple
  34
  35 try:
  36     input = raw_input
  37 except NameError:
  38     pass
  39
  40 CEPH_GIT_VER = "@CEPH_GIT_VER@"
  41 CEPH_GIT_NICE_VER = "@CEPH_GIT_NICE_VER@"
  42 CEPH_RELEASE = "@CEPH_RELEASE@"
  43 CEPH_RELEASE_NAME = "@CEPH_RELEASE_NAME@"
  44 CEPH_RELEASE_TYPE = "@CEPH_RELEASE_TYPE@"
  45
  46 # priorities from src/common/perf_counters.h
  47 PRIO_CRITICAL = 10
  48 PRIO_INTERESTING = 8
  49 PRIO_USEFUL = 5
  50 PRIO_UNINTERESTING = 2
  51 PRIO_DEBUGONLY = 0
  52
  53 PRIO_DEFAULT = PRIO_INTERESTING
  54
  55 # Make life easier on developers:
  56 # If our parent dir contains CMakeCache.txt and bin/init-ceph,
  57 # assume we're running from a build dir (i.e. src/build/bin/ceph)
  58 # and tweak sys.path and LD_LIBRARY_PATH to use built files.
  59 # Since this involves re-execing, if CEPH_DBG is set in the environment
  60 # re-exec with -mpdb.  Also, if CEPH_DEV is in the env, suppress
  61 # the warning message about the DEVELOPER MODE.
  62
  63 MYPATH = os.path.abspath(__file__)
  64 MYDIR = os.path.dirname(MYPATH)
  65 MYPDIR = os.path.dirname(MYDIR)
  66 DEVMODEMSG = '*** DEVELOPER MODE: setting PATH, PYTHONPATH and LD_LIBRARY_PATH ***'
  67
  68
  69 def add_to_ld_path(path_name, path):
  70     paths = re.split('[ :]', os.environ.get(path_name, ''))
  71     if path in paths:
  72         return 0
  73     else:
  74         paths.insert(0, path)
  75         os.environ[path_name] = ':'.join(paths)
  76         return 1
  77
  78
  79 def respawn_in_path(lib_path, pybind_path, pythonlib_path, asan_lib_path):
  80     if platform.system() == "Darwin":
  81         lib_path_var = "DYLD_LIBRARY_PATH"
  82     else:
  83         lib_path_var = "LD_LIBRARY_PATH"
  84
  85     ld_paths_changed = 0
  86     preload_libcxx = os.environ.get('CEPH_PRELOAD_LIBCXX')
  87     if preload_libcxx:
  88         ld_paths_changed += add_to_ld_path('LD_PRELOAD', preload_libcxx)
  89     if asan_lib_path:
  90         ld_paths_changed += add_to_ld_path('LD_PRELOAD', asan_lib_path)
  91     ld_paths_changed += add_to_ld_path(lib_path_var, lib_path)
  92     if ld_paths_changed > 0:
  93         if "CEPH_DEV" not in os.environ:
  94             print(DEVMODEMSG, file=sys.stderr)
  95         execv_cmd = []
  96         if 'CEPH_DBG' in os.environ:
  97             execv_cmd += ['@Python3_EXECUTABLE@', '-mpdb']
  98         execv_cmd += sys.argv
  99         os.execvp(execv_cmd[0], execv_cmd)
 100     else:
 101         sys.path.insert(0, pybind_path)
 102         sys.path.insert(0, pythonlib_path)
 103
 104
 105 def get_pythonlib_dir():
 106     """Returns the name of a distutils build directory"""
 107     return "lib.{version[0]}".format(version=sys.version_info)
 108
 109
 110 def get_cmake_variables(*names):
 111     vars = dict((name, None) for name in names)
 112     for line in open(os.path.join(MYPDIR, "CMakeCache.txt")):
 113         # parse lines like "WITH_ASAN:BOOL=ON"
 114         for name in names:
 115             if line.startswith("{}:".format(name)):
 116                 type_value = line.split(":")[1].strip()
 117                 t, v = type_value.split("=")
 118                 if t == 'BOOL':
 119                     v = v.upper() in ('TRUE', '1', 'Y', 'YES', 'ON')
 120                 vars[name] = v
 121                 break
 122         if all(vars.values()):
 123             break
 124     return [vars[name] for name in names]
 125
 126
 127 if os.path.exists(os.path.join(MYPDIR, "CMakeCache.txt")) \
 128      and os.path.exists(os.path.join(MYPDIR, "bin/init-ceph")):
 129     src_path, with_asan, asan_lib_path = \
 130         get_cmake_variables("ceph_SOURCE_DIR", "WITH_ASAN", "ASAN_LIBRARY")
 131     if src_path is None:
 132         # Huh, maybe we're not really in a cmake environment?
 133         pass
 134     else:
 135         # Developer mode, but in a cmake build dir instead of the src dir
 136         lib_path = os.path.join(MYPDIR, "lib")
 137         bin_path = os.path.join(MYPDIR, "bin")
 138         pybind_path = os.path.join(src_path, "src", "pybind")
 139         pythonlib_path = os.path.join(lib_path,
 140                                       "cython_modules",
 141                                       get_pythonlib_dir())
 142         respawn_in_path(lib_path, pybind_path, pythonlib_path,
 143                         asan_lib_path if with_asan else None)
 144
 145         if 'PATH' in os.environ and bin_path not in os.environ['PATH']:
 146             os.environ['PATH'] = os.pathsep.join([bin_path, os.environ['PATH']])
 147
 148 import argparse
 149 import errno
 150 import json
 151 import rados
 152 import shlex
 153 import signal
 154 import string
 155 import subprocess
 156
 157 from ceph_argparse import \
 158     concise_sig, descsort_key, parse_json_funcsigs, \
 159     validate_command, find_cmd_target, \
 160     json_command, run_in_thread, Flag
 161
 162 from ceph_daemon import admin_socket, DaemonWatcher, Termsize
 163
 164 # just a couple of globals
 165
 166 verbose = False
 167 cluster_handle = None
 168
 169
 170 def raw_write(buf):
 171     sys.stdout.flush()
 172     sys.stdout.buffer.write(buf)
 173
 174
 175 def osdids():
 176     ret, outbuf, outs = json_command(cluster_handle, prefix='osd ls')
 177     if ret:
 178         raise RuntimeError('Can\'t contact mon for osd list')
 179     return [line.decode('utf-8') for line in outbuf.split(b'\n') if line]
 180
 181
 182 def monids():
 183     ret, outbuf, outs = json_command(cluster_handle, prefix='mon dump',
 184                                      argdict={'format': 'json'})
 185     if ret:
 186         raise RuntimeError('Can\'t contact mon for mon list')
 187     d = json.loads(outbuf.decode('utf-8'))
 188     return [m['name'] for m in d['mons']]
 189
 190
 191 def mdsids():
 192     ret, outbuf, outs = json_command(cluster_handle, prefix='fs dump',
 193                                      argdict={'format': 'json'})
 194     if ret:
 195         raise RuntimeError('Can\'t contact mon for mds list')
 196     d = json.loads(outbuf.decode('utf-8'))
 197     l = []
 198     for info in d['standbys']:
 199         l.append(info['name'])
 200     for fs in d['filesystems']:
 201         for info in fs['mdsmap']['info'].values():
 202             l.append(info['name'])
 203     return l
 204
 205
 206 def mgrids():
 207     ret, outbuf, outs = json_command(cluster_handle, prefix='mgr dump',
 208                                      argdict={'format': 'json'})
 209     if ret:
 210         raise RuntimeError('Can\'t contact mon for mgr list')
 211
 212     d = json.loads(outbuf.decode('utf-8'))
 213     l = []
 214     l.append(d['active_name'])
 215     # we can only send tell commands to the active mgr
 216     #for i in d['standbys']:
 217     #    l.append(i['name'])
 218     return l
 219
 220
 221 def ids_by_service(service):
 222     ids = {"mon": monids,
 223            "osd": osdids,
 224            "mds": mdsids,
 225            "mgr": mgrids}
 226     return ids[service]()
 227
 228
 229 def validate_target(target):
 230     """
 231       this function will return true iff target is a correct
 232       target, such as mon.a/osd.2/mds.a/mgr.
 233
 234       target: array, likes ['osd', '2']
 235       return: bool, or raise RuntimeError
 236     """
 237
 238     if len(target) == 2:
 239         # for case "service.id"
 240         service_name, service_id = target[0], target[1]
 241         try:
 242             exist_ids = ids_by_service(service_name)
 243         except KeyError:
 244             print('WARN: {0} is not a legal service name, should be one of mon/osd/mds/mgr'.format(service_name),
 245                   file=sys.stderr)
 246             return False
 247
 248         if service_id in exist_ids or len(exist_ids) > 0 and service_id == '*':
 249             return True
 250         else:
 251             print('WARN: the service id you provided does not exist. service id should '
 252                   'be one of {0}.'.format('/'.join(exist_ids)), file=sys.stderr)
 253             return False
 254
 255     elif len(target) == 1 and target[0] in ['mgr', 'mon']:
 256         return True
 257     else:
 258         print('WARN: \"{0}\" is not a legal target. it should be one of mon.<id>/osd.<int>/mds.<id>/mgr'.format('.'.join(target)), file=sys.stderr)
 259         return False
 260
 261
 262 # these args must be passed to all child programs
 263 GLOBAL_ARGS = {
 264     'client_id': '--id',
 265     'client_name': '--name',
 266     'cluster': '--cluster',
 267     'cephconf': '--conf',
 268 }
 269
 270
 271 def parse_cmdargs(args=None, target='') -> Tuple[argparse.ArgumentParser,
 272                                                  argparse.Namespace,
 273                                                  List[str]]:
 274     """
 275     Consume generic arguments from the start of the ``args``
 276     list.  Call this first to handle arguments that are not
 277     handled by a command description provided by the server.
 278
 279     :returns: three tuple of ArgumentParser instance, Namespace instance
 280               containing parsed values, and list of un-handled arguments
 281     """
 282     # alias: let the line-wrapping be sane
 283     AP = argparse.ArgumentParser
 284
 285     # format our own help
 286     parser = AP(description='Ceph administration tool', add_help=False)
 287
 288     parser.add_argument('--completion', action='store_true',
 289                         help=argparse.SUPPRESS)
 290
 291     parser.add_argument('-h', '--help', help='request mon help',
 292                         action='store_true')
 293
 294     parser.add_argument('-c', '--conf', dest='cephconf',
 295                         help='ceph configuration file')
 296     parser.add_argument('-i', '--in-file', dest='input_file',
 297                         help='input file, or "-" for stdin')
 298     parser.add_argument('-o', '--out-file', dest='output_file',
 299                         help='output file, or "-" for stdout')
 300     parser.add_argument('--setuser', dest='setuser',
 301                         help='set user file permission')
 302     parser.add_argument('--setgroup', dest='setgroup',
 303                         help='set group file permission')
 304     parser.add_argument('--id', '--user', dest='client_id',
 305                         help='client id for authentication')
 306     parser.add_argument('--name', '-n', dest='client_name',
 307                         help='client name for authentication')
 308     parser.add_argument('--cluster', help='cluster name')
 309
 310     parser.add_argument('--admin-daemon', dest='admin_socket',
 311                         help='submit admin-socket command (e.g. "help\" for' \
 312                         'a list of available commands)')
 313
 314     parser.add_argument('-s', '--status', action='store_true',
 315                         help='show cluster status')
 316
 317     parser.add_argument('-w', '--watch', action='store_true',
 318                         help='watch live cluster changes')
 319     parser.add_argument('--watch-debug', action='store_true',
 320                         help='watch debug events')
 321     parser.add_argument('--watch-info', action='store_true',
 322                         help='watch info events')
 323     parser.add_argument('--watch-sec', action='store_true',
 324                         help='watch security events')
 325     parser.add_argument('--watch-warn', action='store_true',
 326                         help='watch warn events')
 327     parser.add_argument('--watch-error', action='store_true',
 328                         help='watch error events')
 329
 330     parser.add_argument('-W', '--watch-channel', dest="watch_channel",
 331                         help="watch live cluster changes on a specific channel "
 332                         "(e.g., cluster, audit, cephadm, or '*' for all)")
 333
 334     parser.add_argument('--version', '-v', action="store_true", help="display version")
 335     parser.add_argument('--verbose', action="store_true", help="make verbose")
 336     parser.add_argument('--concise', dest='verbose', action="store_false",
 337                         help="make less verbose")
 338
 339     parser.add_argument('-f', '--format', choices=['json', 'json-pretty',
 340                         'xml', 'xml-pretty', 'plain', 'yaml'],
 341                         help="Note: yaml is only valid for orch commands", dest='output_format')
 342
 343     parser.add_argument('--connect-timeout', dest='cluster_timeout',
 344                         type=int,
 345                         help='set a timeout for connecting to the cluster')
 346
 347     parser.add_argument('--block', action='store_true',
 348                         help='block until completion (scrub and deep-scrub only)')
 349     parser.add_argument('--period', '-p', default=1, type=float,
 350                         help='polling period, default 1.0 second (for ' \
 351                         'polling commands only)')
 352
 353     # returns a Namespace with the parsed args, and a list of all extras
 354     parsed_args, extras = parser.parse_known_args(args)
 355
 356     return parser, parsed_args, extras
 357
 358
 359 def hdr(s):
 360     print('\n', s, '\n', '=' * len(s))
 361
 362
 363 def do_basic_help(parser, args):
 364     """
 365     Print basic parser help
 366     If the cluster is available, get and print monitor help
 367     """
 368     hdr('General usage:')
 369     parser.print_help()
 370     print_locally_handled_command_help()
 371
 372
 373 def print_locally_handled_command_help():
 374     hdr("Local commands:")
 375     print("""
 376 ping <mon.id>           Send simple presence/life test to a mon
 377                         <mon.id> may be 'mon.*' for all mons
 378 daemon {type.id|path} <cmd>
 379                         Same as --admin-daemon, but auto-find admin socket
 380 daemonperf {type.id | path} [stat-pats] [priority] [<interval>] [<count>]
 381 daemonperf {type.id | path} list|ls [stat-pats] [priority]
 382                         Get selected perf stats from daemon/admin socket
 383                         Optional shell-glob comma-delim match string stat-pats
 384                         Optional selection priority (can abbreviate name):
 385                          critical, interesting, useful, noninteresting, debug
 386                         List shows a table of all available stats
 387                         Run <count> times (default forever),
 388                          once per <interval> seconds (default 1)
 389     """, file=sys.stdout)
 390
 391
 392 def do_extended_help(parser, args, target, partial) -> int:
 393     def help_for_sigs(sigs, partial=None):
 394         try:
 395             while True:
 396                 out = format_help(parse_json_funcsigs(sigs, 'cli'),
 397                                   partial=partial)
 398                 if not out and partial:
 399                     # shorten partial until we get at least one matching command prefix
 400                     partial = ' '.join(partial.split()[:-1])
 401                     continue
 402                 sys.stdout.write(out)
 403                 break
 404         except BrokenPipeError:
 405             pass
 406
 407     def help_for_target(target, partial=None):
 408         # wait for osdmap because we know this is sent after the mgrmap
 409         # and monmap (it's alphabetical).
 410         cluster_handle.wait_for_latest_osdmap()
 411         ret, outbuf, outs = json_command(cluster_handle, target=target,
 412                                          prefix='get_command_descriptions',
 413                                          timeout=10)
 414         if ret:
 415             if (ret == -errno.EPERM or ret == -errno.EACCES) and target[0] in ('osd', 'mds'):
 416                 print("Permission denied.  Check that your user has 'allow *' "
 417                       "capabilities for the target daemon type.", file=sys.stderr)
 418             elif ret == -errno.EPERM:
 419                 print("Permission denied.  Check your user has proper "
 420                       "capabilities configured", file=sys.stderr)
 421             else:
 422                 print("couldn't get command descriptions for {0}: {1} ({2})".
 423                       format(target, outs, ret), file=sys.stderr)
 424             return ret
 425         else:
 426             return help_for_sigs(outbuf.decode('utf-8'), partial)
 427
 428     assert(cluster_handle.state == "connected")
 429     return help_for_target(target, partial)
 430
 431 DONTSPLIT = string.ascii_letters + '{[<>]}'
 432
 433
 434 def wrap(s, width, indent):
 435     """
 436     generator to transform s into a sequence of strings width or shorter,
 437     for wrapping text to a specific column width.
 438     Attempt to break on anything but DONTSPLIT characters.
 439     indent is amount to indent 2nd-through-nth lines.
 440
 441     so "long string long string long string" width=11 indent=1 becomes
 442     'long string', ' long string', ' long string' so that it can be printed
 443     as
 444     long string
 445      long string
 446      long string
 447
 448     Consumes s.
 449     """
 450     result = ''
 451     leader = ''
 452     while len(s):
 453
 454         if len(s) <= width:
 455             # no splitting; just possibly indent
 456             result = leader + s
 457             s = ''
 458             yield result
 459
 460         else:
 461             splitpos = width
 462             while (splitpos > 0) and (s[splitpos-1] in DONTSPLIT):
 463                 splitpos -= 1
 464
 465             if splitpos == 0:
 466                 splitpos = width
 467
 468             if result:
 469                 # prior result means we're mid-iteration, indent
 470                 result = leader
 471             else:
 472                 # first time, set leader and width for next
 473                 leader = ' ' * indent
 474                 width -= 1      # for subsequent space additions
 475
 476             # remove any leading spaces in this chunk of s
 477             result += s[:splitpos].lstrip()
 478             s = s[splitpos:]
 479
 480             yield result
 481
 482
 483 def format_help(cmddict, partial=None) -> str:
 484     """
 485     Formats all the cmdsigs and helptexts from cmddict into a sorted-by-
 486     cmdsig 2-column display, with each column wrapped and indented to
 487     fit into (terminal_width / 2) characters.
 488     """
 489
 490     fullusage = ''
 491     for cmd in sorted(cmddict.values(), key=descsort_key):
 492
 493         if not cmd['help']:
 494             continue
 495         flags = cmd.get('flags', 0)
 496         if flags & (Flag.OBSOLETE | Flag.DEPRECATED | Flag.HIDDEN):
 497             continue
 498         concise = concise_sig(cmd['sig'])
 499         if partial and not concise.startswith(partial):
 500             continue
 501         width = Termsize().cols - 1  # 1 for the line between sig and help
 502         sig_width = int(width / 2)
 503         # make sure width == sig_width + help_width, even (width % 2 > 0)
 504         help_width = int(width / 2) + (width % 2)
 505         siglines = [l for l in wrap(concise, sig_width, 1)]
 506         helplines = [l for l in wrap(cmd['help'], help_width, 1)]
 507
 508         # make lists the same length
 509         maxlen = max(len(siglines), len(helplines))
 510         siglines.extend([''] * (maxlen - len(siglines)))
 511         helplines.extend([''] * (maxlen - len(helplines)))
 512
 513         # so we can zip them for output
 514         for s, h in zip(siglines, helplines):
 515             fullusage += '{s:{w}s} {h}\n'.format(s=s, h=h, w=sig_width)
 516
 517     return fullusage
 518
 519
 520 def ceph_conf(parsed_args, field, name, pid=None):
 521     cmd = 'ceph-conf'
 522     bindir = os.path.dirname(__file__)
 523     if shutil.which(cmd):
 524         args = [cmd]
 525     elif shutil.which(cmd, path=bindir):
 526         args = [os.path.join(bindir, cmd)]
 527     else:
 528         raise RuntimeError('"ceph-conf" not found')
 529
 530     if name:
 531         args.extend(['--name', name])
 532     if pid:
 533         args.extend(['--pid', pid])
 534
 535     # add any args in GLOBAL_ARGS
 536     for key, val in GLOBAL_ARGS.items():
 537         # ignore name in favor of argument name, if any
 538         if name and key == 'client_name':
 539             continue
 540         if getattr(parsed_args, key):
 541             args.extend([val, getattr(parsed_args, key)])
 542
 543     args.extend(['--show-config-value', field])
 544     p = subprocess.Popen(
 545         args,
 546         stdout=subprocess.PIPE,
 547         stderr=subprocess.PIPE)
 548     outdata, errdata = p.communicate()
 549     if p.returncode != 0:
 550         raise RuntimeError('unable to get conf option %s for %s: %s' % (field, name, errdata))
 551     return outdata.rstrip()
 552
 553
 554 PROMPT = 'ceph> '
 555
 556 if sys.stdin.isatty():
 557     def read_input():
 558         while True:
 559             line = input(PROMPT).rstrip()
 560             if line in ['q', 'quit', 'Q', 'exit']:
 561                 return None
 562             if line:
 563                 return line
 564 else:
 565     def read_input():
 566         while True:
 567             line = sys.stdin.readline()
 568             if not line:
 569                 return None
 570             line = line.rstrip()
 571             if line:
 572                 return line
 573
 574
 575 def do_command(parsed_args, target, cmdargs, sigdict, inbuf, verbose):
 576     ''' Validate a command, and handle the polling flag '''
 577
 578     valid_dict = validate_command(sigdict, cmdargs, verbose)
 579     # Validate input args against list of sigs
 580     if valid_dict:
 581         if parsed_args.output_format:
 582             valid_dict['format'] = parsed_args.output_format
 583         if verbose:
 584             print("Submitting command: ", valid_dict, file=sys.stderr)
 585     else:
 586         return -errno.EINVAL, '', 'invalid command'
 587
 588     next_header_print = 0
 589     # Set extra options for polling commands only:
 590     if valid_dict.get('poll', False):
 591         valid_dict['width'] = Termsize().cols
 592     while True:
 593         try:
 594             # Only print the header for polling commands
 595             if next_header_print == 0 and valid_dict.get('poll', False):
 596                 valid_dict['print_header'] = True
 597                 next_header_print = Termsize().rows - 3
 598             next_header_print -= 1
 599             ret, outbuf, outs = json_command(cluster_handle, target=target,
 600                 argdict=valid_dict, inbuf=inbuf, verbose=verbose)
 601             if valid_dict.get('poll', False):
 602                 valid_dict['print_header'] = False
 603             if not valid_dict.get('poll', False):
 604                 # Don't print here if it's not a polling command
 605                 break
 606             if ret:
 607                 ret = abs(ret)
 608                 print('Error: {0} {1}'.format(ret, errno.errorcode.get(ret, 'Unknown')),
 609                         file=sys.stderr)
 610                 break
 611             if outbuf:
 612                 print(outbuf.decode('utf-8'))
 613             if outs:
 614                 print(outs, file=sys.stderr)
 615             if parsed_args.period <= 0:
 616                 break
 617             sleep(parsed_args.period)
 618         except KeyboardInterrupt:
 619             print('Interrupted')
 620             return errno.EINTR, '', ''
 621     if ret == errno.ETIMEDOUT:
 622         ret = -ret
 623         if not outs:
 624             outs = ("Connection timed out. Please check the client's " +
 625                     "permission and connection.")
 626     return ret, outbuf, outs
 627
 628
 629 def new_style_command(parsed_args,
 630                       cmdargs,
 631                       target,
 632                       sigdict,
 633                       inbuf, verbose) -> Tuple[int, bytes, str]:
 634     """
 635     Do new-style command dance.
 636     target: daemon to receive command: mon (any) or osd.N
 637     sigdict - the parsed output from the new monitor describing commands
 638     inbuf - any -i input file data
 639     verbose - bool
 640     """
 641     if verbose:
 642         for cmdtag in sorted(sigdict.keys()):
 643             cmd = sigdict[cmdtag]
 644             sig = cmd['sig']
 645             print('{0}: {1}'.format(cmdtag, concise_sig(sig)))
 646
 647     if cmdargs:
 648         # Non interactive mode
 649         ret, outbuf, outs = do_command(parsed_args, target, cmdargs, sigdict, inbuf, verbose)
 650     else:
 651         # Interactive mode (ceph cli)
 652         if sys.stdin.isatty():
 653             # do the command-interpreter looping
 654             # for input to do readline cmd editing
 655             import readline  # noqa
 656
 657         while True:
 658             try:
 659                 interactive_input = read_input()
 660             except EOFError:
 661                 # leave user an uncluttered prompt
 662                 return 0, b'\n', ''
 663             if interactive_input is None:
 664                 return 0, b'', ''
 665             cmdargs = parse_cmdargs(shlex.split(interactive_input))[2]
 666             try:
 667                 target = find_cmd_target(cmdargs)
 668             except Exception as e:
 669                 print('error handling command target: {0}'.format(e),
 670                       file=sys.stderr)
 671                 continue
 672             if len(cmdargs) and cmdargs[0] == 'tell':
 673                 print('Can not use \'tell\' in interactive mode.',
 674                       file=sys.stderr)
 675                 continue
 676             ret, outbuf, outs = do_command(parsed_args, target, cmdargs,
 677                                            sigdict, inbuf, verbose)
 678             if ret < 0:
 679                 ret = -ret
 680                 errstr = errno.errorcode.get(ret, 'Unknown')
 681                 print('Error {0}: {1}'.format(errstr, outs), file=sys.stderr)
 682             else:
 683                 if outs:
 684                     print(outs, file=sys.stderr)
 685                 if outbuf:
 686                     print(outbuf.decode('utf-8'))
 687
 688     return ret, outbuf, outs
 689
 690
 691 def complete(sigdict, args, target):
 692     """
 693     Command completion.  Match as much of [args] as possible,
 694     and print every possible match separated by newlines.
 695     Return exitcode.
 696     """
 697     # XXX this looks a lot like the front of validate_command().  Refactor?
 698
 699     # Repulsive hack to handle tell: lop off 'tell' and target
 700     # and validate the rest of the command.  'target' is already
 701     # determined in our callers, so it's ok to remove it here.
 702     if len(args) and args[0] == 'tell':
 703         args = args[2:]
 704     # look for best match, accumulate possibles in bestcmds
 705     # (so we can maybe give a more-useful error message)
 706
 707     match_count = 0
 708     comps = []
 709     for cmdtag, cmd in sigdict.items():
 710         flags = cmd.get('flags', 0)
 711         if flags & (Flag.OBSOLETE | Flag.HIDDEN):
 712             continue
 713         sig = cmd['sig']
 714         j = 0
 715         # iterate over all arguments, except last one
 716         for arg in args[0:-1]:
 717             if j > len(sig)-1:
 718                 # an out of argument definitions
 719                 break
 720             found_match = arg in sig[j].complete(arg)
 721             if not found_match and sig[j].req:
 722                 # no elements that match
 723                 break
 724             if not sig[j].N:
 725                 j += 1
 726         else:
 727             # successfully matched all - except last one - arguments
 728             if j < len(sig) and len(args) > 0:
 729                 comps += sig[j].complete(args[-1])
 730
 731             match_count += 1
 732             match_cmd = cmd
 733
 734     if match_count == 1 and len(comps) == 0:
 735         # only one command matched and no hints yet => add help
 736         comps = comps + [' ', '#'+match_cmd['help']]
 737     print('\n'.join(sorted(set(comps))))
 738     return 0
 739
 740
 741 def ping_monitor(cluster_handle, name, timeout):
 742     if 'mon.' not in name:
 743         print('"ping" expects a monitor to ping; try "ping mon.<id>"', file=sys.stderr)
 744         return 1
 745
 746     mon_id = name[len('mon.'):]
 747     if mon_id == '*':
 748         run_in_thread(cluster_handle.connect, timeout=timeout)
 749         for m in monids():
 750             s = run_in_thread(cluster_handle.ping_monitor, m)
 751             if s is None:
 752                 print("mon.{0}".format(m) + '\n' + "Error connecting to monitor.")
 753             else:
 754                 print("mon.{0}".format(m) + '\n' + s)
 755     else:
 756             s = run_in_thread(cluster_handle.ping_monitor, mon_id)
 757             print(s)
 758     return 0
 759
 760
 761 def get_admin_socket(parsed_args, name):
 762     path = ceph_conf(parsed_args, 'admin_socket', name)
 763     try:
 764         if stat.S_ISSOCK(os.stat(path).st_mode):
 765             return path
 766     except OSError:
 767         pass
 768     # try harder, probably the "name" option is in the form of
 769     # "${name}.${pid}"?
 770     parts = name.rsplit('.', 1)
 771     if len(parts) > 1 and parts[-1].isnumeric():
 772         name, pid = parts
 773         return ceph_conf(parsed_args, 'admin_socket', name, pid)
 774     else:
 775         return path
 776
 777
 778 def maybe_daemon_command(parsed_args, childargs):
 779     """
 780     Check if --admin-socket, daemon, or daemonperf command
 781     if it is, returns (boolean handled, return code if handled == True)
 782     """
 783
 784     daemon_perf = False
 785     sockpath = None
 786     if parsed_args.admin_socket:
 787         sockpath = parsed_args.admin_socket
 788     elif len(childargs) > 0 and childargs[0] in ["daemon", "daemonperf"]:
 789         daemon_perf = (childargs[0] == "daemonperf")
 790         # Treat "daemon <path>" or "daemon <name>" like --admin_daemon <path>
 791         # Handle "daemonperf <path>" the same but requires no trailing args
 792         require_args = 2 if daemon_perf else 3
 793         if len(childargs) >= require_args:
 794             if childargs[1].find('/') >= 0:
 795                 sockpath = childargs[1]
 796             else:
 797                 # try resolve daemon name
 798                 try:
 799                     sockpath = get_admin_socket(parsed_args, childargs[1])
 800                 except Exception as e:
 801                     print('Can\'t get admin socket path: ' + str(e), file=sys.stderr)
 802                     return True, errno.EINVAL
 803             # for both:
 804             childargs = childargs[2:]
 805         else:
 806             print('{0} requires at least {1} arguments'.format(childargs[0], require_args),
 807                   file=sys.stderr)
 808             return True, errno.EINVAL
 809
 810     if sockpath and daemon_perf:
 811         return True, daemonperf(childargs, sockpath)
 812     elif sockpath:
 813         try:
 814             raw_write(admin_socket(sockpath, childargs, parsed_args.output_format))
 815         except Exception as e:
 816             print('admin_socket: {0}'.format(e), file=sys.stderr)
 817             return True, errno.EINVAL
 818         return True, 0
 819
 820     return False, 0
 821
 822
 823 def isnum(s):
 824     try:
 825         float(s)
 826         return True
 827     except ValueError:
 828         return False
 829
 830
 831 def daemonperf(childargs: Sequence[str], sockpath: str):
 832     """
 833     Handle daemonperf command; returns errno or 0
 834
 835     daemonperf <daemon> [priority string] [statpats] [interval] [count]
 836     daemonperf <daemon> list|ls [statpats]
 837     """
 838
 839     interval = 1
 840     count = None
 841     statpats = None
 842     priority = None
 843     do_list = False
 844
 845     def prio_from_name(arg):
 846
 847         PRIOMAP = {
 848             'critical': PRIO_CRITICAL,
 849             'interesting': PRIO_INTERESTING,
 850             'useful': PRIO_USEFUL,
 851             'uninteresting': PRIO_UNINTERESTING,
 852             'debugonly': PRIO_DEBUGONLY,
 853         }
 854
 855         if arg in PRIOMAP:
 856             return PRIOMAP[arg]
 857         # allow abbreviation
 858         for name, val in PRIOMAP.items():
 859             if name.startswith(arg):
 860                 return val
 861         return None
 862
 863     # consume and analyze non-numeric args
 864     while len(childargs) and not isnum(childargs[0]):
 865         arg = childargs.pop(0)
 866         # 'list'?
 867         if arg in ['list', 'ls']:
 868             do_list = True
 869             continue
 870         # prio?
 871         prio = prio_from_name(arg)
 872         if prio is not None:
 873             priority = prio
 874             continue
 875         # statpats
 876         statpats = arg.split(',')
 877
 878     if priority is None:
 879         priority = PRIO_DEFAULT
 880
 881     if len(childargs) > 0:
 882         try:
 883             interval = float(childargs.pop(0))
 884             if interval < 0:
 885                 raise ValueError
 886         except ValueError:
 887             print('daemonperf: interval should be a positive number', file=sys.stderr)
 888             return errno.EINVAL
 889
 890     if len(childargs) > 0:
 891         arg = childargs.pop(0)
 892         if (not isnum(arg)) or (int(arg) < 0):
 893             print('daemonperf: count should be a positive integer', file=sys.stderr)
 894             return errno.EINVAL
 895         count = int(arg)
 896
 897     watcher = DaemonWatcher(sockpath, statpats, priority)
 898     if do_list:
 899         watcher.list()
 900     else:
 901         watcher.run(interval, count)
 902
 903     return 0
 904
 905
 906 def get_scrub_timestamps(childargs: Sequence[str]) -> Dict[str,
 907                                                            Tuple[str, str]]:
 908     last_scrub_stamp = "last_" + childargs[1].replace('-', '_') + "_stamp"
 909     results = dict()
 910     scruball = False
 911     if childargs[2] in ['all', 'any', '*']:
 912         scruball = True
 913     devnull = open(os.devnull, 'w')
 914     out = subprocess.check_output(['ceph', 'pg', 'dump', '--format=json-pretty'],
 915                                   stderr=devnull)
 916     try:
 917         pgstats = json.loads(out)['pg_map']['pg_stats']
 918     except KeyError:
 919         pgstats = json.loads(out)['pg_stats']
 920     for stat in pgstats:
 921         if scruball or stat['up_primary'] == int(childargs[2]):
 922             scrub_tuple = (stat['up_primary'], stat[last_scrub_stamp])
 923             results[stat['pgid']] = scrub_tuple
 924     return results
 925
 926
 927 def check_scrub_stamps(waitdata, currdata):
 928     for pg in waitdata.keys():
 929        # Try to handle the case where a pg may not exist in current results
 930        if pg in currdata and waitdata[pg][1] == currdata[pg][1]:
 931            return False
 932     return True
 933
 934
 935 def waitscrub(childargs, waitdata):
 936     print('Waiting for {0} to complete...'.format(childargs[1]), file=sys.stdout)
 937     currdata = get_scrub_timestamps(childargs)
 938     while not check_scrub_stamps(waitdata, currdata):
 939         time.sleep(3)
 940         currdata = get_scrub_timestamps(childargs)
 941     print('{0} completed'.format(childargs[1]), file=sys.stdout)
 942
 943
 944 def wait(childargs: Sequence[str], waitdata):
 945     if childargs[1] in ['scrub', 'deep-scrub']:
 946         waitscrub(childargs, waitdata)
 947
 948
 949 def main():
 950     ceph_args = os.environ.get('CEPH_ARGS')
 951     if ceph_args:
 952         if "injectargs" in sys.argv:
 953             i = sys.argv.index("injectargs")
 954             sys.argv = sys.argv[:i] + ceph_args.split() + sys.argv[i:]
 955         else:
 956             sys.argv.extend([arg for arg in ceph_args.split()
 957                              if '--admin-socket' not in arg])
 958     parser, parsed_args, childargs = parse_cmdargs()
 959
 960     if parsed_args.version:
 961         print('ceph version {0} ({1}) {2} ({3})'.format(
 962             CEPH_GIT_NICE_VER,
 963             CEPH_GIT_VER,
 964             CEPH_RELEASE_NAME,
 965             CEPH_RELEASE_TYPE))  # noqa
 966         return 0
 967
 968     # --watch-channel|-W implies -w
 969     if parsed_args.watch_channel:
 970         parsed_args.watch = True
 971     elif parsed_args.watch and not parsed_args.watch_channel:
 972         parsed_args.watch_channel = 'cluster'
 973
 974     global verbose
 975     verbose = parsed_args.verbose
 976
 977     if verbose:
 978         print("parsed_args: {0}, childargs: {1}".format(parsed_args, childargs), file=sys.stderr)
 979
 980     # pass on --id, --name, --conf
 981     name = 'client.admin'
 982     if parsed_args.client_id:
 983         name = 'client.' + parsed_args.client_id
 984     if parsed_args.client_name:
 985         name = parsed_args.client_name
 986
 987     conffile = rados.Rados.DEFAULT_CONF_FILES
 988     if parsed_args.cephconf:
 989         conffile = parsed_args.cephconf
 990     # For now, --admin-daemon is handled as usual.  Try it
 991     # first in case we can't connect() to the cluster
 992
 993     done, ret = maybe_daemon_command(parsed_args, childargs)
 994     if done:
 995         return ret
 996
 997     timeout = None
 998     if parsed_args.cluster_timeout:
 999         timeout = parsed_args.cluster_timeout
1000
1001     # basic help
1002     if parsed_args.help:
1003         do_basic_help(parser, childargs)
1004
1005     # handle any 'generic' ceph arguments that we didn't parse here
1006     global cluster_handle
1007
1008     # rados.Rados() will call rados_create2, and then read the conf file,
1009     # and then set the keys from the dict.  So we must do these
1010     # "pre-file defaults" first (see common_preinit in librados)
1011     conf_defaults = {
1012         'log_to_stderr': 'true',
1013         'err_to_stderr': 'true',
1014         'log_flush_on_exit': 'true',
1015     }
1016
1017     if 'injectargs' in childargs:
1018         position = childargs.index('injectargs')
1019         injectargs = childargs[position:]
1020         childargs = childargs[:position]
1021         if verbose:
1022             print('Separate childargs {0} from injectargs {1}'.format(childargs, injectargs),
1023                   file=sys.stderr)
1024     else:
1025         injectargs = None
1026
1027     clustername = None
1028     if parsed_args.cluster:
1029         clustername = parsed_args.cluster
1030
1031     try:
1032         cluster_handle = run_in_thread(rados.Rados,
1033                                        name=name, clustername=clustername,
1034                                        conf_defaults=conf_defaults,
1035                                        conffile=conffile)
1036         retargs = run_in_thread(cluster_handle.conf_parse_argv, childargs)
1037     except rados.Error as e:
1038         print('Error initializing cluster client: {0!r}'.format(e), file=sys.stderr)
1039         return 1
1040
1041     childargs = retargs
1042     if not childargs:
1043         childargs = []
1044
1045     # -- means "stop parsing args", but we don't want to see it either
1046     if '--' in childargs:
1047         childargs.remove('--')
1048     if injectargs and '--' in injectargs:
1049         injectargs.remove('--')
1050
1051     block = False
1052     waitdata = dict()
1053     if parsed_args.block:
1054         if (len(childargs) >= 2 and
1055                 childargs[0] == 'osd' and
1056                 childargs[1] in ['deep-scrub', 'scrub']):
1057             block = True
1058             waitdata = get_scrub_timestamps(childargs)
1059
1060     if parsed_args.help:
1061         # short default timeout for -h
1062         if not timeout:
1063             timeout = 5
1064
1065     if childargs and childargs[0] == 'ping' and not parsed_args.help:
1066         if len(childargs) < 2:
1067             print('"ping" requires a monitor name as argument: "ping mon.<id>"', file=sys.stderr)
1068             return 1
1069     if parsed_args.completion:
1070         # for completion let timeout be really small
1071         timeout = 3
1072     try:
1073         if childargs and childargs[0] == 'ping' and not parsed_args.help:
1074             return ping_monitor(cluster_handle, childargs[1], timeout)
1075         result = run_in_thread(cluster_handle.connect, timeout=timeout)
1076         if type(result) is tuple and result[0] == -errno.EINTR:
1077             print('Cluster connection interrupted or timed out', file=sys.stderr)
1078             return 1
1079     except KeyboardInterrupt:
1080         print('Cluster connection aborted', file=sys.stderr)
1081         return 1
1082     except rados.PermissionDeniedError as e:
1083         print(str(e), file=sys.stderr)
1084         return errno.EACCES
1085     except Exception as e:
1086         print(str(e), file=sys.stderr)
1087         return 1
1088
1089     if parsed_args.help:
1090         target = None
1091         if len(childargs) >= 2 and childargs[0] == 'tell':
1092             target = childargs[1].split('.', 1)
1093             if not validate_target(target):
1094                 print('target {0} doesn\'t exist; please pass correct target to tell command (e.g., mon.a, osd.1, mds.a, mgr)'.format(childargs[1]), file=sys.stderr)
1095                 return 1
1096             childargs = childargs[2:]
1097             hdr('Tell %s commands:' % target[0])
1098         else:
1099             hdr('Monitor commands:')
1100             target = ('mon', '')
1101         if verbose:
1102             print('[Contacting monitor, timeout after %d seconds]' % timeout)
1103
1104         return do_extended_help(parser, childargs, target, ' '.join(childargs))
1105
1106     # implement "tell service.id help"
1107     if len(childargs) >= 3 and childargs[0] == 'tell' and childargs[2] == 'help':
1108         target = childargs[1].split('.', 1)
1109         if validate_target(target):
1110             hdr('Tell %s commands' % target[0])
1111             return do_extended_help(parser, childargs, target, None)
1112         else:
1113             print('target {0} doesn\'t exists, please pass correct target to tell command, such as mon.a/'
1114                   'osd.1/mds.a/mgr'.format(childargs[1]), file=sys.stderr)
1115             return 1
1116
1117     # implement -w/--watch_*
1118     # This is ugly, but Namespace() isn't quite rich enough.
1119     level = ''
1120     for k, v in parsed_args._get_kwargs():
1121         if k.startswith('watch') and v:
1122             if k == 'watch':
1123                 level = 'info'
1124             elif k != "watch_channel":
1125                 level = k.replace('watch_', '')
1126     if level:
1127         # an awfully simple callback
1128         def watch_cb(arg, line, channel, name, who, stamp_sec, stamp_nsec, seq, level, msg):
1129             # Filter on channel
1130             channel = channel.decode('utf-8')
1131             if parsed_args.watch_channel in (channel, '*'):
1132                 print(line.decode('utf-8'))
1133                 sys.stdout.flush()
1134
1135         # first do a ceph status
1136         ret, outbuf, outs = json_command(cluster_handle, prefix='status')
1137         if ret:
1138             print("status query failed: ", outs, file=sys.stderr)
1139             return ret
1140         print(outbuf.decode('utf-8'))
1141
1142         # this instance keeps the watch connection alive, but is
1143         # otherwise unused
1144         run_in_thread(cluster_handle.monitor_log2, level, watch_cb, 0)
1145
1146         # loop forever letting watch_cb print lines
1147         try:
1148             signal.pause()
1149         except KeyboardInterrupt:
1150             # or until ^C, at least
1151             return 0
1152
1153     # read input file, if any
1154     inbuf = b''
1155     if parsed_args.input_file:
1156         try:
1157             if parsed_args.input_file == '-':
1158                 inbuf = sys.stdin.buffer.read()
1159             else:
1160                 with open(parsed_args.input_file, 'rb') as f:
1161                     inbuf = f.read()
1162         except Exception as e:
1163             print('Can\'t open input file {0}: {1}'.format(parsed_args.input_file, e), file=sys.stderr)
1164             return 1
1165
1166     # prepare output file, if any
1167     if parsed_args.output_file:
1168         try:
1169             if parsed_args.output_file == '-':
1170                 outf = sys.stdout.buffer
1171             else:
1172                 outf = open(parsed_args.output_file, 'wb')
1173         except Exception as e:
1174             print('Can\'t open output file {0}: {1}'.format(parsed_args.output_file, e), file=sys.stderr)
1175             return 1
1176         if parsed_args.setuser:
1177             try:
1178                 ownerid = pwd.getpwnam(parsed_args.setuser).pw_uid
1179                 os.fchown(outf.fileno(), ownerid, -1)
1180             except OSError as e:
1181                 print('Failed to change user ownership of {0} to {1}: {2}'.format(outf, parsed_args.setuser, e))
1182                 return 1
1183         if parsed_args.setgroup:
1184             try:
1185                 groupid = grp.getgrnam(parsed_args.setgroup).gr_gid
1186                 os.fchown(outf.fileno(), -1, groupid)
1187             except OSError as e:
1188                 print('Failed to change group ownership of {0} to {1}: {2}'.format(outf, parsed_args.setgroup, e))
1189                 return 1
1190
1191     # -s behaves like a command (ceph status).
1192     if parsed_args.status:
1193         childargs.insert(0, 'status')
1194
1195     try:
1196         target = find_cmd_target(childargs)
1197     except Exception as e:
1198         print('error handling command target: {0}'.format(e), file=sys.stderr)
1199         return 1
1200
1201     # Repulsive hack to handle tell: lop off 'tell' and target
1202     # and validate the rest of the command.  'target' is already
1203     # determined in our callers, so it's ok to remove it here.
1204     is_tell = False
1205     if len(childargs) and childargs[0] == 'tell':
1206         childargs = childargs[2:]
1207         is_tell = True
1208
1209     if is_tell:
1210         if injectargs:
1211             childargs = injectargs
1212         if not len(childargs):
1213             print('"{0} tell" requires additional arguments.'.format(sys.argv[0]),
1214                   'Try "{0} tell <name> <command> [options...]" instead.'.format(sys.argv[0]),
1215                   file=sys.stderr)
1216             return errno.EINVAL
1217
1218     # fetch JSON sigs from command
1219     # each line contains one command signature (a placeholder name
1220     # of the form 'cmdNNN' followed by an array of argument descriptors)
1221     # as part of the validated argument JSON object
1222
1223     if target[1] == '*':
1224         service = target[0]
1225         targets = [(service, o) for o in ids_by_service(service)]
1226     else:
1227         targets = [target]
1228
1229     final_ret = 0
1230     for target in targets:
1231         # prettify?  prefix output with target, if there was a wildcard used
1232         prefix = ''
1233         suffix = ''
1234         if not parsed_args.output_file and len(targets) > 1:
1235             prefix = '{0}.{1}: '.format(*target)
1236             suffix = '\n'
1237
1238         ret, outbuf, outs = json_command(cluster_handle, target=target,
1239                                          prefix='get_command_descriptions')
1240         if ret:
1241             where = '{0}.{1}'.format(*target)
1242             if ret > 0:
1243                 raise RuntimeError('Unexpected return code from {0}: {1}'.
1244                                    format(where, ret))
1245             outs = 'problem getting command descriptions from {0}'.format(where)
1246         else:
1247             sigdict = parse_json_funcsigs(outbuf.decode('utf-8'), 'cli')
1248
1249             if parsed_args.completion:
1250                 return complete(sigdict, childargs, target)
1251
1252             ret, outbuf, outs = new_style_command(parsed_args, childargs,
1253                                                   target, sigdict, inbuf,
1254                                                   verbose)
1255
1256             # debug tool: send any successful command *again* to
1257             # verify that it is idempotent.
1258             if not ret and 'CEPH_CLI_TEST_DUP_COMMAND' in os.environ:
1259                 ret, outbuf, outs = new_style_command(parsed_args, childargs,
1260                                                       target, sigdict, inbuf,
1261                                                       verbose)
1262                 if ret < 0:
1263                     ret = -ret
1264                     print(prefix +
1265                           'Second attempt of previously successful command '
1266                           'failed with {0}: {1}'.format(
1267                               errno.errorcode.get(ret, 'Unknown'), outs),
1268                           file=sys.stderr)
1269
1270         sys.stdout.flush()
1271
1272         if parsed_args.output_file:
1273             outf.write(outbuf)
1274         else:
1275             # hack: old code printed status line before many json outputs
1276             # (osd dump, etc.) that consumers know to ignore.  Add blank line
1277             # to satisfy consumers that skip the first line, but not annoy
1278             # consumers that don't.
1279             if parsed_args.output_format and \
1280                parsed_args.output_format.startswith('json'):
1281                 print()
1282
1283             # if we are prettifying things, normalize newlines.  sigh.
1284             if suffix:
1285                 outbuf = outbuf.rstrip()
1286             if outbuf:
1287                 try:
1288                     print(prefix, end='')
1289                     # Write directly to binary stdout
1290                     raw_write(outbuf)
1291                     print(suffix, end='')
1292                 except IOError as e:
1293                     if e.errno != errno.EPIPE:
1294                         raise e
1295         final_e = None
1296         try:
1297             sys.stdout.flush()
1298         except IOError as e:
1299             if e.errno != errno.EPIPE:
1300                 final_e = e
1301
1302         if ret < 0:
1303             ret = -ret
1304             errstr = errno.errorcode.get(ret, 'Unknown')
1305             print('Error {0}: {1}'.format(errstr, outs), file=sys.stderr)
1306             final_ret = ret
1307         elif outs:
1308             print(prefix + outs, file=sys.stderr)
1309
1310         if final_e:
1311             raise final_e
1312
1313     # Block until command completion (currently scrub and deep scrub only)
1314     if block:
1315         wait(childargs, waitdata)
1316
1317     if parsed_args.output_file and parsed_args.output_file != '-':
1318         outf.close()
1319
1320     if final_ret:
1321         return final_ret
1322
1323     return 0
1324
1325 if __name__ == '__main__':
1326     try:
1327         retval = main()
1328         # shutdown explicitly; Rados() does not
1329         if retval == 0 and cluster_handle:
1330             run_in_thread(cluster_handle.shutdown)
1331     except KeyboardInterrupt:
1332         print('Interrupted')
1333         retval = errno.EINTR
1334
1335     if retval:
1336         # flush explicitly because we aren't exiting in the usual way
1337         sys.stdout.flush()
1338         sys.stderr.flush()
1339         os._exit(retval)
1340     else:
1341         sys.exit(retval)