import time
import platform
+from typing import Dict, List, Sequence, Tuple
+
try:
input = raw_input
except NameError:
}
-def parse_cmdargs(args=None, target=''):
+def parse_cmdargs(args=None, target='') -> Tuple[argparse.ArgumentParser,
+ argparse.Namespace,
+ List[str]]:
"""
Consume generic arguments from the start of the ``args``
list. Call this first to handle arguments that are not
""", file=sys.stdout)
-def do_extended_help(parser, args, target, partial):
+def do_extended_help(parser, args, target, partial) -> int:
def help_for_sigs(sigs, partial=None):
try:
sys.stdout.write(format_help(parse_json_funcsigs(sigs, 'cli'),
yield result
-def format_help(cmddict, partial=None):
+def format_help(cmddict, partial=None) -> str:
"""
Formats all the cmdsigs and helptexts from cmddict into a sorted-by-
cmdsig 2-column display, with each column wrapped and indented to
return ret, outbuf, outs
-def new_style_command(parsed_args, cmdargs, target, sigdict, inbuf, verbose):
+def new_style_command(parsed_args,
+ cmdargs,
+ target,
+ sigdict,
+ inbuf, verbose) -> Tuple[int, bytes, str]:
"""
Do new-style command dance.
target: daemon to receive command: mon (any) or osd.N
return False
-def daemonperf(childargs, sockpath):
+def daemonperf(childargs: Sequence[str], sockpath: str):
"""
Handle daemonperf command; returns errno or 0
return 0
-def get_scrub_timestamps(childargs):
+
+def get_scrub_timestamps(childargs: Sequence[str]) -> Dict[str,
+ Tuple[str, str]]:
last_scrub_stamp = "last_" + childargs[1].replace('-', '_') + "_stamp"
results = dict()
scruball = False
results[stat['pgid']] = scrub_tuple
return results
+
def check_scrub_stamps(waitdata, currdata):
for pg in waitdata.keys():
# Try to handle the case where a pg may not exist in current results
return False
return True
+
def waitscrub(childargs, waitdata):
print('Waiting for {0} to complete...'.format(childargs[1]), file=sys.stdout)
currdata = get_scrub_timestamps(childargs)
currdata = get_scrub_timestamps(childargs)
print('{0} completed'.format(childargs[1]), file=sys.stdout)
-def wait(childargs, waitdata):
+
+def wait(childargs: Sequence[str], waitdata):
if childargs[1] in ['scrub', 'deep-scrub']:
waitscrub(childargs, waitdata)
import threading
import uuid
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
+
+
# Flags are from MonCommand.h
class Flag:
NOFORWARD = (1 << 0)
return cmp(descsort_key(sh1), descsort_key(sh2))
-def parse_funcsig(sig):
+def parse_funcsig(sig: Sequence[Union[str, Dict[str, str]]]) -> List[argdesc]:
"""
parse a single descriptor (array of strings or dicts) into a
dict of function descriptor/validators (objects of CephXXX type)
return newsig
-def parse_json_funcsigs(s, consumer):
+def parse_json_funcsigs(s: str,
+ consumer: str) -> Dict[str, Dict[str, List[argdesc]]]:
"""
A function signature is mostly an array of argdesc; it's represented
in JSON as
return matchcnt
-def store_arg(desc, d):
+ValidatedArgs = Dict[str, Union[bool, int, float, str,
+ Tuple[str, str],
+ Sequence[str]]]
+
+
+def store_arg(desc: argdesc, d: ValidatedArgs):
'''
Store argument described by, and held in, thanks to valid(),
desc into the dictionary d, keyed by desc.name. Three cases:
d[desc.name] = desc.instance.val
-def validate(args, signature, flags=0, partial=False):
+def validate(args: List[str],
+ signature: Sequence[argdesc],
+ flags: Optional[int] = 0,
+ partial: Optional[bool] = False) -> ValidatedArgs:
"""
validate(args, signature, flags=0, partial=False)
return d
-def validate_command(sigdict, args, verbose=False):
+def validate_command(sigdict: Dict[str, Dict[str, Any]],
+ args: Sequence[str],
+ verbose: Optional[bool] = False) -> ValidatedArgs:
"""
Parse positional arguments into a parameter dict, according to
the command descriptions.
return valid_dict
-def find_cmd_target(childargs):
+def find_cmd_target(childargs: List[str]) -> Tuple[str, str]:
"""
Using a minimal validation, figure out whether the command
should be sent to a monitor or an osd. We do this before even
self.exception = e
-def run_in_thread(func, *args, **kwargs):
+def run_in_thread(func: Callable[[Any, Any], int],
+ *args: Any, **kwargs: Any) -> int:
timeout = kwargs.pop('timeout', 0)
if timeout == 0 or timeout == None:
# python threading module will just get blocked if timeout is `None`,
return t.retval
-def send_command_retry(*args, **kwargs):
+def send_command_retry(*args: Any, **kwargs: Any) -> Tuple[int, bytes, str]:
while True:
try:
return send_command(*args, **kwargs)
else:
raise
-def send_command(cluster, target=('mon', ''), cmd=None, inbuf=b'', timeout=0,
- verbose=False):
+
+def send_command(cluster,
+ target: Optional[Tuple[str, str]] = ('mon', ''),
+ cmd: Optional[List[str]] = None,
+ inbuf: Optional[bytes] = b'',
+ timeout: Optional[int] = 0,
+ verbose: Optional[bool] = False) -> Tuple[int, bytes, str]:
"""
Send a command to a daemon using librados's
mon_command, osd_command, mgr_command, or pg_command. Any bulk input data
return ret, outbuf, outs
-def json_command(cluster, target=('mon', ''), prefix=None, argdict=None,
- inbuf=b'', timeout=0, verbose=False):
+def json_command(cluster,
+ target: Optional[Tuple[str, str]] = ('mon', ''),
+ prefix: Optional[str] = None,
+ argdict: Optional[Dict[str, str]] = None,
+ inbuf: Optional[bytes] = b'',
+ timeout: Optional[int] = 0,
+ verbose: Optional[bool] = False) -> Tuple[int, bytes, str]:
"""
Serialize a command and up a JSON command and send it with send_command() above.
Prefix may be supplied separately or in argdict. Any bulk input
projects / ceph-ci.git / commitdiff