Pluggable API/Features Limits

.github/workflows/airflow-apis-tests-3_9.yml

@@ -100,7 +100,7 @@ jobs:
     - name: Start Server and Ingest Sample Data
       uses: nick-fields/retry@v2.8.3
       env:
-        INGESTION_DEPENDENCY: "mysql,elasticsearch"
+        INGESTION_DEPENDENCY: "mysql,elasticsearch,sample-data"
       with:
         timeout_minutes: 60
         max_attempts: 2

common/pom.xml

@@ -18,7 +18,7 @@
   <parent>
     <artifactId>platform</artifactId>
     <groupId>org.open-metadata</groupId>
-    <version>1.5.0-SNAPSHOT</version>
+    <version>1.4.3</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <properties>

conf/openmetadata.yaml

@@ -370,6 +370,11 @@ email:
   password: ${SMTP_SERVER_PWD:-""}
   transportationStrategy: ${SMTP_SERVER_STRATEGY:-"SMTP_TLS"}
 
+limits:
+  enable: ${LIMITS_ENABLED:-false}
+  className: ${LIMITS_CLASS_NAME:-"org.openmetadata.service.limits.DefaultLimits"}
+  limitsConfigFile: ${LIMITS_CONFIG_FILE:-""}
+
 web:
   uriPath: ${WEB_CONF_URI_PATH:-"/api"}
   hsts:

docker/docker-compose-ingestion/docker-compose-ingestion.yml

@@ -18,7 +18,7 @@ volumes:
 services:  
   ingestion:
     container_name: openmetadata_ingestion
-    image: docker.getcollate.io/openmetadata/ingestion:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/ingestion:1.4.3
     environment:
       AIRFLOW__API__AUTH_BACKENDS: "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
       AIRFLOW__CORE__EXECUTOR: LocalExecutor

docker/docker-compose-openmetadata/docker-compose-openmetadata.yml

@@ -14,7 +14,7 @@ services:
   execute-migrate-all:
     container_name: execute_migrate_all
     command: "./bootstrap/openmetadata-ops.sh migrate"
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
       SERVER_PORT: ${SERVER_PORT:-8585}
@@ -227,7 +227,7 @@ services:
   openmetadata-server:
     container_name: openmetadata_server
     restart: always
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
       SERVER_PORT: ${SERVER_PORT:-8585}

docker/docker-compose-quickstart/Dockerfile

@@ -11,7 +11,7 @@
 
 # Build stage
 FROM alpine:3.19 AS build
-ARG RI_VERSION="1.5.0-SNAPSHOT"
+ARG RI_VERSION="1.4.3"
 ENV RELEASE_URL="https://github.com/open-metadata/OpenMetadata/releases/download/${RI_VERSION}-release/openmetadata-${RI_VERSION}.tar.gz"
 
 RUN mkdir -p /opt/openmetadata && \
@@ -21,7 +21,7 @@ RUN mkdir -p /opt/openmetadata && \
 
 # Final stage
 FROM alpine:3.19
-ARG RI_VERSION="1.5.0-SNAPSHOT"
+ARG RI_VERSION="1.4.3"
 ARG BUILD_DATE
 ARG COMMIT_ID
 LABEL maintainer="OpenMetadata"

docker/docker-compose-quickstart/docker-compose-postgres.yml

@@ -18,7 +18,7 @@ volumes:
 services:
   postgresql:
     container_name: openmetadata_postgresql
-    image: docker.getcollate.io/openmetadata/postgresql:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/postgresql:1.4.3
     restart: always
     command: "--work_mem=10MB"
     environment:
@@ -61,7 +61,7 @@ services:
 
   execute-migrate-all:
     container_name: execute_migrate_all
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     command: "./bootstrap/openmetadata-ops.sh migrate"
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
@@ -275,7 +275,7 @@ services:
   openmetadata-server:
     container_name: openmetadata_server
     restart: always
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
       SERVER_PORT: ${SERVER_PORT:-8585}
@@ -483,7 +483,7 @@ services:
 
   ingestion:
     container_name: openmetadata_ingestion
-    image: docker.getcollate.io/openmetadata/ingestion:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/ingestion:1.4.3
     depends_on:
       elasticsearch:
         condition: service_started

docker/docker-compose-quickstart/docker-compose.yml

@@ -18,7 +18,7 @@ volumes:
 services:
   mysql:
     container_name: openmetadata_mysql
-    image: docker.getcollate.io/openmetadata/db:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/db:1.4.3
     command: "--sort_buffer_size=10M"
     restart: always
     environment:
@@ -59,7 +59,7 @@ services:
 
   execute-migrate-all:
     container_name: execute_migrate_all
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     command: "./bootstrap/openmetadata-ops.sh migrate"
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
@@ -273,7 +273,7 @@ services:
   openmetadata-server:
     container_name: openmetadata_server
     restart: always
-    image: docker.getcollate.io/openmetadata/server:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/server:1.4.3
     environment:
       OPENMETADATA_CLUSTER_NAME: ${OPENMETADATA_CLUSTER_NAME:-openmetadata}
       SERVER_PORT: ${SERVER_PORT:-8585}
@@ -482,7 +482,7 @@ services:
 
   ingestion:
     container_name: openmetadata_ingestion
-    image: docker.getcollate.io/openmetadata/ingestion:1.5.0-SNAPSHOT
+    image: docker.getcollate.io/openmetadata/ingestion:1.4.3
     depends_on:
       elasticsearch:
         condition: service_started

docker/mysql/mysql-script.sql

@@ -5,5 +5,6 @@ CREATE USER 'airflow_user'@'%' IDENTIFIED BY 'airflow_pass';
 GRANT ALL PRIVILEGES ON openmetadata_db.* TO 'openmetadata_user'@'%' WITH GRANT OPTION;
 GRANT PROCESS, USAGE ON *.* TO 'openmetadata_user'@'%';
 GRANT ALL PRIVILEGES ON airflow_db.* TO 'airflow_user'@'%' WITH GRANT OPTION;
+GRANT PROCESS, USAGE ON *.* TO 'openmetadata_user'@'%';
 FLUSH PRIVILEGES;
 commit;

ingestion/Dockerfile

@@ -68,7 +68,7 @@ ARG INGESTION_DEPENDENCY="all"
 ENV PIP_NO_CACHE_DIR=1
 # Make pip silent
 ENV PIP_QUIET=1
-ARG RI_VERSION="1.5.0.0.dev0"
+ARG RI_VERSION="1.4.3.0"
 RUN pip install --upgrade pip
 RUN pip install "openmetadata-managed-apis~=${RI_VERSION}" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.7.3/constraints-3.10.txt"
 RUN pip install "openmetadata-ingestion[${INGESTION_DEPENDENCY}]~=${RI_VERSION}"

ingestion/operators/docker/Dockerfile

@@ -73,7 +73,7 @@ ENV PIP_QUIET=1
 RUN pip install --upgrade pip
 
 ARG INGESTION_DEPENDENCY="all"
-ARG RI_VERSION="1.5.0.0.dev0"
+ARG RI_VERSION="1.4.3.0"
 RUN pip install --upgrade pip
 RUN pip install "openmetadata-ingestion[airflow]~=${RI_VERSION}"
 RUN pip install "openmetadata-ingestion[${INGESTION_DEPENDENCY}]~=${RI_VERSION}"

ingestion/pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 # since it helps us organize and isolate version management
 [project]
 name = "openmetadata-ingestion"
-version = "1.5.0.0.dev0"
+version = "1.4.3.0"
 dynamic = ["readme", "dependencies", "optional-dependencies"]
 authors = [
     {name = "OpenMetadata Committers"}

ingestion/src/metadata/data_quality/validations/runtime_param_setter/table_diff_params_setter.py

@@ -131,21 +131,19 @@ def get_parameter(test_case: TestCase, key: str, default=None):
     @staticmethod
     def get_data_diff_url(service_url: str, table_fqn) -> str:
         url = urlparse(service_url)
-        # remove the drivername from the url becuase table-diff doesn't support it
+        # remove the driver name from the url because table-diff doesn't support it
         kwargs = {"scheme": url.scheme.split("+")[0]}
-        service, database, schema, table = fqn.split(  # pylint: disable=unused-variable
-            table_fqn
-        )
+        # pylint: disable=unbalanced-tuple-unpacking
+        _, database, schema, _ = fqn.split(table_fqn)
         # path needs to include the database AND schema in some of the connectors
         if kwargs["scheme"] in ["mssql"]:
             kwargs["path"] = f"/{database}/{schema}"
         return url._replace(**kwargs).geturl()
 
     @staticmethod
     def get_data_diff_table_path(table_fqn: str):
-        service, database, schema, table = fqn.split(  # pylint: disable=unused-variable
-            table_fqn
-        )
+        # pylint: disable=unbalanced-tuple-unpacking
+        _, _, schema, table = fqn.split(table_fqn)
         return fqn._build(  # pylint: disable=protected-access
             "___SERVICE___", "__DATABASE__", schema, table
         ).replace("___SERVICE___.__DATABASE__.", "")

ingestion/src/metadata/ingestion/ometa/client.py

@@ -21,6 +21,7 @@
 
 from metadata.config.common import ConfigModel
 from metadata.ingestion.ometa.credentials import URL, get_api_version
+from metadata.ingestion.ometa.ttl_cache import TTLCache
 from metadata.utils.execution_time_tracker import calculate_execution_time
 from metadata.utils.logger import ometa_logger
 
@@ -33,6 +34,12 @@ class RetryException(Exception):
     """
 
 
+class LimitsException(Exception):
+    """
+    API Client Feature Limit exception
+    """
+
+
 class APIError(Exception):
     """
     Represent API related error.
@@ -97,7 +104,8 @@ class ClientConfig(ConfigModel):
     api_version: Optional[str] = "v1"
     retry: Optional[int] = 3
     retry_wait: Optional[int] = 30
-    retry_codes: List[int] = [429, 504]
+    limit_codes: List[int] = [429]
+    retry_codes: List[int] = [504]
     auth_token: Optional[Callable] = None
     access_token: Optional[str] = None
     expires_in: Optional[int] = None
@@ -107,8 +115,10 @@ class ClientConfig(ConfigModel):
     allow_redirects: Optional[bool] = False
     auth_token_mode: Optional[str] = "Bearer"
     verify: Optional[Union[bool, str]] = None
+    ttl_cache: int = 60
 
 
+# pylint: disable=too-many-instance-attributes
 class REST:
     """
     REST client wrapper to manage requests with
@@ -124,10 +134,13 @@ def __init__(self, config: ClientConfig):
         self._retry = self.config.retry
         self._retry_wait = self.config.retry_wait
         self._retry_codes = self.config.retry_codes
+        self._limit_codes = self.config.limit_codes
         self._auth_token = self.config.auth_token
         self._auth_token_mode = self.config.auth_token_mode
         self._verify = self.config.verify
 
+        self._limits_reached = TTLCache(config.ttl_cache)
+
     def _request(  # pylint: disable=too-many-arguments
         self,
         method,
@@ -139,6 +152,9 @@ def _request(  # pylint: disable=too-many-arguments
         headers: dict = None,
     ):
         # pylint: disable=too-many-locals
+        if path in self._limits_reached:
+            raise LimitsException(f"Skipping request - limits reached for {path}")
+
         if not headers:
             headers = {"Content-type": "application/json"}
         base_url = base_url or self._base_url
@@ -194,6 +210,10 @@ def _request(  # pylint: disable=too-many-arguments
         while retry >= 0:
             try:
                 return self._one_request(method, url, opts, retry)
+            except LimitsException as exc:
+                logger.error(f"Feature limit exceeded for {url}")
+                self._limits_reached.add(path)
+                raise exc
             except RetryException:
                 retry_wait = self._retry_wait * (total_retries - retry + 1)
                 logger.warning(
@@ -217,6 +237,7 @@ def _one_request(self, method: str, url: URL, opts: dict, retry: int):
         Returns the body json in the 200 status.
         """
         retry_codes = self._retry_codes
+        limit_codes = self._limit_codes
         try:
             resp = self._session.request(method, url, **opts)
             resp.raise_for_status()
@@ -234,6 +255,8 @@ def _one_request(self, method: str, url: URL, opts: dict, retry: int):
             # retry if we hit Rate Limit
             if resp.status_code in retry_codes and retry > 0:
                 raise RetryException() from http_error
+            if resp.status_code in limit_codes:
+                raise LimitsException() from http_error
             if "code" in resp.text:
                 error = resp.json()
                 if "code" in error:

ingestion/src/metadata/ingestion/ometa/ttl_cache.py

@@ -0,0 +1,47 @@
+#  Copyright 2021 Collate
+#  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.
+
+"""
+Simple dictionary implementation for keys with TTL
+"""
+from datetime import datetime
+from typing import Dict
+
+
+class TTLCache:
+    """
+    class to handle ttl cache
+    """
+
+    def __init__(self, ttl: int):
+        self._ttl = ttl
+        # The key will be the object, and the value the created time to check the TTL
+        self._cache: Dict[str, int] = {}
+
+    @staticmethod
+    def _now() -> int:
+        return int(datetime.now().timestamp())
+
+    def __contains__(self, item) -> bool:
+        if item in self._cache:
+            created_at = self._cache[item]
+            if self._now() - created_at > self._ttl:
+                self.delete(item)
+                return False
+            return True
+        return False
+
+    def add(self, value: str):
+        if value not in self._cache:
+            self._cache[value] = self._now()
+
+    def delete(self, key):
+        self._cache.pop(key, None)