Merge branch 'logicalclocks:main' into dedup-basic-hopsworks

Author: aversey

Diff for: python/hopsworks/connection.py

@@ -244,6 +244,9 @@ def connect(self):
                     self._host,
                     self._port,
                     self._project,
+                    None,
+                    None,
+                    None,
                     self._hostname_verification,
                     self._trust_store_path,
                     self._cert_folder,

Diff for: python/hopsworks/constants.py

@@ -1,5 +1,5 @@
 #
-#   Copyright 2022 Logical Clocks AB
+#   Copyright 2024 Hopsworks AB
 #
 #   Licensed under the Apache License, Version 2.0 (the "License");
 #   you may not use this file except in compliance with the License.
@@ -14,124 +14,59 @@
 #   limitations under the License.
 #
 
-
-class JOBS:
-    SUCCESS_STATES = ["FINISHED", "SUCCEEDED"]
-    ERROR_STATES = [
-        "FAILED",
-        "KILLED",
-        "FRAMEWORK_FAILURE",
-        "APP_MASTER_START_FAILED",
-        "INITIALIZATION_FAILED",
-    ]
-
-
-class GIT:
-    SUCCESS_STATES = ["SUCCESS"]
-    ERROR_STATES = ["FAILED", "KILLED", "INITIALIZATION_FAILED", "TIMEDOUT"]
-
-
-class SERVICES:
-    LIST = ["JOBS", "KAFKA", "JUPYTER", "HIVE", "SERVING", "FEATURESTORE", "AIRFLOW"]
-
-
-class OPENSEARCH_CONFIG:
-    SSL_CONFIG = "es.net.ssl"
-    NODES_WAN_ONLY = "es.nodes.wan.only"
-    NODES = "es.nodes"
-    SSL_KEYSTORE_LOCATION = "es.net.ssl.keystore.location"
-    SSL_KEYSTORE_PASSWORD = "es.net.ssl.keystore.pass"
-    SSL_TRUSTSTORE_LOCATION = "es.net.ssl.truststore.location"
-    SSL_TRUSTSTORE_PASSWORD = "es.net.ssl.truststore.pass"
-    HTTP_AUTHORIZATION = "es.net.http.header.Authorization"
-    INDEX = "es.resource"
-    HOSTS = "hosts"
-    HTTP_COMPRESS = "http_compress"
-    HEADERS = "headers"
-    USE_SSL = "use_ssl"
-    VERIFY_CERTS = "verify_certs"
-    SSL_ASSERT_HOSTNAME = "ssl_assert_hostname"
-    CA_CERTS = "ca_certs"
-
-
-class KAFKA_SSL_CONFIG:
-    """
-    Kafka SSL constant strings for configuration
-    """
-
-    SSL = "SSL"
-    SSL_TRUSTSTORE_LOCATION_CONFIG = "ssl.truststore.location"
-    SSL_TRUSTSTORE_LOCATION_DOC = "The location of the trust store file. "
-    SSL_TRUSTSTORE_PASSWORD_CONFIG = "ssl.truststore.password"
-    SSL_TRUSTSTORE_PASSWORD_DOC = "The password for the trust store file. If a password is not set access to the truststore is still available, but integrity checking is disabled."
-    SSL_KEYSTORE_LOCATION_CONFIG = "ssl.keystore.location"
-    SSL_KEYSTORE_PASSWORD_CONFIG = "ssl.keystore.password"
-    SSL_KEY_PASSWORD_CONFIG = "ssl.key.password"
-    SECURITY_PROTOCOL_CONFIG = "security.protocol"
-    SSL_CERTIFICATE_LOCATION_CONFIG = "ssl.certificate.location"
-    SSL_CA_LOCATION_CONFIG = "ssl.ca.location"
-    SSL_PRIVATE_KEY_LOCATION_CONFIG = "ssl.key.location"
-    SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG = (
-        "ssl.endpoint.identification.algorithm"
-    )
-
-
-class KAFKA_PRODUCER_CONFIG:
-    """
-    Constant strings for Kafka producers
-    """
-
-    BOOTSTRAP_SERVERS_CONFIG = "bootstrap.servers"
-    KEY_SERIALIZER_CLASS_CONFIG = "key.serializer"
-    VALUE_SERIALIZER_CLASS_CONFIG = "value.serializer"
-
-
-class KAFKA_CONSUMER_CONFIG:
-    """
-    Constant strings for Kafka consumers
-    """
-
-    GROUP_ID_CONFIG = "group.id"
-    CLIENT_ID_CONFIG = "client.id"
-    ENABLE_AUTO_COMMIT_CONFIG = "enable.auto.commit"
-    AUTO_COMMIT_INTERVAL_MS_CONFIG = "auto.commit.interval.ms"
-    SESSION_TIMEOUT_MS_CONFIG = "session.timeout.ms"
-    KEY_DESERIALIZER_CLASS_CONFIG = "key.deserializer"
-    VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer"
-    AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset"
-    ENABLE_AUTO_COMMIT_CONFIG = "enable.auto.commit"
-    KEY_DESERIALIZER_CLASS_CONFIG = "key.deserializer"
-    VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer"
-
-
-class ENV_VARS:
-    """
-    Constant strings for environment variables
-    """
-
-    KAFKA_BROKERS = "KAFKA_BROKERS"
-    ELASTIC_ENDPOINT_ENV_VAR = "ELASTIC_ENDPOINT"
-
-
-class SSL_CONFIG:
-    """
-    General SSL configuration constants for Hops-TLS
-    """
-
-    KEYSTORE_SUFFIX = "__kstore.jks"
-    TRUSTSTORE_SUFFIX = "__tstore.jks"
-    PASSWORD_SUFFIX = "__cert.key"
-
-    K_CERTIFICATE_CONFIG = "k_certificate"
-    T_CERTIFICATE_CONFIG = "t_certificate"
-    PEM_CLIENT_CERTIFICATE_CONFIG = "client.pem"
-    PEM_CLIENT_KEY_CONFIG = "client_key.pem"
-    PEM_CA_CHAIN_CERTIFICATE_CONFIG = "ca_chain.pem"
-    DOMAIN_CA_TRUSTSTORE = "domain_ca_truststore"
-    CRYPTO_MATERIAL_PASSWORD = "material_passwd"
-    PEM_CA_ROOT_CERT = "/srv/hops/kagent/host-certs/hops_root_ca.pem"
-    SSL_ENABLED = "ipc.server.ssl.enabled"
-
-
-class HOSTS:
-    APP_HOST = "c.app.hopsworks.ai"
+from hopsworks_common.constants import (
+    ARTIFACT_VERSION,
+    DEFAULT,
+    DEPLOYABLE_COMPONENT,
+    DEPLOYMENT,
+    ENV_VARS,
+    GIT,
+    HOSTS,
+    INFERENCE_BATCHER,
+    INFERENCE_ENDPOINTS,
+    INFERENCE_LOGGER,
+    JOBS,
+    KAFKA_CONSUMER_CONFIG,
+    KAFKA_PRODUCER_CONFIG,
+    KAFKA_SSL_CONFIG,
+    KAFKA_TOPIC,
+    MODEL,
+    MODEL_REGISTRY,
+    MODEL_SERVING,
+    OPENSEARCH_CONFIG,
+    PREDICTOR,
+    PREDICTOR_STATE,
+    RESOURCES,
+    SERVICES,
+    SSL_CONFIG,
+    Default,
+)
+
+
+__all__ = [
+    ARTIFACT_VERSION,
+    DEFAULT,
+    DEPLOYABLE_COMPONENT,
+    DEPLOYMENT,
+    ENV_VARS,
+    GIT,
+    HOSTS,
+    INFERENCE_BATCHER,
+    INFERENCE_ENDPOINTS,
+    INFERENCE_LOGGER,
+    JOBS,
+    KAFKA_CONSUMER_CONFIG,
+    KAFKA_PRODUCER_CONFIG,
+    KAFKA_SSL_CONFIG,
+    KAFKA_TOPIC,
+    MODEL,
+    MODEL_REGISTRY,
+    MODEL_SERVING,
+    OPENSEARCH_CONFIG,
+    PREDICTOR,
+    PREDICTOR_STATE,
+    RESOURCES,
+    SERVICES,
+    SSL_CONFIG,
+    Default,
+]

Diff for: python/hopsworks_common/client/__init__.py

@@ -56,6 +56,8 @@ def init(
                 api_key_file,
                 api_key_value,
             )
+    elif isinstance(_client, external.Client) and not _client._project_name:
+        _client._hsfs_post_init(project, engine, region_name)
 
 
 def get_instance() -> Union[hopsworks.Client, external.Client]:

Diff for: python/hopsworks_common/client/external.py

@@ -65,14 +65,6 @@ def __init__(
         self._port = port
         self._base_url = "https://" + self._host + ":" + str(self._port)
         _logger.info("Base URL: %s", self._base_url)
-        self._project_name = project
-        if project is not None:
-            project_info = self._get_project_info(project)
-            self._project_id = str(project_info["projectId"])
-            _logger.debug("Setting Project ID: %s", self._project_id)
-        else:
-            self._project_id = None
-        _logger.debug("Project name: %s", self._project_name)
         self._region_name = region_name or self.DEFAULT_REGION
         _logger.debug("Region name: %s", self._region_name)
 
@@ -99,6 +91,19 @@ def __init__(
         self._cert_folder_base = cert_folder
         self._cert_folder = None
 
+        self._hsfs_post_init(project, engine, region_name)
+
+    def _hsfs_post_init(self, project, engine, region_name):
+        self._region_name = region_name or self._region_name or self.DEFAULT_REGION
+        self._project_name = project
+        if project is not None:
+            project_info = self._get_project_info(project)
+            self._project_id = str(project_info["projectId"])
+            _logger.debug("Setting Project ID: %s", self._project_id)
+        else:
+            self._project_id = None
+        _logger.debug("Project name: %s", self._project_name)
+
         if project is None:
             return
 
@@ -151,7 +156,7 @@ def __init__(
                 _spark_session._jsc.hadoopConfiguration().set(conf_key, conf_value)
 
     def download_certs(self, project):
-        res = self._materialize_certs(self, project)
+        res = self._materialize_certs(project)
         self._write_pem_file(res["caChain"], self._get_ca_chain_path())
         self._write_pem_file(res["clientCert"], self._get_client_cert_path())
         self._write_pem_file(res["clientKey"], self._get_client_key_path())

Diff for: python/hopsworks_common/client/hopsworks.py

@@ -14,9 +14,7 @@
 #   limitations under the License.
 #
 
-import base64
 import os
-import textwrap
 from pathlib import Path
 
 import requests
@@ -85,7 +83,14 @@ def _get_trust_store_path(self):
         """Convert truststore from jks to pem and return the location"""
         ca_chain_path = Path(self.PEM_CA_CHAIN)
         if not ca_chain_path.exists():
-            self._write_ca_chain(ca_chain_path)
+            keystore_pw = self._cert_key
+            ks = jks.KeyStore.load(
+                self._get_jks_key_store_path(), keystore_pw, try_decrypt_keys=True
+            )
+            ts = jks.KeyStore.load(
+                self._get_jks_trust_store_path(), keystore_pw, try_decrypt_keys=True
+            )
+            self._write_ca_chain(ks, ts, ca_chain_path)
         return str(ca_chain_path)
 
     def _get_ca_chain_path(self, project_name=None) -> str:
@@ -97,64 +102,6 @@ def _get_client_cert_path(self, project_name=None) -> str:
     def _get_client_key_path(self, project_name=None) -> str:
         return os.path.join("/tmp", "client_key.pem")
 
-    def _write_ca_chain(self, ca_chain_path):
-        """
-        Converts JKS trustore file into PEM to be compatible with Python libraries
-        """
-        keystore_pw = self._cert_key
-        keystore_ca_cert = self._convert_jks_to_pem(
-            self._get_jks_key_store_path(), keystore_pw
-        )
-        truststore_ca_cert = self._convert_jks_to_pem(
-            self._get_jks_trust_store_path(), keystore_pw
-        )
-
-        with ca_chain_path.open("w") as f:
-            f.write(keystore_ca_cert + truststore_ca_cert)
-
-    def _convert_jks_to_pem(self, jks_path, keystore_pw):
-        """
-        Converts a keystore JKS that contains client private key,
-         client certificate and CA certificate that was used to
-         sign the certificate to PEM format and returns the CA certificate.
-        Args:
-        :jks_path: path to the JKS file
-        :pw: password for decrypting the JKS file
-        Returns:
-             strings: (ca_cert)
-        """
-        # load the keystore and decrypt it with password
-        ks = jks.KeyStore.load(jks_path, keystore_pw, try_decrypt_keys=True)
-        ca_certs = ""
-
-        # Convert CA Certificates into PEM format and append to string
-        for _alias, c in ks.certs.items():
-            ca_certs = ca_certs + self._bytes_to_pem_str(c.cert, "CERTIFICATE")
-        return ca_certs
-
-    def _bytes_to_pem_str(self, der_bytes, pem_type):
-        """
-        Utility function for creating PEM files
-
-        Args:
-            der_bytes: DER encoded bytes
-            pem_type: type of PEM, e.g Certificate, Private key, or RSA private key
-
-        Returns:
-            PEM String for a DER-encoded certificate or private key
-        """
-        pem_str = ""
-        pem_str = pem_str + "-----BEGIN {}-----".format(pem_type) + "\n"
-        pem_str = (
-            pem_str
-            + "\r\n".join(
-                textwrap.wrap(base64.b64encode(der_bytes).decode("ascii"), 64)
-            )
-            + "\n"
-        )
-        pem_str = pem_str + "-----END {}-----".format(pem_type) + "\n"
-        return pem_str
-
     def _get_jks_trust_store_path(self):
         """
         Get truststore location