[HWORKS-975] OpenSearchApi in hopsworks-api should not rely on ELASTIC_ENDPOINT

python/hopsworks/core/kafka_api.py

@@ -349,9 +349,6 @@ def get_default_config(self):
         # Raises
             `RestAPIError`: If unable to get the kafka configuration.
         """
-        _client = client.get_instance()
-        if type(_client) == Client:
-            _client.download_certs(self._project_name)
 
         config = {
             constants.KAFKA_SSL_CONFIG.SECURITY_PROTOCOL_CONFIG: self._get_security_protocol(),
@@ -368,7 +365,7 @@ def get_default_config(self):
             constants.KAFKA_CONSUMER_CONFIG.GROUP_ID_CONFIG: "my-group-id",
             constants.KAFKA_SSL_CONFIG.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG: "none",
         }
-
+        _client = client.get_instance()
         if type(_client) == Client:
             config[constants.KAFKA_PRODUCER_CONFIG.BOOTSTRAP_SERVERS_CONFIG] = ",".join(
                 [

python/hopsworks/core/opensearch_api.py

@@ -14,10 +14,11 @@
 #   limitations under the License.
 #
 
-import os
 from furl import furl
 
 from hopsworks import client, constants
+from hopsworks.core import variable_api
+from hopsworks.client.exceptions import OpenSearchException
 
 
 class OpenSearchApi:
@@ -28,9 +29,27 @@ def __init__(
     ):
         self._project_id = project_id
         self._project_name = project_name
+        self._variable_api = variable_api.VariableApi()
 
     def _get_opensearch_url(self):
-        return os.environ[constants.ENV_VARS.ELASTIC_ENDPOINT_ENV_VAR]
+        if isinstance(client.get_instance(), client.external.Client):
+            external_domain = self._variable_api.get_variable(
+                "loadbalancer_external_domain"
+            )
+            if external_domain == "":
+                # fallback to use hostname of head node
+                external_domain = client.get_instance().host
+            return f"https://{external_domain}:9200"
+        else:
+            service_discovery_domain = self._variable_api.get_variable(
+                "service_discovery_domain"
+            )
+            if service_discovery_domain == "":
+                raise OpenSearchException(
+                    "Client could not locate service_discovery_domain "
+                    "in cluster configuration or variable is empty."
+                )
+            return f"https://rest.elastic.service.{service_discovery_domain}:9200"
 
     def get_project_index(self, index):
         """

python/hopsworks/project.py

@@ -176,6 +176,9 @@ def get_kafka_api(self):
         # Returns
             `KafkaApi`: The Kafka Api handle
         """
+        _client = client.get_instance()
+        if type(_client) == Client:
+            _client.download_certs(self.name)
         return self._kafka_api
 
     def get_opensearch_api(self):
@@ -184,6 +187,9 @@ def get_opensearch_api(self):
         # Returns
             `OpenSearchApi`: The OpenSearch Api handle
         """
+        _client = client.get_instance()
+        if type(_client) == Client:
+            _client.download_certs(self.name)
         return self._opensearch_api
 
     def get_jobs_api(self):