feat: use supervisord to manage the processes

.github/workflows/integration-test.yml

@@ -48,6 +48,8 @@ jobs:
               - 'project.toml'
               - 'requirements.txt'
               - 'logger_config.yaml'
+              - 'logger_config_em.yaml'
+              - 'supervisord.conf'
               - '.github/workflows/integration-test.yml'
 
   integration:
@@ -181,6 +183,7 @@ jobs:
           pip install -r requirements.txt
           cp example.env .env
           echo "NEXTCLOUD_URL=http://localhost:8080" >> .env
+          python3 -u ./main_em.py > em_backend_logs 2>&1 &
           python3 -u ./main.py > backend_logs 2>&1 &
           echo $! > ../pid.txt          # Save the process ID (PID)
 
@@ -260,7 +263,8 @@ jobs:
         run: |
           cat data/nextcloud.log
           echo '--------------------------------------------------'
-          cat context_chat_backend/backend_logs || echo "No backend logs"
+          cat context_chat_backend/backend_logs || echo "No main backend logs"
+          cat context_chat_backend/em_backend_logs || echo "No main backend logs"
           echo '--------------------------------------------------'
           tail -v -n +1 context_chat_backend/persistent_storage/logs/* || echo "No logs in logs directory"
 

Dockerfile

@@ -46,6 +46,8 @@ COPY main.py .
 COPY main_em.py .
 COPY config.?pu.yaml .
 COPY logger_config.yaml .
+COPY logger_config_em.yaml .
 COPY hwdetect.sh .
+COPY supervisord.conf /etc/supervisor/supervisord.conf
 
-ENTRYPOINT [ "./dockerfile_scripts/entrypoint.sh" ]
+ENTRYPOINT ["supervisord", "-c", "/etc/supervisor/supervisord.conf"]

context_chat_backend/dyn_loader.py

@@ -6,17 +6,11 @@
 
 import gc
 import logging
-import multiprocessing as mp
-import os
-import signal
-import subprocess
 from abc import ABC, abstractmethod
-from datetime import datetime
 from time import sleep, time
 from typing import Any
 
 import httpx
-import psutil
 import torch
 from fastapi import FastAPI
 from langchain.llms.base import LLM
@@ -39,45 +33,21 @@ def load(self) -> Any:
 	def offload(self):
 		...
 
-pid = mp.Value('i', 0)
 
 class EmbeddingModelLoader(Loader):
 	def __init__(self, config: TConfig):
 		self.config = config
-		logfile_path = os.path.join(
-			os.environ['EM_SERVER_LOG_PATH'],
-			f'embedding_server_{datetime.now().strftime("%Y-%m-%d")}.log',
-		)
-		self.logfile = open(logfile_path, 'a+')
 
 	def load(self):
-		global pid
-
 		emconf = self.config.embedding
-
-		# start the embedding server if workers are > 0
-		if emconf.workers > 0:
-			# compare with None, as PID can be 0, you never know
-			if pid.value > 0 and psutil.pid_exists(pid.value):
-				return
-
-			proc = subprocess.Popen(  # noqa: S603
-				['./main_em.py'],
-				stdout=self.logfile,
-				stderr=self.logfile,
-				stdin=None,
-				close_fds=True,
-				env=os.environ,
-			)
-			pid.value = proc.pid
-
 		last_resp, last_exc = None, None
 		# poll for heartbeat
 		try_ = 0
-		while try_ < 20:
-			with httpx.Client() as client:
+		with httpx.Client() as client:
+			while try_ < 20:
 				try:
 					# test the server is up
+					# todo: replace with a tcp connection check
 					response = client.post(
 						f'{emconf.protocol}://{emconf.host}:{emconf.port}/v1/embeddings',
 						json={'input': 'hello'},
@@ -98,10 +68,7 @@ def load(self):
 		) from last_exc
 
 	def offload(self):
-		global pid
-		if pid.value > 0 and psutil.pid_exists(pid.value):
-			os.kill(pid.value, signal.SIGTERM)
-		self.logfile.close()
+		...
 
 
 class VectorDBLoader(Loader):

context_chat_backend/logger.py

@@ -84,8 +84,8 @@ def _prepare_log_dict(self, record: logging.LogRecord):
 		return message
 
 
-def get_logging_config() -> dict:
-	with open('logger_config.yaml') as f:
+def get_logging_config(config_path: str) -> dict:
+	with open(config_path) as f:
 		try:
 			yaml = YAML(typ='safe')
 			config: dict = yaml.load(f)

context_chat_backend/ocs_utils.py

@@ -15,7 +15,7 @@
 
 logger = logging.getLogger('ccb.ocs_utils')
 
-def _sign_request(headers: dict, username: str = '') -> None:
+def sign_request(headers: dict, username: str = '') -> None:
 	headers['EX-APP-ID'] = getenv('APP_ID')
 	headers['EX-APP-VERSION'] = getenv('APP_VERSION')
 	headers['OCS-APIRequest'] = 'true'
@@ -124,7 +124,7 @@ def ocs_call(
 		headers.update({'Content-Type': 'application/json'})
 		data_bytes = json.dumps(json_data).encode('utf-8')
 
-	_sign_request(headers, kwargs.get('username', ''))
+	sign_request(headers, kwargs.get('username', ''))
 
 	with httpx.Client(verify=verify_ssl) as client:
 		ret = client.request(

dockerfile_scripts/install_deps.sh

@@ -4,4 +4,4 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 #
 apt-get update
-apt-get install -y --no-install-recommends vim git pciutils libgomp1 libreoffice
+apt-get install -y --no-install-recommends vim git pciutils libgomp1 libreoffice supervisor

hwdetect.sh

@@ -47,6 +47,7 @@ if [ "$1" = "config" ]; then
 		exit 0
 	fi
 
+# todo: use fallback config instead of copying it to persistent storage
 	echo "Copying config file to the persistent storage..."
 	if [ "$accel" = "CUDA" ]; then
 		cp "config.gpu.yaml" "$APP_PERSISTENT_STORAGE/config.yaml"

logger_config_em.yaml

@@ -0,0 +1,57 @@
+#
+# SPDX-FileCopyrightText: 2022 MCODING, LLC
+# SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+
+version: 1
+disable_existing_loggers: false
+
+formatters:
+  default:
+    format: '%(asctime)s: [%(levelname)s|%(module)s]: %(message)s'
+    datefmt: '%Y-%m-%dT%H:%M:%S%z'
+
+  json:
+    (): context_chat_backend.logger.JSONFormatter
+    fmt_keys:
+      timestamp: timestamp
+      level: levelname
+      logger: name
+      message: message
+      filename: filename
+      function: funcName
+      line: lineno
+      thread_name: threadName
+      pid: process
+
+
+handlers:
+  stderr:
+    class: logging.StreamHandler
+    level: WARNING
+    formatter: default
+    stream: ext://sys.stderr
+
+  file_json:
+    class: logging.handlers.RotatingFileHandler
+    level: DEBUG
+    formatter: json
+    filename: logs/em_server.log
+    maxBytes: 20971520
+    backupCount: 10
+
+
+loggers:
+  root:
+    level: WARNING
+    handlers:
+    - stderr
+    - file_json
+
+  emserver:
+    level: DEBUG
+    handlers:
+    - stderr
+    - file_json
+    propagate: false

main.py

@@ -13,6 +13,8 @@
 from context_chat_backend.utils import to_int  # isort: skip
 from context_chat_backend.logger import get_logging_config, setup_logging  # isort: skip
 
+LOGGER_CONFIG_NAME = 'logger_config.yaml'
+
 def _setup_log_levels(debug: bool):
 	'''
 	Set log levels for the modules at once for a cleaner usage later.
@@ -40,7 +42,7 @@ def _setup_log_levels(debug: bool):
 
 
 if __name__ == '__main__':
-	logging_config = get_logging_config()
+	logging_config = get_logging_config(LOGGER_CONFIG_NAME)
 	setup_logging(logging_config)
 	app_config: TConfig = app.extra['CONFIG']
 	_setup_log_levels(app_config.debug)