Merge pull request #54 from d4rkstar/main

Etcd parameters and milvus setup

Author: sciabarracom

deploy/nuvolaris-permissions/whisk-crd.yaml

@@ -702,6 +702,12 @@ spec:
                     replicas:
                       description: number of total postgres replicas (1 primary, N-1 replicas). Defaulted to 1
                       type: integer
+                    auto-compaction-retention:
+                      description: defines how long to retain historical versions of keys before automatically compacting them
+                      type: string
+                    quota-backend-bytes:
+                      description: the etcd backend database has a default size limit of 2 GiB. This can be increased explicitly
+                      type: integer
                     root:
                       description: ETCD root user credentials
                       type: object

nuvolaris/milvus_admin_client.py

@@ -18,13 +18,12 @@
 import logging
 
 import nuvolaris.config as cfg
-# from pymilvus import MilvusClient, connections,  db
 from nuvolaris.milvus_simple_client import MilvusSimpleClient as MilvusClient
 
 
 class MilvusAdminClient:
     """
-    Simple Milvus Client used to perform Mivlus administration Tasks
+    Simple Milvus Client used to perform Milvus administration Tasks
     """
     def __init__(self, db_name="default"):
         self.admin_username   = cfg.get("milvus.admin.user", "MILVUS_ROOT_USER", "root")

nuvolaris/milvus_standalone.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+import time
 
 import kopf, logging
 import nuvolaris.kube as kube
@@ -89,6 +90,13 @@ def create(owner=None):
         kube.apply(mspec)
         util.wait_for_pod_ready(
             r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')].metadata.name}")
+
+        milvus_api_host = cfg.get("milvus.host", "MILVUS_API_HOST", "nuvolaris-milvus")
+        milvus_api_port = cfg.get("milvus.host", "MILVUS_API_PORT", "19530")
+
+        logging.info("*** waiting for milvus api to be available")
+        util.wait_for_http(f"http://{milvus_api_host}:{milvus_api_port}", up_statuses=[200,401], timeout=30)
+
         res = create_default_milvus_database(data)
         logging.info("*** created a milvus standalone instance")
 

nuvolaris/templates/etcd-sts.yaml

@@ -40,6 +40,10 @@ spec:
         env:
         - name: ETCD_INITIAL_CLUSTER
           value: {{etc_initial_cluster}}
+        - name: ETCD_AUTO_COMPACTION_RETENTION
+          value: "{{etcd_auto_compaction_retention}}"
+        - name: ETCD_QUOTA_BACKEND_BYTES
+          value: "{{etcd_quota_backend_bytes}}"
         image: docker.io/bitnami/etcd:3.5.17-debian-12-r0
         imagePullPolicy: IfNotPresent
   volumeClaimTemplates:

nuvolaris/util.py

@@ -16,13 +16,22 @@
 # under the License.
 #
 # this module wraps utilities functions
-import nuvolaris.kube as kube
 import logging
-import time, random, math, os
-import nuvolaris.config as cfg
+import math
+import random
+import time
 import uuid
-import nuvolaris.apihost_util as apihost_util
 from base64 import b64decode, b64encode
+from typing import List, Union
+from urllib.parse import urlparse
+
+import urllib3
+from urllib3.exceptions import NewConnectionError, MaxRetryError, ProtocolError
+
+import nuvolaris.apihost_util as apihost_util
+import nuvolaris.config as cfg
+import nuvolaris.kube as kube
+
 
 # Implements truncated exponential backoff from
 # https://cloud.google.com/storage/docs/retry-strategy#exponential-backoff
@@ -190,6 +199,74 @@ def wait_for_pod_ready(pod_name_jsonpath, timeout="600s", namespace="nuvolaris")
     except Exception as e:
         logging.error(e)
 
+
+def status_matches(code: int, allowed: List[Union[int, str]]) -> bool:
+    """Check if the status code matches any allowed pattern."""
+    for pattern in allowed:
+        if isinstance(pattern, int) and code == pattern:
+            return True
+        if isinstance(pattern, str) and len(pattern) == 3 and pattern.endswith("XX"):
+            if int(pattern[0]) == code // 100:
+                return True
+    return False
+
+def status_matches(code: int, allowed: List[Union[int, str]]) -> bool:
+    """Check if the status code matches any allowed pattern."""
+    for pattern in allowed:
+        if isinstance(pattern, int) and code == pattern:
+            return True
+        if isinstance(pattern, str) and len(pattern) == 3 and pattern.endswith("XX"):
+            if int(pattern[0]) == code // 100:
+                return True
+    return False
+
+def wait_for_http(url: str, timeout: int = 60, up_statuses: List[Union[int, str]] = [200]):
+    """Wait until an HTTP endpoint becomes available with an accepted status code.
+
+    Args:
+        url (str): Full URL to check (e.g. http://milvus:9091/healthz)
+        timeout (int): Total seconds to wait before giving up.
+        up_statuses (List[Union[int, str]]): Status codes or patterns considered as 'UP'.
+
+    Raises:
+        TimeoutError: If the endpoint doesn't respond with a valid status within the timeout.
+    """
+    parsed = urlparse(url)
+    scheme = parsed.scheme
+    host = parsed.hostname
+    port = parsed.port or (443 if scheme == "https" else 80)
+    path = parsed.path or "/"
+
+    if scheme == "https":
+        conn = urllib3.connectionpool.HTTPSConnectionPool(host, port=port,
+                                                          timeout=urllib3.util.Timeout(connect=5.0, read=5.0),
+                                                          retries=False)
+    else:
+        conn = urllib3.connectionpool.HTTPConnectionPool(host, port=port,
+                                                         timeout=urllib3.util.Timeout(connect=5.0, read=5.0),
+                                                         retries=False)
+
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        try:
+            response = conn.request("GET", path)
+            if status_matches(response.status, up_statuses):
+                logging.info(f"Service is up: {url} (status {response.status})")
+                return
+            else:
+                logging.warning(f"Service responded with {response.status}, not in {up_statuses}. Waiting...")
+        except (NewConnectionError, MaxRetryError):
+            logging.warning(f"Cannot connect to {url}, retrying...")
+        except ProtocolError as e:
+            if "Connection reset by peer" in str(e):
+                logging.warning("Connection reset by peer. Sleeping 2 seconds...")
+                time.sleep(2)
+                continue
+            else:
+                logging.error(f"Protocol error: {e}")
+        time.sleep(1)
+
 # return mongodb configuration parameter with default valued if not configured
 def get_mongodb_config_data():
     data = {
@@ -684,6 +761,8 @@ def get_etcd_config_data():
         "storageClass": cfg.get("nuvolaris.storageclass"),
         "root_password":cfg.get("etcd.root.password") or "s0meP@ass3wd",
         "etcd_replicas":get_etcd_replica(),
+        "etcd_auto_compaction_retention": cfg.get("etcd.auto_compaction_retention") or "1",
+        "etcd_quota_backend_bytes": cfg.get("etcd.quota-backend-bytes") or "2147483648",
         "namespace":"nuvolaris",
         "container_cpu_req": cfg.get('etcd.resources.cpu-req') or "250m",
         "container_cpu_lim": cfg.get('etcd.resources.cpu-lim') or "375m",

tests/kind/milvus_standalone_test.ipy

@@ -28,7 +28,12 @@ from nuvolaris.milvus_simple_client import MilvusSimpleClient as MilvusClient
 from nuvolaris.user_config import UserConfig
 from nuvolaris.user_metadata import UserMetadata
 
-# tu.enable_debug_logging()
+
+#tu.enable_debug_logging()
+### Initial cleanup
+!kubectl -n nuvolaris delete all --all
+!kubectl -n nuvolaris delete pvc --all
+
 # test
 assert(cfg.configure(tu.load_sample_config()))
 assert(cfg.detect_storage()["nuvolaris.storageclass"])
@@ -87,3 +92,8 @@ assert(milvus.delete_ow_milvus(ucfg))
 assert(milvus.delete())
 assert(etcd.delete())
 assert(minio.delete())
+
+# final cleanup
+!kubectl -n nuvolaris delete all --all
+# etcd pvc in particular
+!kubectl -n nuvolaris delete pvc --all

tests/kind/whisk-full.yaml

@@ -73,6 +73,13 @@ spec:
     invoker:
       user: controller_admin
       password: s0meP@ass2
+  etcd:
+    volume-size: 5
+    replicas: 3
+    root:
+      password: 0therPa55
+    auto-compaction-retention: 1
+    quota-backend-bytes: 8589934592
   kafka:
     host: kafka
     volume-size: 5