Fix Langflow DLS ingestion preflight

Author: phact

Makefile

@@ -791,6 +791,7 @@ test-ci: ensure-langflow-data ensure-backend-volumes ## Start infra, run integra
 	GOOGLE_OAUTH_CLIENT_ID="" \
 	GOOGLE_OAUTH_CLIENT_SECRET="" \
 	OPENSEARCH_HOST=localhost OPENSEARCH_PORT=9200 \
+	LANGFLOW_OPENSEARCH_HOST=opensearch LANGFLOW_OPENSEARCH_PORT=9200 \
 	OPENSEARCH_USERNAME=admin OPENSEARCH_PASSWORD=$${OPENSEARCH_PASSWORD} \
 	DISABLE_STARTUP_INGEST=$${DISABLE_STARTUP_INGEST:-true} \
 	uv run pytest tests/integration/core -vv -s -o log_cli=true --log-cli-level=DEBUG; \
@@ -915,6 +916,7 @@ test-ci-local: ensure-langflow-data ensure-backend-volumes ## Same as test-ci bu
 	GOOGLE_OAUTH_CLIENT_ID="" \
 	GOOGLE_OAUTH_CLIENT_SECRET="" \
 	OPENSEARCH_HOST=localhost OPENSEARCH_PORT=9200 \
+	LANGFLOW_OPENSEARCH_HOST=opensearch LANGFLOW_OPENSEARCH_PORT=9200 \
 	OPENSEARCH_USERNAME=admin OPENSEARCH_PASSWORD=$${OPENSEARCH_PASSWORD} \
 	DISABLE_STARTUP_INGEST=$${DISABLE_STARTUP_INGEST:-true} \
 	uv run pytest tests/integration/core -vv -s -o log_cli=true --log-cli-level=DEBUG; \

flows/components/opensearch_multimodal.py

@@ -7,9 +7,6 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any
 
-from opensearchpy import OpenSearch, helpers
-from opensearchpy.exceptions import OpenSearchException, RequestError
-
 from lfx.base.vectorstores.model import LCVectorStoreComponent, check_cached_vector_store
 from lfx.base.vectorstores.vector_store_connection_decorator import vector_store_connection
 from lfx.io import (
@@ -25,6 +22,8 @@
 )
 from lfx.log import logger
 from lfx.schema.data import Data
+from opensearchpy import OpenSearch, helpers
+from opensearchpy.exceptions import OpenSearchException, RequestError
 
 REQUEST_TIMEOUT = 60
 MAX_RETRIES = 5
@@ -813,6 +812,13 @@ def _bulk_ingest_embeddings(
         helpers.bulk(client, requests, max_chunk_bytes=max_chunk_bytes)
         return return_ids
 
+    def _log_index_admin_skip(self, operation: str, error: Exception) -> None:
+        """Log index-admin operations that may be blocked under filter-level DLS."""
+        logger.warning(
+            f"[OpenSearchMultimodel] Could not run index-admin operation '{operation}': {error}. "
+            "Assuming the backend pre-created the required index/mapping and continuing."
+        )
+
     # ---------- param helpers ----------
     def _parse_int_param(self, attr_name: str, default: int) -> int:
         """Parse a string attribute to int, returning *default* on failure."""
@@ -1228,8 +1234,14 @@ def _embed(text: str) -> list[float]:
         )
 
         # Ensure index exists with baseline mapping (index.knn: true is required for vector search)
+        index_exists = True
+        try:
+            index_exists = bool(client.indices.exists(index=self.index_name))
+        except OpenSearchException as exists_error:
+            self._log_index_admin_skip("indices.exists", exists_error)
+
         try:
-            if not client.indices.exists(index=self.index_name):
+            if not index_exists:
                 self.log(f"Creating index '{self.index_name}' with base mapping")
                 client.indices.create(index=self.index_name, body=mapping)
         except RequestError as creation_error:

flows/ingestion_flow.json

@@ -22,6 +22,7 @@ def __init__(self, flows_service=None, docling_service=None):
         self.flows_service = flows_service
         self.docling_service = docling_service
         self.flow_id_url_ingest = LANGFLOW_URL_INGEST_FLOW_ID
+        self._embedding_dimension_cache: dict[str, int] = {}
 
     _TRANSIENT_STATUS_CODES = {408, 429, 500, 502, 503, 504}
 
@@ -71,6 +72,77 @@ def merge_ui_ingest_settings_into_tweaks(
 
         return final_tweaks
 
+    async def _detect_embedding_dimensions(
+        self,
+        embedding_model: str,
+        embedding_provider: str | None,
+    ) -> int:
+        """Generate one probe embedding so mapping dimensions match the provider."""
+        from services.models_service import ModelsService
+
+        cache_key = f"{embedding_provider or ''}:{embedding_model}"
+        cached = self._embedding_dimension_cache.get(cache_key)
+        if cached:
+            return cached
+
+        litellm_model_name = await ModelsService().get_litellm_model_name(
+            embedding_model,
+            provider=embedding_provider,
+        )
+        response = await clients.patched_embedding_client.embeddings.create(
+            model=litellm_model_name,
+            input=["dimension probe"],
+        )
+        if not response.data:
+            raise RuntimeError("Embedding provider returned no data for dimension probe")
+
+        first = response.data[0]
+        embedding = first["embedding"] if isinstance(first, dict) else first.embedding
+        dimensions = len(embedding)
+        if dimensions <= 0:
+            raise RuntimeError("Embedding provider returned an empty dimension probe")
+
+        self._embedding_dimension_cache[cache_key] = dimensions
+        return dimensions
+
+    async def _ensure_langflow_ingest_index(self, embedding_model: str | None) -> None:
+        """Pre-create index mappings Langflow cannot manage with a DLS JWT."""
+        if clients.opensearch is None:
+            logger.debug("[LF] OpenSearch admin client unavailable; skipping ingest preflight")
+            return
+
+        try:
+            from config.embedding_constants import OPENAI_DEFAULT_EMBEDDING_MODEL
+            from config.settings import get_index_name, get_openrag_config
+            from utils.embedding_fields import ensure_embedding_field_exists
+            from utils.embeddings import create_index_body
+
+            config = get_openrag_config()
+            index_name = get_index_name()
+            model_name = embedding_model or OPENAI_DEFAULT_EMBEDDING_MODEL
+            embedding_dimensions = await self._detect_embedding_dimensions(
+                model_name,
+                config.knowledge.embedding_provider,
+            )
+            if not await clients.opensearch.indices.exists(index=index_name):
+                await clients.opensearch.indices.create(
+                    index=index_name,
+                    body=await create_index_body(model_name, embedding_dimensions),
+                )
+
+            await ensure_embedding_field_exists(
+                clients.opensearch,
+                model_name,
+                index_name,
+                embedding_dimensions,
+            )
+        except Exception as e:
+            logger.warning(
+                "[LF] Failed to preconfigure OpenSearch index before Langflow ingest",
+                embedding_model=embedding_model,
+                error=str(e),
+            )
+
     async def upload_user_file(self, file_tuple, jwt_token: str | None = None) -> dict[str, Any]:
         """Upload a file using Langflow Files API v2: POST /api/v2/files.
         Returns JSON with keys: id, name, path, size, provider.
@@ -229,6 +301,7 @@ async def run_ingestion_flow(
         await add_provider_credentials_to_headers(
             headers, config, flows_service=self.flows_service, jwt_token=jwt_token
         )
+        await self._ensure_langflow_ingest_index(embedding_model)
         start_time = time.time()
         logger.info(
             "[INGEST] Run started",
@@ -353,6 +426,7 @@ async def run_url_ingestion_flow(
         await add_provider_credentials_to_headers(
             headers, config, flows_service=self.flows_service, jwt_token=jwt_token
         )
+        await self._ensure_langflow_ingest_index(embedding_model)
 
         logger.info(
             "[LF] Running URL ingestion flow",

flows/openrag_agent.json

@@ -1,37 +1,58 @@
+from utils.embedding_fields import build_knn_vector_field, get_embedding_field_name
 from utils.logging_config import get_logger
 
 logger = get_logger(__name__)
 
 
-async def create_index_body() -> dict:
+async def create_index_body(
+    embedding_model: str | None = None,
+    embedding_dimensions: int | None = None,
+) -> dict:
     """Create a static index body configuration.
 
     Returns:
         OpenSearch index body configuration
     """
+    from config.embedding_constants import OPENAI_DEFAULT_EMBEDDING_MODEL
+    from config.settings import VECTOR_DIM, get_openrag_config
+
+    resolved_embedding_model = (
+        embedding_model
+        or get_openrag_config().knowledge.embedding_model
+        or OPENAI_DEFAULT_EMBEDDING_MODEL
+    )
+
+    properties = {
+        "document_id": {"type": "keyword"},
+        "filename": {"type": "keyword"},
+        "mimetype": {"type": "keyword"},
+        "page": {"type": "integer"},
+        "text": {"type": "text"},
+        # Legacy field - kept for backward compatibility and for clusters where
+        # Langflow cannot perform mapping updates with a DLS-filtered JWT.
+        "chunk_embedding": build_knn_vector_field(VECTOR_DIM),
+        # Track which embedding model was used for this chunk
+        "embedding_model": {"type": "keyword"},
+        "embedding_dimensions": {"type": "integer"},
+        "source_url": {"type": "keyword"},
+        "connector_type": {"type": "keyword"},
+        "owner": {"type": "keyword"},
+        "owner_email": {"type": "keyword"},
+        "allowed_users": {"type": "keyword"},
+        "allowed_groups": {"type": "keyword"},
+        "allowed_principals": {"type": "keyword"},
+        "created_time": {"type": "date"},
+        "modified_time": {"type": "date"},
+        "indexed_time": {"type": "date"},
+        "metadata": {"type": "object"},
+    }
+
+    if embedding_dimensions:
+        properties[get_embedding_field_name(resolved_embedding_model)] = build_knn_vector_field(
+            embedding_dimensions
+        )
 
     return {
         "settings": {"index": {"knn": True}, "number_of_shards": 1, "number_of_replicas": 0},
-        "mappings": {
-            "properties": {
-                "document_id": {"type": "keyword"},
-                "filename": {"type": "keyword"},
-                "mimetype": {"type": "keyword"},
-                "page": {"type": "integer"},
-                "text": {"type": "text"},
-                # Track which embedding model was used for this chunk
-                "embedding_model": {"type": "keyword"},
-                "embedding_dimensions": {"type": "integer"},
-                "source_url": {"type": "keyword"},
-                "connector_type": {"type": "keyword"},
-                "owner": {"type": "keyword"},
-                "allowed_users": {"type": "keyword"},
-                "allowed_groups": {"type": "keyword"},
-                "allowed_principals": {"type": "keyword"},
-                "created_time": {"type": "date"},
-                "modified_time": {"type": "date"},
-                "indexed_time": {"type": "date"},
-                "metadata": {"type": "object"},
-            }
-        },
+        "mappings": {"properties": properties},
     }

flows/openrag_nudges.json

@@ -8,11 +8,12 @@
 JVector/DiskANN method configuration with only the dimension varying per
 embedding model.
 """
-from typing import Any, Dict
+from types import SimpleNamespace
+from typing import Any
 
 import pytest
 
-from utils.embedding_fields import build_knn_vector_field
+from utils.embedding_fields import build_knn_vector_field, get_embedding_field_name
 
 
 class TestBuildKnnVectorFieldStructure:
@@ -99,6 +100,28 @@ class TestBuildKnnVectorFieldCallSitesMatch:
     def test_index_body_uses_helper_output(self) -> None:
         from config.settings import INDEX_BODY, VECTOR_DIM
 
-        chunk_field: Dict[str, Any] = INDEX_BODY["mappings"]["properties"]["chunk_embedding"]
+        chunk_field: dict[str, Any] = INDEX_BODY["mappings"]["properties"][
+            "chunk_embedding"
+        ]
         expected = build_knn_vector_field(VECTOR_DIM)
         assert chunk_field == expected
+
+    @pytest.mark.asyncio
+    async def test_create_index_body_precreates_configured_embedding_field(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            "config.settings.get_openrag_config",
+            lambda: SimpleNamespace(
+                knowledge=SimpleNamespace(embedding_model="text-embedding-3-large")
+            ),
+        )
+
+        from utils.embeddings import create_index_body
+
+        body = await create_index_body("text-embedding-3-large", 3072)
+        properties = body["mappings"]["properties"]
+        embedding_field = get_embedding_field_name("text-embedding-3-large")
+
+        assert properties[embedding_field] == build_knn_vector_field(3072)
+        assert properties["owner_email"] == {"type": "keyword"}

flows/openrag_url_mcp.json

@@ -6,6 +6,7 @@
 Docling fails / expires / times out.
 """
 
+from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
@@ -231,6 +232,37 @@ def test_processor_accepts_injected_polling_service():
     assert processor.docling_polling_service is injected
 
 
+@pytest.mark.asyncio
+async def test_langflow_preflight_detects_embedding_dimensions_with_probe(monkeypatch):
+    class FakeEmbeddings:
+        def __init__(self):
+            self.calls = []
+
+        async def create(self, model, input):
+            self.calls.append((model, input))
+            return SimpleNamespace(data=[SimpleNamespace(embedding=[0.1] * 7)])
+
+    fake_embeddings = FakeEmbeddings()
+    fake_client = SimpleNamespace(embeddings=fake_embeddings)
+
+    async def fake_get_litellm_model_name(self, model_name, provider=None, strict=False):
+        assert model_name == "provider/model"
+        assert provider == "provider"
+        return "provider/provider/model"
+
+    monkeypatch.setattr("config.settings.clients._patched_async_client", fake_client)
+    monkeypatch.setattr(
+        "services.models_service.ModelsService.get_litellm_model_name",
+        fake_get_litellm_model_name,
+    )
+
+    svc = LangflowFileService(docling_service=AsyncMock())
+
+    assert await svc._detect_embedding_dimensions("provider/model", "provider") == 7
+    assert await svc._detect_embedding_dimensions("provider/model", "provider") == 7
+    assert fake_embeddings.calls == [("provider/provider/model", ["dimension probe"])]
+
+
 @pytest.mark.asyncio
 async def test_task_service_threads_polling_service_to_processor(monkeypatch):
     """TaskService.create_langflow_upload_task must forward its injected