[pull] main from onyx-dot-app:main

backend/ee/onyx/configs/license_enforcement_config.py

@@ -78,7 +78,7 @@
     "/manage/admin/standard-answer": Tier.ENTERPRISE,
     "/admin/token-rate-limits": Tier.ENTERPRISE,
     "/admin/hooks": Tier.ENTERPRISE,  # outbound webhooks
-    "/admin/log-export": Tier.ENTERPRISE,  # container-local log download
+    "/admin/log-export": Tier.ENTERPRISE,  # deployment-wide log export
     "/analytics": Tier.ENTERPRISE,  # non-admin analytics (e.g. assistant stats)
     "/evals": Tier.ENTERPRISE,
     "/scim": Tier.ENTERPRISE,  # SCIM protocol

backend/ee/onyx/server/log_export/api.py

@@ -8,10 +8,7 @@
 from fastapi.responses import StreamingResponse
 from starlette.background import BackgroundTask
 
-from ee.onyx.server.log_export.collection import (
-    build_log_zip,
-    get_default_log_directories,
-)
+from ee.onyx.server.log_export.collection import get_default_log_directories
 from ee.onyx.server.log_export.models import (
     LogExportManifest,
     LogExportStartResponse,
@@ -46,24 +43,18 @@
 
 router = APIRouter()
 
-API_SERVER_SCOPE_NOTE = (
-    "Scope: this export contains log files from the api_server container only. "
-    "Logs from background workers and other services are not included; use "
-    "'docker logs <container>' or 'kubectl logs <pod>' to retrieve those."
-)
-
 
 class _ExpiringLock:
     """Non-blocking lock whose hold expires after a TTL.
 
-    Guards against leaked holds: release hooks tied to the response lifecycle
-    are skipped by Starlette on some exit paths (a body iterator raising, or
-    client disconnects under ASGI >= 2.4), so a plain ``threading.Lock`` could
-    stay held until process restart. Expiry bounds any such leak.
+    Guards against leaked holds: when no code path releases (nobody polls the
+    status endpoint, or the poll lands on another replica), a plain
+    ``threading.Lock`` would stay held until process restart. Expiry bounds any
+    such leak.
 
     ``try_acquire`` returns a token; ``release`` is a no-op unless the token
-    belongs to the current hold, so a stale holder (or a duplicate call from a
-    second cleanup hook) can never release a successor's hold.
+    belongs to the current hold, so a stale holder (or a duplicate release) can
+    never free a successor's hold.
     """
 
     def __init__(
@@ -97,79 +88,6 @@ def held(self) -> bool:
             return self._held_until is not None and self._clock() < self._held_until
 
 
-# Serializes exports process-wide: each one burns seconds of CPU on compression
-# and holds a temp file until streaming ends, and concurrent exports of the same
-# logs are pure waste. The TTL comfortably exceeds build time plus a slow
-# streaming session (nginx's ``proxy_read_timeout`` defaults to 300s of idle).
-_EXPORT_LOCK_TTL_SECONDS = 15 * 60
-_EXPORT_LOCK = _ExpiringLock(ttl_seconds=_EXPORT_LOCK_TTL_SECONDS)
-
-
-@router.get("/admin/log-export/download")
-def download_api_server_logs(
-    _: User = Depends(require_permission(Permission.FULL_ADMIN_PANEL_ACCESS)),
-) -> StreamingResponse:
-    if MULTI_TENANT:
-        raise OnyxError(
-            OnyxErrorCode.SINGLE_TENANT_ONLY,
-            "Log export is only available on self-hosted deployments.",
-        )
-
-    token = _EXPORT_LOCK.try_acquire()
-    if token is None:
-        raise OnyxError(
-            OnyxErrorCode.RATE_LIMITED,
-            "A log export is already in progress. Try again once it completes.",
-        )
-
-    handed_off = False
-    try:
-        # The archive is fully materialized before streaming, so its exact size
-        # is known and an explicit Content-Length can be sent.
-        built = build_log_zip(get_default_log_directories(), API_SERVER_SCOPE_NOTE)
-        zip_buffer = built.zip_buffer
-
-        def cleanup() -> None:
-            # Wired to both the generator's ``finally`` and the response
-            # background task because neither alone covers every exit path:
-            # Starlette skips background tasks when the body iterator raises
-            # (and on client disconnects under ASGI >= 2.4), while a generator
-            # ``finally`` never runs if the generator is closed before its first
-            # iteration. Double invocation is safe: ``close`` tolerates repeats
-            # and ``release`` ignores stale or duplicate tokens.
-            try:
-                zip_buffer.close()
-            finally:
-                _EXPORT_LOCK.release(token)
-
-        def iter_zip() -> Generator[bytes, None, None]:
-            try:
-                while chunk := zip_buffer.read(STANDARD_CHUNK_SIZE):
-                    yield chunk
-            finally:
-                cleanup()
-
-        timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d_%H-%M-%S")
-        response = StreamingResponse(
-            content=iter_zip(),
-            media_type="application/zip",
-            headers={
-                "Content-Disposition": (
-                    f"attachment; filename=onyx_api_server_logs_{timestamp}.zip"
-                ),
-                "Content-Length": str(built.size_bytes),
-            },
-            background=BackgroundTask(cleanup),
-        )
-        handed_off = True
-        return response
-    finally:
-        # Once the response exists, its cleanup hooks own the release; until
-        # then, any exit (including BaseException) releases here.
-        if not handed_off:
-            _EXPORT_LOCK.release(token)
-
-
 API_SERVER_WORKER_NAME = "api_server"
 
 # One collector task per worker type, each routed to a queue that worker
@@ -302,8 +220,6 @@ def start_log_export(
             _ASYNC_EXPORT_LOCK.release(token)
 
 
-# Declared after the sync ``/admin/log-export/download`` route above so that
-# literal path keeps matching before ``{export_id}``.
 @router.get("/admin/log-export/{export_id}")
 def get_log_export_status(
     export_id: str,
@@ -352,8 +268,11 @@ def download_log_export(
 
     def cleanup() -> None:
         # Wired to both the generator's ``finally`` and the response background
-        # task, matching ``download_api_server_logs``; ``close`` tolerates
-        # repeats. No lock is involved here.
+        # task because neither alone covers every exit path: Starlette skips
+        # background tasks when the body iterator raises (and on client
+        # disconnects under ASGI >= 2.4), while a generator ``finally`` never
+        # runs if the generator is closed before its first iteration. Double
+        # invocation is safe: ``close`` tolerates repeats.
         zip_buffer.close()
 
     def iter_zip() -> Generator[bytes, None, None]:

backend/onyx/background/celery/tasks/docfetching/tasks.py

@@ -53,6 +53,7 @@
 from onyx.redis.redis_connector import RedisConnector
 from onyx.server.metrics.connector_health_metrics import on_index_attempt_status_change
 from onyx.utils.logger import setup_logger
+from onyx.utils.os_reaper import reap_children_before_exit
 from onyx.utils.variable_functionality import global_version
 from shared_configs.configs import SENTRY_CELERY_TRACES_SAMPLE_RATE, SENTRY_DSN
 
@@ -272,6 +273,8 @@ def _docfetching_task(
         cc_pair_id,
         search_settings_id,
     )
+    # os._exit bypasses the drain in _initializer's finally, so reap here.
+    reap_children_before_exit()
     os._exit(0)  # ensure process exits cleanly
 
 

backend/onyx/background/indexing/job_client.py

@@ -16,6 +16,11 @@
 from onyx.configs.constants import POSTGRES_CELERY_WORKER_INDEXING_CHILD_APP_NAME
 from onyx.db.engine.sql_engine import SqlEngine
 from onyx.utils.logger import setup_logger
+from onyx.utils.os_reaper import (
+    become_child_subreaper,
+    install_sigterm_drain,
+    reap_children_before_exit,
+)
 from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA, TENANT_ID_PREFIX
 from shared_configs.contextvars import CURRENT_TENANT_ID_CONTEXTVAR
 
@@ -55,6 +60,12 @@ def _initializer(
         kwargs = {}
 
     logger.info("Initializing spawned worker child process.")
+
+    # adopt orphans (e.g. Chromium helpers) instead of leaking zombies to PID 1,
+    # and drain them even when a watchdog cancels this child with SIGTERM
+    become_child_subreaper()
+    install_sigterm_drain()
+
     # 1. Get tenant_id from args or fallback to default
     tenant_id = POSTGRES_DEFAULT_SCHEMA
     for arg in reversed(args):
@@ -94,6 +105,11 @@ def _initializer(
     finally:
         CURRENT_TENANT_ID_CONTEXTVAR.reset(token)
 
+        # os._exit entrypoints skip this finally and drain themselves
+        reaped = reap_children_before_exit()
+        if reaped:
+            logger.info("Spawned worker child reaped %s orphaned processes.", reaped)
+
 
 def _run_in_process(
     func: Callable,

backend/onyx/document_index/opensearch/client.py

@@ -160,11 +160,27 @@ def __init__(
 _VERSION_CONFLICT_ERROR_TYPE = "version_conflict_engine_exception"
 # Raised by a search whose PIT has expired/been deleted; we re-open and retry.
 _SEARCH_CONTEXT_MISSING_ERROR_TYPE = "search_context_missing"
+# Rejection by an index/cluster block, e.g. the read_only_allow_delete block
+# OpenSearch applies when disk usage crosses the flood-stage watermark.
+_CLUSTER_BLOCK_ERROR_TYPE = "cluster_block_exception"
 # Chunks per PIT-scan page. A port doc-batch is small (INDEX_BATCH_SIZE docs), so
 # one page covers a batch; paging still protects against a pathological doc.
 _PIT_SCAN_PAGE_SIZE = 1000
 
 
+def is_cluster_block_error(e: Exception) -> bool:
+    """True when a request was rejected by an index/cluster block rather than a
+    problem with the request itself."""
+    return isinstance(e, TransportError) and _CLUSTER_BLOCK_ERROR_TYPE in str(e.error)
+
+
+class OpenSearchIndexWriteBlockedError(Exception):
+    """An existing index rejected a metadata write because of a block (e.g.
+    read_only_allow_delete applied at the disk flood-stage watermark). The
+    index is still fully readable — callers that can serve degraded may catch
+    this. Never raised for a missing index or a blocked index creation."""
+
+
 class OpenSearchServerSideTimeout(Exception):
     """
     A server-side timeout occurred when searching an OpenSearch index.

backend/onyx/document_index/opensearch/opensearch_document_index.py

@@ -39,7 +39,9 @@
     OpenSearchClient,
     OpenSearchDocumentMissingError,
     OpenSearchIndexClient,
+    OpenSearchIndexWriteBlockedError,
     SearchHit,
+    is_cluster_block_error,
 )
 from onyx.document_index.opensearch.cluster_settings import OPENSEARCH_CLUSTER_SETTINGS
 from onyx.document_index.opensearch.constants import OpenSearchSearchType
@@ -308,10 +310,24 @@ def __init__(
             and VERIFY_CREATE_OPENSEARCH_INDEX_ON_INIT_MT
             and index_name not in _verified_index_names_for_current_process
         ):
-            self.verify_and_create_index_if_necessary(
-                embedding_dim=embedding_dim, embedding_precision=embedding_precision
-            )
-            _verified_index_names_for_current_process.add(index_name)
+            try:
+                self.verify_and_create_index_if_necessary(
+                    embedding_dim=embedding_dim, embedding_precision=embedding_precision
+                )
+            except OpenSearchIndexWriteBlockedError as e:
+                # Existing index, still readable — don't fail the caller. Not
+                # cached as verified, so a later init retries the mapping
+                # refresh once the block clears.
+                logger.error(
+                    "Index %s is write-blocked; continuing without the mapping "
+                    "refresh. Search still works, but indexing will fail until "
+                    "the block is cleared (usually by freeing disk space below "
+                    "the flood-stage watermark). Error: %s",
+                    index_name,
+                    e,
+                )
+            else:
+                _verified_index_names_for_current_process.add(index_name)
 
     def verify_and_create_index_if_necessary(
         self,
@@ -372,6 +388,15 @@ def verify_and_create_index_if_necessary(
                 try:
                     self._client.put_mapping(expected_mappings)
                 except Exception as e:
+                    if is_cluster_block_error(e):
+                        # The index exists and is readable; only this metadata
+                        # write was rejected. Raise the targeted type so
+                        # callers that can serve degraded can catch exactly
+                        # this case (never a missing index / blocked create).
+                        raise OpenSearchIndexWriteBlockedError(
+                            f"Index {self._index_name} is write-blocked; the mapping "
+                            "refresh was rejected."
+                        ) from e
                     logger.error(
                         "Failed to update mappings for index %s. This likely means a field type was changed which requires reindexing. Error: %s",
                         self._index_name,

backend/onyx/setup.py

@@ -47,6 +47,7 @@
 from onyx.document_index.interfaces_new import DocumentIndex
 from onyx.document_index.opensearch.client import (
     OpenSearchClient,
+    OpenSearchIndexWriteBlockedError,
     wait_for_opensearch_with_timeout,
 )
 from onyx.document_index.opensearch.opensearch_document_index import set_cluster_state
@@ -235,6 +236,22 @@ def setup_document_indices(
                 )
                 document_index_setup_success = True
                 break
+            except OpenSearchIndexWriteBlockedError as e:
+                # The index exists but is write-blocked (typically the
+                # read_only_allow_delete block applied at the disk flood-stage
+                # watermark). It is still readable, so start up degraded rather
+                # than crash-loop until the block clears. A missing index or
+                # blocked creation raises a different error and still fails.
+                logger.error(
+                    "Document index %s is write-blocked; continuing startup without "
+                    "the mapping refresh. Search still works, but indexing will fail "
+                    "until the block is cleared (usually by freeing disk space below "
+                    "the flood-stage watermark). Error: %s",
+                    document_index.__class__.__name__,
+                    e,
+                )
+                document_index_setup_success = True
+                break
             except Exception:
                 logger.exception(
                     "Document index %s setup did not succeed. The relevant service may not be ready yet. Retrying in %s seconds.",