diff --git a/admin/server/config.py b/admin/server/config.py index 9982b9964c..eac8047974 100644 --- a/admin/server/config.py +++ b/admin/server/config.py @@ -285,6 +285,50 @@ def load_configurations(config_path: str) -> list[BaseConfig]: config = MinioConfig(id=id_count, name=name, host=host, port=port, user=user, password=password, service_type="file_store", store_type="minio", detail_func_name="check_minio_alive") configurations.append(config) id_count += 1 + case "s3": + # AWS S3 (or any S3-compatible service: MinIO, R2, ...). + # The config block uses `endpoint_url` instead of `host:port`, + # so parse the URL to derive host/port for the status page. + name: str = "s3" + endpoint_url = v.get("endpoint_url") or "" + if endpoint_url: + try: + parsed = urlparse(endpoint_url) + # `parsed.port` raises ValueError on non-numeric or + # out-of-range ports; `urlparse` itself raises + # ValueError on malformed IPv6 URLs. Fall back to + # the raw endpoint as host and the scheme's + # default port so config loading completes. + host: str = parsed.hostname or endpoint_url + port: int = parsed.port or (443 if parsed.scheme == "https" else 80) + logging.debug( + "Selected S3 host=%s port=%d for endpoint %r.", + host, + port, + endpoint_url, + ) + except ValueError: + logging.warning( + "Could not parse S3 endpoint_url %r; using raw value as host with default port.", + endpoint_url, + ) + host = endpoint_url + port = 443 if endpoint_url.startswith("https://") else 80 + else: + host: str = "s3.amazonaws.com" + port: int = 443 + logging.debug("No S3 endpoint_url configured; defaulting to AWS S3 at %s:%d.", host, port) + config = FileStoreConfig( + id=id_count, + name=name, + host=host, + port=port, + service_type="file_store", + store_type="s3", + detail_func_name="check_s3_alive", + ) + configurations.append(config) + id_count += 1 case "redis": name: str = "redis" url = v["host"] diff --git a/admin/server/services.py b/admin/server/services.py index 57ca3ff4cf..748ebe1cc6 100644 --- a/admin/server/services.py +++ b/admin/server/services.py @@ -273,6 +273,11 @@ class ServiceMgr: @staticmethod def get_all_services(): doc_engine = os.getenv("DOC_ENGINE", "elasticsearch") + # Map STORAGE_IMPL (e.g. "AWS_S3", "MINIO", "OSS") to the lowercase + # `store_type` we use in FileStoreConfig.store_type. The "AWS_" + # prefix is stripped so AWS_S3 matches store_type "s3". + storage_impl = os.getenv("STORAGE_IMPL", "MINIO") + active_store_type = storage_impl.lower().removeprefix("aws_") result = [] configs = SERVICE_CONFIGS.configs for service_id, config in enumerate(configs): @@ -280,6 +285,12 @@ class ServiceMgr: if config_dict["service_type"] == "retrieval": if config_dict["extra"]["retrieval_type"] != doc_engine: continue + if config_dict["service_type"] == "file_store": + # Only show the file-store backend that's actually active. + # Without this filter, a stale minio entry from service_conf.yaml + # is returned even when STORAGE_IMPL=AWS_S3 (see #17294). + if config_dict.get("extra", {}).get("store_type") != active_store_type: + continue try: service_detail = ServiceMgr.get_service_details(service_id) if "status" in service_detail: diff --git a/api/utils/health_utils.py b/api/utils/health_utils.py index 8e8a574f5f..896f62c132 100644 --- a/api/utils/health_utils.py +++ b/api/utils/health_utils.py @@ -15,6 +15,7 @@ # from datetime import datetime import json +import logging import os import requests from timeit import default_timer as timer @@ -246,6 +247,21 @@ def check_minio_alive(): } +def check_s3_alive(): + """ + Check AWS S3 (or any S3-compatible) liveness via the active + storage backend's `.health()` method. Delegates to the generic + ``check_storage`` so the same check works for AWS S3, MinIO, + R2, and any other S3-compatible endpoint. See #17294. + """ + ok, payload = check_storage() + if ok: + logging.debug("check_s3_alive: ok, elapsed=%s ms", payload.get("elapsed", "?")) + return {"status": "alive", "message": f"Confirm elapsed: {payload.get('elapsed', '?')} ms."} + logging.debug("check_s3_alive: failed, error=%s", payload.get("error", "unknown")) + return {"status": "timeout", "message": f"error: {payload.get('error', 'unknown')}"} + + def get_redis_info(): try: return {"status": "alive", "message": REDIS_CONN.info()} diff --git a/test/unit_test/admin/conftest.py b/test/unit_test/admin/conftest.py new file mode 100644 index 0000000000..71f173f138 --- /dev/null +++ b/test/unit_test/admin/conftest.py @@ -0,0 +1,38 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +""" +Conftest for admin unit tests. + +The admin package is invoked as a script (`python admin/server/admin_server.py`) +and its internal modules use top-level imports like `from config import +SERVICE_CONFIGS`. To make those modules importable from pytest, we prepend +``admin/server`` to ``sys.path`` for the duration of the test session. +""" + +import os +import sys + +_ADMIN_SERVER = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", + "..", + "..", + "admin", + "server", +) +_ADMIN_SERVER = os.path.normpath(_ADMIN_SERVER) +if _ADMIN_SERVER not in sys.path: + sys.path.insert(0, _ADMIN_SERVER) diff --git a/test/unit_test/admin/test_get_all_services.py b/test/unit_test/admin/test_get_all_services.py new file mode 100644 index 0000000000..51e72ed322 --- /dev/null +++ b/test/unit_test/admin/test_get_all_services.py @@ -0,0 +1,241 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +""" +Unit tests for ``ServiceMgr.get_all_services`` in +``admin/server/services.py``. + +Specifically covers the new ``STORAGE_IMPL`` filter that hides +inactive file_store backends. See #17294 (Admin Service status still +reports MinIO when object storage is configured as AWS S3). +""" + +from unittest.mock import patch + +import pytest + +from config import ( + ElasticsearchConfig, + FileStoreConfig, + MinioConfig, + SERVICE_CONFIGS, +) + + +def _minio_config(): + return MinioConfig( + id=0, + name="minio", + host="minio", + port=9000, + user="u", + password="p", + service_type="file_store", + store_type="minio", + detail_func_name="check_minio_alive", + ) + + +def _s3_config(): + return FileStoreConfig( + id=1, + name="s3", + host="s3.us-east-1.amazonaws.com", + port=443, + service_type="file_store", + store_type="s3", + detail_func_name="check_s3_alive", + ) + + +def _es_config(retrieval_type="elasticsearch"): + return ElasticsearchConfig( + id=2, + name="elasticsearch", + host="es", + port=9200, + service_type="retrieval", + retrieval_type=retrieval_type, + username="", + password="", + detail_func_name="get_es_cluster_stats", + ) + + +@pytest.fixture +def install_configs(): + """Install a clean ``SERVICE_CONFIGS.configs`` for each test, then + restore the previous value. Required because the admin module uses + ``SERVICE_CONFIGS`` as a mutable namespace, not an instance.""" + previous = list(getattr(SERVICE_CONFIGS, "configs", [])) + yield + SERVICE_CONFIGS.configs = previous + + +class TestFileStoreFilter: + """The new filter: only the active file_store backend is returned.""" + + def test_minio_shown_when_storage_impl_is_minio(self, install_configs, monkeypatch): + monkeypatch.setenv("STORAGE_IMPL", "MINIO") + SERVICE_CONFIGS.configs = [_minio_config(), _s3_config()] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + stores = [s for s in result if s["service_type"] == "file_store"] + assert len(stores) == 1 + assert stores[0]["name"] == "minio" + + def test_s3_shown_when_storage_impl_is_aws_s3(self, install_configs, monkeypatch): + monkeypatch.setenv("STORAGE_IMPL", "AWS_S3") + SERVICE_CONFIGS.configs = [_minio_config(), _s3_config()] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + stores = [s for s in result if s["service_type"] == "file_store"] + assert len(stores) == 1 + assert stores[0]["name"] == "s3" + + def test_no_file_store_shown_when_active_backend_not_configured(self, install_configs, monkeypatch): + """If the active backend has no corresponding config block, + nothing is shown for file_store. We never fall back to a + stale minio entry.""" + monkeypatch.setenv("STORAGE_IMPL", "AWS_S3") + # Only the minio block is present. + SERVICE_CONFIGS.configs = [_minio_config()] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + stores = [s for s in result if s["service_type"] == "file_store"] + assert stores == [] + + +class TestRetrivalFilterStillWorks: + """Regression guard: the existing DOC_ENGINE filter for retrieval + must keep working — the new file_store filter is additive.""" + + def test_elasticsearch_shown_when_doc_engine_is_elasticsearch(self, install_configs, monkeypatch): + monkeypatch.setenv("DOC_ENGINE", "elasticsearch") + monkeypatch.setenv("STORAGE_IMPL", "MINIO") + from config import InfinityConfig + + SERVICE_CONFIGS.configs = [ + _es_config(retrieval_type="elasticsearch"), + InfinityConfig( + id=3, + name="infinity", + host="inf", + port=23800, + service_type="retrieval", + retrieval_type="infinity", + db_name="default_db", + detail_func_name="get_infinity_status", + ), + _minio_config(), + ] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + retrievals = [s for s in result if s["service_type"] == "retrieval"] + assert len(retrievals) == 1 + assert retrievals[0]["name"] == "elasticsearch" + + def test_infinity_filtered_when_doc_engine_is_elasticsearch(self, install_configs, monkeypatch): + """When DOC_ENGINE=elasticsearch, an infinity retrieval config + must be filtered out.""" + monkeypatch.setenv("DOC_ENGINE", "elasticsearch") + monkeypatch.setenv("STORAGE_IMPL", "MINIO") + from config import InfinityConfig + + SERVICE_CONFIGS.configs = [ + _es_config(retrieval_type="elasticsearch"), + InfinityConfig( + id=3, + name="infinity", + host="inf", + port=23800, + service_type="retrieval", + retrieval_type="infinity", + db_name="default_db", + detail_func_name="get_infinity_status", + ), + ] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + names = [s["name"] for s in result if s["service_type"] == "retrieval"] + assert "infinity" not in names + assert "elasticsearch" in names + + +class TestStorageImplNameMapping: + """The STORAGE_IMPL env var uses upper-case + underscores (e.g. + ``AWS_S3``). The ``store_type`` we record is lower-case (``s3``). + The filter must translate correctly so ``AWS_S3`` matches + ``s3``, not ``aws_s3``.""" + + @pytest.mark.parametrize( + "storage_impl,expected_store", + [ + ("MINIO", "minio"), + ("AWS_S3", "s3"), + ("OSS", "oss"), + ("GCS", "gcs"), + ], + ) + def test_env_var_maps_to_store_type(self, install_configs, monkeypatch, storage_impl, expected_store): + monkeypatch.setenv("STORAGE_IMPL", storage_impl) + active = FileStoreConfig( + id=0, + name=expected_store, + host="x", + port=443, + service_type="file_store", + store_type=expected_store, + detail_func_name="check_storage", + ) + inactive = FileStoreConfig( + id=1, + name="other", + host="x", + port=443, + service_type="file_store", + store_type="other", + detail_func_name="check_storage", + ) + SERVICE_CONFIGS.configs = [active, inactive] + + with patch("services.ServiceMgr.get_service_details", return_value={"status": "alive"}): + from services import ServiceMgr + + result = ServiceMgr.get_all_services() + + stores = [s for s in result if s["service_type"] == "file_store"] + assert len(stores) == 1 + assert stores[0]["name"] == expected_store diff --git a/test/unit_test/admin/test_load_configurations.py b/test/unit_test/admin/test_load_configurations.py new file mode 100644 index 0000000000..ce8c87629f --- /dev/null +++ b/test/unit_test/admin/test_load_configurations.py @@ -0,0 +1,178 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +""" +Unit tests for ``load_configurations`` in ``admin/server/config.py``. + +Specifically covers the new ``case "s3"`` branch that the admin status +page needs to render AWS S3 as the active file_store backend. See #17294 +(``Admin Service status still reports MinIO when object storage is +configured as AWS S3``). + +``read_config`` in ``common.config_utils`` interprets its argument as a +filename relative to the project ``conf/`` directory, so we patch it +to return a controlled mapping per test. +""" + +from unittest.mock import patch + +import pytest + +from config import FileStoreConfig, MinioConfig, load_configurations + + +def _read_config_returns(mapping): + """Build a patch that makes ``read_config`` return ``mapping``.""" + return patch("config.read_config", return_value=mapping) + + +class TestLoadS3Configuration: + """``case "s3"`` branch of load_configurations.""" + + def test_s3_endpoint_with_explicit_https(self): + with _read_config_returns( + { + "s3": { + "access_key": "AKIA...", + "secret_key": "...", + "region_name": "us-east-1", + "endpoint_url": "https://s3.us-east-1.amazonaws.com", + "bucket": "my-bucket", + } + } + ): + configs = load_configurations("/unused/service_conf.yaml") + s3 = next(c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3") + assert s3.name == "s3" + assert s3.service_type == "file_store" + assert s3.detail_func_name == "check_s3_alive" + assert s3.host == "s3.us-east-1.amazonaws.com" + assert s3.port == 443 + + def test_s3_endpoint_with_explicit_http(self): + with _read_config_returns( + { + "s3": { + "endpoint_url": "http://minio.local:9000", + "bucket": "test", + } + } + ): + configs = load_configurations("/unused/service_conf.yaml") + s3 = next(c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3") + assert s3.host == "minio.local" + assert s3.port == 9000 + + def test_s3_endpoint_with_non_default_https_port(self): + with _read_config_returns( + { + "s3": { + "endpoint_url": "https://s3.custom.example.com:8443", + "bucket": "x", + } + } + ): + configs = load_configurations("/unused/service_conf.yaml") + s3 = next(c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3") + assert s3.host == "s3.custom.example.com" + assert s3.port == 8443 + + def test_s3_endpoint_omitted_uses_aws_default(self): + with _read_config_returns({"s3": {"bucket": "my-bucket"}}): + configs = load_configurations("/unused/service_conf.yaml") + s3 = next(c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3") + # No endpoint_url → fall back to standard AWS S3 endpoint. + assert s3.host == "s3.amazonaws.com" + assert s3.port == 443 + + def test_s3_id_count_strictly_increments(self): + """Each backend added by the loader must receive a unique id.""" + with _read_config_returns( + { + "minio": {"host": "minio:9000", "user": "u", "password": "p"}, + "s3": {"endpoint_url": "https://s3.us-east-1.amazonaws.com", "bucket": "x"}, + "redis": {"host": "redis:6379", "password": "", "db": 0}, + } + ): + configs = load_configurations("/unused/service_conf.yaml") + ids = [c.id for c in configs] + assert ids == sorted(set(ids)), f"Duplicate ids: {ids}" + + +class TestLoadS3Regressions: + """Make sure the new branch does not break existing backends.""" + + def test_minio_still_creates_minio_config(self): + with _read_config_returns( + { + "minio": { + "user": "rag_flow", + "password": "infini_rag_flow", + "host": "minio:9000", + } + } + ): + configs = load_configurations("/unused/service_conf.yaml") + minio = next(c for c in configs if isinstance(c, MinioConfig)) + assert minio.store_type == "minio" + assert minio.service_type == "file_store" + assert minio.detail_func_name == "check_minio_alive" + assert minio.host == "minio" + assert minio.port == 9000 + + def test_both_minio_and_s3_in_same_config(self): + """When the user has both backends configured, the admin loader + surfaces both. The service-layer filter at #17294 is responsible + for hiding the inactive one.""" + with _read_config_returns( + { + "minio": {"host": "minio:9000", "user": "u", "password": "p"}, + "s3": {"endpoint_url": "https://s3.us-east-1.amazonaws.com", "bucket": "x"}, + } + ): + configs = load_configurations("/unused/service_conf.yaml") + minio = [c for c in configs if isinstance(c, MinioConfig)] + s3 = [c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3"] + assert len(minio) == 1 + assert len(s3) == 1 + + +class TestLoadS3HandlesBadEndpoint: + """The endpoint_url parser must never raise on bad input — the admin + UI would otherwise crash and stop showing all services. The fix + degrades to a best-effort host string.""" + + @pytest.mark.parametrize( + "endpoint", + [ + "not a url", + "://no-scheme", + "https://", + # `parsed.port` raises ValueError on these: + "https://s3.example.com:not-a-port", + "https://s3.example.com:99999999999999", + # `urlparse` itself raises ValueError on malformed IPv6: + "https://[::1", + ], + ) + def test_malformed_endpoint_does_not_raise(self, endpoint): + with _read_config_returns({"s3": {"endpoint_url": endpoint, "bucket": "x"}}): + configs = load_configurations("/unused/service_conf.yaml") + s3 = next(c for c in configs if isinstance(c, FileStoreConfig) and c.store_type == "s3") + # The contract is "does not raise". The exact host/port for + # pathological input is best-effort. + assert s3 is not None + # port is always an int (or 0 for empty string) — never raises. + assert isinstance(s3.port, int) diff --git a/test/unit_test/api/utils/test_health_utils_s3.py b/test/unit_test/api/utils/test_health_utils_s3.py new file mode 100644 index 0000000000..fe86346876 --- /dev/null +++ b/test/unit_test/api/utils/test_health_utils_s3.py @@ -0,0 +1,84 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +""" +Unit tests for the S3 file-store health check (check_s3_alive). +Closes #17294 — Admin Service status used to silently show MinIO when +STORAGE_IMPL=AWS_S3, because the only health check wired into the +admin config loader was check_minio_alive. +""" + +from unittest.mock import patch + + +class TestCheckS3Alive: + """Test check_s3_alive delegates to the generic check_storage.""" + + @patch("api.utils.health_utils.check_storage") + def test_returns_alive_when_storage_healthy(self, mock_check_storage): + mock_check_storage.return_value = (True, {"elapsed": "12.3"}) + from api.utils.health_utils import check_s3_alive + + result = check_s3_alive() + assert result["status"] == "alive" + assert "elapsed" in result["message"] + mock_check_storage.assert_called_once_with() + + @patch("api.utils.health_utils.check_storage") + def test_returns_timeout_when_storage_unhealthy(self, mock_check_storage): + mock_check_storage.return_value = (False, {"elapsed": "9.0", "error": "Connection refused"}) + from api.utils.health_utils import check_s3_alive + + result = check_s3_alive() + assert result["status"] == "timeout" + assert "Connection refused" in result["message"] + + @patch("api.utils.health_utils.check_storage") + def test_returns_timeout_when_error_missing(self, mock_check_storage): + # Storage health returned False with no error key — must still + # surface a non-empty message so the admin UI doesn't render + # an empty status. + mock_check_storage.return_value = (False, {"elapsed": "1.0"}) + from api.utils.health_utils import check_s3_alive + + result = check_s3_alive() + assert result["status"] == "timeout" + assert "unknown" in result["message"] + + +class TestCheckStorageDelegateContract: + """Verify check_s3_alive and check_storage stay in sync. + + The S3 health check is a thin wrapper over check_storage so the + same code path works for AWS S3, MinIO, R2, etc. If check_storage's + return shape ever changes, these tests will fail and force a + companion update to check_s3_alive. + """ + + @patch("api.utils.health_utils.check_storage") + def test_message_includes_elapsed_from_storage(self, mock_check_storage): + mock_check_storage.return_value = (True, {"elapsed": "42.0"}) + from api.utils.health_utils import check_s3_alive + + result = check_s3_alive() + assert "42.0" in result["message"] + + @patch("api.utils.health_utils.check_storage") + def test_message_includes_error_string(self, mock_check_storage): + mock_check_storage.return_value = (False, {"elapsed": "0.5", "error": "boom"}) + from api.utils.health_utils import check_s3_alive + + result = check_s3_alive() + assert "boom" in result["message"]