浏览代码

Hue [core] Hue log file randomly gets rotated and Hue log lines
goes missing. After using gunicorn base WSGI server, noticed the issue.

Gunicorn server starts multiple worker processes to handle requests.
Python logging module does not support logging to a single file from
multiple processes is not supported. In Python, there is no standard way
to serialize access to a single file across multiple processes in Python.
If you need to log to a single file from multiple processes, one way of
doing this is to have all the processes log to a SocketHandler, and have
a separate process that implements a socket server that reads from the
socket and logs to the file.

This change implements the fix. It uses a Unix domain socket file and
logging SocketHandler.

Issues Fixed:
- logfile gets rotated properly. logs messages are not overwritten.
UDS is local to the machine(not exposing TCPIP port)

Testing:
Execute listener command:
"sudo -u hue ./desktop/core/src/desktop/loglistener.py"

To generate log using hue code, execute below command:
"hue testloglistener"

Check the logs:
/var/log/hue/rungunicornserver.log

Prakash Ranade 2 年之前
父节点
当前提交
b15b1d4350

+ 115 - 0
desktop/conf/gunicorn_log.conf

@@ -0,0 +1,115 @@
+########################################
+# Definition for the different objects
+# - FOR DEVELOPMENT ONLY -
+#
+# Directories where log files are kept must already exist.
+# That's why we pick /tmp.
+#
+# The loggers are configured to write to the log files ONLY.
+# Developers may set the DESKTOP_DEBUG environment variable to
+# enable stderr logging output.
+########################################
+
+[logger_root]
+handlers=logfile,errorlog
+
+[logger_access]
+handlers=accesslog
+qualname=access
+
+[logger_django_auth_ldap]
+handlers=accesslog
+qualname=django_auth_ldap
+
+[logger_kazoo_client]
+level=INFO
+handlers=errorlog
+qualname=kazoo.client
+
+[logger_djangosaml2]
+level=INFO
+handlers=errorlog
+qualname=djangosaml2
+
+[logger_requests_packages_urllib3_connectionpool]
+level=DEBUG
+handlers=errorlog
+qualname=requests.packages.urllib3.connectionpool
+
+[logger_django_db]
+level=DEBUG
+handlers=errorlog
+qualname=django.db.backends
+
+[logger_boto]
+level=ERROR
+handlers=errorlog
+qualname=boto
+
+[logger_gunicorn.error]
+level=INFO
+handlers=errorlog
+propagate=1
+qualname=gunicorn.error
+
+[logger_gunicorn.access]
+level=INFO
+handlers=accesslog
+propagate=0
+qualname=gunicorn.access
+
+[handler_stderr]
+class=StreamHandler
+formatter=default
+level=DEBUG
+args=(sys.stderr,)
+
+[handler_stdout]
+class=StreamHandler
+formatter=default
+level=DEBUG
+args=(sys.stdout,)
+
+[handler_accesslog]
+class=handlers.RotatingFileHandler
+level=INFO
+propagate=True
+formatter=access
+args=('%LOG_DIR%/access.log', 'a', 5000000, 5)
+
+[handler_errorlog]
+class=handlers.RotatingFileHandler
+level=ERROR
+formatter=default
+args=('%LOG_DIR%/error.log', 'a', 5000000, 5)
+
+[handler_logfile]
+class=handlers.RotatingFileHandler
+# Choices are DEBUG, INFO, WARNING, ERROR, CRITICAL
+level=DEBUG
+formatter=default
+args=('%LOG_DIR%/rungunicornserver.log', 'a', 5000000, 5)
+
+[formatter_default]
+class=logging.Formatter
+format=[%(asctime)s] %(module)-12s %(levelname)-8s %(message)s
+datefmt=%d/%b/%Y %H:%M:%S %z
+
+[formatter_access]
+class=logging.Formatter
+format=[%(asctime)s] %(levelname)-8s %(message)s
+datefmt=%d/%b/%Y %H:%M:%S %z
+
+
+########################################
+# A summary of loggers, handlers and formatters
+########################################
+
+[loggers]
+keys=root,access,django_auth_ldap,kazoo_client,requests_packages_urllib3_connectionpool,djangosaml2,django_db,boto,gunicorn.error,gunicorn.access
+
+[handlers]
+keys=stderr,logfile,accesslog,errorlog,stdout
+
+[formatters]
+keys=default,access

+ 6 - 85
desktop/conf/log.conf

@@ -9,107 +9,28 @@
 # Developers may set the DESKTOP_DEBUG environment variable to
 # enable stderr logging output.
 ########################################
-
 [logger_root]
-handlers=logfile,errorlog
-
-[logger_access]
-handlers=accesslog
-qualname=access
-
-[logger_django_auth_ldap]
-handlers=accesslog
-qualname=django_auth_ldap
-
-[logger_kazoo_client]
-level=INFO
-handlers=errorlog
-qualname=kazoo.client
-
-[logger_djangosaml2]
-level=INFO
-handlers=errorlog
-qualname=djangosaml2
-
-[logger_requests_packages_urllib3_connectionpool]
-level=DEBUG
-handlers=errorlog
-qualname=requests.packages.urllib3.connectionpool
-
-[logger_django_db]
-level=DEBUG
-handlers=errorlog
-qualname=django.db.backends
-
-[logger_boto]
-level=ERROR
-handlers=errorlog
-qualname=boto
-
-[logger_gunicorn.error]
-level=INFO
-handlers=errorlog
-propagate=1
-qualname=gunicorn.error
-
-[logger_gunicorn.access]
-level=INFO
-handlers=accesslog
-propagate=0
-qualname=gunicorn.access
-
-[handler_stderr]
-class=StreamHandler
-formatter=default
-level=DEBUG
-args=(sys.stderr,)
-
-[handler_stdout]
-class=StreamHandler
-formatter=default
-level=DEBUG
-args=(sys.stdout,)
-
-[handler_accesslog]
-class=handlers.RotatingFileHandler
-level=INFO
-propagate=True
-formatter=access
-args=('%LOG_DIR%/access.log', 'a', 1000000, 3)
-
-[handler_errorlog]
-class=handlers.RotatingFileHandler
-level=ERROR
-formatter=default
-args=('%LOG_DIR%/error.log', 'a', 1000000, 3)
+handlers=logfile
 
 [handler_logfile]
-class=handlers.RotatingFileHandler
-# Choices are DEBUG, INFO, WARNING, ERROR, CRITICAL
 level=DEBUG
+class=handlers.SocketHandler
 formatter=default
-args=('%LOG_DIR%/%PROC_NAME%.log', 'a', 1000000, 3)
+args=("/tmp/hue.uds",None,)
 
 [formatter_default]
 class=desktop.log.formatter.Formatter
 format=[%(asctime)s] %(module)-12s %(levelname)-8s %(message)s
 datefmt=%d/%b/%Y %H:%M:%S %z
 
-[formatter_access]
-class=desktop.log.formatter.Formatter
-format=[%(asctime)s] %(levelname)-8s %(message)s
-datefmt=%d/%b/%Y %H:%M:%S %z
-
-
 ########################################
 # A summary of loggers, handlers and formatters
 ########################################
-
 [loggers]
-keys=root,access,django_auth_ldap,kazoo_client,requests_packages_urllib3_connectionpool,djangosaml2,django_db,boto,gunicorn.error,gunicorn.access
+keys=root
 
 [handlers]
-keys=stderr,logfile,accesslog,errorlog,stdout
+keys=logfile
 
 [formatters]
-keys=default,access
+keys=default

+ 214 - 0
desktop/core/src/desktop/loglistener.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3.8
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import argparse
+import logging
+import logging.config
+import os
+import pickle
+import re
+import select
+import signal
+import socket
+import stat
+import struct
+import sys
+import threading
+
+from io import StringIO as string_io
+from socketserver import ThreadingTCPServer, StreamRequestHandler
+#
+#   The following code implements a socket listener for on-the-fly
+#   reconfiguration of logging.
+#
+#   _listener holds the server object doing the listening
+_udslistener = None
+def udslisten(server_address='/tmp/hue.uds', verify=None):
+  """
+  Start up a socket server on the specified unix domain socket, and listen for new
+  configurations.
+  """
+  class ConfigStreamHandler(StreamRequestHandler):
+    """
+    Handler for a streaming logging request.
+    This basically logs the record using whatever logging policy is
+    configured locally.
+    """
+
+    def handle(self):
+      """
+      Handle multiple requests - each expected to be a 4-byte length,
+      followed by the LogRecord in pickle format. Logs the record
+      according to whatever policy is configured locally.
+      """
+      while True:
+        chunk = self.connection.recv(4)
+        if len(chunk) < 4:
+          break
+        slen = struct.unpack('>L', chunk)[0]
+        chunk = self.connection.recv(slen)
+        while len(chunk) < slen:
+          chunk = chunk + self.connection.recv(slen - len(chunk))
+        obj = self.unPickle(chunk)
+        record = logging.makeLogRecord(obj)
+        self.handleLogRecord(record)
+
+    def unPickle(self, data):
+      return pickle.loads(data)
+
+    def handleLogRecord(self, record):
+      # if a name is specified, we use the named logger rather than the one
+      # implied by the record.
+      if self.server.logname is not None:
+        name = self.server.logname
+      else:
+        name = record.name
+      logger = logging.getLogger(name)
+      logger.handle(record)
+
+  class ConfigSocketReceiver(ThreadingTCPServer):
+    """
+    A simple TCP socket-based logging config receiver.
+    """
+    request_queue_size = 1
+
+    def __init__(self, server_address='/tmp/hue.uds', handler=None,
+                 ready=None, verify=None):
+      ThreadingTCPServer.__init__(self, server_address, handler)
+      logging._acquireLock()
+      self.abort = 0
+      logging._releaseLock()
+      self.timeout = 1
+      self.ready = ready
+      self.verify = verify
+      self.logname = None
+      self.server_address = server_address
+
+    def server_bind(self):
+      # Create a UDS socket
+      self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+      self.socket.bind(self.server_address)
+      st = os.stat(self.server_address)
+      os.chmod(self.server_address, st.st_mode | stat.S_IWOTH)
+
+    def server_activate(self):
+      self.socket.listen(self.request_queue_size)
+
+    def serve_until_stopped(self):
+      abort = 0
+      while not abort:
+        rd, wr, ex = select.select([self.socket.fileno()],
+                       [], [],
+                       self.timeout)
+        if rd:
+          self.handle_request()
+        logging._acquireLock()
+        abort = self.abort
+        logging._releaseLock()
+      self.server_close()
+
+  class Server(threading.Thread):
+    def __init__(self, rcvr, hdlr, server_address, verify):
+      super(Server, self).__init__()
+      self.rcvr = rcvr
+      self.hdlr = hdlr
+      self.server_address = server_address
+      self.verify = verify
+      self.ready = threading.Event()
+
+    def run(self):
+      server = self.rcvr(server_address='/tmp/hue.uds', handler=self.hdlr,
+                         ready=self.ready, verify=self.verify)
+      self.ready.set()
+      global _udslistener
+      logging._acquireLock()
+      _udslistener = server
+      logging._releaseLock()
+      server.serve_until_stopped()
+
+  return Server(ConfigSocketReceiver, ConfigStreamHandler, server_address, verify)
+
+def udsstopListening():
+  """
+  Stop the listening server which was created with a call to listen().
+  """
+  global _udslistener
+  logging._acquireLock()
+  try:
+    if _udslistener:
+      _udslistener.abort = 1
+      _udslistener = None
+  finally:
+    logging._releaseLock()
+
+def argprocessing(args=[], options={}):
+  parser = argparse.ArgumentParser(prog='runloglistener', description='What this program does', epilog='Text at the bottom of help')
+  parser.add_argument('-s', '--socket', dest='socket', action='store', default='/tmp/hue.uds')
+
+  opts = parser.parse_args()
+  if opts.socket:
+    options['socket'] = opts.socket
+
+def enable_logging(args, options):
+  CONF_RE = re.compile('%LOG_DIR%')
+  CONF_FILE = os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(__file__)),
+                                           '..', '..', 'conf', 'gunicorn_log.conf'))
+  if os.path.exists(CONF_FILE):
+    log_dir = os.getenv("DESKTOP_LOG_DIR", "/var/log/hue")
+    raw = open(CONF_FILE).read()
+    def _repl(match):
+      if match.group(0) == '%LOG_DIR%':
+        return log_dir
+    sio = string_io(CONF_RE.sub(_repl, raw))
+    logging.config.fileConfig(sio)
+  root_logger = logging.getLogger()
+  root_logger.info("Welcome to Hue from Listener server ")
+
+class LogException(Exception):
+  def __init__(self, e):
+    super(LogException, self).__init__(e)
+    self.message = e.status.message
+
+  def __str__(self):
+    return self.message
+
+rt = None
+def signal_handler(sig, frame):
+  print("Received %s" % sig)
+  global rt
+  udsstopListening()
+  rt.join()
+
+def start_listener(args, options):
+  global rt
+  try:
+    os.unlink(options["socket"])
+  except OSError:
+    if os.path.exists(options["socket"]):
+      raise
+  signal.signal(signal.SIGTERM, signal_handler)
+  signal.signal(signal.SIGHUP, signal_handler)
+  signal.signal(signal.SIGINT, signal_handler)
+  signal.signal(signal.SIGQUIT, signal_handler)
+  enable_logging(args, options)
+  rt = udslisten(options["socket"], verify=False)
+  rt.start()
+
+if __name__ == '__main__':
+  args = sys.argv[1:]
+  options = {}
+  argprocessing(args=args, options=options)
+  start_listener(args, options)

+ 67 - 32
desktop/core/src/desktop/management/commands/rungunicornserver.py

@@ -14,34 +14,27 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 from __future__ import unicode_literals
-
-from OpenSSL import crypto
-
 import atexit
+import desktop.log
+import gunicorn.app.base
+import logging
+import logging.config
 import os
-import sys
+import pkg_resources
 import ssl
-import multiprocessing
+import sys
 import tempfile
 
-import gunicorn.app.base
-
+from OpenSSL import crypto
+from multiprocessing.util import _exit_function
 from desktop import conf
-from desktop import supervisor, metrics
 from django.core.management.base import BaseCommand
 from django.core.wsgi import get_wsgi_application
+from django.utils.translation import gettext as _
 from gunicorn import util
-from multiprocessing.util import _exit_function
 from six import iteritems
 
-if sys.version_info[0] > 2:
-  from django.utils.translation import gettext as _
-else:
-  from django.utils.translation import ugettext as _
-
-
 GUNICORN_SERVER_HELP = r"""
   Run Hue using the Gunicorn WSGI server in asynchronous mode.
 """
@@ -49,26 +42,47 @@ GUNICORN_SERVER_HELP = r"""
 class Command(BaseCommand):
   help = _("Gunicorn Web server for Hue.")
 
+  def add_arguments(self, parser):
+    parser.add_argument('--bind', help=_("Bind Address"), action='store', default=None)
+
   def handle(self, *args, **options):
-    rungunicornserver()
+    start_server(args, options)
 
   def usage(self, subcommand):
     return GUNICORN_SERVER_HELP
 
+def activate_translation():
+  from django.conf import settings
+  from django.utils import translation
+
+  # Activate the current language, because it won't get activated later.
+  try:
+    translation.activate(settings.LANGUAGE_CODE)
+  except AttributeError:
+    pass
+
 def number_of_workers():
   return (multiprocessing.cpu_count() * 2) + 1
 
-
 def handler_app(environ, start_response):
   os.environ.setdefault("DJANGO_SETTINGS_MODULE", "desktop.settings")
   return get_wsgi_application()
 
-def post_worker_init(worker):
-  atexit.unregister(_exit_function)
+def enable_logging(args, options):
+  HUE_DESKTOP_VERSION = pkg_resources.get_distribution("desktop").version or "Unknown"
+  # Start basic logging as soon as possible.
+  if "HUE_PROCESS_NAME" not in os.environ:
+    _proc = os.path.basename(len(sys.argv) > 1 and sys.argv[1] or sys.argv[0])
+    os.environ["HUE_PROCESS_NAME"] = _proc
 
+  desktop.log.basic_logging(os.environ["HUE_PROCESS_NAME"], log_dir=None, conf_file='log.conf')
+  logging.info("Welcome to Hue from Gunicorn server " + HUE_DESKTOP_VERSION)
 
-class StandaloneApplication(gunicorn.app.base.BaseApplication):
+def post_fork(server, worker):
+  with open("/tmp/gunicorn_workers.pid", "a") as f:
+    f.write("%s\n"%worker.pid)
 
+class StandaloneApplication(gunicorn.app.base.BaseApplication):
   def __init__(self, app, options=None):
     self.options = options or {}
     self.app_uri = 'desktop.wsgi:application'
@@ -97,8 +111,12 @@ class StandaloneApplication(gunicorn.app.base.BaseApplication):
   def load(self):
     return self.load_wsgiapp()
 
-def rungunicornserver():
-  bind_addr = conf.HTTP_HOST.get() + ":" + str(conf.HTTP_PORT.get())
+def argprocessing(args=[], options={}):
+  if options['bind']:
+    bind_addr = options['bind']
+  else:
+    bind_addr = conf.HTTP_HOST.get() + ":" + str(conf.HTTP_PORT.get())
+  options['bind_addr'] = bind_addr
 
   # Currently gunicorn does not support passphrase suppored SSL Keyfile
   # https://github.com/benoitc/gunicorn/issues/2410
@@ -115,11 +133,13 @@ def rungunicornserver():
           ssl_keyfile = tf.name
     else:
       ssl_keyfile = conf.SSL_PRIVATE_KEY.get()
+  options['ssl_keyfile'] = ssl_keyfile
 
-  options = {
+def rungunicornserver(args=[], options={}):
+  gunicorn_options = {
       'accesslog': "-",
       'backlog': 2048,
-      'bind': [bind_addr],
+      'bind': [options['bind_addr']],
       'ca_certs': conf.SSL_CACERTS.get(),     # CA certificates file
       'capture_output': True,
       'cert_reqs': None,                      # Whether client certificate is required (see stdlib ssl module)
@@ -137,11 +157,11 @@ def rungunicornserver():
       'group': conf.SERVER_GROUP.get(),
       'initgroups': None,
       'keepalive': 120,                       # seconds to wait for requests on a keep-alive connection.
-      'keyfile': ssl_keyfile,                 # SSL key file
+      'keyfile': options['ssl_keyfile'],      # SSL key file
       'limit_request_field_size': conf.LIMIT_REQUEST_FIELD_SIZE.get(),
       'limit_request_fields': conf.LIMIT_REQUEST_FIELDS.get(),
       'limit_request_line': conf.LIMIT_REQUEST_LINE.get(),
-      'logconfig': None,
+      'logconfig': '/etc/hue/conf/log.conf',
       'loglevel': 'info',
       'max_requests': 1200,                   # The maximum number of requests a worker will process before restarting.
       'max_requests_jitter': 0,
@@ -156,7 +176,7 @@ def rungunicornserver():
       'raw_paste_global_conf': None,
       'reload': None,
       'reload_engine': None,
-      'sendfile': None,
+      'sendfile': True,
       'spew': None,
       'ssl_version': ssl.PROTOCOL_TLSv1_2,    # SSL version to use
       'statsd_host': None,
@@ -173,10 +193,25 @@ def rungunicornserver():
       'worker_class': conf.GUNICORN_WORKER_CLASS.get(),
       'worker_connections': 1000,
       'worker_tmp_dir': None,
-      'workers': conf.GUNICORN_NUMBER_OF_WORKERS.get() if conf.GUNICORN_NUMBER_OF_WORKERS.get() is not None else number_of_workers(),
-      'post_worker_init': post_worker_init
+      'workers': conf.GUNICORN_NUMBER_OF_WORKERS.get() if conf.GUNICORN_NUMBER_OF_WORKERS.get() is not None else 5,
+      'post_fork': post_fork
   }
-  StandaloneApplication(handler_app, options).run()
+  StandaloneApplication(handler_app, gunicorn_options).run()
+
+def start_server(args, options):
+  argprocessing(args, options)
+
+  # Hide the Server software version in the response body
+  gunicorn.SERVER_SOFTWARE = "apache"
+  os.environ["SERVER_SOFTWARE"] = gunicorn.SERVER_SOFTWARE
+
+  # Activate django translation
+  activate_translation()
+  enable_logging(args, options)
+  with open("/tmp/gunicorn_workers.pid", "w") as f:
+    f.write("%s\n"%os.getpid())
+  atexit.unregister(_exit_function)
+  rungunicornserver(args, options)
 
 if __name__ == '__main__':
-  rungunicornserver()
+  start_server(args=sys.argv[1:], options={})

+ 75 - 0
desktop/core/src/desktop/management/commands/testloglistener.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import argparse
+import desktop.log
+import logging
+import logging.config
+import os
+import pkg_resources
+import sys
+
+from django.core.management.base import BaseCommand
+from django.utils.translation import gettext as _
+
+SERVER_HELP = r"""
+  Run Python log tester
+"""
+def argprocessing(args=[], options={}):
+  parser = argparse.ArgumentParser(prog='testloglistener', description='What this program does', epilog='Text at the bottom of help')
+  parser.add_argument('-s', '--socket', dest='socket', action='store', default='/tmp/hue.uds')
+
+  opts = parser.parse_args()
+  if opts.socket:
+    options['socket'] = opts.socket
+
+def enable_logging(args, options):
+  HUE_DESKTOP_VERSION = pkg_resources.get_distribution("desktop").version or "Unknown"
+  # Start basic logging as soon as possible.
+  if "HUE_PROCESS_NAME" not in os.environ:
+    _proc = os.path.basename(len(sys.argv) > 1 and sys.argv[1] or sys.argv[0])
+    os.environ["HUE_PROCESS_NAME"] = _proc
+
+  desktop.log.basic_logging(os.environ["HUE_PROCESS_NAME"])
+  logging.info("Welcome to Hue from Listener server " + HUE_DESKTOP_VERSION)
+
+class Command(BaseCommand):
+  help = _("Web server for Hue.")
+
+  def add_arguments(self, parser):
+    parser.add_argument('--socket', help=_("Unix Domain Socket file"), action='store', default=None)
+
+  def handle(self, *args, **options):
+    start_testing(args, options)
+
+  def usage(self, subcommand):
+    return SERVER_HELP
+
+def start_testing(args, options):
+  enable_logging(args, options)
+  logger = logging.getLogger()
+  i = 0
+  import time
+  while i < 100:
+    logger.info("Hue log testing %s" % i)
+    i += 1
+    time.sleep(1)
+
+if __name__ == '__main__':
+  args = sys.argv[1:]
+  options = {}
+  argprocessing(args=args, options=options)
+  start_testing(args, options)