Merge pull request #23522 from mmaslankaprv/polaris-catalog

Polaris catalog in ducktape

tests/docker/Dockerfile

@@ -236,6 +236,12 @@ RUN /ocsf-server && rm /ocsf-server
 
 #################################
 
+FROM base AS polaris
+COPY --chown=0:0 --chmod=0755 tests/docker/ducktape-deps/polaris /
+RUN /polaris && rm /polaris
+
+#################################
+
 FROM librdkafka AS final
 
 COPY --chown=0:0 --chmod=0755 tests/docker/ducktape-deps/python-deps /
@@ -325,6 +331,7 @@ COPY --from=keycloak /opt/keycloak/ /opt/keycloak/
 COPY --from=wasi-transforms /opt/transforms/ /opt/transforms/
 COPY --from=ocsf /opt/ocsf-schema/ /opt/ocsf-schema/
 COPY --from=ocsf /opt/ocsf-server/ /opt/ocsf-server/
+COPY --from=polaris /opt/polaris/ /opt/polaris/
 COPY --from=flink /opt/flink/ /opt/flink/
 
 RUN ldconfig

tests/docker/ducktape-deps/java-dev-tools

@@ -4,6 +4,7 @@ apt update
 apt install -y \
   build-essential \
   openjdk-17-jdk \
+  openjdk-21-jdk \
   git \
   maven \
   cmake \
@@ -14,3 +15,18 @@ SCRIPTPATH="$(
   pwd -P
 )"
 $SCRIPTPATH/protobuf
+
+update-java-alternatives -s java-1.17.0-openjdk-amd64
+mkdir /opt/java
+
+cat <<EOF >/opt/java/java-21
+JAVA_HOME="/usr/lib/jvm/java-21-openjdk-amd64"
+/usr/lib/jvm/java-21-openjdk-amd64/bin/java "\$@"
+EOF
+chmod +x /opt/java/java-21
+
+cat <<EOF >/opt/java/java-17
+JAVA_HOME="/usr/lib/jvm/java-17-openjdk-amd64"
+/usr/lib/jvm/java-17-openjdk-amd64/bin/java "\$@"
+EOF
+chmod +x /opt/java/java-17

tests/docker/ducktape-deps/polaris

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+set -e
+update-java-alternatives -s java-1.21.0-openjdk-amd64
+git -C /opt clone https://github.com/apache/polaris.git
+cd /opt/polaris
+git reset --hard 1a6b3eb3963355f78c5ca916cc1d66ecd1493092
+./gradlew --no-daemon --info shadowJar
+update-java-alternatives -s java-1.17.0-openjdk-amd64

tests/rptest/services/polaris_catalog.py

@@ -0,0 +1,165 @@
+# Copyright 2020 Vectorized, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+
+import os
+import json
+import collections
+import re
+from typing import Optional, Any
+
+from ducktape.services.service import Service
+from ducktape.utils.util import wait_until
+from ducktape.cluster.cluster import ClusterNode
+
+from polaris.management.api_client import ApiClient
+from polaris.management.configuration import Configuration
+from polaris.management.api.polaris_default_api import PolarisDefaultApi
+
+
+class PolarisCatalog(Service):
+    """Polaris Catalog service
+    
+    The polaris catalog service maintain lifecycle of catalog process on the nodes. 
+    The service deploys polaris in a test mode with in-memory storage which is intended 
+    to be used for dev/test purposes.
+    """
+    PERSISTENT_ROOT = "/var/lib/polaris"
+    INSTALL_PATH = "/opt/polaris"
+    JAR = "polaris-service-1.0.0-all.jar"
+    JAR_PATH = os.path.join(INSTALL_PATH, "polaris-service/build/libs", JAR)
+    LOG_FILE = os.path.join(PERSISTENT_ROOT, "polaris.log")
+    POLARIS_CONFIG = os.path.join(PERSISTENT_ROOT, "polaris-server.yml")
+    logs = {
+        # Includes charts/ and results/ directories along with benchmark.log
+        "polaris_logs": {
+            "path": LOG_FILE,
+            "collect_default": True
+        },
+    }
+    # the only way to access polaris credentials running with the in-memory
+    # storage is to parse them from standard output
+    credentials_pattern = re.compile(
+        "realm: default-realm root principal credentials: (?P<client_id>.+):(?P<password>.+)"
+    )
+
+    nodes: list[ClusterNode]
+
+    def _cmd(self, node):
+        java = "/opt/java/java-21"
+        return f"{java} -jar {PolarisCatalog.JAR_PATH} server  {PolarisCatalog.POLARIS_CONFIG} \
+            1>> {PolarisCatalog.LOG_FILE} 2>> {PolarisCatalog.LOG_FILE} &"
+
+    def __init__(self, ctx, node: ClusterNode | None = None):
+        super(PolarisCatalog, self).__init__(ctx, num_nodes=0 if node else 1)
+
+        if node:
+            self.nodes = [node]
+        self._ctx = ctx
+        # catalog API url
+        self.catalog_url = None
+        # polaris management api url
+        self.management_url = None
+        self.client_id = None
+        self.password = None
+
+    def _parse_credentials(self, node):
+        line = node.account.ssh_output(
+            f"grep 'root principal credentials' {PolarisCatalog.LOG_FILE}"
+        ).decode('utf-8')
+        m = PolarisCatalog.credentials_pattern.match(line)
+        if m is None:
+            raise Exception(f"Unable to find credentials in line: {line}")
+        self.client_id = m['client_id']
+        self.password = m['password']
+
+    def start_node(self, node, timeout_sec=60, **kwargs):
+        node.account.ssh("mkdir -p %s" % PolarisCatalog.PERSISTENT_ROOT,
+                         allow_fail=False)
+        # polaris server settings
+        cfg_yaml = self.render("polaris-server.yml")
+        node.account.create_file(PolarisCatalog.POLARIS_CONFIG, cfg_yaml)
+        cmd = self._cmd(node)
+        self.logger.info(
+            f"Starting polaris catalog service on {node.name} with command {cmd}"
+        )
+        node.account.ssh(cmd, allow_fail=False)
+
+        # wait for the healthcheck to return 200
+        def _polaris_ready():
+            out = node.account.ssh_output(
+                "curl -s -o /dev/null -w '%{http_code}' http://localhost:8182/healthcheck"
+            )
+            status_code = int(out.decode('utf-8'))
+            self.logger.info(f"health check result status code: {status_code}")
+            return status_code == 200
+
+        wait_until(_polaris_ready,
+                   timeout_sec=timeout_sec,
+                   backoff_sec=0.4,
+                   err_msg="Error waiting for polaris catalog to start",
+                   retry_on_exc=True)
+
+        # setup urls and credentials
+        self.catalog_url = f"http://{node.account.hostname}:8181/api/catalog/v1"
+        self.management_url = f'http://{node.account.hostname}:8181/api/management/v1'
+        self._parse_credentials(node)
+        self.logger.info(
+            f"Polaris catalog ready, credentials - client_id: {self.client_id}, password: {self.password}"
+        )
+
+    def _get_token(self) -> str:
+        client = ApiClient(configuration=Configuration(host=self.catalog_url))
+        response = client.call_api('POST',
+                                   f'{self.catalog_url}/oauth/tokens',
+                                   header_params={
+                                       'Content-Type':
+                                       'application/x-www-form-urlencoded'
+                                   },
+                                   post_params={
+                                       'grant_type': 'client_credentials',
+                                       'client_id': self.client_id,
+                                       'client_secret': self.password,
+                                       'scope': 'PRINCIPAL_ROLE:ALL'
+                                   }).response.data
+
+        if 'access_token' not in json.loads(response):
+            raise Exception('Failed to get access token')
+        return json.loads(response)['access_token']
+
+    def management_client(self) -> ApiClient:
+        token = self._get_token()
+        return ApiClient(configuration=Configuration(host=self.management_url,
+                                                     access_token=token))
+
+    def catalog_client(self) -> ApiClient:
+        token = self._get_token()
+        return ApiClient(configuration=Configuration(host=self.catalog_url,
+                                                     access_token=token))
+
+    def wait_node(self, node, timeout_sec=None):
+        ## unused as there is nothing to wait for here
+        return False
+
+    def stop_node(self, node, allow_fail=False, **_):
+
+        node.account.kill_java_processes(PolarisCatalog.JAR,
+                                         allow_fail=allow_fail)
+
+        def _stopped():
+            out = node.account.ssh_output("jcmd").decode('utf-8')
+            return PolarisCatalog.JAR not in out
+
+        wait_until(_stopped,
+                   timeout_sec=10,
+                   backoff_sec=1,
+                   err_msg="Error stopping Polaris")
+
+    def clean_node(self, node, **_):
+        self.stop_node(node, allow_fail=True)
+        node.account.remove(PolarisCatalog.PERSISTENT_ROOT, allow_fail=True)

tests/rptest/services/templates/polaris-server.yml

@@ -0,0 +1,121 @@
+#
+# Copyright (c) 2024 Snowflake Computing Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+server:
+  # Maximum number of threads.
+  maxThreads: 200
+
+  # Minimum number of thread to keep alive.
+  minThreads: 10
+  applicationConnectors:
+    # HTTP-specific options.
+    - type: http
+
+      # The port on which the HTTP server listens for service requests.
+      port: 8181
+
+  adminConnectors:
+    - type: http
+      port: 8182
+
+  # The hostname of the interface to which the HTTP server socket wil be found. If omitted, the
+  # socket will listen on all interfaces.
+  #bindHost: localhost
+
+  #  ssl:
+  #    keyStore: ./example.keystore
+  #    keyStorePassword: example
+  #
+  #    keyStoreType: JKS # (optional, JKS is default)
+
+  # HTTP request log settings
+  requestLog:
+    appenders:
+      - type: console
+# Either 'jdbc' or 'polaris'; specifies the underlying delegate catalog
+baseCatalogType: "polaris"
+
+featureConfiguration:
+  ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING: false
+  DISABLE_TOKEN_GENERATION_FOR_USER_PRINCIPALS: true
+  SUPPORTED_CATALOG_STORAGE_TYPES:
+    - S3
+    - GCS
+    - AZURE
+    - FILE
+
+# Whether we want to enable Snowflake OAuth locally. Setting this to true requires
+# that you go through the setup outlined in the `README.md` file, specifically the
+# `OAuth + Snowflake: Local Testing And Then Some` section
+callContextResolver:
+  type: default
+
+realmContextResolver:
+  type: default
+
+defaultRealms:
+  - default-realm
+
+metaStoreManager:
+  type: in-memory
+
+# TODO - avoid duplicating token broker config
+oauth2:
+  type: test
+#  type: default # - uncomment to support Auth0 JWT tokens
+#  tokenBroker:
+#    type: symmetric-key
+#    secret: polaris
+
+authenticator:
+  class: io.polaris.service.auth.TestInlineBearerTokenPolarisAuthenticator
+  # class: io.polaris.service.auth.DefaultPolarisAuthenticator # - uncomment to support Auth0 JWT tokens
+  # tokenBroker:
+  #  type: symmetric-key
+  #  secret: polaris
+
+cors:
+  allowed-origins:
+    - http://localhost:8080
+  allowed-timing-origins:
+    - http://localhost:8080
+  allowed-methods:
+    - PATCH
+    - POST
+    - DELETE
+    - GET
+    - PUT
+  allowed-headers:
+    - "*"
+  exposed-headers:
+    - "*"
+  preflight-max-age: 600
+  allowed-credentials: true
+
+# Logging settings.
+
+logging:
+  # The default level of all loggers. Can be OFF, ERROR, WARN, INFO, DEBUG, TRACE, or ALL.
+  level: INFO
+
+  # Logger-specific levels.
+  loggers:
+    org.apache.iceberg.rest: DEBUG
+    io.polaris: DEBUG
+
+  appenders:
+    - type: console
+      threshold: ALL
+      logFormat: "%-5p [%d{ISO8601} - %-6r] [%t] [%X{aid}%X{sid}%X{tid}%X{wid}%X{oid}%X{srv}%X{job}%X{rid}] %c{30}: %m %kvp%n%ex"

tests/rptest/tests/polaris_catalog_smoke_test.py

@@ -0,0 +1,53 @@
+# Copyright 2020 Redpanda Data, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+
+from rptest.services.cluster import cluster
+
+from rptest.services.polaris_catalog import PolarisCatalog
+from rptest.tests.polaris_catalog_test import PolarisCatalogTest
+import polaris.catalog
+
+from polaris.management.api.polaris_default_api import PolarisDefaultApi
+from polaris.management.models.create_catalog_request import CreateCatalogRequest
+from polaris.management.models.catalog_properties import CatalogProperties
+from polaris.management.models.catalog import Catalog
+from polaris.management.models.storage_config_info import StorageConfigInfo
+
+
+class PolarisCatalogSmokeTest(PolarisCatalogTest):
+    def __init__(self, test_ctx, *args, **kwargs):
+        super(PolarisCatalogSmokeTest, self).__init__(test_ctx,
+                                                      num_brokers=1,
+                                                      *args,
+                                                      extra_rp_conf={},
+                                                      **kwargs)
+
+    """
+    Validates if the polaris catalog is accessible from ducktape tests harness
+    """
+
+    @cluster(num_nodes=2)
+    def test_creating_catalog(self):
+        """The very basic test checking interaction with polaris catalog
+        """
+        polaris_api = PolarisDefaultApi(self.polaris.management_client())
+        catalog = Catalog(
+            type="INTERNAL",
+            name="test-catalog",
+            properties=CatalogProperties(
+                default_base_location=
+                f"file://{PolarisCatalog.PERSISTENT_ROOT}/catalog_data",
+                additional_properties={}),
+            storageConfigInfo=StorageConfigInfo(storageType="FILE"))
+
+        polaris_api.create_catalog(CreateCatalogRequest(catalog=catalog))
+        resp = polaris_api.list_catalogs()
+
+        assert len(resp.catalogs) == 1
+        assert resp.catalogs[0].name == "test-catalog"