python/ovs/unixctl.py

   1 # Copyright (c) 2012 Nicira Networks
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import copy
  16 import errno
  17 import os
  18 import types
  19
  20 import ovs.daemon
  21 import ovs.dirs
  22 import ovs.jsonrpc
  23 import ovs.stream
  24 import ovs.util
  25 import ovs.version
  26 import ovs.vlog
  27
  28 Message = ovs.jsonrpc.Message
  29 vlog = ovs.vlog.Vlog("unixctl")
  30 commands = {}
  31 strtypes = types.StringTypes
  32
  33
  34 class _UnixctlCommand(object):
  35     def __init__(self, usage, min_args, max_args, callback, aux):
  36         self.usage = usage
  37         self.min_args = min_args
  38         self.max_args = max_args
  39         self.callback = callback
  40         self.aux = aux
  41
  42
  43 def _unixctl_help(conn, unused_argv, unused_aux):
  44     assert isinstance(conn, UnixctlConnection)
  45     reply = "The available commands are:\n"
  46     command_names = sorted(commands.keys())
  47     for name in command_names:
  48         reply += "  "
  49         usage = commands[name].usage
  50         if usage:
  51             reply += "%-23s %s" % (name, usage)
  52         else:
  53             reply += name
  54         reply += "\n"
  55     conn.reply(reply)
  56
  57
  58 def _unixctl_version(conn, unused_argv, unused_aux):
  59     assert isinstance(conn, UnixctlConnection)
  60     version = "%s (Open vSwitch) %s" % (ovs.util.PROGRAM_NAME,
  61                                         ovs.version.VERSION)
  62     conn.reply(version)
  63
  64
  65 def command_register(name, usage, min_args, max_args, callback, aux):
  66     """ Registers a command with the given 'name' to be exposed by the
  67     UnixctlServer. 'usage' describes the arguments to the command; it is used
  68     only for presentation to the user in "help" output.
  69
  70     'callback' is called when the command is received.  It is passed a
  71     UnixctlConnection object, the list of arguments as unicode strings, and
  72     'aux'.  Normally 'callback' should reply by calling
  73     UnixctlConnection.reply() or UnixctlConnection.reply_error() before it
  74     returns, but if the command cannot be handled immediately, then it can
  75     defer the reply until later.  A given connection can only process a single
  76     request at a time, so a reply must be made eventually to avoid blocking
  77     that connection."""
  78
  79     assert isinstance(name, strtypes)
  80     assert isinstance(usage, strtypes)
  81     assert isinstance(min_args, int)
  82     assert isinstance(max_args, int)
  83     assert isinstance(callback, types.FunctionType)
  84
  85     if name not in commands:
  86         commands[name] = _UnixctlCommand(usage, min_args, max_args, callback,
  87                                          aux)
  88
  89
  90 def socket_name_from_target(target):
  91     assert isinstance(target, strtypes)
  92
  93     if target.startswith("/"):
  94         return 0, target
  95
  96     pidfile_name = "%s/%s.pid" % (ovs.dirs.RUNDIR, target)
  97     pid = ovs.daemon.read_pidfile(pidfile_name)
  98     if pid < 0:
  99         return -pid, "cannot read pidfile \"%s\"" % pidfile_name
 100
 101     return 0, "%s/%s.%d.ctl" % (ovs.dirs.RUNDIR, target, pid)
 102
 103
 104 class UnixctlConnection(object):
 105     def __init__(self, rpc):
 106         assert isinstance(rpc, ovs.jsonrpc.Connection)
 107         self._rpc = rpc
 108         self._request_id = None
 109
 110     def run(self):
 111         self._rpc.run()
 112         error = self._rpc.get_status()
 113         if error or self._rpc.get_backlog():
 114             return error
 115
 116         for _ in range(10):
 117             if error or self._request_id:
 118                 break
 119
 120             error, msg = self._rpc.recv()
 121             if msg:
 122                 if msg.type == Message.T_REQUEST:
 123                     self._process_command(msg)
 124                 else:
 125                     # XXX: rate-limit
 126                     vlog.warn("%s: received unexpected %s message"
 127                               % (self._rpc.name,
 128                                  Message.type_to_string(msg.type)))
 129                     error = errno.EINVAL
 130
 131             if not error:
 132                 error = self._rpc.get_status()
 133
 134         return error
 135
 136     def reply(self, body):
 137         self._reply_impl(True, body)
 138
 139     def reply_error(self, body):
 140         self._reply_impl(False, body)
 141
 142     # Called only by unixctl classes.
 143     def _close(self):
 144         self._rpc.close()
 145         self._request_id = None
 146
 147     def _wait(self, poller):
 148         self._rpc.wait(poller)
 149         if not self._rpc.get_backlog():
 150             self._rpc.recv_wait(poller)
 151
 152     def _reply_impl(self, success, body):
 153         assert isinstance(success, bool)
 154         assert body is None or isinstance(body, strtypes)
 155
 156         assert self._request_id is not None
 157
 158         if body is None:
 159             body = ""
 160
 161         if body and not body.endswith("\n"):
 162             body += "\n"
 163
 164         if success:
 165             reply = Message.create_reply(body, self._request_id)
 166         else:
 167             reply = Message.create_error(body, self._request_id)
 168
 169         self._rpc.send(reply)
 170         self._request_id = None
 171
 172     def _process_command(self, request):
 173         assert isinstance(request, ovs.jsonrpc.Message)
 174         assert request.type == ovs.jsonrpc.Message.T_REQUEST
 175
 176         self._request_id = request.id
 177
 178         error = None
 179         params = request.params
 180         method = request.method
 181         command = commands.get(method)
 182         if command is None:
 183             error = '"%s" is not a valid command' % method
 184         elif len(params) < command.min_args:
 185             error = '"%s" command requires at least %d arguments' \
 186                     % (method, command.min_args)
 187         elif len(params) > command.max_args:
 188             error = '"%s" command takes at most %d arguments' \
 189                     % (method, command.max_args)
 190         else:
 191             for param in params:
 192                 if not isinstance(param, strtypes):
 193                     error = '"%s" command has non-string argument' % method
 194                     break
 195
 196             if error is None:
 197                 unicode_params = [unicode(p) for p in params]
 198                 command.callback(self, unicode_params, command.aux)
 199
 200         if error:
 201             self.reply_error(error)
 202
 203
 204 class UnixctlServer(object):
 205     def __init__(self, listener):
 206         assert isinstance(listener, ovs.stream.PassiveStream)
 207         self._listener = listener
 208         self._conns = []
 209
 210     def run(self):
 211         for _ in range(10):
 212             error, stream = self._listener.accept()
 213             if not error:
 214                 rpc = ovs.jsonrpc.Connection(stream)
 215                 self._conns.append(UnixctlConnection(rpc))
 216             elif error == errno.EAGAIN:
 217                 break
 218             else:
 219                 # XXX: rate-limit
 220                 vlog.warn("%s: accept failed: %s" % (self._listener.name,
 221                                                      os.strerror(error)))
 222
 223         for conn in copy.copy(self._conns):
 224             error = conn.run()
 225             if error and error != errno.EAGAIN:
 226                 conn._close()
 227                 self._conns.remove(conn)
 228
 229     def wait(self, poller):
 230         self._listener.wait(poller)
 231         for conn in self._conns:
 232             conn._wait(poller)
 233
 234     def close(self):
 235         for conn in self._conns:
 236             conn._close()
 237         self._conns = None
 238
 239         self._listener.close()
 240         self._listener = None
 241
 242     @staticmethod
 243     def create(path):
 244         assert path is None or isinstance(path, strtypes)
 245
 246         if path is not None:
 247             path = "punix:%s" % ovs.util.abs_file_name(ovs.dirs.RUNDIR, path)
 248         else:
 249             path = "punix:%s/%s.%d.ctl" % (ovs.dirs.RUNDIR,
 250                                            ovs.util.PROGRAM_NAME, os.getpid())
 251
 252         error, listener = ovs.stream.PassiveStream.open(path)
 253         if error:
 254             ovs.util.ovs_error(error, "could not initialize control socket %s"
 255                                % path)
 256             return error, None
 257
 258         command_register("help", "", 0, 0, _unixctl_help, None)
 259         command_register("version", "", 0, 0, _unixctl_version, None)
 260
 261         return 0, UnixctlServer(listener)
 262
 263
 264 class UnixctlClient(object):
 265     def __init__(self, conn):
 266         assert isinstance(conn, ovs.jsonrpc.Connection)
 267         self._conn = conn
 268
 269     def transact(self, command, argv):
 270         assert isinstance(command, strtypes)
 271         assert isinstance(argv, list)
 272         for arg in argv:
 273             assert isinstance(arg, strtypes)
 274
 275         request = Message.create_request(command, argv)
 276         error, reply = self._conn.transact_block(request)
 277
 278         if error:
 279             vlog.warn("error communicating with %s: %s"
 280                       % (self._conn.name, os.strerror(error)))
 281             return error, None, None
 282
 283         if reply.error is not None:
 284             return 0, str(reply.error), None
 285         else:
 286             assert reply.result is not None
 287             return 0, None, str(reply.result)
 288
 289     def close(self):
 290         self._conn.close()
 291         self.conn = None
 292
 293     @staticmethod
 294     def create(path):
 295         assert isinstance(path, str)
 296
 297         unix = "unix:%s" % ovs.util.abs_file_name(ovs.dirs.RUNDIR, path)
 298         error, stream = ovs.stream.Stream.open_block(
 299             ovs.stream.Stream.open(unix))
 300
 301         if error:
 302             vlog.warn("failed to connect to %s" % path)
 303             return error, None
 304
 305         return 0, UnixctlClient(ovs.jsonrpc.Connection(stream))