1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
#!/usr/bin/env python3
import logging
import argparse
import sys
import shlex
try:
from rpc.client import print_dict, JSONRPCException
import rpc
except ImportError:
print("SPDK RPC library missing. Please add spdk/scripts/ directory to PYTHONPATH:")
print("'export PYTHONPATH=$PYTHONPATH:./spdk/scripts/'")
exit(1)
try:
from shlex import quote
except ImportError:
from pipes import quote
def print_array(a):
print(" ".join((quote(v) for v in a)))
def perform_tests_func(client, name=None):
"""
Args:
name: bdev name to perform bdevio tests on (optional; if omitted, test all bdevs)
Returns:
Number of failures in tests. 0 means no errors found.
"""
params = {}
if name:
params['name'] = name
return client.call('perform_tests', params)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description='SPDK RPC command line interface. NOTE: spdk/scripts/ is expected in PYTHONPATH')
parser.add_argument('-s', dest='server_addr',
help='RPC domain socket path or IP address', default='/var/tmp/spdk.sock')
parser.add_argument('-p', dest='port',
help='RPC port number (if server_addr is IP address)',
default=5260, type=int)
parser.add_argument('-t', dest='timeout',
help='Timeout as a floating point number expressed in seconds waiting for response. Default: 60.0',
default=60.0, type=float)
parser.add_argument('-v', dest='verbose', action='store_const', const="INFO",
help='Set verbose mode to INFO', default="ERROR")
parser.add_argument('--verbose', dest='verbose', choices=['DEBUG', 'INFO', 'ERROR'],
help="""Set verbose level. """)
subparsers = parser.add_subparsers(help='RPC methods')
def perform_tests(args):
print_dict(perform_tests_func(args.client, name=args.name))
p = subparsers.add_parser('perform_tests', help='Perform all bdevio tests on select bdev')
p.add_argument('-b', '--name', help="Name of the Blockdev. Example: Nvme0n1")
p.set_defaults(func=perform_tests)
def call_rpc_func(args):
try:
args.func(args)
except JSONRPCException as ex:
print(ex.message)
exit(1)
def execute_script(parser, client, fd):
for rpc_call in map(str.rstrip, fd):
if not rpc_call.strip():
continue
args = parser.parse_args(shlex.split(rpc_call))
args.client = client
call_rpc_func(args)
args = parser.parse_args()
args.client = rpc.client.JSONRPCClient(args.server_addr, args.port, args.timeout, log_level=getattr(logging, args.verbose.upper()))
if hasattr(args, 'func'):
call_rpc_func(args)
elif sys.stdin.isatty():
# No arguments and no data piped through stdin
parser.print_help()
exit(1)
else:
execute_script(parser, args.client, sys.stdin)
|
