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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
#!/usr/bin/python3
# Copyright (c) 2021, Dell Inc. or its subsidiaries. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
# See the LICENSE file for details.
#
# This file is part of NVMe STorage Appliance Services (nvme-stas).
#
# Authors: Martin Belanger <Martin.Belanger@dell.com>
#
''' STorage Appliance Finder Daemon
'''
import sys
from argparse import ArgumentParser
from staslib import defs
# ******************************************************************************
def parse_args(conf_file: str):
'''Parse command line options'''
parser = ArgumentParser(description='STorage Appliance Finder (STAF). Must be root to run this program.')
parser.add_argument(
'-f',
'--conf-file',
action='store',
help='Configuration file (default: %(default)s)',
default=conf_file,
type=str,
metavar='FILE',
)
parser.add_argument(
'-s',
'--syslog',
action='store_true',
help='Send messages to syslog instead of stdout. Use this when running %(prog)s as a daemon. (default: %(default)s)',
default=False,
)
parser.add_argument('--tron', action='store_true', help='Trace ON. (default: %(default)s)', default=False)
parser.add_argument('-v', '--version', action='store_true', help='Print version, then exit', default=False)
return parser.parse_args()
ARGS = parse_args(defs.STAFD_CONF_FILE)
if ARGS.version:
print(f'nvme-stas {defs.VERSION}')
print(f'libnvme {defs.LIBNVME_VERSION}')
sys.exit(0)
# ******************************************************************************
if __name__ == '__main__':
import json
import logging
import dasbus.server.interface
from staslib import log, service, stas, udev # pylint: disable=ungrouped-imports
# Before going any further, make sure the script is allowed to run.
stas.check_if_allowed_to_continue()
class Dbus:
'''This is the DBus interface that external programs can use to
communicate with stafd.
'''
__dbus_xml__ = stas.load_idl('stafd.idl')
@dasbus.server.interface.dbus_signal
def log_pages_changed( # pylint: disable=too-many-arguments
self,
transport: str,
traddr: str,
trsvcid: str,
host_traddr: str,
host_iface: str,
subsysnqn: str,
device: str,
):
'''@brief Signal sent when log pages have changed.'''
@dasbus.server.interface.dbus_signal
def dc_removed(self):
'''@brief Signal sent when log pages have changed.'''
@property
def tron(self):
'''@brief Get Trace ON property'''
return STAF.tron
@tron.setter
def tron(self, value): # pylint: disable=no-self-use
'''@brief Set Trace ON property'''
STAF.tron = value
@property
def log_level(self) -> str:
'''@brief Get Log Level property'''
return log.level()
def process_info(self) -> str:
'''@brief Get status info (for debug)
@return A string representation of a json object.
'''
info = {
'tron': STAF.tron,
'log-level': self.log_level,
}
info.update(STAF.info())
return json.dumps(info)
def controller_info( # pylint: disable=no-self-use,too-many-arguments
self, transport, traddr, trsvcid, host_traddr, host_iface, subsysnqn
) -> str:
'''@brief D-Bus method used to return information about a controller'''
controller = STAF.get_controller(transport, traddr, trsvcid, host_traddr, host_iface, subsysnqn)
return json.dumps(controller.info()) if controller else '{}'
def get_log_pages( # pylint: disable=no-self-use,too-many-arguments
self, transport, traddr, trsvcid, host_traddr, host_iface, subsysnqn
) -> list:
'''@brief D-Bus method used to retrieve the discovery log pages from one controller'''
controller = STAF.get_controller(transport, traddr, trsvcid, host_traddr, host_iface, subsysnqn)
return controller.log_pages() if controller else list()
def get_all_log_pages(self, detailed) -> str: # pylint: disable=no-self-use
'''@brief D-Bus method used to retrieve the discovery log pages from all controllers'''
log_pages = list()
for controller in STAF.get_controllers():
log_pages.append(
{
'discovery-controller': controller.details() if detailed else controller.controller_id_dict(),
'log-pages': controller.log_pages(),
}
)
return json.dumps(log_pages)
def list_controllers(self, detailed) -> list: # pylint: disable=no-self-use
'''@brief Return the list of discovery controller IDs'''
return [
controller.details() if detailed else controller.controller_id_dict()
for controller in STAF.get_controllers()
]
log.init(ARGS.syslog)
STAF = service.Staf(ARGS, Dbus())
STAF.run()
STAF = None
ARGS = None
udev.shutdown()
logging.shutdown()
|
