python/ovs/unixctl.py

   1 # Copyright (c) 2012 Nicira, Inc.
   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, version):
  59     assert isinstance(conn, UnixctlConnection)
  60     version = "%s (Open vSwitch) %s" % (ovs.util.PROGRAM_NAME, version)
  61     conn.reply(version)
  62
  63
  64 def command_register(name, usage, min_args, max_args, callback, aux):
  65     """ Registers a command with the given 'name' to be exposed by the
  66     UnixctlServer. 'usage' describes the arguments to the command; it is used
  67     only for presentation to the user in "help" output.
  68
  69     'callback' is called when the command is received.  It is passed a
  70     UnixctlConnection object, the list of arguments as unicode strings, and
  71     'aux'.  Normally 'callback' should reply by calling
  72     UnixctlConnection.reply() or UnixctlConnection.reply_error() before it
  73     returns, but if the command cannot be handled immediately, then it can
  74     defer the reply until later.  A given connection can only process a single
  75     request at a time, so a reply must be made eventually to avoid blocking
  76     that connection."""
  77
  78     assert isinstance(name, strtypes)
  79     assert isinstance(usage, strtypes)
  80     assert isinstance(min_args, int)
  81     assert isinstance(max_args, int)
  82     assert isinstance(callback, types.FunctionType)
  83
  84     if name not in commands:
  85         commands[name] = _UnixctlCommand(usage, min_args, max_args, callback,
  86                                          aux)
  87
  88
  89 def socket_name_from_target(target):
  90     assert isinstance(target, strtypes)
  91
  92     if target.startswith("/"):
  93         return 0, target
  94
  95     pidfile_name = "%s/%s.pid" % (ovs.dirs.RUNDIR, target)
  96     pid = ovs.daemon.read_pidfile(pidfile_name)
  97     if pid < 0:
  98         return -pid, "cannot read pidfile \"%s\"" % pidfile_name
  99
 100     return 0, "%s/%s.%d.ctl" % (ovs.dirs.RUNDIR, target, pid)
 101
 102
 103 class UnixctlConnection(object):
 104     def __init__(self, rpc):
 105         assert isinstance(rpc, ovs.jsonrpc.Connection)
 106         self._rpc = rpc
 107         self._request_id = None
 108
 109     def run(self):
 110         self._rpc.run()
 111         error = self._rpc.get_status()
 112         if error or self._rpc.get_backlog():
 113             return error
 114
 115         for _ in range(10):
 116             if error or self._request_id:
 117                 break
 118
 119             error, msg = self._rpc.recv()
 120             if msg:
 121                 if msg.type == Message.T_REQUEST:
 122                     self._process_command(msg)
 123                 else:
 124                     # XXX: rate-limit
 125                     vlog.warn("%s: received unexpected %s message"
 126                               % (self._rpc.name,
 127                                  Message.type_to_string(msg.type)))
 128                     error = errno.EINVAL
 129
 130             if not error:
 131                 error = self._rpc.get_status()
 132
 133         return error
 134
 135     def reply(self, body):
 136         self._reply_impl(True, body)
 137
 138     def reply_error(self, body):
 139         self._reply_impl(False, body)
 140
 141     # Called only by unixctl classes.
 142     def _close(self):
 143         self._rpc.close()
 144         self._request_id = None
 145
 146     def _wait(self, poller):
 147         self._rpc.wait(poller)
 148         if not self._rpc.get_backlog():
 149             self._rpc.recv_wait(poller)
 150
 151     def _reply_impl(self, success, body):
 152         assert isinstance(success, bool)
 153         assert body is None or isinstance(body, strtypes)
 154
 155         assert self._request_id is not None
 156
 157         if body is None:
 158             body = ""
 159
 160         if body and not body.endswith("\n"):
 161             body += "\n"
 162
 163         if success:
 164             reply = Message.create_reply(body, self._request_id)
 165         else:
 166             reply = Message.create_error(body, self._request_id)
 167
 168         self._rpc.send(reply)
 169         self._request_id = None
 170
 171     def _process_command(self, request):
 172         assert isinstance(request, ovs.jsonrpc.Message)
 173         assert request.type == ovs.jsonrpc.Message.T_REQUEST
 174
 175         self._request_id = request.id
 176
 177         error = None
 178         params = request.params
 179         method = request.method
 180         command = commands.get(method)
 181         if command is None:
 182             error = '"%s" is not a valid command' % method
 183         elif len(params) < command.min_args:
 184             error = '"%s" command requires at least %d arguments' \
 185                     % (method, command.min_args)
 186         elif len(params) > command.max_args:
 187             error = '"%s" command takes at most %d arguments' \
 188                     % (method, command.max_args)
 189         else:
 190             for param in params:
 191                 if not isinstance(param, strtypes):
 192                     error = '"%s" command has non-string argument' % method
 193                     break
 194
 195             if error is None:
 196                 unicode_params = [unicode(p) for p in params]
 197                 command.callback(self, unicode_params, command.aux)
 198
 199         if error:
 200             self.reply_error(error)
 201
 202
 203 class UnixctlServer(object):
 204     def __init__(self, listener):
 205         assert isinstance(listener, ovs.stream.PassiveStream)
 206         self._listener = listener
 207         self._conns = []
 208
 209     def run(self):
 210         for _ in range(10):
 211             error, stream = self._listener.accept()
 212             if not error:
 213                 rpc = ovs.jsonrpc.Connection(stream)
 214                 self._conns.append(UnixctlConnection(rpc))
 215             elif error == errno.EAGAIN:
 216                 break
 217             else:
 218                 # XXX: rate-limit
 219                 vlog.warn("%s: accept failed: %s" % (self._listener.name,
 220                                                      os.strerror(error)))
 221
 222         for conn in copy.copy(self._conns):
 223             error = conn.run()
 224             if error and error != errno.EAGAIN:
 225                 conn._close()
 226                 self._conns.remove(conn)
 227
 228     def wait(self, poller):
 229         self._listener.wait(poller)
 230         for conn in self._conns:
 231             conn._wait(poller)
 232
 233     def close(self):
 234         for conn in self._conns:
 235             conn._close()
 236         self._conns = None
 237
 238         self._listener.close()
 239         self._listener = None
 240
 241     @staticmethod
 242     def create(path, version=None):
 243         """Creates a new UnixctlServer which listens on a unixctl socket
 244         created at 'path'.  If 'path' is None, the default path is chosen.
 245         'version' contains the version of the server as reported by the unixctl
 246         version command.  If None, ovs.version.VERSION is used."""
 247
 248         assert path is None or isinstance(path, strtypes)
 249
 250         if path is not None:
 251             path = "punix:%s" % ovs.util.abs_file_name(ovs.dirs.RUNDIR, path)
 252         else:
 253             path = "punix:%s/%s.%d.ctl" % (ovs.dirs.RUNDIR,
 254                                            ovs.util.PROGRAM_NAME, os.getpid())
 255
 256         if version is None:
 257             version = ovs.version.VERSION
 258
 259         error, listener = ovs.stream.PassiveStream.open(path)
 260         if error:
 261             ovs.util.ovs_error(error, "could not initialize control socket %s"
 262                                % path)
 263             return error, None
 264
 265         command_register("help", "", 0, 0, _unixctl_help, None)
 266         command_register("version", "", 0, 0, _unixctl_version, version)
 267
 268         return 0, UnixctlServer(listener)
 269
 270
 271 class UnixctlClient(object):
 272     def __init__(self, conn):
 273         assert isinstance(conn, ovs.jsonrpc.Connection)
 274         self._conn = conn
 275
 276     def transact(self, command, argv):
 277         assert isinstance(command, strtypes)
 278         assert isinstance(argv, list)
 279         for arg in argv:
 280             assert isinstance(arg, strtypes)
 281
 282         request = Message.create_request(command, argv)
 283         error, reply = self._conn.transact_block(request)
 284
 285         if error:
 286             vlog.warn("error communicating with %s: %s"
 287                       % (self._conn.name, os.strerror(error)))
 288             return error, None, None
 289
 290         if reply.error is not None:
 291             return 0, str(reply.error), None
 292         else:
 293             assert reply.result is not None
 294             return 0, None, str(reply.result)
 295
 296     def close(self):
 297         self._conn.close()
 298         self.conn = None
 299
 300     @staticmethod
 301     def create(path):
 302         assert isinstance(path, str)
 303
 304         unix = "unix:%s" % ovs.util.abs_file_name(ovs.dirs.RUNDIR, path)
 305         error, stream = ovs.stream.Stream.open_block(
 306             ovs.stream.Stream.open(unix))
 307
 308         if error:
 309             vlog.warn("failed to connect to %s" % path)
 310             return error, None
 311
 312         return 0, UnixctlClient(ovs.jsonrpc.Connection(stream))