Merge pull request #545 from GSA/add-opensearch

Add opensearch

Author: FuhuXia

docker-compose.yml

@@ -25,6 +25,19 @@ services:
         interval: 5s
         timeout: 5s
         retries: 5
+  opensearch:
+      image: opensearchproject/opensearch:2.11.1
+      environment:
+        discovery.type: single-node
+      ports:
+        - "9200:9200"
+        - "9600:9600"
+      healthcheck:
+        test: ['CMD-SHELL', 'curl -sk -u admin:admin https://localhost:9200/_cluster/health?wait_for_status=yellow\&timeout=1s | grep -q ''"timed_out":false''']
+        interval: 5s
+        timeout: 5s
+        retries: 12
+        start_period: 10s
   transformer:
       image: ghcr.io/gsa/mdtranslator:latest
       restart: always
@@ -45,11 +58,14 @@ services:
     depends_on:
       db:
         condition: service_healthy
+      opensearch:
+        condition: service_healthy
       transformer:
         condition: service_healthy
     volumes:
       - .:/app
     environment:
+      OPENSEARCH_HOST: opensearch
       DATABASE_URI: postgresql+psycopg://${DATABASE_USER}:${DATABASE_PASSWORD}@db:${DATABASE_PORT}/${DATABASE_NAME}
       CF_SERVICE_USER: ${CF_SERVICE_USER}
       CF_SERVICE_AUTH: ${CF_SERVICE_AUTH}

harvester/harvest.py

@@ -99,6 +99,8 @@ class HarvestSource:
     internal_records: dict = field(default_factory=lambda: {}, repr=False)
 
     deletions: set = field(default_factory=lambda: set(), repr=False)
+    _opensearch: object = field(default=None, init=False, repr=False)
+    _opensearch_initialized: bool = field(default=False, init=False, repr=False)
 
     def __post_init__(self) -> None:
         self._db_interface: HarvesterDBInterface = db_interface
@@ -154,6 +156,27 @@ def records(self, value) -> None:
     def reporter(self):
         return self._reporter
 
+    @property
+    def opensearch(self):
+        if self._opensearch_initialized:
+            return self._opensearch
+
+        self._opensearch_initialized = True
+        opensearch_host = os.getenv("OPENSEARCH_HOST")
+        if not opensearch_host:
+            self._opensearch = None
+            return None
+
+        try:
+            from harvester.opensearch import OpenSearchInterface
+
+            self._opensearch = OpenSearchInterface.from_environment()
+        except Exception as e:
+            logger.exception("Failed to initialize OpenSearch client: %s", e)
+            self._opensearch = None
+
+        return self._opensearch
+
     @property
     def source_attrs(self) -> List:
         return self._source_attrs
@@ -1051,6 +1074,41 @@ def _dataset_payload(self, metadata: dict) -> dict:
 
         return payload
 
+    def _index_dataset_in_opensearch(self, dataset) -> None:
+        client = self.harvest_source.opensearch
+        if client is None or dataset is None:
+            return
+        try:
+            succeeded, failed, errors = client.index_datasets([dataset])
+            if failed:
+                logger.error(
+                    "OpenSearch indexing failed for dataset %s (slug %s): %s",
+                    dataset.id,
+                    dataset.slug,
+                    errors,
+                )
+        except Exception as e:
+            logger.exception(
+                "OpenSearch indexing error for dataset %s (slug %s): %s",
+                dataset.id,
+                dataset.slug,
+                e,
+            )
+
+    def _delete_dataset_from_opensearch(self, dataset) -> None:
+        client = self.harvest_source.opensearch
+        if client is None or dataset is None:
+            return
+        try:
+            client.delete_dataset_by_id(dataset.id)
+        except Exception as e:
+            logger.exception(
+                "OpenSearch delete error for dataset %s (slug %s): %s",
+                dataset.id,
+                dataset.slug,
+                e,
+            )
+
     def sync(self):
         try:
             if self.status == "error":
@@ -1068,13 +1126,21 @@ def sync(self):
             if self.action in ("create", "update") and metadata is not None:
                 dataset_payload = self._dataset_payload(metadata)
                 if self.action == "create":
-                    self._insert_dataset_with_unique_slug(dataset_payload)
+                    dataset = self._insert_dataset_with_unique_slug(dataset_payload)
                 else:
-                    self.harvest_source.db_interface.upsert_dataset(dataset_payload)
+                    dataset = self.harvest_source.db_interface.upsert_dataset(
+                        dataset_payload
+                    )
+                self._index_dataset_in_opensearch(dataset)
             elif self.action == "delete" and self.dataset_slug:
-                self.harvest_source.db_interface.delete_dataset_by_slug(
+                dataset = self.harvest_source.db_interface.get_dataset_by_slug(
                     self.dataset_slug
                 )
+                deleted = self.harvest_source.db_interface.delete_dataset_by_slug(
+                    self.dataset_slug
+                )
+                if deleted:
+                    self._delete_dataset_from_opensearch(dataset)
 
             return True
 
@@ -1103,13 +1169,12 @@ def update_self_in_db(self) -> bool:
             data,
         )
 
-    def _insert_dataset_with_unique_slug(self, dataset_payload: dict) -> None:
+    def _insert_dataset_with_unique_slug(self, dataset_payload: dict) -> object:
         """Persist a dataset, retrying with a unique slug when needed."""
 
         while True:
             try:
-                self.harvest_source.db_interface.insert_dataset(dataset_payload)
-                return
+                return self.harvest_source.db_interface.insert_dataset(dataset_payload)
             except IntegrityError as error:
                 if not self._is_slug_unique_violation(error):
                     raise