fix: milvus setup

Fixed a bug where the milvus pod seems to be ready but api still not available.

Author: d4rkstar

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/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 = {