bin/netsvc.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 ##############################################################################
   4 #
   5 #    OpenERP, Open Source Management Solution
   6 #    Copyright (C) 2004-2009 Tiny SPRL (<http://tiny.be>). All Rights Reserved
   7 #    The refactoring about the OpenSSL support come from Tryton
   8 #    Copyright (C) 2007-2009 Cédric Krier.
   9 #    Copyright (C) 2007-2009 Bertrand Chenal.
  10 #    Copyright (C) 2008 B2CK SPRL.
  11 #
  12 #    This program is free software: you can redistribute it and/or modify
  13 #    it under the terms of the GNU General Public License as published by
  14 #    the Free Software Foundation, either version 3 of the License, or
  15 #    (at your option) any later version.
  16 #
  17 #    This program is distributed in the hope that it will be useful,
  18 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  19 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  20 #    GNU General Public License for more details.
  21 #
  22 #    You should have received a copy of the GNU General Public License
  23 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  24 #
  25 ##############################################################################
  26
  27 import errno
  28 import logging
  29 import logging.handlers
  30 import os
  31 import socket
  32 import sys
  33 import threading
  34 import time
  35 import release
  36 from pprint import pformat
  37 import warnings
  38
  39 class Service(object):
  40     """ Base class for *Local* services
  41
  42         Functionality here is trusted, no authentication.
  43     """
  44     _services = {}
  45     def __init__(self, name, audience=''):
  46         Service._services[name] = self
  47         self.__name = name
  48         self._methods = {}
  49
  50     def joinGroup(self, name):
  51         raise Exception("No group for local services")
  52         #GROUPS.setdefault(name, {})[self.__name] = self
  53
  54     @classmethod
  55     def exists(cls, name):
  56         return name in cls._services
  57
  58     @classmethod
  59     def remove(cls, name):
  60         if cls.exists(name):
  61             cls._services.pop(name)
  62
  63     def exportMethod(self, method):
  64         if callable(method):
  65             self._methods[method.__name__] = method
  66
  67     def abortResponse(self, error, description, origin, details):
  68         if not tools.config['debug_mode']:
  69             raise Exception("%s -- %s\n\n%s"%(origin, description, details))
  70         else:
  71             raise
  72
  73 class LocalService(object):
  74     """ Proxy for local services.
  75
  76         Any instance of this class will behave like the single instance
  77         of Service(name)
  78     """
  79     __logger = logging.getLogger('service')
  80     def __init__(self, name):
  81         self.__name = name
  82         try:
  83             self._service = Service._services[name]
  84             for method_name, method_definition in self._service._methods.items():
  85                 setattr(self, method_name, method_definition)
  86         except KeyError, keyError:
  87             self.__logger.error('This service does not exist: %s' % (str(keyError),) )
  88             raise
  89
  90     def __call__(self, method, *params):
  91         return getattr(self, method)(*params)
  92
  93 class ExportService(object):
  94     """ Proxy for exported services.
  95
  96     All methods here should take an AuthProxy as their first parameter. It
  97     will be appended by the calling framework.
  98
  99     Note that this class has no direct proxy, capable of calling
 100     eservice.method(). Rather, the proxy should call
 101     dispatch(method,auth,params)
 102     """
 103
 104     _services = {}
 105     _groups = {}
 106     _logger = logging.getLogger('web-services')
 107
 108     def __init__(self, name, audience=''):
 109         ExportService._services[name] = self
 110         self.__name = name
 111         self._logger.debug("Registered an exported service: %s" % name)
 112
 113     def joinGroup(self, name):
 114         ExportService._groups.setdefault(name, {})[self.__name] = self
 115
 116     @classmethod
 117     def getService(cls,name):
 118         return cls._services[name]
 119
 120     def dispatch(self, method, auth, params):
 121         raise Exception("stub dispatch at %s" % self.__name)
 122
 123     def new_dispatch(self,method,auth,params):
 124         raise Exception("stub dispatch at %s" % self.__name)
 125
 126     def abortResponse(self, error, description, origin, details):
 127         if not tools.config['debug_mode']:
 128             raise Exception("%s -- %s\n\n%s"%(origin, description, details))
 129         else:
 130             raise
 131
 132 LOG_NOTSET = 'notset'
 133 LOG_DEBUG_SQL = 'debug_sql'
 134 LOG_DEBUG_RPC = 'debug_rpc'
 135 LOG_DEBUG = 'debug'
 136 LOG_TEST = 'test'
 137 LOG_INFO = 'info'
 138 LOG_WARNING = 'warn'
 139 LOG_ERROR = 'error'
 140 LOG_CRITICAL = 'critical'
 141
 142 logging.DEBUG_RPC = logging.DEBUG - 2
 143 logging.addLevelName(logging.DEBUG_RPC, 'DEBUG_RPC')
 144 logging.DEBUG_SQL = logging.DEBUG_RPC - 2
 145 logging.addLevelName(logging.DEBUG_SQL, 'DEBUG_SQL')
 146
 147 logging.TEST = logging.INFO - 5
 148 logging.addLevelName(logging.TEST, 'TEST')
 149
 150 BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE, _NOTHING, DEFAULT = range(10)
 151 #The background is set with 40 plus the number of the color, and the foreground with 30
 152 #These are the sequences need to get colored ouput
 153 RESET_SEQ = "\033[0m"
 154 COLOR_SEQ = "\033[1;%dm"
 155 BOLD_SEQ = "\033[1m"
 156 COLOR_PATTERN = "%s%s%%s%s" % (COLOR_SEQ, COLOR_SEQ, RESET_SEQ)
 157 LEVEL_COLOR_MAPPING = {
 158     logging.DEBUG_SQL: (WHITE, MAGENTA),
 159     logging.DEBUG_RPC: (BLUE, WHITE),
 160     logging.DEBUG: (BLUE, DEFAULT),
 161     logging.INFO: (GREEN, DEFAULT),
 162     logging.TEST: (WHITE, BLUE),
 163     logging.WARNING: (YELLOW, DEFAULT),
 164     logging.ERROR: (RED, DEFAULT),
 165     logging.CRITICAL: (WHITE, RED),
 166 }
 167
 168 class DBFormatter(logging.Formatter):
 169     def format(self, record):
 170         record.dbname = getattr(threading.current_thread(), 'dbname', '?')
 171         return logging.Formatter.format(self, record)
 172
 173 class ColoredFormatter(DBFormatter):
 174     def format(self, record):
 175         fg_color, bg_color = LEVEL_COLOR_MAPPING[record.levelno]
 176         record.levelname = COLOR_PATTERN % (30 + fg_color, 40 + bg_color, record.levelname)
 177         return DBFormatter.format(self, record)
 178
 179 def init_logger():
 180     import os
 181     from tools.translate import resetlocale
 182     resetlocale()
 183
 184     # create a format for log messages and dates
 185     format = '[%(asctime)s][%(dbname)s] %(levelname)s:%(name)s:%(message)s'
 186
 187     if tools.config['syslog']:
 188         # SysLog Handler
 189         if os.name == 'nt':
 190             handler = logging.handlers.NTEventLogHandler("%s %s" % (release.description, release.version))
 191         else:
 192             handler = logging.handlers.SysLogHandler('/dev/log')
 193         format = '%s %s' % (release.description, release.version) \
 194                 + ':%(dbname)s:%(levelname)s:%(name)s:%(message)s'
 195
 196     elif tools.config['logfile']:
 197         # LogFile Handler
 198         logf = tools.config['logfile']
 199         try:
 200             dirname = os.path.dirname(logf)
 201             if dirname and not os.path.isdir(dirname):
 202                 os.makedirs(dirname)
 203             if tools.config['logrotate'] is not False:
 204                 handler = logging.handlers.TimedRotatingFileHandler(logf,'D',1,30)
 205             elif os.name == 'posix':
 206                 handler = logging.handlers.WatchedFileHandler(logf)
 207             else:
 208                 handler = logging.handlers.FileHandler(logf)
 209         except Exception, ex:
 210             sys.stderr.write("ERROR: couldn't create the logfile directory. Logging to the standard output.\n")
 211             handler = logging.StreamHandler(sys.stdout)
 212     else:
 213         # Normal Handler on standard output
 214         handler = logging.StreamHandler(sys.stdout)
 215
 216     if isinstance(handler, logging.StreamHandler) and os.isatty(handler.stream.fileno()):
 217         formatter = ColoredFormatter(format)
 218     else:
 219         formatter = DBFormatter(format)
 220     handler.setFormatter(formatter)
 221
 222     # add the handler to the root logger
 223     logger = logging.getLogger()
 224     logger.addHandler(handler)
 225     logger.setLevel(int(tools.config['log_level'] or '0'))
 226
 227
 228 class Logger(object):
 229     def __init__(self):
 230         warnings.warn("The netsvc.Logger API shouldn't be used anymore, please "
 231                       "use the standard `logging.getLogger` API instead",
 232                       PendingDeprecationWarning, stacklevel=2)
 233         super(Logger, self).__init__()
 234
 235     def notifyChannel(self, name, level, msg):
 236         warnings.warn("notifyChannel API shouldn't be used anymore, please use "
 237                       "the standard `logging` module instead",
 238                       PendingDeprecationWarning, stacklevel=2)
 239         from service.web_services import common
 240
 241         log = logging.getLogger(tools.ustr(name))
 242
 243         if level in [LOG_DEBUG_RPC, LOG_TEST] and not hasattr(log, level):
 244             fct = lambda msg, *args, **kwargs: log.log(getattr(logging, level.upper()), msg, *args, **kwargs)
 245             setattr(log, level, fct)
 246
 247
 248         level_method = getattr(log, level)
 249
 250         if isinstance(msg, Exception):
 251             msg = tools.exception_to_unicode(msg)
 252
 253         try:
 254             msg = tools.ustr(msg).strip()
 255             if level in (LOG_ERROR, LOG_CRITICAL) and tools.config.get_misc('debug','env_info',False):
 256                 msg = common().exp_get_server_environment() + "\n" + msg
 257
 258             result = msg.split('\n')
 259         except UnicodeDecodeError:
 260             result = msg.strip().split('\n')
 261         try:
 262             if len(result)>1:
 263                 for idx, s in enumerate(result):
 264                     level_method('[%02d]: %s' % (idx+1, s,))
 265             elif result:
 266                 level_method(result[0])
 267         except IOError,e:
 268             # TODO: perhaps reset the logger streams?
 269             #if logrotate closes our files, we end up here..
 270             pass
 271         except:
 272             # better ignore the exception and carry on..
 273             pass
 274
 275     def set_loglevel(self, level, logger=None):
 276         if logger is not None:
 277             log = logging.getLogger(str(logger))
 278         else:
 279             log = logging.getLogger()
 280         log.setLevel(logging.INFO) # make sure next msg is printed
 281         log.info("Log level changed to %s" % logging.getLevelName(level))
 282         log.setLevel(level)
 283
 284     def shutdown(self):
 285         logging.shutdown()
 286
 287 import tools
 288 init_logger()
 289
 290 class Agent(object):
 291     _timers = {}
 292     _logger = Logger()
 293
 294     __logger = logging.getLogger('timer')
 295
 296     def setAlarm(self, fn, dt, db_name, *args, **kwargs):
 297         wait = dt - time.time()
 298         if wait > 0:
 299             self.__logger.debug("Job scheduled in %.3g seconds for %s.%s" % (wait, fn.im_class.__name__, fn.func_name))
 300             timer = threading.Timer(wait, fn, args, kwargs)
 301             timer.start()
 302             self._timers.setdefault(db_name, []).append(timer)
 303
 304         for db in self._timers:
 305             for timer in self._timers[db]:
 306                 if not timer.isAlive():
 307                     self._timers[db].remove(timer)
 308
 309     @classmethod
 310     def cancel(cls, db_name):
 311         """Cancel all timers for a given database. If None passed, all timers are cancelled"""
 312         for db in cls._timers:
 313             if db_name is None or db == db_name:
 314                 for timer in cls._timers[db]:
 315                     timer.cancel()
 316
 317     @classmethod
 318     def quit(cls):
 319         cls.cancel(None)
 320
 321 import traceback
 322
 323 class Server:
 324     """ Generic interface for all servers with an event loop etc.
 325         Override this to impement http, net-rpc etc. servers.
 326
 327         Servers here must have threaded behaviour. start() must not block,
 328         there is no run().
 329     """
 330     __is_started = False
 331     __servers = []
 332     __starter_threads = []
 333
 334     # we don't want blocking server calls (think select()) to
 335     # wait forever and possibly prevent exiting the process,
 336     # but instead we want a form of polling/busy_wait pattern, where
 337     # _server_timeout should be used as the default timeout for
 338     # all I/O blocking operations
 339     _busywait_timeout = 0.5
 340
 341
 342     __logger = logging.getLogger('server')
 343
 344     def __init__(self):
 345         Server.__servers.append(self)
 346         if Server.__is_started:
 347             # raise Exception('All instances of servers must be inited before the startAll()')
 348             # Since the startAll() won't be called again, allow this server to
 349             # init and then start it after 1sec (hopefully). Register that
 350             # timer thread in a list, so that we can abort the start if quitAll
 351             # is called in the meantime
 352             t = threading.Timer(1.0, self._late_start)
 353             t.name = 'Late start timer for %s' % str(self.__class__)
 354             Server.__starter_threads.append(t)
 355             t.start()
 356
 357     def start(self):
 358         self.__logger.debug("called stub Server.start")
 359
 360     def _late_start(self):
 361         self.start()
 362         for thr in Server.__starter_threads:
 363             if thr.finished.is_set():
 364                 Server.__starter_threads.remove(thr)
 365
 366     def stop(self):
 367         self.__logger.debug("called stub Server.stop")
 368
 369     def stats(self):
 370         """ This function should return statistics about the server """
 371         return "%s: No statistics" % str(self.__class__)
 372
 373     @classmethod
 374     def startAll(cls):
 375         if cls.__is_started:
 376             return
 377         cls.__logger.info("Starting %d services" % len(cls.__servers))
 378         for srv in cls.__servers:
 379             srv.start()
 380         cls.__is_started = True
 381
 382     @classmethod
 383     def quitAll(cls):
 384         if not cls.__is_started:
 385             return
 386         cls.__logger.info("Stopping %d services" % len(cls.__servers))
 387         for thr in cls.__starter_threads:
 388             if not thr.finished.is_set():
 389                 thr.cancel()
 390             cls.__starter_threads.remove(thr)
 391
 392         for srv in cls.__servers:
 393             srv.stop()
 394         cls.__is_started = False
 395
 396     @classmethod
 397     def allStats(cls):
 398         res = ["Servers %s" % ('stopped', 'started')[cls.__is_started]]
 399         res.extend(srv.stats() for srv in cls.__servers)
 400         return '\n'.join(res)
 401
 402     def _close_socket(self):
 403         if os.name != 'nt':
 404             try:
 405                 self.socket.shutdown(getattr(socket, 'SHUT_RDWR', 2))
 406             except socket.error, e:
 407                 if e.errno != errno.ENOTCONN: raise
 408                 # OSX, socket shutdowns both sides if any side closes it
 409                 # causing an error 57 'Socket is not connected' on shutdown
 410                 # of the other side (or something), see
 411                 # http://bugs.python.org/issue4397
 412                 self.__logger.debug(
 413                     '"%s" when shutting down server socket, '
 414                     'this is normal under OS X', e)
 415         self.socket.close()
 416
 417 class OpenERPDispatcherException(Exception):
 418     def __init__(self, exception, traceback):
 419         self.exception = exception
 420         self.traceback = traceback
 421
 422 class OpenERPDispatcher:
 423     def log(self, title, msg):
 424         logger = logging.getLogger(title)
 425         if logger.isEnabledFor(logging.DEBUG_RPC):
 426             for line in pformat(msg).split('\n'):
 427                 logger.log(logging.DEBUG_RPC, line)
 428
 429     def dispatch(self, service_name, method, params):
 430         try:
 431             self.log('service', service_name)
 432             self.log('method', method)
 433             self.log('params', params)
 434             auth = getattr(self, 'auth_provider', None)
 435             result = ExportService.getService(service_name).dispatch(method, auth, params)
 436             self.log('result', result)
 437             # We shouldn't marshall None,
 438             if result == None:
 439                 result = False
 440             return result
 441         except Exception, e:
 442             self.log('exception', tools.exception_to_unicode(e))
 443             tb = getattr(e, 'traceback', sys.exc_info())
 444             tb_s = "".join(traceback.format_exception(*tb))
 445             if tools.config['debug_mode']:
 446                 import pdb
 447                 pdb.post_mortem(tb[2])
 448             raise OpenERPDispatcherException(e, tb_s)
 449
 450 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: