Merge pull request #442 from langflow-ai/persist-opensearch-data

Persist opensearch data

.env.example

@@ -20,6 +20,10 @@ NUDGES_FLOW_ID=ebc01d31-1976-46ce-a385-b0240327226c
 The password must contain at least 8 characters, and must contain at least one uppercase letter, one lowercase letter, one digit, and one special character.
 OPENSEARCH_PASSWORD=
 
+# Path to persist OpenSearch data (indices, documents, cluster state)
+# Default: ./opensearch-data
+OPENSEARCH_DATA_PATH=./opensearch-data
+
 # make here https://console.cloud.google.com/apis/credentials
 GOOGLE_OAUTH_CLIENT_ID=
 GOOGLE_OAUTH_CLIENT_SECRET=

.gitignore

@@ -26,3 +26,6 @@ wheels/
 config/
 
 .docling.pid
+
+# OpenSearch data directory
+opensearch-data/

docker-compose-cpu.yml

@@ -13,6 +13,9 @@ services:
     # Run security setup in background after OpenSearch starts
     command: >
       bash -c "
+        # Ensure data directory has correct permissions
+        sudo chown -R opensearch:opensearch /usr/share/opensearch/data || true
+        
         # Start OpenSearch in background
         /usr/share/opensearch/opensearch-docker-entrypoint.sh opensearch &
 
@@ -25,6 +28,8 @@ services:
     ports:
       - "9200:9200"
       - "9600:9600"
+    volumes:
+      - ${OPENSEARCH_DATA_PATH:-./opensearch-data}:/usr/share/opensearch/data:Z
 
   dashboards:
     image: opensearchproject/opensearch-dashboards:3.0.0

docker-compose.yml

@@ -13,6 +13,9 @@ services:
     # Run security setup in background after OpenSearch starts
     command: >
       bash -c "
+        # Ensure data directory has correct permissions
+        sudo chown -R opensearch:opensearch /usr/share/opensearch/data || true
+        
         # Start OpenSearch in background
         /usr/share/opensearch/opensearch-docker-entrypoint.sh opensearch &
 
@@ -25,6 +28,8 @@ services:
     ports:
       - "9200:9200"
       - "9600:9600"
+    volumes:
+      - ${OPENSEARCH_DATA_PATH:-./opensearch-data}:/usr/share/opensearch/data:Z
 
   dashboards:
     image: opensearchproject/opensearch-dashboards:3.0.0

src/tui/managers/env_manager.py

@@ -59,6 +59,9 @@ class EnvConfig:
     # Document paths (comma-separated)
     openrag_documents_paths: str = "./documents"
 
+    # OpenSearch data path
+    opensearch_data_path: str = "./opensearch-data"
+
     # Validation errors
     validation_errors: Dict[str, str] = field(default_factory=dict)
 
@@ -142,6 +145,7 @@ class EnvManager:
                             "AWS_SECRET_ACCESS_KEY": "aws_secret_access_key",
                             "LANGFLOW_PUBLIC_URL": "langflow_public_url",
                             "OPENRAG_DOCUMENTS_PATHS": "openrag_documents_paths",
+                            "OPENSEARCH_DATA_PATH": "opensearch_data_path",
                             "LANGFLOW_AUTO_LOGIN": "langflow_auto_login",
                             "LANGFLOW_NEW_USER_IS_ACTIVE": "langflow_new_user_is_active",
                             "LANGFLOW_ENABLE_SUPERUSER_CLI": "langflow_enable_superuser_cli",
@@ -291,6 +295,9 @@ class EnvManager:
                 f.write(
                     f"OPENRAG_DOCUMENTS_PATHS={self._quote_env_value(self.config.openrag_documents_paths)}\n"
                 )
+                f.write(
+                    f"OPENSEARCH_DATA_PATH={self._quote_env_value(self.config.opensearch_data_path)}\n"
+                )
                 f.write("\n")
 
                 # Ingestion settings

src/tui/screens/config.py

@@ -387,6 +387,28 @@ class ConfigScreen(Screen):
         self.inputs["openrag_documents_paths"] = input_widget
         yield Static(" ")
 
+        # OpenSearch Data Path
+        yield Label("OpenSearch Data Path")
+        yield Static(
+            "Directory to persist OpenSearch indices across upgrades",
+            classes="helper-text",
+        )
+        current_value = getattr(self.env_manager.config, "opensearch_data_path", "./opensearch-data")
+        input_widget = Input(
+            placeholder="./opensearch-data",
+            value=current_value,
+            id="input-opensearch_data_path",
+        )
+        yield input_widget
+        # Actions row with pick button
+        yield Horizontal(
+            Button("Pick…", id="pick-opensearch-data-btn"),
+            id="opensearch-data-path-actions",
+            classes="controls-row",
+        )
+        self.inputs["opensearch_data_path"] = input_widget
+        yield Static(" ")
+
         # Langflow Auth Settings - These are automatically configured based on password presence
         # Not shown in UI; set in env_manager.setup_secure_defaults()
 
@@ -514,6 +536,8 @@ class ConfigScreen(Screen):
             self.action_back()
         elif event.button.id == "pick-docs-btn":
             self.action_pick_documents_path()
+        elif event.button.id == "pick-opensearch-data-btn":
+            self.action_pick_opensearch_data_path()
         elif event.button.id == "toggle-opensearch-password":
             # Toggle OpenSearch password visibility
             input_widget = self.inputs.get("opensearch_password")
@@ -658,6 +682,62 @@ class ConfigScreen(Screen):
             self._docs_pick_callback = _append_path  # type: ignore[attr-defined]
             self.app.push_screen(picker)
 
+    def action_pick_opensearch_data_path(self) -> None:
+        """Open textual-fspicker to select OpenSearch data directory."""
+        try:
+            import importlib
+
+            fsp = importlib.import_module("textual_fspicker")
+        except Exception:
+            self.notify("textual-fspicker not available", severity="warning")
+            return
+
+        # Determine starting path from current input if possible
+        input_widget = self.inputs.get("opensearch_data_path")
+        start = Path.home()
+        if input_widget and input_widget.value:
+            path_str = input_widget.value.strip()
+            if path_str:
+                candidate = Path(path_str).expanduser()
+                # If path doesn't exist, use parent or fallback to home
+                if candidate.exists():
+                    start = candidate
+                elif candidate.parent.exists():
+                    start = candidate.parent
+
+        # Prefer SelectDirectory for directories; fallback to FileOpen
+        PickerClass = getattr(fsp, "SelectDirectory", None) or getattr(
+            fsp, "FileOpen", None
+        )
+        if PickerClass is None:
+            self.notify(
+                "No compatible picker found in textual-fspicker", severity="warning"
+            )
+            return
+        try:
+            picker = PickerClass(location=start)
+        except Exception:
+            try:
+                picker = PickerClass(start)
+            except Exception:
+                self.notify("Could not initialize textual-fspicker", severity="warning")
+                return
+
+        def _set_path(result) -> None:
+            if not result:
+                return
+            path_str = str(result)
+            if input_widget is None:
+                return
+            input_widget.value = path_str
+
+        # Push with callback when supported; otherwise, use on_screen_dismissed fallback
+        try:
+            self.app.push_screen(picker, _set_path)  # type: ignore[arg-type]
+        except TypeError:
+            self._opensearch_data_pick_callback = _set_path  # type: ignore[attr-defined]
+            self.app.push_screen(picker)
+
     def on_screen_dismissed(self, event) -> None:  # type: ignore[override]
         try:
             # textual-fspicker screens should dismiss with a result; hand to callback if present
@@ -668,6 +748,15 @@ class ConfigScreen(Screen):
                     delattr(self, "_docs_pick_callback")
                 except Exception:
                     pass
+            
+            # Handle OpenSearch data path picker callback
+            cb = getattr(self, "_opensearch_data_pick_callback", None)
+            if cb is not None:
+                cb(getattr(event, "result", None))
+                try:
+                    delattr(self, "_opensearch_data_pick_callback")
+                except Exception:
+                    pass
         except Exception:
             pass
 

tests/conftest.py

@@ -28,11 +28,21 @@ async def onboard_system():
     so that tests can use the /settings endpoint.
     """
     from pathlib import Path
+    import shutil
 
     # Delete any existing config to ensure clean onboarding
     config_file = Path("config/config.yaml")
     if config_file.exists():
         config_file.unlink()
+    
+    # Clean up OpenSearch data directory to ensure fresh state for tests
+    opensearch_data_path = Path(os.getenv("OPENSEARCH_DATA_PATH", "./opensearch-data"))
+    if opensearch_data_path.exists():
+        try:
+            shutil.rmtree(opensearch_data_path)
+            print(f"[DEBUG] Cleaned up OpenSearch data directory: {opensearch_data_path}")
+        except Exception as e:
+            print(f"[DEBUG] Could not clean OpenSearch data directory: {e}")
 
     # Initialize clients
     await clients.initialize()