Browse Source

[backend] Add minimal test suite for loglistener functionality

To run : ./build/env/bin/python3.11 -m unittest desktop.lib.gunicorn_loglistener_minimal_test -v

- Created 5 essential tests covering core loglistener features
- Tests log record processing, thread lifecycle, start/stop operations, recovery files, and signal handlers
Prakash Ranade 2 months ago
parent
commit
f7d517ea89

+ 252 - 252
desktop/core/src/desktop/lib/gunicorn_cleanup_utils.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python
 # Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
+# or more contributor license agreements. See the NOTICE file
 # distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# 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
+# with the License. You may obtain a copy of the License at
 #
-#     http://www.apache.org/licenses/LICENSE-2.0
+#   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,
@@ -28,289 +28,289 @@ import os
 
 
 def write_process_recovery_file(main_pid=None, socket_path=None, pid_file=None):
-    """
-    Write process information to a recovery file for post-SIGKILL cleanup
-    """
-    try:
-        import psutil
-
-        recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
-
-        if main_pid is None:
-            current_process = psutil.Process()
-            main_pid = current_process.pid
-        else:
-            current_process = psutil.Process(main_pid)
-
-        process_info = {
-            "main_pid": main_pid,
-            "start_time": current_process.create_time(),
-            "socket_path": socket_path,
-            "pid_file": pid_file,
-            "children": []
-        }
-
-        # Record all child processes
-        for child in current_process.children(recursive=True):
-            try:
-                process_info["children"].append({
-                    "pid": child.pid,
-                    "name": child.name(),
-                    "cmdline": child.cmdline(),
-                    "start_time": child.create_time()
-                })
-            except (psutil.NoSuchProcess, psutil.AccessDenied):
-                continue
-
-        with open(recovery_file, 'w') as f:
-            json.dump(process_info, f, indent=2)
-
-        logging.debug(f"Written recovery file: {recovery_file}")
-
-    except Exception as e:
-        logging.error(f"Error writing recovery file: {e}")
+  """
+  Write process information to a recovery file for post-SIGKILL cleanup
+  """
+  try:
+    import psutil
+
+    recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
+
+    if main_pid is None:
+      current_process = psutil.Process()
+      main_pid = current_process.pid
+    else:
+      current_process = psutil.Process(main_pid)
+
+    process_info = {
+      "main_pid": main_pid,
+      "start_time": current_process.create_time(),
+      "socket_path": socket_path,
+      "pid_file": pid_file,
+      "children": []
+    }
+
+    # Record all child processes
+    for child in current_process.children(recursive=True):
+      try:
+        process_info["children"].append({
+          "pid": child.pid,
+          "name": child.name(),
+          "cmdline": child.cmdline(),
+          "start_time": child.create_time()
+        })
+      except (psutil.NoSuchProcess, psutil.AccessDenied):
+        continue
+
+    with open(recovery_file, 'w') as f:
+      json.dump(process_info, f, indent=2)
+
+    logging.debug(f"Written recovery file: {recovery_file}")
+
+  except Exception as e:
+    logging.error(f"Error writing recovery file: {e}")
 
 
 def cleanup_from_recovery_file():
-    """
-    Clean up processes using information from recovery file (for post-SIGKILL cleanup)
-    """
-    try:
-        import psutil
+  """
+  Clean up processes using information from recovery file (for post-SIGKILL cleanup)
+  """
+  try:
+    import psutil
 
-        recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
+    recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
 
-        if not os.path.exists(recovery_file):
-            logging.info("No recovery file found")
-            return
+    if not os.path.exists(recovery_file):
+      logging.info("No recovery file found")
+      return
 
-        with open(recovery_file, 'r') as f:
-            process_info = json.load(f)
+    with open(recovery_file, 'r') as f:
+      process_info = json.load(f)
 
-        logging.info(f"Processing recovery file for main PID {process_info['main_pid']}")
+    logging.info(f"Processing recovery file for main PID {process_info['main_pid']}")
 
-        # Check if main process is still running
-        try:
-            main_proc = psutil.Process(process_info['main_pid'])
-            if main_proc.is_running():
-                logging.info(f"Main process {process_info['main_pid']} is still running, skipping cleanup")
-                return
-        except psutil.NoSuchProcess:
-            logging.info(f"Main process {process_info['main_pid']} is no longer running, proceeding with cleanup")
-
-        # Clean up child processes
-        cleaned_count = 0
-        for child_info in process_info['children']:
-            try:
-                child_proc = psutil.Process(child_info['pid'])
-                if child_proc.is_running():
-                    # Verify it's the same process (check start time)
-                    if abs(child_proc.create_time() - child_info['start_time']) < 1.0:
-                        logging.info(f"Terminating orphaned child process: PID {child_info['pid']}, CMD: {' '.join(child_info['cmdline'])}")
-                        child_proc.terminate()
-                        try:
-                            child_proc.wait(timeout=5)
-                            cleaned_count += 1
-                        except psutil.TimeoutExpired:
-                            logging.warning(f"Force killing child process: PID {child_info['pid']}")
-                            child_proc.kill()
-                            cleaned_count += 1
-            except (psutil.NoSuchProcess, psutil.AccessDenied):
-                logging.info(f"Child process {child_info['pid']} is no longer running, proceeding with cleanup")
-                continue
-
-        # Clean up socket file
-        if process_info.get('socket_path') and os.path.exists(process_info['socket_path']):
-            try:
-                os.unlink(process_info['socket_path'])
-                logging.info(f"Cleaned up socket file: {process_info['socket_path']}")
-            except OSError as e:
-                logging.warning(f"Could not clean up socket file: {e}")
-
-        # Clean up PID file
-        if process_info.get('pid_file') and os.path.exists(process_info['pid_file']):
+    # Check if main process is still running
+    try:
+      main_proc = psutil.Process(process_info['main_pid'])
+      if main_proc.is_running():
+        logging.info(f"Main process {process_info['main_pid']} is still running, skipping cleanup")
+        return
+    except psutil.NoSuchProcess:
+      logging.info(f"Main process {process_info['main_pid']} is no longer running, proceeding with cleanup")
+
+    # Clean up child processes
+    cleaned_count = 0
+    for child_info in process_info['children']:
+      try:
+        child_proc = psutil.Process(child_info['pid'])
+        if child_proc.is_running():
+          # Verify it's the same process (check start time)
+          if abs(child_proc.create_time() - child_info['start_time']) < 1.0:
+            logging.info(f"Terminating orphaned child process: PID {child_info['pid']}, CMD: {' '.join(child_info['cmdline'])}")
+            child_proc.terminate()
             try:
-                os.unlink(process_info['pid_file'])
-                logging.info(f"Cleaned up PID file: {process_info['pid_file']}")
-            except OSError as e:
-                logging.warning(f"Could not clean up PID file: {e}")
-
-        # Remove recovery file
-        os.unlink(recovery_file)
-        logging.info(f"Recovery cleanup completed: {cleaned_count} processes cleaned up")
-
-    except Exception as e:
-        logging.error(f"Error during recovery cleanup: {e}")
+              child_proc.wait(timeout=5)
+              cleaned_count += 1
+            except psutil.TimeoutExpired:
+              logging.warning(f"Force killing child process: PID {child_info['pid']}")
+              child_proc.kill()
+              cleaned_count += 1
+      except (psutil.NoSuchProcess, psutil.AccessDenied):
+        logging.info(f"Child process {child_info['pid']} is no longer running, proceeding with cleanup")
+        continue
+
+    # Clean up socket file
+    if process_info.get('socket_path') and os.path.exists(process_info['socket_path']):
+      try:
+        os.unlink(process_info['socket_path'])
+        logging.info(f"Cleaned up socket file: {process_info['socket_path']}")
+      except OSError as e:
+        logging.warning(f"Could not clean up socket file: {e}")
+
+    # Clean up PID file
+    if process_info.get('pid_file') and os.path.exists(process_info['pid_file']):
+      try:
+        os.unlink(process_info['pid_file'])
+        logging.info(f"Cleaned up PID file: {process_info['pid_file']}")
+      except OSError as e:
+        logging.warning(f"Could not clean up PID file: {e}")
+
+    # Remove recovery file
+    os.unlink(recovery_file)
+    logging.info(f"Recovery cleanup completed: {cleaned_count} processes cleaned up")
+
+  except Exception as e:
+    logging.error(f"Error during recovery cleanup: {e}")
 
 
 def cleanup_orphaned_processes():
-    """
-    Clean up any orphaned processes that might be left behind
-    """
-    try:
-        import psutil
-        current_pid = os.getpid()
-
-        # Look for processes that might be related to this Hue instance
-        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
-            try:
-                if proc.info['pid'] == current_pid:
-                    continue
-
-                cmdline = ' '.join(proc.info['cmdline'] or [])
-
-                # Check if this looks like a related Hue process
-                if ('rungunicornserver' in cmdline or
-                    'hue' in proc.info['name'].lower() or
-                    ('python' in proc.info['name'].lower() and 'hue' in cmdline)):
-
-                    # Check if it's actually orphaned (parent is init or doesn't exist)
-                    try:
-                        parent = proc.parent()
-                        if parent is None or parent.pid == 1:
-                            logging.warning(f"Found orphaned process: PID {proc.info['pid']}, CMD: {cmdline}")
-                            proc.terminate()
-                            # Give it time to terminate gracefully
-                            try:
-                                proc.wait(timeout=5)
-                                logging.info(f"Successfully terminated orphaned process: PID {proc.info['pid']}")
-                            except psutil.TimeoutExpired:
-                                logging.warning(f"Force killing orphaned process: PID {proc.info['pid']}")
-                                proc.kill()
-                    except (psutil.NoSuchProcess, psutil.AccessDenied):
-                        continue
-
-            except (psutil.NoSuchProcess, psutil.AccessDenied, IndexError):
-                continue
-
-    except Exception as e:
-        logging.error(f"Error during orphaned process cleanup: {e}")
+  """
+  Clean up any orphaned processes that might be left behind
+  """
+  try:
+    import psutil
+    current_pid = os.getpid()
+
+    # Look for processes that might be related to this Hue instance
+    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
+      try:
+        if proc.info['pid'] == current_pid:
+          continue
+
+        cmdline = ' '.join(proc.info['cmdline'] or [])
+
+        # Check if this looks like a related Hue process
+        if ('rungunicornserver' in cmdline or
+          'hue' in proc.info['name'].lower() or
+          ('python' in proc.info['name'].lower() and 'hue' in cmdline)):
+
+          # Check if it's actually orphaned (parent is init or doesn't exist)
+          try:
+            parent = proc.parent()
+            if parent is None or parent.pid == 1:
+              logging.warning(f"Found orphaned process: PID {proc.info['pid']}, CMD: {cmdline}")
+              proc.terminate()
+              # Give it time to terminate gracefully
+              try:
+                proc.wait(timeout=5)
+                logging.info(f"Successfully terminated orphaned process: PID {proc.info['pid']}")
+              except psutil.TimeoutExpired:
+                logging.warning(f"Force killing orphaned process: PID {proc.info['pid']}")
+                proc.kill()
+          except (psutil.NoSuchProcess, psutil.AccessDenied):
+            continue
+
+      except (psutil.NoSuchProcess, psutil.AccessDenied, IndexError):
+        continue
+
+  except Exception as e:
+    logging.error(f"Error during orphaned process cleanup: {e}")
 
 
 def unified_cleanup():
-    """
-    Unified cleanup function that handles both recovery file cleanup and general orphan cleanup
-    """
-    logging.info("Starting unified cleanup process...")
-
-    # Step 1: Try recovery file cleanup first (more precise)
-    try:
-        recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
-        if os.path.exists(recovery_file):
-            logging.info("Found recovery file, attempting recovery cleanup...")
-            cleanup_from_recovery_file()
-            return  # If recovery cleanup worked, we're done
-        else:
-            logging.info("No recovery file found, proceeding with general cleanup...")
-    except Exception as e:
-        logging.warning(f"Recovery cleanup failed: {e}, falling back to general cleanup...")
-
-    # Step 2: Fall back to general orphan cleanup
-    cleanup_orphaned_processes()
+  """
+  Unified cleanup function that handles both recovery file cleanup and general orphan cleanup
+  """
+  logging.info("Starting unified cleanup process...")
+
+  # Step 1: Try recovery file cleanup first (more precise)
+  try:
+    recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
+    if os.path.exists(recovery_file):
+      logging.info("Found recovery file, attempting recovery cleanup...")
+      cleanup_from_recovery_file()
+      return  # If recovery cleanup worked, we're done
+    else:
+      logging.info("No recovery file found, proceeding with general cleanup...")
+  except Exception as e:
+    logging.warning(f"Recovery cleanup failed: {e}, falling back to general cleanup...")
+
+  # Step 2: Fall back to general orphan cleanup
+  cleanup_orphaned_processes()
 
 
 def cleanup_recovery_file():
-    """
-    Clean up recovery file on normal shutdown
-    """
-    try:
-        recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
-        if os.path.exists(recovery_file):
-            os.unlink(recovery_file)
-            logging.info("Cleaned up recovery file")
-    except Exception as e:
-        logging.warning(f"Could not clean up recovery file: {e}")
+  """
+  Clean up recovery file on normal shutdown
+  """
+  try:
+    recovery_file = os.path.join(os.getenv("DESKTOP_LOG_DIR", "/var/log/hue"), "hue_recovery.json")
+    if os.path.exists(recovery_file):
+      os.unlink(recovery_file)
+      logging.info("Cleaned up recovery file")
+  except Exception as e:
+    logging.warning(f"Could not clean up recovery file: {e}")
 
 
 def terminate_child_processes():
-    """
-    Terminate all child processes (gunicorn workers) with timeout and force-kill fallback
-    """
-    try:
-        import psutil
-        current_process = psutil.Process()
-        children = current_process.children(recursive=True)
-
-        if children:
-            logging.info(f"Terminating {len(children)} child processes...")
-            for child in children:
-                try:
-                    child.terminate()
-                except psutil.NoSuchProcess:
-                    pass
-
-            # Wait for children to terminate gracefully
-            gone, alive = psutil.wait_procs(children, timeout=10)
-
-            # Force kill any remaining processes
-            if alive:
-                logging.warning(f"Force killing {len(alive)} remaining child processes...")
-                for child in alive:
-                    try:
-                        child.kill()
-                    except psutil.NoSuchProcess:
-                        pass
-    except Exception as e:
-        logging.error(f"Error terminating child processes: {e}")
+  """
+  Terminate all child processes (gunicorn workers) with timeout and force-kill fallback
+  """
+  try:
+    import psutil
+    current_process = psutil.Process()
+    children = current_process.children(recursive=True)
+
+    if children:
+      logging.info(f"Terminating {len(children)} child processes...")
+      for child in children:
+        try:
+          child.terminate()
+        except psutil.NoSuchProcess:
+          pass
+
+      # Wait for children to terminate gracefully
+      gone, alive = psutil.wait_procs(children, timeout=10)
+
+      # Force kill any remaining processes
+      if alive:
+        logging.warning(f"Force killing {len(alive)} remaining child processes...")
+        for child in alive:
+          try:
+            child.kill()
+          except psutil.NoSuchProcess:
+            pass
+  except Exception as e:
+    logging.error(f"Error terminating child processes: {e}")
 
 
 def cleanup_pid_file(pid_file):
-    """
-    Clean up PID file
-    """
-    if pid_file and os.path.exists(pid_file):
-        try:
-            os.unlink(pid_file)
-            logging.info(f"Cleaned up PID file: {pid_file}")
-        except OSError as e:
-            logging.warning(f"Could not clean up PID file {pid_file}: {e}")
+  """
+  Clean up PID file
+  """
+  if pid_file and os.path.exists(pid_file):
+    try:
+      os.unlink(pid_file)
+      logging.info(f"Cleaned up PID file: {pid_file}")
+    except OSError as e:
+      logging.warning(f"Could not clean up PID file {pid_file}: {e}")
 
 
 def cleanup_socket_file(socket_path):
-    """
-    Clean up socket file
-    """
-    if socket_path and os.path.exists(socket_path):
-        try:
-            os.unlink(socket_path)
-            logging.info(f"Cleaned up socket file: {socket_path}")
-        except OSError as e:
-            logging.warning(f"Could not clean up socket file {socket_path}: {e}")
+  """
+  Clean up socket file
+  """
+  if socket_path and os.path.exists(socket_path):
+    try:
+      os.unlink(socket_path)
+      logging.info(f"Cleaned up socket file: {socket_path}")
+    except OSError as e:
+      logging.warning(f"Could not clean up socket file {socket_path}: {e}")
 
 
 def create_signal_handler(log_listener_thread, socket_path, pid_file):
+  """
+  Create a signal handler function with the necessary cleanup context
+  """
+  def signal_handler_with_log_listener(sig, frame):
     """
-    Create a signal handler function with the necessary cleanup context
+    Signal handler that properly shuts down both gunicorn and log listener
     """
-    def signal_handler_with_log_listener(sig, frame):
-        """
-        Signal handler that properly shuts down both gunicorn and log listener
-        """
-        try:
-            logging.info(f"Received signal {sig}, shutting down log listener and server...")
-
-            # Stop log listener first
-            if log_listener_thread:
-                logging.info("Stopping log listener thread...")
-                log_listener_thread.stop()
-                log_listener_thread.join(timeout=5)  # Wait up to 5 seconds for clean shutdown
-                logging.info("Log listener thread stopped")
-        except Exception as e:
-            logging.error(f"Error stopping log listener: {e}")
+    try:
+      logging.info(f"Received signal {sig}, shutting down log listener and server...")
+
+      # Stop log listener first
+      if log_listener_thread:
+        logging.info("Stopping log listener thread...")
+        log_listener_thread.stop()
+        log_listener_thread.join(timeout=5)  # Wait up to 5 seconds for clean shutdown
+        logging.info("Log listener thread stopped")
+    except Exception as e:
+      logging.error(f"Error stopping log listener: {e}")
 
-        # Clean up socket file
-        cleanup_socket_file(socket_path)
+    # Clean up socket file
+    cleanup_socket_file(socket_path)
 
-        # Clean up PID file
-        cleanup_pid_file(pid_file)
+    # Clean up PID file
+    cleanup_pid_file(pid_file)
 
-        # Terminate all child processes (gunicorn workers)
-        terminate_child_processes()
+    # Terminate all child processes (gunicorn workers)
+    terminate_child_processes()
 
-        # Clean up recovery file on normal shutdown
-        cleanup_recovery_file()
+    # Clean up recovery file on normal shutdown
+    cleanup_recovery_file()
 
-        logging.info("Graceful shutdown completed")
-        os._exit(0)  # Use os._exit to avoid cleanup handlers that might hang
+    logging.info("Graceful shutdown completed")
+    os._exit(0)  # Use os._exit to avoid cleanup handlers that might hang
 
-    return signal_handler_with_log_listener
+  return signal_handler_with_log_listener

+ 177 - 177
desktop/core/src/desktop/lib/gunicorn_log_utils.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python
 # Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
+# or more contributor license agreements. See the NOTICE file
 # distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# 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
+# with the License. You may obtain a copy of the License at
 #
-#     http://www.apache.org/licenses/LICENSE-2.0
+#   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,
@@ -43,199 +43,199 @@ LOG_LISTENER_SOCKET_PATH = None
 
 
 class ConfigStreamHandler(StreamRequestHandler):
+  """
+  Handler for a streaming logging request.
+  This basically logs the record using whatever logging policy is
+  configured locally.
+  """
+  def handle(self):
     """
-    Handler for a streaming logging request.
-    This basically logs the record using whatever logging policy is
-    configured locally.
+    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.
     """
-    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 self.server.ready:
-            try:
-                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 = pickle.loads(chunk)
-                record = logging.makeLogRecord(obj)
-                self.handleLogRecord(record)
-            except Exception as e:
-                logging.debug(f"Error handling log record: {e}")
-                break
-
-    def handleLogRecord(self, record):
-        """Handle a single log 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)
+    while self.server.ready:
+      try:
+        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 = pickle.loads(chunk)
+        record = logging.makeLogRecord(obj)
+        self.handleLogRecord(record)
+      except Exception as e:
+        logging.debug(f"Error handling log record: {e}")
+        break
+
+  def handleLogRecord(self, record):
+    """Handle a single log 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='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 and bind Unix Domain 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):
-        """Activate the server"""
-        self.socket.listen(self.request_queue_size)
-
-    def serve_until_stopped(self):
-        """Serve requests until stopped"""
-        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()
+  """
+  A simple TCP socket-based logging config receiver.
+  """
+  request_queue_size = 1
+
+  def __init__(self, server_address='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 and bind Unix Domain 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):
+    """Activate the server"""
+    self.socket.listen(self.request_queue_size)
+
+  def serve_until_stopped(self):
+    """Serve requests until stopped"""
+    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 LogListenerThread(threading.Thread):
-    """
-    Thread that runs the log listener server
-    """
-    def __init__(self, server_address, verify=None):
-        super(LogListenerThread, self).__init__()
-        self.server_address = server_address
-        self.verify = verify
-        self.ready = threading.Event()
-        self.daemon = True  # Daemon thread will be killed when main process exits
-
-    def run(self):
-        """Run the log listener server"""
-        global LOG_LISTENER_SERVER
-        server = ConfigSocketReceiver(server_address=self.server_address,
-                                      handler=ConfigStreamHandler,
-                                      ready=self.ready, verify=self.verify)
-        self.ready.set()
-        logging._acquireLock()
-        LOG_LISTENER_SERVER = server
-        logging._releaseLock()
-        server.serve_until_stopped()
-
-    def stop(self):
-        """Stop the log listener"""
-        stop_log_listener()
-        self.ready.clear()
-
-    def stopped(self):
-        """Check if the listener is stopped"""
-        return not self.ready.is_set()
-
-
-def stop_log_listener():
-    """
-    Stop the listening server which was created with start_log_listener().
-    """
+  """
+  Thread that runs the log listener server
+  """
+  def __init__(self, server_address, verify=None):
+    super(LogListenerThread, self).__init__()
+    self.server_address = server_address
+    self.verify = verify
+    self.ready = threading.Event()
+    self.daemon = True  # Daemon thread will be killed when main process exits
+
+  def run(self):
+    """Run the log listener server"""
     global LOG_LISTENER_SERVER
+    server = ConfigSocketReceiver(server_address=self.server_address,
+                   handler=ConfigStreamHandler,
+                   ready=self.ready, verify=self.verify)
+    self.ready.set()
     logging._acquireLock()
-    try:
-        if LOG_LISTENER_SERVER:
-            LOG_LISTENER_SERVER.abort = 1
-            LOG_LISTENER_SERVER = None
-    finally:
-        logging._releaseLock()
+    LOG_LISTENER_SERVER = server
+    logging._releaseLock()
+    server.serve_until_stopped()
 
+  def stop(self):
+    """Stop the log listener"""
+    stop_log_listener()
+    self.ready.clear()
+
+  def stopped(self):
+    """Check if the listener is stopped"""
+    return not self.ready.is_set()
 
-def start_log_listener(socket_path=None):
-    """
-    Start up a socket server on the specified unix domain socket, and listen for new
-    configurations from gunicorn worker processes.
-    """
-    global LOG_LISTENER_THREAD, LOG_LISTENER_SOCKET_PATH
 
-    if not socket_path:
-        socket_dir = os.getenv("DESKTOP_LOG_DIR", "/var/log/hue")
-        socket_name = conf.LOG_LISTENER_SOCKET_NAME.get()
-        socket_path = os.path.join(socket_dir, socket_name)
+def stop_log_listener():
+  """
+  Stop the listening server which was created with start_log_listener().
+  """
+  global LOG_LISTENER_SERVER
+  logging._acquireLock()
+  try:
+    if LOG_LISTENER_SERVER:
+      LOG_LISTENER_SERVER.abort = 1
+      LOG_LISTENER_SERVER = None
+  finally:
+    logging._releaseLock()
 
-    # Store socket path globally for signal handler access
-    LOG_LISTENER_SOCKET_PATH = socket_path
 
-    # Remove existing socket file if it exists
+def start_log_listener(socket_path=None):
+  """
+  Start up a socket server on the specified unix domain socket, and listen for new
+  configurations from gunicorn worker processes.
+  """
+  global LOG_LISTENER_THREAD, LOG_LISTENER_SOCKET_PATH
+
+  if not socket_path:
+    socket_dir = os.getenv("DESKTOP_LOG_DIR", "/var/log/hue")
+    socket_name = conf.LOG_LISTENER_SOCKET_NAME.get()
+    socket_path = os.path.join(socket_dir, socket_name)
+
+  # Store socket path globally for signal handler access
+  LOG_LISTENER_SOCKET_PATH = socket_path
+
+  # Remove existing socket file if it exists
+  try:
+    os.unlink(socket_path)
+  except OSError:
+    if os.path.exists(socket_path):
+      logging.warning(f"Could not remove existing socket file: {socket_path}")
+
+  # Ensure the directory exists
+  socket_dir = os.path.dirname(socket_path)
+  if not os.path.exists(socket_dir):
     try:
-        os.unlink(socket_path)
-    except OSError:
-        if os.path.exists(socket_path):
-            logging.warning(f"Could not remove existing socket file: {socket_path}")
-
-    # Ensure the directory exists
-    socket_dir = os.path.dirname(socket_path)
-    if not os.path.exists(socket_dir):
-        try:
-            os.makedirs(socket_dir)
-            logging.info(f"Created log directory: {socket_dir}")
-        except OSError as e:
-            logging.error(f"Failed to create log directory {socket_dir}: {e}")
-            raise
+      os.makedirs(socket_dir)
+      logging.info(f"Created log directory: {socket_dir}")
+    except OSError as e:
+      logging.error(f"Failed to create log directory {socket_dir}: {e}")
+      raise
 
-    # Configure logging for the listener
-    enable_log_listener_logging(socket_path)
+  # Configure logging for the listener
+  enable_log_listener_logging(socket_path)
 
-    # Start the listener thread
-    LOG_LISTENER_THREAD = LogListenerThread(server_address=socket_path, verify=None)
-    LOG_LISTENER_THREAD.start()
+  # Start the listener thread
+  LOG_LISTENER_THREAD = LogListenerThread(server_address=socket_path, verify=None)
+  LOG_LISTENER_THREAD.start()
 
-    # Wait for the thread to be ready
-    LOG_LISTENER_THREAD.ready.wait()
+  # Wait for the thread to be ready
+  LOG_LISTENER_THREAD.ready.wait()
 
-    logging.info(f"Log listener started on socket: {socket_path}")
-    return LOG_LISTENER_THREAD
+  logging.info(f"Log listener started on socket: {socket_path}")
+  return LOG_LISTENER_THREAD
 
 
 def enable_log_listener_logging(socket_path):
-    """
-    Configure logging for the log listener functionality
-    """
-    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("Starting integrated Hue Log Listener using socket file %s" % socket_path)
-    root_logger.info("Using logging.conf file %s" % CONF_FILE)
+  """
+  Configure logging for the log listener functionality
+  """
+  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("Starting integrated Hue Log Listener using socket file %s" % socket_path)
+  root_logger.info("Using logging.conf file %s" % CONF_FILE)

+ 215 - 0
desktop/core/src/desktop/lib/gunicorn_loglistener_minimal_test.py

@@ -0,0 +1,215 @@
+#!/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.
+
+"""
+Minimal test suite for log listener functionality - 5 essential tests
+"""
+
+import logging
+import os
+import pickle
+import struct
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+
+from desktop.lib.gunicorn_cleanup_utils import cleanup_from_recovery_file, create_signal_handler, write_process_recovery_file
+from desktop.lib.gunicorn_log_utils import ConfigStreamHandler, LogListenerThread, start_log_listener, stop_log_listener
+
+
+class TestMinimalLogListener(unittest.TestCase):
+  """Minimal test suite covering essential loglistener functionality"""
+
+  def setUp(self):
+    self.temp_dir = tempfile.mkdtemp()
+    self.socket_path = os.path.join(self.temp_dir, 'test.sock')
+
+  def tearDown(self):
+    try:
+      stop_log_listener()
+    except Exception:
+      pass
+
+    import shutil
+    shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+  def test_1_log_record_handling(self):
+    """Test 1: Core log record processing functionality"""
+    # Test ConfigStreamHandler can process log records
+    mock_connection = Mock()
+    mock_server = Mock()
+    mock_server.ready = True  # Allow processing
+    mock_server.logname = None
+
+    handler = ConfigStreamHandler(
+      mock_connection,
+      ('127.0.0.1', 12345),
+      mock_server
+    )
+
+    # Create a test log record
+    record = logging.LogRecord(
+      name='test.logger',
+      level=logging.INFO,
+      pathname='/test/path.py',
+      lineno=42,
+      msg='Test message',
+      args=(),
+      exc_info=None
+    )
+
+    # Serialize the record
+    pickled_record = pickle.dumps(record.__dict__)
+    record_length = struct.pack('>L', len(pickled_record))
+
+    # Mock connection.recv to return our data, then stop server
+    def mock_recv_side_effect(size):
+      if not hasattr(mock_recv_side_effect, 'call_count'):
+        mock_recv_side_effect.call_count = 0
+      mock_recv_side_effect.call_count += 1
+
+      if mock_recv_side_effect.call_count == 1:
+        return record_length
+      elif mock_recv_side_effect.call_count == 2:
+        return pickled_record
+      else:
+        # Stop server after processing
+        mock_server.ready = False
+        return b''
+
+    mock_connection.recv.side_effect = mock_recv_side_effect
+
+    # Mock the logger
+    with patch('logging.getLogger') as mock_get_logger:
+      mock_logger = Mock()
+      mock_get_logger.return_value = mock_logger
+
+      # Call handle method
+      handler.handle()
+
+      # Verify logger was called
+      mock_get_logger.assert_called_with('test.logger')
+      mock_logger.handle.assert_called_once()
+
+  def test_2_thread_lifecycle(self):
+    """Test 2: Log listener thread start/stop lifecycle"""
+    thread = LogListenerThread(self.socket_path)
+
+    # Test initialization
+    self.assertEqual(thread.server_address, self.socket_path)
+    self.assertFalse(thread.ready.is_set())
+    self.assertTrue(thread.daemon)
+
+    # Test stopped status
+    self.assertTrue(thread.stopped())
+
+  @patch('desktop.lib.gunicorn_log_utils.LogListenerThread')
+  @patch('desktop.lib.gunicorn_log_utils.enable_log_listener_logging')
+  def test_3_start_stop_log_listener(self, mock_enable_logging, mock_thread_class):
+    """Test 3: Start and stop log listener functionality"""
+    # Mock the thread
+    mock_thread = Mock()
+    mock_thread.ready.wait.return_value = True
+    mock_thread_class.return_value = mock_thread
+
+    # Start log listener
+    thread = start_log_listener(self.socket_path)
+
+    # Verify listener was created and started
+    mock_thread_class.assert_called_once_with(server_address=self.socket_path, verify=None)
+    self.assertEqual(thread, mock_thread)
+    mock_thread.start.assert_called_once()
+    mock_thread.ready.wait.assert_called_once()
+    mock_enable_logging.assert_called_once_with(self.socket_path)
+
+    # Test stop functionality
+    stop_log_listener()  # Should complete without errors
+
+  @patch('psutil.Process')
+  def test_4_recovery_file_operations(self, mock_process_class):
+    """Test 4: Process recovery file creation and cleanup"""
+    # Mock process
+    mock_process = Mock()
+    mock_process.pid = 12345
+    mock_process.create_time.return_value = 1234567890.0
+    mock_process.children.return_value = []
+    mock_process_class.return_value = mock_process
+
+    recovery_file = os.path.join(self.temp_dir, 'hue_recovery.json')
+
+    # Test writing recovery file
+    with patch.dict(os.environ, {'DESKTOP_LOG_DIR': self.temp_dir}):
+      write_process_recovery_file(
+        socket_path=self.socket_path,
+        pid_file='/tmp/test.pid'
+      )
+
+    # Verify recovery file was created
+    self.assertTrue(os.path.exists(recovery_file))
+
+    # Test cleanup from recovery file when no process is running
+    mock_process.is_running.return_value = False
+    mock_process_class.side_effect = lambda pid: mock_process
+
+    with patch.dict(os.environ, {'DESKTOP_LOG_DIR': self.temp_dir}):
+      cleanup_from_recovery_file()
+
+    # Recovery file should be cleaned up
+    self.assertFalse(os.path.exists(recovery_file))
+
+  def test_5_signal_handler_integration(self):
+    """Test 5: Signal handler creation and integration"""
+    mock_log_listener = Mock()
+    test_socket = os.path.join(self.temp_dir, 'signal_test.sock')
+    test_pid_file = os.path.join(self.temp_dir, 'signal_test.pid')
+
+    # Create test files
+    with open(test_socket, 'w') as f:
+      f.write('')
+    with open(test_pid_file, 'w') as f:
+      f.write('12345')
+
+    # Create signal handler
+    handler = create_signal_handler(
+      mock_log_listener,
+      test_socket,
+      test_pid_file
+    )
+
+    # Verify handler is callable
+    self.assertTrue(callable(handler))
+
+    # Test signal handler execution
+    with patch('os._exit') as mock_exit:
+      with patch('desktop.lib.gunicorn_cleanup_utils.terminate_child_processes'):
+        with patch('desktop.lib.gunicorn_cleanup_utils.cleanup_recovery_file'):
+          handler(15, None)  # SIGTERM
+
+    # Verify log listener was stopped
+    mock_log_listener.stop.assert_called_once()
+    mock_log_listener.join.assert_called_with(timeout=5)
+
+    # Verify files were cleaned up
+    self.assertFalse(os.path.exists(test_socket))
+    self.assertFalse(os.path.exists(test_pid_file))
+
+    # Verify exit was called
+    mock_exit.assert_called_with(0)
+
+
+if __name__ == '__main__':
+  unittest.main()

+ 263 - 263
desktop/core/src/desktop/lib/gunicorn_server_utils.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python
 # Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
+# or more contributor license agreements. See the NOTICE file
 # distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# 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
+# with the License. You may obtain a copy of the License at
 #
-#     http://www.apache.org/licenses/LICENSE-2.0
+#   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,
@@ -49,318 +49,318 @@ from filebrowser.utils import parse_broker_url
 
 
 def activate_translation():
-    """Activate Django translation"""
-    from django.conf import settings
-    from django.utils import translation
+  """Activate Django 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
+  # Activate the current language, because it won't get activated later.
+  try:
+    translation.activate(settings.LANGUAGE_CODE)
+  except AttributeError:
+    pass
 
 
 def number_of_workers():
-    """Calculate the number of workers for Gunicorn"""
-    return (multiprocessing.cpu_count() * 2) + 1
+  """Calculate the number of workers for Gunicorn"""
+  return (multiprocessing.cpu_count() * 2) + 1
 
 
 def handler_app(environ, start_response):
-    """WSGI application handler"""
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "desktop.settings")
-    return get_wsgi_application()
+  """WSGI application handler"""
+  os.environ.setdefault("DJANGO_SETTINGS_MODULE", "desktop.settings")
+  return get_wsgi_application()
 
 
 def post_fork(server, worker):
-    """Post-fork worker initialization"""
-    global PID_FILE
-    with open(PID_FILE, "a") as f:
-        f.write("%s\n" % worker.pid)
+  """Post-fork worker initialization"""
+  global PID_FILE
+  with open(PID_FILE, "a") as f:
+    f.write("%s\n" % worker.pid)
 
 
 def post_worker_init(worker):
-    """Initialize worker after fork"""
-    connection.connect()
+  """Initialize worker after fork"""
+  connection.connect()
 
 
 def worker_int(worker):
-    """Worker interrupt handler"""
-    connection.close()
+  """Worker interrupt handler"""
+  connection.close()
 
 
 def enable_logging(args, options):
-    """Enable basic logging for the process"""
-    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
+  """Enable basic logging for the process"""
+  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 Gunicorn server " + HUE_DESKTOP_VERSION)
+  desktop.log.basic_logging(os.environ["HUE_PROCESS_NAME"])
+  logging.info("Welcome to Hue from Gunicorn server " + HUE_DESKTOP_VERSION)
 
 
 def initialize_free_disk_space_in_redis():
-    """Initialize free disk space tracking in Redis"""
-    conn_success = False
-    for retries in range(5):
-        try:
-            redis_client = parse_broker_url(desktop.conf.TASK_SERVER_V2.BROKER_URL.get())
-            free_space = psutil.disk_usage('/tmp').free
-            available_space = redis_client.get('upload_available_space')
-            if available_space is None:
-                available_space = free_space
-            else:
-                available_space = int(available_space)
-            upload_keys_exist = any(redis_client.scan_iter('upload__*'))
-            redis_client.delete('upload_available_space')
-            if not upload_keys_exist:
-                redis_client.setnx('upload_available_space', free_space)
-            else:
-                redis_client.setnx('upload_available_space', min(free_space, available_space))
-            logging.info("Successfully initialized free disk space in Redis.")
-            conn_success = True
-            break
-        except redis.ConnectionError as e:
-            logging.error(f"Redis connection error: {e}")
-            time.sleep(10)
-        except Exception as e:
-            logging.error(f"Error while initializing free disk space in Redis: {e}")
-            time.sleep(10)
-    if not conn_success:
-        logging.error("Failed to initialize free disk space in Redis after 5 retries.")
+  """Initialize free disk space tracking in Redis"""
+  conn_success = False
+  for retries in range(5):
+    try:
+      redis_client = parse_broker_url(desktop.conf.TASK_SERVER_V2.BROKER_URL.get())
+      free_space = psutil.disk_usage('/tmp').free
+      available_space = redis_client.get('upload_available_space')
+      if available_space is None:
+        available_space = free_space
+      else:
+        available_space = int(available_space)
+      upload_keys_exist = any(redis_client.scan_iter('upload__*'))
+      redis_client.delete('upload_available_space')
+      if not upload_keys_exist:
+        redis_client.setnx('upload_available_space', free_space)
+      else:
+        redis_client.setnx('upload_available_space', min(free_space, available_space))
+      logging.info("Successfully initialized free disk space in Redis.")
+      conn_success = True
+      break
+    except redis.ConnectionError as e:
+      logging.error(f"Redis connection error: {e}")
+      time.sleep(10)
+    except Exception as e:
+      logging.error(f"Error while initializing free disk space in Redis: {e}")
+      time.sleep(10)
+  if not conn_success:
+    logging.error("Failed to initialize free disk space in Redis after 5 retries.")
 
 
 class StandaloneApplication(gunicorn.app.base.BaseApplication):
-    """Standalone Gunicorn application"""
+  """Standalone Gunicorn application"""
 
-    def __init__(self, app, options=None):
-        self.options = options or {}
-        self.app_uri = 'desktop.wsgi:application'
-        super(StandaloneApplication, self).__init__()
+  def __init__(self, app, options=None):
+    self.options = options or {}
+    self.app_uri = 'desktop.wsgi:application'
+    super(StandaloneApplication, self).__init__()
 
-    def load_config(self):
-        config = dict([(key, value) for key, value in iteritems(self.options)
-                       if key in self.cfg.settings and value is not None])
-        for key, value in iteritems(config):
-            self.cfg.set(key.lower(), value)
+  def load_config(self):
+    config = dict([(key, value) for key, value in iteritems(self.options)
+            if key in self.cfg.settings and value is not None])
+    for key, value in iteritems(config):
+      self.cfg.set(key.lower(), value)
 
-    def chdir(self):
-        # chdir to the configured path before loading,
-        # default is the current dir
-        os.chdir(self.cfg.chdir)
+  def chdir(self):
+    # chdir to the configured path before loading,
+    # default is the current dir
+    os.chdir(self.cfg.chdir)
 
-        # add the path to sys.path
-        sys.path.insert(0, self.cfg.chdir)
+    # add the path to sys.path
+    sys.path.insert(0, self.cfg.chdir)
 
-    def load_wsgiapp(self):
-        self.chdir()
+  def load_wsgiapp(self):
+    self.chdir()
 
-        # load the app
-        return util.import_app(self.app_uri)
+    # load the app
+    return util.import_app(self.app_uri)
 
-    def load(self):
-        return self.load_wsgiapp()
+  def load(self):
+    return self.load_wsgiapp()
 
 
 def process_arguments(args=[], options={}):
-    """Process command line arguments and options"""
-    global PID_FILE
-
-    ipv6_enabled = conf.ENABLE_IPV6.get()  # This is already a bool
-
-    if options.get('bind'):
-        http_port = "8888"
-        bind_addr = options['bind']
-        if ":" in bind_addr:
-            http_port = bind_addr.split(":")[-1]  # Use last part in case of IPv6
-        PID_FILE = f"/tmp/hue_{http_port}.pid"
+  """Process command line arguments and options"""
+  global PID_FILE
+
+  ipv6_enabled = conf.ENABLE_IPV6.get()  # This is already a bool
+
+  if options.get('bind'):
+    http_port = "8888"
+    bind_addr = options['bind']
+    if ":" in bind_addr:
+      http_port = bind_addr.split(":")[-1]  # Use last part in case of IPv6
+    PID_FILE = f"/tmp/hue_{http_port}.pid"
+  else:
+    http_host = conf.HTTP_HOST.get()
+    http_port = str(conf.HTTP_PORT.get())
+
+    if ipv6_enabled:
+      bind_addr = fetch_ipv6_bind_address(http_host, http_port)
     else:
-        http_host = conf.HTTP_HOST.get()
-        http_port = str(conf.HTTP_PORT.get())
-
-        if ipv6_enabled:
-            bind_addr = fetch_ipv6_bind_address(http_host, http_port)
-        else:
-            bind_addr = f"{http_host}:{http_port}"
-            logging.info(f"IPv6 disabled, using standard format: {bind_addr}")
+      bind_addr = f"{http_host}:{http_port}"
+      logging.info(f"IPv6 disabled, using standard format: {bind_addr}")
 
-        PID_FILE = f"/tmp/hue_{http_port}.pid"
+    PID_FILE = f"/tmp/hue_{http_port}.pid"
 
-    options['bind_addr'] = bind_addr
-    options['pid_file'] = PID_FILE  # Store PID file in options for return
+  options['bind_addr'] = bind_addr
+  options['pid_file'] = PID_FILE  # Store PID file in options for return
 
-    # Currently gunicorn does not support passphrase suppored SSL Keyfile
-    # https://github.com/benoitc/gunicorn/issues/2410
-    ssl_keyfile = None
+  # Currently gunicorn does not support passphrase suppored SSL Keyfile
+  # https://github.com/benoitc/gunicorn/issues/2410
+  ssl_keyfile = None
 
-    # HUE_RUN_DIR is set for env variable. If unset, fall back to previous behaviour (for base).
-    worker_tmp_dir = os.environ.get("HUE_RUN_DIR")
+  # HUE_RUN_DIR is set for env variable. If unset, fall back to previous behaviour (for base).
+  worker_tmp_dir = os.environ.get("HUE_RUN_DIR")
+  if not worker_tmp_dir:
+    worker_tmp_dir = os.environ.get("HUE_CONF_DIR", get_desktop_root("conf"))
     if not worker_tmp_dir:
-        worker_tmp_dir = os.environ.get("HUE_CONF_DIR", get_desktop_root("conf"))
-        if not worker_tmp_dir:
-            worker_tmp_dir = "/tmp"
-    options['worker_tmp_dir'] = worker_tmp_dir
-
-    # Gunicorn needs chown privileges if the OS user/group does not match Gunicorn's user/group.
-    # This can happen when the OS user is not in our control to set (OpenShift).
-    if not os.environ.get("GUNICORN_USE_OS_USER"):
-        options['user'] = conf.SERVER_USER.get()
-        options['group'] = conf.SERVER_GROUP.get()
-
-    if conf.SSL_CERTIFICATE.get() and conf.SSL_PRIVATE_KEY.get():
-        ssl_password = str.encode(conf.get_ssl_password()) if conf.get_ssl_password() is not None else None
-        if ssl_password:
-            with open(conf.SSL_PRIVATE_KEY.get(), 'r') as f:
-                with tempfile.NamedTemporaryFile(dir=worker_tmp_dir, delete=False) as tf:
-                    tf.write(crypto.dump_privatekey(crypto.FILETYPE_PEM,
-                                                    crypto.load_privatekey(crypto.FILETYPE_PEM,
-                                                                           f.read(), ssl_password)))
-                    ssl_keyfile = tf.name
-        else:
-            ssl_keyfile = conf.SSL_PRIVATE_KEY.get()
-    options['ssl_keyfile'] = ssl_keyfile
+      worker_tmp_dir = "/tmp"
+  options['worker_tmp_dir'] = worker_tmp_dir
+
+  # Gunicorn needs chown privileges if the OS user/group does not match Gunicorn's user/group.
+  # This can happen when the OS user is not in our control to set (OpenShift).
+  if not os.environ.get("GUNICORN_USE_OS_USER"):
+    options['user'] = conf.SERVER_USER.get()
+    options['group'] = conf.SERVER_GROUP.get()
+
+  if conf.SSL_CERTIFICATE.get() and conf.SSL_PRIVATE_KEY.get():
+    ssl_password = str.encode(conf.get_ssl_password()) if conf.get_ssl_password() is not None else None
+    if ssl_password:
+      with open(conf.SSL_PRIVATE_KEY.get(), 'r') as f:
+        with tempfile.NamedTemporaryFile(dir=worker_tmp_dir, delete=False) as tf:
+          tf.write(crypto.dump_privatekey(crypto.FILETYPE_PEM,
+                          crypto.load_privatekey(crypto.FILETYPE_PEM,
+                                      f.read(), ssl_password)))
+          ssl_keyfile = tf.name
+    else:
+      ssl_keyfile = conf.SSL_PRIVATE_KEY.get()
+  options['ssl_keyfile'] = ssl_keyfile
 
-    return options
+  return options
 
 
 def create_gunicorn_options(options):
-    """Create Gunicorn configuration options"""
-    # Get TLS configuration
-    tls_settings = get_tls_settings()
+  """Create Gunicorn configuration options"""
+  # Get TLS configuration
+  tls_settings = get_tls_settings()
 
-    def gunicorn_ssl_context(conf, default_ssl_context_factory):
-        """
-        Create and configure SSL context for Gunicorn based on TLS settings.
-        """
-        context = default_ssl_context_factory()
-
-        try:
-            # Validate TLS settings before applying
-            if "error" in tls_settings:
-                logging.warning(f"TLS configuration error: {tls_settings['error']}")
-                return context
-
-            # Configure maximum TLS version
-            max_version = tls_settings.get("tls_maximum_version")
-            if max_version == "TLSv1.3":
-                if hasattr(ssl, "HAS_TLSv1_3") and ssl.HAS_TLSv1_3 and hasattr(ssl, "TLSVersion"):
-                    context.maximum_version = ssl.TLSVersion.TLSv1_3
-                    logging.info("Set maximum TLS version to TLSv1.3")
-                else:
-                    logging.warning("TLS 1.3 requested but not supported by system, falling back to TLS 1.2")
-                    if hasattr(ssl, "TLSVersion"):
-                        context.maximum_version = ssl.TLSVersion.TLSv1_2
-            elif max_version == "TLSv1.2":
-                if hasattr(ssl, "TLSVersion"):
-                    context.maximum_version = ssl.TLSVersion.TLSv1_2
-                    logging.info("Set maximum TLS version to TLSv1.2")
-
-            # Configure minimum TLS version
-            min_version = tls_settings.get("tls_minimum_version")
-            if min_version == "TLSv1.3":
-                if hasattr(ssl, "HAS_TLSv1_3") and ssl.HAS_TLSv1_3 and hasattr(ssl, "TLSVersion"):
-                    context.minimum_version = ssl.TLSVersion.TLSv1_3
-                    logging.info("Set minimum TLS version to TLSv1.3")
-                else:
-                    logging.warning("TLS 1.3 minimum requested but not supported, falling back to TLS 1.2")
-                    if hasattr(ssl, "TLSVersion"):
-                        context.minimum_version = ssl.TLSVersion.TLSv1_2
-            elif min_version == "TLSv1.2":
-                if hasattr(ssl, "TLSVersion"):
-                    context.minimum_version = ssl.TLSVersion.TLSv1_2
-                    logging.info("Set minimum TLS version to TLSv1.2")
-
-            # Configure ciphers (only for TLS 1.2)
-            ciphers = tls_settings.get("ciphers", "")
-            if ciphers and ciphers.strip():  # Only set if ciphers is not empty
-                try:
-                    context.set_ciphers(ciphers)
-                    logging.info(f"Successfully configured ciphers: {ciphers}")
-                except ssl.SSLError as e:
-                    logging.error(f"Invalid cipher configuration '{ciphers}': {e}")
-                except Exception as e:
-                    logging.error(f"Unexpected error setting ciphers '{ciphers}': {e}")
-            elif max_version == "TLSv1.3":
-                logging.info("TLS 1.3 enabled - cipher configuration handled automatically")
-            else:
-                logging.info("No custom ciphers configured, using system defaults")
-
-        except Exception as e:
-            logging.error(f"Error configuring SSL context: {e}")
-            logging.info("Using default SSL context configuration")
+  def gunicorn_ssl_context(conf, default_ssl_context_factory):
+    """
+    Create and configure SSL context for Gunicorn based on TLS settings.
+    """
+    context = default_ssl_context_factory()
 
+    try:
+      # Validate TLS settings before applying
+      if "error" in tls_settings:
+        logging.warning(f"TLS configuration error: {tls_settings['error']}")
         return context
 
-    gunicorn_options = {
-        'accesslog': "-",
-        'access_log_format': "%({x-forwarded-for}i)s %(h)s %(l)s %(u)s %(t)s '%(r)s' %(s)s %(b)s '%(f)s' '%(a)s'",
-        'backlog': 2048,
-        '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)
-        'certfile': conf.SSL_CERTIFICATE.get(),  # SSL certificate file
-        'chdir': None,
-        'check_config': None,
-        'config': None,
-        'daemon': None,
-        'do_handshake_on_connect': False,       # Whether to perform SSL handshake on socket connect.
-        'enable_stdio_inheritance': None,
-        'errorlog': "-",
-        'forwarded_allow_ips': None,
-        'graceful_timeout': conf.GUNICORN_WORKER_GRACEFUL_TIMEOUT.get(),
-        'group': options.get('group'),
-        'initgroups': None,
-        'keepalive': 120,                       # seconds to wait for requests on a keep-alive connection.
-        '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(),
-        'loglevel': 'DEBUG' if conf.DJANGO_DEBUG_MODE.get() else 'INFO',
-        'max_requests': 1200,                   # The maximum number of requests a worker will process before restarting.
-        'max_requests_jitter': 0,
-        'paste': None,
-        'pidfile': None,
-        'preload_app': False,
-        'proc_name': "hue",
-        'proxy_allow_ips': None,
-        'proxy_protocol': None,
-        'pythonpath': None,
-        'raw_env': None,
-        'raw_paste_global_conf': None,
-        'reload': None,
-        'reload_engine': None,
-        'sendfile': True,
-        'spew': None,
-        'ssl_context': gunicorn_ssl_context,
-        'statsd_host': None,
-        'statsd_prefix': None,
-        'suppress_ragged_eofs': None,           # Suppress ragged EOFs (see stdlib ssl module)
-        'syslog': None,
-        'syslog_addr': None,
-        'syslog_facility': None,
-        'syslog_prefix': None,
-        'threads': conf.CHERRYPY_SERVER_THREADS.get(),
-        'timeout': conf.GUNICORN_WORKER_TIMEOUT.get(),
-        'umask': None,
-        'user': options.get('user'),
-        'worker_class': conf.GUNICORN_WORKER_CLASS.get(),
-        'worker_connections': 1000,
-        'worker_tmp_dir': options['worker_tmp_dir'],
-        'workers': conf.GUNICORN_NUMBER_OF_WORKERS.get() if conf.GUNICORN_NUMBER_OF_WORKERS.get() is not None else 5,
-        'post_fork': post_fork,
-        'post_worker_init': post_worker_init,
-        'worker_int': worker_int
-    }
-
-    return gunicorn_options
+      # Configure maximum TLS version
+      max_version = tls_settings.get("tls_maximum_version")
+      if max_version == "TLSv1.3":
+        if hasattr(ssl, "HAS_TLSv1_3") and ssl.HAS_TLSv1_3 and hasattr(ssl, "TLSVersion"):
+          context.maximum_version = ssl.TLSVersion.TLSv1_3
+          logging.info("Set maximum TLS version to TLSv1.3")
+        else:
+          logging.warning("TLS 1.3 requested but not supported by system, falling back to TLS 1.2")
+          if hasattr(ssl, "TLSVersion"):
+            context.maximum_version = ssl.TLSVersion.TLSv1_2
+      elif max_version == "TLSv1.2":
+        if hasattr(ssl, "TLSVersion"):
+          context.maximum_version = ssl.TLSVersion.TLSv1_2
+          logging.info("Set maximum TLS version to TLSv1.2")
+
+      # Configure minimum TLS version
+      min_version = tls_settings.get("tls_minimum_version")
+      if min_version == "TLSv1.3":
+        if hasattr(ssl, "HAS_TLSv1_3") and ssl.HAS_TLSv1_3 and hasattr(ssl, "TLSVersion"):
+          context.minimum_version = ssl.TLSVersion.TLSv1_3
+          logging.info("Set minimum TLS version to TLSv1.3")
+        else:
+          logging.warning("TLS 1.3 minimum requested but not supported, falling back to TLS 1.2")
+          if hasattr(ssl, "TLSVersion"):
+            context.minimum_version = ssl.TLSVersion.TLSv1_2
+      elif min_version == "TLSv1.2":
+        if hasattr(ssl, "TLSVersion"):
+          context.minimum_version = ssl.TLSVersion.TLSv1_2
+          logging.info("Set minimum TLS version to TLSv1.2")
+
+      # Configure ciphers (only for TLS 1.2)
+      ciphers = tls_settings.get("ciphers", "")
+      if ciphers and ciphers.strip():  # Only set if ciphers is not empty
+        try:
+          context.set_ciphers(ciphers)
+          logging.info(f"Successfully configured ciphers: {ciphers}")
+        except ssl.SSLError as e:
+          logging.error(f"Invalid cipher configuration '{ciphers}': {e}")
+        except Exception as e:
+          logging.error(f"Unexpected error setting ciphers '{ciphers}': {e}")
+      elif max_version == "TLSv1.3":
+        logging.info("TLS 1.3 enabled - cipher configuration handled automatically")
+      else:
+        logging.info("No custom ciphers configured, using system defaults")
+
+    except Exception as e:
+      logging.error(f"Error configuring SSL context: {e}")
+      logging.info("Using default SSL context configuration")
+
+    return context
+
+  gunicorn_options = {
+    'accesslog': "-",
+    'access_log_format': "%({x-forwarded-for}i)s %(h)s %(l)s %(u)s %(t)s '%(r)s' %(s)s %(b)s '%(f)s' '%(a)s'",
+    'backlog': 2048,
+    '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)
+    'certfile': conf.SSL_CERTIFICATE.get(),  # SSL certificate file
+    'chdir': None,
+    'check_config': None,
+    'config': None,
+    'daemon': None,
+    'do_handshake_on_connect': False,    # Whether to perform SSL handshake on socket connect.
+    'enable_stdio_inheritance': None,
+    'errorlog': "-",
+    'forwarded_allow_ips': None,
+    'graceful_timeout': conf.GUNICORN_WORKER_GRACEFUL_TIMEOUT.get(),
+    'group': options.get('group'),
+    'initgroups': None,
+    'keepalive': 120,            # seconds to wait for requests on a keep-alive connection.
+    '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(),
+    'loglevel': 'DEBUG' if conf.DJANGO_DEBUG_MODE.get() else 'INFO',
+    'max_requests': 1200,          # The maximum number of requests a worker will process before restarting.
+    'max_requests_jitter': 0,
+    'paste': None,
+    'pidfile': None,
+    'preload_app': False,
+    'proc_name': "hue",
+    'proxy_allow_ips': None,
+    'proxy_protocol': None,
+    'pythonpath': None,
+    'raw_env': None,
+    'raw_paste_global_conf': None,
+    'reload': None,
+    'reload_engine': None,
+    'sendfile': True,
+    'spew': None,
+    'ssl_context': gunicorn_ssl_context,
+    'statsd_host': None,
+    'statsd_prefix': None,
+    'suppress_ragged_eofs': None,      # Suppress ragged EOFs (see stdlib ssl module)
+    'syslog': None,
+    'syslog_addr': None,
+    'syslog_facility': None,
+    'syslog_prefix': None,
+    'threads': conf.CHERRYPY_SERVER_THREADS.get(),
+    'timeout': conf.GUNICORN_WORKER_TIMEOUT.get(),
+    'umask': None,
+    'user': options.get('user'),
+    'worker_class': conf.GUNICORN_WORKER_CLASS.get(),
+    'worker_connections': 1000,
+    'worker_tmp_dir': options['worker_tmp_dir'],
+    'workers': conf.GUNICORN_NUMBER_OF_WORKERS.get() if conf.GUNICORN_NUMBER_OF_WORKERS.get() is not None else 5,
+    'post_fork': post_fork,
+    'post_worker_init': post_worker_init,
+    'worker_int': worker_int
+  }
+
+  return gunicorn_options
 
 
 def run_gunicorn_server(args=[], options={}):
-    """Run the Gunicorn server with the given options"""
-    gunicorn_options = create_gunicorn_options(options)
-    StandaloneApplication(handler_app, gunicorn_options).run()
+  """Run the Gunicorn server with the given options"""
+  gunicorn_options = create_gunicorn_options(options)
+  StandaloneApplication(handler_app, gunicorn_options).run()
 
 
 # Global variable for PID file (used by post_fork)