fix: solr keywords field type change (#1153)

Author: olevski

components/renku_data_services/search/core.py

@@ -41,9 +41,12 @@
 logger = logging.getLogger(__name__)
 
 
-async def update_solr(search_updates_repo: SearchUpdatesRepo, solr_client: SolrClient, batch_size: int) -> None:
+async def update_solr(
+    search_updates_repo: SearchUpdatesRepo, solr_client: SolrClient, batch_size: int
+) -> list[Exception]:
     """Selects entries from the search staging table and updates SOLR."""
     counter = 0
+    output: list[Exception] = []
     while True:
         entries = await search_updates_repo.select_next(batch_size)
         if entries == []:
@@ -69,17 +72,22 @@ async def update_solr(search_updates_repo: SearchUpdatesRepo, solr_client: SolrC
                 await solr_client.delete(DeleteDoc.solr_query())
             except Exception as de:
                 logger.error("Error when removing soft-deleted documents", exc_info=de)
+                output.append(de)
 
         except Exception as e:
+            output.append(e)
             logger.error(f"Error while updating solr with entities {ids}", exc_info=e)
             try:
                 await search_updates_repo.mark_failed(ids)
             except Exception as e2:
+                output.append(e2)
                 logger.error("Error while setting search entities to failed", exc_info=e2)
 
     if counter > 0:
         logger.info(f"Updated {counter} entries in SOLR")
 
+    return output
+
 
 async def _renku_query(
     authz_client: AuthzClient, ctx: Context, uq: SolrUserQuery, limit: int, offset: int

components/renku_data_services/search/reprovision.py

@@ -17,7 +17,7 @@
 from renku_data_services.project.models import Project
 from renku_data_services.search.db import SearchUpdatesRepo
 from renku_data_services.solr import entity_schema
-from renku_data_services.solr.solr_client import DefaultSolrClient, SolrClientConfig
+from renku_data_services.solr.solr_client import DefaultSolrAdminClient, DefaultSolrClient, SolrClientConfig
 from renku_data_services.solr.solr_migrate import SchemaMigrator
 from renku_data_services.users.db import UserRepo
 from renku_data_services.users.models import UserInfo
@@ -103,7 +103,21 @@ def log_counter(c: int) -> None:
             started = datetime.now()
             await self._search_updates_repo.clear_all()
             async with DefaultSolrClient(self._solr_config) as client:
-                await client.delete("_type:*")
+                res = await client.delete("_type:*")
+                if res.status_code != 200:
+                    logger.error(
+                        f"Failed to delete all documents in solr during reprovisioning: {res.text}, "
+                        f"status_code: {res.status_code}",
+                        exc_info=False,
+                    )
+                async with DefaultSolrAdminClient(self._solr_config) as admin_client:
+                    res = await admin_client.reload(None)
+                    if res.status_code != 200:
+                        logger.error(
+                            f"Failed to reload solr core during reprovisioning: {res.text}, "
+                            f"status code: {res.status_code}",
+                            exc_info=False,
+                        )
 
             if migrate_solr_schema:
                 await migrator.migrate(entity_schema.all_migrations)

components/renku_data_services/solr/entity_schema.py

@@ -126,8 +126,8 @@ class FieldTypes:
     AddCommand(Field.of(Fields.description, FieldTypes.text)),
     AddCommand(Field.of(Fields.created_by, FieldTypes.id)),
     AddCommand(Field.of(Fields.creation_date, FieldTypes.date_time)),
-    # text all
     AddCommand(FieldTypes.text_all),
+    # text all
     AddCommand(Field.of(Fields.content_all, FieldTypes.text_all).make_multi_valued()),
     AddCommand(CopyFieldRule(source=Fields.name, dest=Fields.content_all)),
     AddCommand(CopyFieldRule(source=Fields.description, dest=Fields.content_all)),

components/renku_data_services/solr/solr_client.py

@@ -626,9 +626,6 @@ async def delete(self, query: str) -> Response:
 class DefaultSolrClient(SolrClient):
     """Default implementation of the solr client."""
 
-    delegate: AsyncClient
-    config: SolrClientConfig
-
     def __init__(self, cfg: SolrClientConfig):
         self.config = cfg
         url_parsed = list(urlparse(cfg.base_url))
@@ -758,23 +755,22 @@ async def create(self, core_name: str | None) -> None:
         """Create a core."""
         ...
 
+    @abstractmethod
+    async def reload(self, core_name: str | None) -> Response:
+        """Reload a core."""
+        ...
+
 
 class DefaultSolrAdminClient(SolrAdminClient):
     """A client to the core admin api.
 
     Url: https://solr.apache.org/guide/solr/latest/configuration-guide/coreadmin-api.html
     """
 
-    delegate: AsyncClient
-    config: SolrClientConfig
-
     def __init__(self, cfg: SolrClientConfig):
         self.config = cfg
-        url_parsed = list(urlparse(cfg.base_url))
-        url_parsed[2] = urljoin(url_parsed[2], "/api/cores")
-        burl = urlunparse(url_parsed)
         bauth = BasicAuth(username=cfg.user.username, password=cfg.user.password) if cfg.user is not None else None
-        self.delegate = AsyncClient(auth=bauth, base_url=burl, timeout=cfg.timeout)
+        self.delegate = AsyncClient(auth=bauth, base_url=self.config.base_url, timeout=cfg.timeout)
 
     async def __aenter__(self) -> Self:
         await self.delegate.__aenter__()
@@ -788,7 +784,7 @@ async def __aexit__(
     async def core_status(self, core_name: str | None) -> dict[str, Any] | None:
         """Return the status of the connected core or the one given by `core_name`."""
         core = core_name or self.config.core
-        resp = await self.delegate.get(f"/{core}")
+        resp = await self.delegate.get(f"/api/cores/{core}")
         if not resp.is_success:
             raise SolrClientStatusException(self.config, resp)
         else:
@@ -800,8 +796,16 @@ async def create(self, core_name: str | None) -> None:
         """Create a core with the given `core_name` or the name provided in the config object."""
         core = core_name or self.config.core
         data = {"create": {"name": core, "configSet": self.config.configset}}
-        resp = await self.delegate.post("", json=data)
+        resp = await self.delegate.post("/api/cores", json=data)
         if not resp.is_success:
             raise SolrClientCreateCoreException(core, resp)
         else:
             return None
+
+    async def reload(self, core_name: str | None) -> Response:
+        """Reload a core with the given `core_name` or the name provided in the config object."""
+        core = core_name or self.config.core
+        return await self.delegate.post(
+            f"/v2/cores/{core}/reload",
+            headers={"Content-Type": "application/json"},
+        )

components/renku_data_services/solr/solr_schema.py

@@ -315,7 +315,12 @@ def to_dict(self) -> dict[str, Any]:
 @dataclass
 @final
 class ReplaceCommand(SchemaCommand):
-    """Replace a field, field type or dynamic field."""
+    """Replace a field, field type or dynamic field.
+
+    Use this with care if you are changing the field type.
+    We have seen issues when we use this command to change the field type
+    that do not occur if you call 'delete' and then 'add' rather than just 'replace'.
+    """
 
     value: FieldType | Field | DynamicFieldRule
 

test/bases/renku_data_services/data_api/conftest.py

@@ -35,7 +35,7 @@
 from renku_data_services.secrets_storage_api.app import register_all_handlers as register_secrets_handlers
 from renku_data_services.secrets_storage_api.dependencies import DependencyManager as SecretsDependencyManager
 from renku_data_services.solr import entity_schema
-from renku_data_services.solr.solr_client import DefaultSolrClient
+from renku_data_services.solr.solr_client import DefaultSolrAdminClient, DefaultSolrClient
 from renku_data_services.solr.solr_migrate import SchemaMigrator
 from renku_data_services.storage.rclone import RCloneValidator
 from renku_data_services.users.dummy_kc_api import DummyKeycloakAPI
@@ -295,21 +295,26 @@ async def search_reprovision(search_push_updates) -> SearchReprovisionCall:
     admin = InternalServiceAdmin(id=ServiceAdminId.search_reprovision)
 
     async def search_reprovision_helper(
-        app_manager_instance: DependencyManager, migrate_solr_schema: bool = True
+        app_manager_instance: DependencyManager, migrate_solr_schema: bool = True, clear_index: bool = False
     ) -> None:
         await app_manager_instance.search_reprovisioning.run_reprovision(admin, migrate_solr_schema)
-        await search_push_updates(app_manager_instance, clear_index=False)
+        await search_push_updates(app_manager_instance, clear_index=clear_index)
 
     return search_reprovision_helper
 
 
 @pytest_asyncio.fixture
 async def search_push_updates():
-    async def search_push_updates_helper(app_manager_instance: DependencyManager, clear_index: bool = True) -> None:
+    async def search_push_updates_helper(app_manager_instance: DependencyManager, clear_index: bool = False) -> None:
         async with DefaultSolrClient(app_manager_instance.config.solr) as client:
             if clear_index:
-                await client.delete("*:*")
-            await search_core.update_solr(app_manager_instance.search_updates_repo, client, 10)
+                res = await client.delete("_type:*")
+                assert res.status_code == 200, res.text
+                async with DefaultSolrAdminClient(app_manager_instance.config.solr) as admin_client:
+                    res = await admin_client.reload(None)
+                    assert res.status_code == 200, res.text
+            responses = await search_core.update_solr(app_manager_instance.search_updates_repo, client, 10)
+            assert len(responses) == 0, responses
 
     return search_push_updates_helper
 

test/bases/renku_data_services/data_api/test_search_migrations.py

@@ -83,8 +83,7 @@ def _helper(latest_version: int | None = None) -> list[SchemaMigration]:
     ("start_solr_version", "end_solr_version"),
     [
         (12, 13),
-        # TODO: Fix this migration
-        # (13, 14),
+        (13, 14),
     ],
 )
 @pytest.mark.xdist_group("search")
@@ -139,7 +138,7 @@ async def test_search_schema_upgrade(
     assert_search_result(result, [p1, p3])
     res = await solr_migrator.migrate(get_solr_schemas(end_solr_version))
     assert res.migrations_run == end_solr_version - start_solr_version, res
-    await search_reprovision(app_manager, migrate_solr_schema=False)
+    await search_reprovision(app_manager, migrate_solr_schema=True, clear_index=True)
     result = await search_query(sanic_client, "type:project", user=wout)
     assert_search_result(result, [p1, p3])
     result = await search_query(sanic_client, "keyword:test-keyword", user=wout)