feat: Using opensearch instead of elasticsearch (#116)

* feat: add AUDIT log level for upload

* chore: update outdated tests

* fix: allow empty str as RESULT_PATH_PREFIX & replace w/ default val

* fix: allowing optional original stac item

* fix: using opensearch

* fix: remove unused class

* fix: backward compatible

* fix: use_ssl option re-added

* fix: replace lsmd pointer with uds pointer

Author: wphyojpl

mdps_ds_lib/lib/aws/es_abstract.py

@@ -5,8 +5,9 @@
 
 
 class ESAbstract(ABC):
+
     @abstractmethod
-    def migrate_index_data(self, old_index, new_index):
+    def migrate_index_data(self, old_index, new_index, remove_old_data=True):
         return
     @abstractmethod
     def create_index(self, index_name, index_body):
@@ -62,11 +63,11 @@ def query_with_scroll(self, dsl, querying_index=None):
         return
 
     @abstractmethod
-    def query(self, dsl, querying_index=None):
+    def delete_by_query(self, dsl, querying_index=None):
         return
 
     @abstractmethod
-    def delete_by_query(self, dsl, querying_index=None):
+    def query(self, dsl, querying_index=None):
         return
 
     @abstractmethod

mdps_ds_lib/lib/aws/es_factory.py

@@ -4,15 +4,20 @@
 class ESFactory(FactoryAbstract):
     NO_AUTH = 'NO_AUTH'
     AWS = 'AWS'
+    AWS_ES = 'AWS_ES'
+
+    def get_instance_from_env(self, **kwargs):
+        raise NotImplementedError(f'not yet')
 
     def get_instance(self, class_type, **kwargs):
-        if 'use_ssl' not in kwargs:
-            kwargs['use_ssl'] = True
         ct = class_type.upper()
         if ct == self.NO_AUTH:
-            from mdps_ds_lib.lib.aws.es_middleware import ESMiddleware
-            return ESMiddleware(kwargs['index'], kwargs['base_url'], port=kwargs['port'], use_ssl=kwargs['use_ssl'])
+            from mdps_ds_lib.lib.aws.es_middleware_no_auth import ESMiddlewareNoAuth
+            return ESMiddlewareNoAuth(kwargs['index'], kwargs['base_url'], port=kwargs['port'], use_ssl=kwargs['use_ssl'])
         if ct == self.AWS:
+            from mdps_ds_lib.lib.aws.os_middleware_aws import OsMiddlewareAws
+            return OsMiddlewareAws(kwargs['index'], kwargs['base_url'], port=kwargs['port'], use_ssl=kwargs['use_ssl'])
+        if ct == self.AWS_ES:
             from mdps_ds_lib.lib.aws.es_middleware_aws import EsMiddlewareAws
             return EsMiddlewareAws(kwargs['index'], kwargs['base_url'], port=kwargs['port'], use_ssl=kwargs['use_ssl'])
         raise ModuleNotFoundError(f'cannot find ES class for {ct}')

mdps_ds_lib/lib/aws/es_middleware.py renamed to mdps_ds_lib/lib/aws/es_middleware_abstract.py

@@ -1,22 +1,15 @@
 import logging
-
-from elasticsearch import Elasticsearch
-from elasticsearch.exceptions import NotFoundError
-
 from mdps_ds_lib.lib.aws.es_abstract import ESAbstract, DEFAULT_TYPE
 
 LOGGER = logging.getLogger(__name__)
 
 
-class ESMiddleware(ESAbstract):
-
+class ESMiddlewareAbstract(ESAbstract):
     def __init__(self, index, base_url, port=443, use_ssl=True) -> None:
         if any([k is None for k in [index, base_url]]):
             raise ValueError(f'index or base_url is None')
         self.__index = index
-        base_url = base_url.replace('https://', '')  # hide https
-        # https://elasticsearch-py.readthedocs.io/en/v7.13.4/api.html#elasticsearch.Elasticsearch
-        self._engine = Elasticsearch(hosts=[{'host': base_url, 'port': port, 'use_ssl': use_ssl}])
+        self._engine = None
 
     def __validate_index(self, index):
         if index is not None:
@@ -73,9 +66,11 @@ def create_index(self, index_name, index_body):
     def get_index_mapping(self, index_name):
         try:
             result = self._engine.indices.get_mapping(index=index_name)
-        except NotFoundError as e:
-            return None
-        return result
+        except Exception as e:
+            if e.error == 'index_not_found_exception':
+                return None
+            raise e
+        return result[index_name]
 
     def has_index(self, index_name):
         result = self._engine.indices.exists(index=index_name)
@@ -84,18 +79,21 @@ def has_index(self, index_name):
     def swap_index_for_alias(self, alias_name, old_index_name, new_index_name):
         try:
             temp_result = self._engine.indices.delete_alias(index=old_index_name, name=alias_name)
-        except NotFoundError as ee:
-            LOGGER.exception(f'error while unlinking {old_index_name} from {alias_name}')
-            temp_result = {}
+        except Exception as ee:
+            if 'NotFoundError' in str(ee):
+                return {}
+            raise ee
         result = self.create_alias(new_index_name, alias_name)
         return result
 
     def get_alias(self, alias_name):
         # /Users/wphyo/anaconda3/envs/cumulus_py_3.9/lib/python3.9/site-packages/elasticsearch-7.13.4-py3.9.egg/elasticsearch/client/indices.py
         try:
             result = self._engine.indices.get_alias(name=alias_name)
-        except NotFoundError as ee:
-            return {}
+        except Exception as ee:
+            if 'NotFoundError' in str(ee):
+                return {}
+            raise ee
         return result
 
     def create_alias(self, index_name, alias_name):
@@ -124,7 +122,7 @@ def index_many(self, docs=None, doc_ids=None, doc_dict=None, index=None):
                                               body=body, doc_type=DEFAULT_TYPE)
             LOGGER.info('indexed. result: {}'.format(index_result))
             return self.__check_errors_for_bulk(index_result)
-        except:
+        except Exception as e:
             LOGGER.exception('cannot add indices with ids: {} for index: {}'.format(list(doc_dict.keys()), index))
             return doc_dict
         return
@@ -136,9 +134,9 @@ def index_one(self, doc, doc_id, index=None):
                                               body=doc, doc_type=DEFAULT_TYPE, id=doc_id)
             LOGGER.info('indexed. result: {}'.format(index_result))
             pass
-        except Exception as e:
+        except:
             LOGGER.exception('cannot add a new index with id: {} for index: {}'.format(doc_id, index))
-            raise e
+            return None
         return self
 
     def update_many(self, docs=None, doc_ids=None, doc_dict=None, index=None):
@@ -208,13 +206,13 @@ def query_with_scroll(self, dsl, querying_index=None):
                 first_batch['hits']['hits'].extend(scrolled_result['hits']['hits'])
         return first_batch
 
-    def query(self, dsl, querying_index=None):
+    def delete_by_query(self, dsl, querying_index=None):
         index = self.__validate_index(querying_index)
-        return self._engine.search(body=dsl, index=index)
+        return self._engine.delete_by_query(body=dsl, index=index, conflicts='proceed', request_timeout=120)
 
-    def delete_by_query(self, dsl, querying_index=None):
+    def query(self, dsl, querying_index=None):
         index = self.__validate_index(querying_index)
-        return self._engine.delete_by_query(body=dsl, index=index)
+        return self._engine.search(body=dsl, index=index)
 
     def __is_querying_next_page(self, targeted_size: int, current_size: int, total_size: int):
         if targeted_size < 0:

mdps_ds_lib/lib/aws/es_middleware_aws.py

@@ -1,18 +1,17 @@
 import logging
 
-from requests_aws4auth import AWS4Auth
-
 from mdps_ds_lib.lib.aws.aws_cred import AwsCred
-from mdps_ds_lib.lib.aws.es_middleware import ESMiddleware
 from elasticsearch import Elasticsearch, RequestsHttpConnection
+from requests_aws4auth import AWS4Auth
+from mdps_ds_lib.lib.aws.es_middleware_abstract import ESMiddlewareAbstract
 
 LOGGER = logging.getLogger(__name__)
 
 
-class EsMiddlewareAws(ESMiddleware):
+class EsMiddlewareAws(ESMiddlewareAbstract):
 
     def __init__(self, index, base_url, port=443, use_ssl=True) -> None:
-        super().__init__(index, base_url, port, use_ssl)
+        super().__init__(index, base_url, port)
         base_url = base_url.replace('https://', '')  # hide https
         self._index = index
         aws_cred = AwsCred()
@@ -23,7 +22,7 @@ def __init__(self, index, base_url, port=443, use_ssl=True) -> None:
         self._engine = Elasticsearch(
             hosts=[{'host': base_url, 'port': port}],
             http_auth=aws_auth,
-            use_ssl=True,
+            use_ssl=use_ssl,
             verify_certs=True,
             connection_class=RequestsHttpConnection
         )

mdps_ds_lib/lib/aws/es_middleware_no_auth.py

@@ -0,0 +1,14 @@
+import logging
+
+from elasticsearch import Elasticsearch
+from mdps_ds_lib.lib.aws.es_middleware_abstract import ESMiddlewareAbstract
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ESMiddlewareNoAuth(ESMiddlewareAbstract):
+
+    def __init__(self, index, base_url, port=443, use_ssl=True) -> None:
+        super().__init__(index, base_url, port)
+        base_url = base_url.replace('https://', '')  # hide https
+        self._engine = Elasticsearch(hosts=[{'host': base_url, 'port': port}])

mdps_ds_lib/lib/aws/os_middleware_aws.py

@@ -0,0 +1,29 @@
+import logging
+
+from mdps_ds_lib.lib.aws.aws_cred import AwsCred
+from opensearchpy import AWSV4SignerAuth, OpenSearch, RequestsHttpConnection
+from mdps_ds_lib.lib.aws.es_middleware_abstract import ESMiddlewareAbstract
+
+LOGGER = logging.getLogger(__name__)
+
+
+class OsMiddlewareAws(ESMiddlewareAbstract):
+
+    def __init__(self, index, base_url, port=443, use_ssl=True) -> None:
+        super().__init__(index, base_url, port)
+        base_url = base_url.replace('https://', '')  # hide https
+        self._index = index
+        aws_cred = AwsCred()
+        service = 'es'
+        credentials = aws_cred.get_session().get_credentials()
+        # https://opensearch.org/blog/aws-sigv4-support-for-clients/
+        # This works
+        auth = AWSV4SignerAuth(credentials, aws_cred.region)
+
+        self._engine = OpenSearch(
+            hosts=[{'host': base_url, 'port': port}],
+            http_auth=auth,
+            use_ssl=use_ssl,
+            verify_certs=True,
+            connection_class=RequestsHttpConnection
+        )