♻️ nexent-data-process now takes less memory, elasticsearch takes more available memory

Phinease · web-flow · commit d56ed70f4877 · 2025-11-14T18:42:45.000+08:00
diff --git a/backend/consts/const.py b/backend/consts/const.py
@@ -106,11 +106,14 @@
 RAY_DASHBOARD_PORT = int(os.getenv("RAY_DASHBOARD_PORT", "8265"))
 RAY_DASHBOARD_HOST = os.getenv("RAY_DASHBOARD_HOST", "0.0.0.0")
 RAY_NUM_CPUS = os.getenv("RAY_NUM_CPUS")
-RAY_PLASMA_DIRECTORY = os.getenv("RAY_PLASMA_DIRECTORY", "/tmp")
 RAY_OBJECT_STORE_MEMORY_GB = float(
-    os.getenv("RAY_OBJECT_STORE_MEMORY_GB", "2.0"))
+    os.getenv("RAY_OBJECT_STORE_MEMORY_GB", "0.25"))
 RAY_TEMP_DIR = os.getenv("RAY_TEMP_DIR", "/tmp/ray")
 RAY_LOG_LEVEL = os.getenv("RAY_LOG_LEVEL", "INFO").upper()
+# Disable plasma preallocation to reduce idle memory usage
+# When set to false, Ray will allocate object store memory on-demand instead of preallocating
+RAY_preallocate_plasma = os.getenv(
+    "RAY_preallocate_plasma", "false").lower() == "true"
 
 
 # Service Control Flags
diff --git a/backend/data_process/ray_config.py b/backend/data_process/ray_config.py
@@ -9,10 +9,9 @@
 import ray
 
 from consts.const import (
-    RAY_NUM_CPUS,
     RAY_OBJECT_STORE_MEMORY_GB,
-    RAY_PLASMA_DIRECTORY,
     RAY_TEMP_DIR,
+    RAY_preallocate_plasma,
 )
 
 logger = logging.getLogger("data_process.ray_config")
@@ -25,9 +24,9 @@ class RayConfig:
     """Ray configuration manager"""
 
     def __init__(self):
-        self.plasma_directory = RAY_PLASMA_DIRECTORY
         self.object_store_memory_gb = RAY_OBJECT_STORE_MEMORY_GB
         self.temp_dir = RAY_TEMP_DIR
+        self.preallocate_plasma = RAY_preallocate_plasma
 
     def get_init_params(
             self,
@@ -52,7 +51,6 @@ def get_init_params(
         """
         params = {
             "ignore_reinit_error": True,
-            "_plasma_directory": self.plasma_directory,
         }
 
         if address:
@@ -70,9 +68,16 @@ def get_init_params(
             # Temp directory configuration
             params["_temp_dir"] = self.temp_dir
 
+            # Object spilling directory (stable API)
+            # This allows Ray to spill objects to disk when memory is full
+            params["object_spilling_directory"] = self.temp_dir
+
             # Dashboard configuration
+            # Always pass include_dashboard explicitly because Ray's default is True.
+            # If we omit this parameter when include_dashboard is False,
+            # Ray will still start the dashboard by default.
+            params["include_dashboard"] = include_dashboard
             if include_dashboard:
-                params["include_dashboard"] = True
                 params["dashboard_host"] = dashboard_host
                 params["dashboard_port"] = dashboard_port
 
@@ -93,30 +98,49 @@ def init_ray(self, **kwargs) -> bool:
                 logger.info("Ray already initialized, skipping...")
                 return True
 
+            # Set RAY_preallocate_plasma environment variable before initialization
+            # Ray reads this environment variable during initialization
+            os.environ["RAY_preallocate_plasma"] = str(
+                self.preallocate_plasma).lower()
+
             params = self.get_init_params(**kwargs)
 
             # Log the attempt to initialize
-            logger.debug("Initializing Ray cluster...")
-            logger.debug("Ray configuration parameters:")
+            logger.info("Initializing Ray cluster...")
+            logger.info("Ray memory optimization configuration:")
+            logger.info(
+                f"  RAY_preallocate_plasma: {self.preallocate_plasma}")
+            logger.info(
+                f"  Object store memory: {self.object_store_memory_gb} GB")
             for key, value in params.items():
                 if key.startswith('_'):
                     logger.debug(f"  {key}: {value}")
                 elif key == 'object_store_memory':
-                    logger.debug(f"  {key}: {value / (1024 ** 3):.1f} GB")
+                    logger.info(f"  {key}: {value / (1024 ** 3):.2f} GB")
+                elif key == 'object_spilling_directory':
+                    logger.info(f"  {key}: {value}")
                 else:
                     logger.debug(f"  {key}: {value}")
 
             ray.init(**params)
             logger.info("✅ Ray initialization successful")
 
-            # Display cluster information
+            # Display cluster information and verify memory configuration
             try:
                 if hasattr(ray, 'cluster_resources'):
                     resources = ray.cluster_resources()
-                    logger.debug(f"Ray cluster resources: {resources}")
+                    logger.info(f"Ray cluster resources: {resources}")
+
+                    # Log memory-related resources
+                    if 'memory' in resources:
+                        logger.info(
+                            f"  Total cluster memory: {resources['memory'] / (1024**3):.2f} GB")
+                    if 'object_store_memory' in resources:
+                        logger.info(
+                            f"  Object store memory: {resources['object_store_memory'] / (1024**3):.2f} GB")
             except Exception as e:
-                logger.error(
-                    f"Failed to get cluster resources information: {e}")
+                logger.warning(
+                    f"Could not retrieve cluster resources information: {e}")
 
             return True
 
@@ -139,9 +163,17 @@ def connect_to_cluster(self, address: str = "auto") -> bool:
                 logger.debug("Ray already initialized, skipping...")
                 return True
 
+            # Set RAY_preallocate_plasma environment variable before initialization
+            # Note: When connecting to existing cluster, this setting may not take effect
+            # as the cluster was already initialized with its own settings
+            os.environ["RAY_preallocate_plasma"] = str(
+                self.preallocate_plasma).lower()
+
             params = self.get_init_params(address=address)
 
             logger.debug(f"Connecting to Ray cluster: {address}")
+            logger.debug(
+                f"  RAY_preallocate_plasma: {self.preallocate_plasma}")
             ray.init(**params)
             logger.info("✅ Successfully connected to Ray cluster")
 
@@ -183,6 +215,7 @@ def log_configuration(self):
         logger.debug(f"  Plasma directory: {self.plasma_directory}")
         logger.debug(f"  ObjectStore memory: {self.object_store_memory_gb} GB")
         logger.debug(f"  Temp directory: {self.temp_dir}")
+        logger.debug(f"  Preallocate plasma: {self.preallocate_plasma}")
 
     @classmethod
     def init_ray_for_worker(cls, address: str = "auto") -> bool:
diff --git a/backend/data_process/tasks.py b/backend/data_process/tasks.py
@@ -22,6 +22,7 @@
     REDIS_BACKEND_URL,
     FORWARD_REDIS_RETRY_DELAY_S,
     FORWARD_REDIS_RETRY_MAX,
+    DISABLE_RAY_DASHBOARD,
 )
 
 
@@ -36,24 +37,28 @@ def init_ray_in_worker():
     Initializes Ray within a Celery worker, ensuring it is done only once.
     This function is designed to be called from within a task.
     """
-    if not ray.is_initialized():
-        logger.info(
-            "Ray not initialized. Initializing Ray for Celery worker...")
-        try:
-            # `configure_logging=False` prevents Ray from setting up its own loggers,
-            # which can interfere with Celery's logging.
-            # `faulthandler=False` is critical to prevent the `AttributeError: 'LoggingProxy' object has no attribute 'fileno'`
-            # error when running inside a Celery worker.
-            ray.init(
-                configure_logging=False,
-                faulthandler=False
-            )
-            logger.info("Ray initialized successfully for Celery worker.")
-        except Exception as e:
-            logger.error(f"Failed to initialize Ray for Celery worker: {e}")
-            raise
-    else:
+    if ray.is_initialized():
         logger.debug("Ray is already initialized.")
+        return
+
+    logger.info("Ray not initialized. Initializing Ray for Celery worker...")
+    try:
+        # `configure_logging=False` prevents Ray from setting up its own loggers,
+        # which can interfere with Celery's logging.
+        # `faulthandler=False` is critical to prevent the
+        # `AttributeError: 'LoggingProxy' object has no attribute 'fileno'`
+        # error when running inside a Celery worker.
+        # We also explicitly control the Ray dashboard behavior here to ensure
+        # that Celery workers respect the global DISABLE_RAY_DASHBOARD setting.
+        ray.init(
+            configure_logging=False,
+            faulthandler=False,
+            include_dashboard=not DISABLE_RAY_DASHBOARD,
+        )
+        logger.info("Ray initialized successfully for Celery worker.")
+    except Exception as e:
+        logger.error(f"Failed to initialize Ray for Celery worker: {e}")
+        raise
 
 
 def run_async(coro):
diff --git a/backend/data_process/worker.py b/backend/data_process/worker.py
@@ -40,7 +40,7 @@
     ELASTICSEARCH_SERVICE,
     QUEUES,
     RAY_ADDRESS,
-    RAY_PLASMA_DIRECTORY,
+    RAY_preallocate_plasma,
     REDIS_URL,
     WORKER_CONCURRENCY,
     WORKER_NAME,
@@ -94,13 +94,16 @@ def setup_worker_environment(**kwargs):
             ray_address = RAY_ADDRESS
 
             try:
+                os.environ["RAY_preallocate_plasma"] = str(
+                    RAY_preallocate_plasma).lower()
+
                 # Initialize Ray using the centralized RayConfig helper
                 if not RayConfig.init_ray_for_worker(ray_address):
+                    logger.warning("Warning: fallback to direct ray.init")
                     # Fallback to direct ray.init if helper fails
                     ray.init(
                         address=ray_address,
                         ignore_reinit_error=True,
-                        _plasma_directory=RAY_PLASMA_DIRECTORY
                     )
 
                 logger.info(
diff --git a/backend/data_process_service.py b/backend/data_process_service.py
@@ -19,7 +19,7 @@
 from consts.const import (
     REDIS_URL, REDIS_PORT, FLOWER_PORT, RAY_DASHBOARD_PORT, RAY_DASHBOARD_HOST,
     RAY_ACTOR_NUM_CPUS, RAY_NUM_CPUS, DISABLE_RAY_DASHBOARD, DISABLE_CELERY_FLOWER,
-    DOCKER_ENVIRONMENT
+    DOCKER_ENVIRONMENT, RAY_OBJECT_STORE_MEMORY_GB, RAY_preallocate_plasma, RAY_TEMP_DIR
 )
 
 # Load environment variables
@@ -111,16 +111,30 @@ def start_ray_cluster(self):
             if not success:
                 # Fallback to direct Ray initialization
                 try:
+                    # Set RAY_preallocate_plasma environment variable before initialization
+                    os.environ["RAY_preallocate_plasma"] = str(
+                        RAY_preallocate_plasma).lower()
+
+                    # Calculate object store memory in bytes
+                    object_store_memory = int(
+                        RAY_OBJECT_STORE_MEMORY_GB * 1024 * 1024 * 1024)
+
+                    logger.info(
+                        f"Fallback: Initializing Ray with object_store_memory={RAY_OBJECT_STORE_MEMORY_GB}GB, preallocate_plasma={RAY_preallocate_plasma}")
+
                     ray.init(
                         num_cpus=num_cpus,
-                        _plasma_directory="/tmp",
+                        object_store_memory=object_store_memory,
+                        _temp_dir=RAY_TEMP_DIR,
+                        object_spilling_directory=RAY_TEMP_DIR,
                         include_dashboard=include_dashboard,
                         dashboard_host=dashboard_host,
                         dashboard_port=self.ray_dashboard_port,
                         ignore_reinit_error=True
                     )
                     success = True
-                except Exception:
+                except Exception as e:
+                    logger.error(f"Fallback Ray initialization failed: {e}")
                     success = False
             
             if success:
diff --git a/docker/.env.example b/docker/.env.example
@@ -27,7 +27,7 @@ ELASTICSEARCH_HOST=http://nexent-elasticsearch:9200
 ELASTIC_PASSWORD=nexent@2025
 
 # Elasticsearch Memory Configuration
-ES_JAVA_OPTS="-Xms1g -Xmx1g"
+ES_JAVA_OPTS="-Xms2g -Xmx2g"
 
 # Elasticsearch Disk Watermark Configuration
 ES_DISK_WATERMARK_LOW=85%
@@ -111,15 +111,14 @@ FLOWER_PORT=5555
 RAY_ACTOR_NUM_CPUS=2
 RAY_DASHBOARD_PORT=8265
 RAY_DASHBOARD_HOST=0.0.0.0
-RAY_NUM_CPUS=
-RAY_PLASMA_DIRECTORY=/tmp
-RAY_OBJECT_STORE_MEMORY_GB=2.0
+RAY_NUM_CPUS=4
+RAY_OBJECT_STORE_MEMORY_GB=0.25
 RAY_TEMP_DIR=/tmp/ray
 RAY_LOG_LEVEL=INFO
 
 # Service Control Flags
-DISABLE_RAY_DASHBOARD=false
-DISABLE_CELERY_FLOWER=false
+DISABLE_RAY_DASHBOARD=true
+DISABLE_CELERY_FLOWER=true
 DOCKER_ENVIRONMENT=false
 
 # Celery Configuration
diff --git a/test/backend/data_process/test_ray_config.py b/test/backend/data_process/test_ray_config.py
diff --git a/test/backend/data_process/test_tasks.py b/test/backend/data_process/test_tasks.py
diff --git a/test/backend/data_process/test_worker.py b/test/backend/data_process/test_worker.py
