qemu-img: add support for rate limit in qemu-img commit
[qemu.git] / python / qemu / qmp.py
1 """ QEMU Monitor Protocol Python class """
2 # Copyright (C) 2009, 2010 Red Hat Inc.
3 #
4 # Authors:
5 # Luiz Capitulino <lcapitulino@redhat.com>
6 #
7 # This work is licensed under the terms of the GNU GPL, version 2. See
8 # the COPYING file in the top-level directory.
9
10 import errno
11 import json
12 import logging
13 import socket
14 from types import TracebackType
15 from typing import (
16 Any,
17 Dict,
18 List,
19 Optional,
20 TextIO,
21 Tuple,
22 Type,
23 Union,
24 cast,
25 )
26
27
28 # QMPMessage is a QMP Message of any kind.
29 # e.g. {'yee': 'haw'}
30 #
31 # QMPReturnValue is the inner value of return values only.
32 # {'return': {}} is the QMPMessage,
33 # {} is the QMPReturnValue.
34 QMPMessage = Dict[str, Any]
35 QMPReturnValue = Dict[str, Any]
36
37 InternetAddrT = Tuple[str, str]
38 UnixAddrT = str
39 SocketAddrT = Union[InternetAddrT, UnixAddrT]
40
41
42 class QMPError(Exception):
43 """
44 QMP base exception
45 """
46
47
48 class QMPConnectError(QMPError):
49 """
50 QMP connection exception
51 """
52
53
54 class QMPCapabilitiesError(QMPError):
55 """
56 QMP negotiate capabilities exception
57 """
58
59
60 class QMPTimeoutError(QMPError):
61 """
62 QMP timeout exception
63 """
64
65
66 class QMPProtocolError(QMPError):
67 """
68 QMP protocol error; unexpected response
69 """
70
71
72 class QMPResponseError(QMPError):
73 """
74 Represents erroneous QMP monitor reply
75 """
76 def __init__(self, reply: QMPMessage):
77 try:
78 desc = reply['error']['desc']
79 except KeyError:
80 desc = reply
81 super().__init__(desc)
82 self.reply = reply
83
84
85 class QEMUMonitorProtocol:
86 """
87 Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then
88 allow to handle commands and events.
89 """
90
91 #: Logger object for debugging messages
92 logger = logging.getLogger('QMP')
93
94 def __init__(self, address: SocketAddrT,
95 server: bool = False,
96 nickname: Optional[str] = None):
97 """
98 Create a QEMUMonitorProtocol class.
99
100 @param address: QEMU address, can be either a unix socket path (string)
101 or a tuple in the form ( address, port ) for a TCP
102 connection
103 @param server: server mode listens on the socket (bool)
104 @raise OSError on socket connection errors
105 @note No connection is established, this is done by the connect() or
106 accept() methods
107 """
108 self.__events: List[QMPMessage] = []
109 self.__address = address
110 self.__sock = self.__get_sock()
111 self.__sockfile: Optional[TextIO] = None
112 self._nickname = nickname
113 if self._nickname:
114 self.logger = logging.getLogger('QMP').getChild(self._nickname)
115 if server:
116 self.__sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
117 self.__sock.bind(self.__address)
118 self.__sock.listen(1)
119
120 def __get_sock(self) -> socket.socket:
121 if isinstance(self.__address, tuple):
122 family = socket.AF_INET
123 else:
124 family = socket.AF_UNIX
125 return socket.socket(family, socket.SOCK_STREAM)
126
127 def __negotiate_capabilities(self) -> QMPMessage:
128 greeting = self.__json_read()
129 if greeting is None or "QMP" not in greeting:
130 raise QMPConnectError
131 # Greeting seems ok, negotiate capabilities
132 resp = self.cmd('qmp_capabilities')
133 if resp and "return" in resp:
134 return greeting
135 raise QMPCapabilitiesError
136
137 def __json_read(self, only_event: bool = False) -> Optional[QMPMessage]:
138 assert self.__sockfile is not None
139 while True:
140 data = self.__sockfile.readline()
141 if not data:
142 return None
143 # By definition, any JSON received from QMP is a QMPMessage,
144 # and we are asserting only at static analysis time that it
145 # has a particular shape.
146 resp: QMPMessage = json.loads(data)
147 if 'event' in resp:
148 self.logger.debug("<<< %s", resp)
149 self.__events.append(resp)
150 if not only_event:
151 continue
152 return resp
153
154 def __get_events(self, wait: Union[bool, float] = False) -> None:
155 """
156 Check for new events in the stream and cache them in __events.
157
158 @param wait (bool): block until an event is available.
159 @param wait (float): If wait is a float, treat it as a timeout value.
160
161 @raise QMPTimeoutError: If a timeout float is provided and the timeout
162 period elapses.
163 @raise QMPConnectError: If wait is True but no events could be
164 retrieved or if some other error occurred.
165 """
166
167 # Current timeout and blocking status
168 current_timeout = self.__sock.gettimeout()
169
170 # Check for new events regardless and pull them into the cache:
171 self.__sock.settimeout(0) # i.e. setblocking(False)
172 try:
173 self.__json_read()
174 except OSError as err:
175 # EAGAIN: No data available; not critical
176 if err.errno != errno.EAGAIN:
177 raise
178 finally:
179 self.__sock.settimeout(current_timeout)
180
181 # Wait for new events, if needed.
182 # if wait is 0.0, this means "no wait" and is also implicitly false.
183 if not self.__events and wait:
184 if isinstance(wait, float):
185 self.__sock.settimeout(wait)
186 try:
187 ret = self.__json_read(only_event=True)
188 except socket.timeout as err:
189 raise QMPTimeoutError("Timeout waiting for event") from err
190 except Exception as err:
191 msg = "Error while reading from socket"
192 raise QMPConnectError(msg) from err
193 finally:
194 self.__sock.settimeout(current_timeout)
195
196 if ret is None:
197 raise QMPConnectError("Error while reading from socket")
198
199 def __enter__(self) -> 'QEMUMonitorProtocol':
200 # Implement context manager enter function.
201 return self
202
203 def __exit__(self,
204 # pylint: disable=duplicate-code
205 # see https://github.com/PyCQA/pylint/issues/3619
206 exc_type: Optional[Type[BaseException]],
207 exc_val: Optional[BaseException],
208 exc_tb: Optional[TracebackType]) -> None:
209 # Implement context manager exit function.
210 self.close()
211
212 def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
213 """
214 Connect to the QMP Monitor and perform capabilities negotiation.
215
216 @return QMP greeting dict, or None if negotiate is false
217 @raise OSError on socket connection errors
218 @raise QMPConnectError if the greeting is not received
219 @raise QMPCapabilitiesError if fails to negotiate capabilities
220 """
221 self.__sock.connect(self.__address)
222 self.__sockfile = self.__sock.makefile(mode='r')
223 if negotiate:
224 return self.__negotiate_capabilities()
225 return None
226
227 def accept(self, timeout: Optional[float] = 15.0) -> QMPMessage:
228 """
229 Await connection from QMP Monitor and perform capabilities negotiation.
230
231 @param timeout: timeout in seconds (nonnegative float number, or
232 None). The value passed will set the behavior of the
233 underneath QMP socket as described in [1].
234 Default value is set to 15.0.
235 @return QMP greeting dict
236 @raise OSError on socket connection errors
237 @raise QMPConnectError if the greeting is not received
238 @raise QMPCapabilitiesError if fails to negotiate capabilities
239
240 [1]
241 https://docs.python.org/3/library/socket.html#socket.socket.settimeout
242 """
243 self.__sock.settimeout(timeout)
244 self.__sock, _ = self.__sock.accept()
245 self.__sockfile = self.__sock.makefile(mode='r')
246 return self.__negotiate_capabilities()
247
248 def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
249 """
250 Send a QMP command to the QMP Monitor.
251
252 @param qmp_cmd: QMP command to be sent as a Python dict
253 @return QMP response as a Python dict
254 """
255 self.logger.debug(">>> %s", qmp_cmd)
256 self.__sock.sendall(json.dumps(qmp_cmd).encode('utf-8'))
257 resp = self.__json_read()
258 if resp is None:
259 raise QMPConnectError("Unexpected empty reply from server")
260 self.logger.debug("<<< %s", resp)
261 return resp
262
263 def cmd(self, name: str,
264 args: Optional[Dict[str, Any]] = None,
265 cmd_id: Optional[Any] = None) -> QMPMessage:
266 """
267 Build a QMP command and send it to the QMP Monitor.
268
269 @param name: command name (string)
270 @param args: command arguments (dict)
271 @param cmd_id: command id (dict, list, string or int)
272 """
273 qmp_cmd: QMPMessage = {'execute': name}
274 if args:
275 qmp_cmd['arguments'] = args
276 if cmd_id:
277 qmp_cmd['id'] = cmd_id
278 return self.cmd_obj(qmp_cmd)
279
280 def command(self, cmd: str, **kwds: Any) -> QMPReturnValue:
281 """
282 Build and send a QMP command to the monitor, report errors if any
283 """
284 ret = self.cmd(cmd, kwds)
285 if 'error' in ret:
286 raise QMPResponseError(ret)
287 if 'return' not in ret:
288 raise QMPProtocolError(
289 "'return' key not found in QMP response '{}'".format(str(ret))
290 )
291 return cast(QMPReturnValue, ret['return'])
292
293 def pull_event(self,
294 wait: Union[bool, float] = False) -> Optional[QMPMessage]:
295 """
296 Pulls a single event.
297
298 @param wait (bool): block until an event is available.
299 @param wait (float): If wait is a float, treat it as a timeout value.
300
301 @raise QMPTimeoutError: If a timeout float is provided and the timeout
302 period elapses.
303 @raise QMPConnectError: If wait is True but no events could be
304 retrieved or if some other error occurred.
305
306 @return The first available QMP event, or None.
307 """
308 self.__get_events(wait)
309
310 if self.__events:
311 return self.__events.pop(0)
312 return None
313
314 def get_events(self, wait: bool = False) -> List[QMPMessage]:
315 """
316 Get a list of available QMP events.
317
318 @param wait (bool): block until an event is available.
319 @param wait (float): If wait is a float, treat it as a timeout value.
320
321 @raise QMPTimeoutError: If a timeout float is provided and the timeout
322 period elapses.
323 @raise QMPConnectError: If wait is True but no events could be
324 retrieved or if some other error occurred.
325
326 @return The list of available QMP events.
327 """
328 self.__get_events(wait)
329 return self.__events
330
331 def clear_events(self) -> None:
332 """
333 Clear current list of pending events.
334 """
335 self.__events = []
336
337 def close(self) -> None:
338 """
339 Close the socket and socket file.
340 """
341 if self.__sock:
342 self.__sock.close()
343 if self.__sockfile:
344 self.__sockfile.close()
345
346 def settimeout(self, timeout: Optional[float]) -> None:
347 """
348 Set the socket timeout.
349
350 @param timeout (float): timeout in seconds (non-zero), or None.
351 @note This is a wrap around socket.settimeout
352
353 @raise ValueError: if timeout was set to 0.
354 """
355 if timeout == 0:
356 msg = "timeout cannot be 0; this engages non-blocking mode."
357 msg += " Use 'None' instead to disable timeouts."
358 raise ValueError(msg)
359 self.__sock.settimeout(timeout)
360
361 def get_sock_fd(self) -> int:
362 """
363 Get the socket file descriptor.
364
365 @return The file descriptor number.
366 """
367 return self.__sock.fileno()
368
369 def is_scm_available(self) -> bool:
370 """
371 Check if the socket allows for SCM_RIGHTS.
372
373 @return True if SCM_RIGHTS is available, otherwise False.
374 """
375 return self.__sock.family == socket.AF_UNIX