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