Merge pull request #43 from d4rkstar/main

more milvus tests

Author: Francesco Timperi Tiberi

nuvolaris/milvus_standalone.py

@@ -29,38 +29,52 @@
 from nuvolaris.user_config import UserConfig
 from nuvolaris.user_metadata import UserMetadata
 
+
 def patchEntries(data: dict):
-    tplp = ["milvus-cfg-base.yaml","milvus.yaml"]
+    tplp = ["milvus-cfg-base.yaml", "milvus.yaml"]
 
-    if(data['affinity'] or data['tolerations']):
+    if (data['affinity'] or data['tolerations']):
         tplp.append("affinity-tolerance-dep-core-attach.yaml")
 
     kust = kus.patchTemplates("milvus", tplp, data)
-    kust += kus.patchGenericEntry("Secret","nuvolaris-milvus-etcd-secret","/data/username",util.b64_encode(data['milvus_etcd_username']))
-    kust += kus.patchGenericEntry("Secret","nuvolaris-milvus-etcd-secret","/data/password",util.b64_encode(data['milvus_etcd_password']))
-
-    kust += kus.patchGenericEntry("Secret","nuvolaris-milvus-s3-secret","/stringData/accesskey",data['milvus_s3_username'])
-    kust += kus.patchGenericEntry("Secret","nuvolaris-milvus-s3-secret","/stringData/secretkey",data['milvus_s3_password'])
-
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus","/spec/storageClassName",data['storageClass'])
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus","/spec/resources/requests/storage",f"{data['size']}Gi")
-
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-zookeeper","/spec/storageClassName",data['storageClass'])
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-zookeeper","/spec/resources/requests/storage",f"{data['zookeeper_size']}Gi")        
+    kust += kus.patchGenericEntry("Secret", "nuvolaris-milvus-etcd-secret", "/data/username",
+                                  util.b64_encode(data['milvus_etcd_username']))
+    kust += kus.patchGenericEntry("Secret", "nuvolaris-milvus-etcd-secret", "/data/password",
+                                  util.b64_encode(data['milvus_etcd_password']))
+
+    kust += kus.patchGenericEntry("Secret", "nuvolaris-milvus-s3-secret", "/stringData/accesskey",
+                                  data['milvus_s3_username'])
+    kust += kus.patchGenericEntry("Secret", "nuvolaris-milvus-s3-secret", "/stringData/secretkey",
+                                  data['milvus_s3_password'])
+
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus", "/spec/storageClassName",
+                                  data['storageClass'])
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus", "/spec/resources/requests/storage",
+                                  f"{data['size']}Gi")
+
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-zookeeper", "/spec/storageClassName",
+                                  data['storageClass'])
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-zookeeper",
+                                  "/spec/resources/requests/storage", f"{data['zookeeper_size']}Gi")
+
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-bookie-journal", "/spec/storageClassName",
+                                  data['storageClass'])
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-bookie-journal",
+                                  "/spec/resources/requests/storage", f"{data['bookie_journal_size']}Gi")
+
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-bookie-ledgers", "/spec/storageClassName",
+                                  data['storageClass'])
+    kust += kus.patchGenericEntry("PersistentVolumeClaim", "nuvolaris-milvus-bookie-ledgers",
+                                  "/spec/resources/requests/storage", f"{data['bookie_ledgers_size']}Gi")
+    return kust
 
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-bookie-journal","/spec/storageClassName",data['storageClass'])
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-bookie-journal","/spec/resources/requests/storage",f"{data['bookie_journal_size']}Gi")
-
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-bookie-ledgers","/spec/storageClassName",data['storageClass'])
-    kust += kus.patchGenericEntry("PersistentVolumeClaim","nuvolaris-milvus-bookie-ledgers","/spec/resources/requests/storage",f"{data['bookie_ledgers_size']}Gi")
-    return kust 
 
 def create(owner=None):
     """
     Deploys the milvus vector db in standalone mode.
     """
-    data = util.get_milvus_config_data()  
-    res = create_milvus_accounts(data) 
+    data = util.get_milvus_config_data()
+    res = create_milvus_accounts(data)
 
     if res:
         logging.info("*** creating a milvus standalone instance")
@@ -72,153 +86,171 @@ def create(owner=None):
         else:
             cfg.put("state.milvus.spec", mspec)
 
-        res = kube.apply(mspec)
-        util.wait_for_pod_ready(r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')].metadata.name}")
-        res=create_default_milvus_database(data)
+        kube.apply(mspec)
+        util.wait_for_pod_ready(
+            r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')].metadata.name}")
+        res = create_default_milvus_database(data)
         logging.info("*** created a milvus standalone instance")
 
+
     return res
 
-def create_milvus_accounts(data:dict):
+
+def create_milvus_accounts(data: dict):
     """"
     Creates technical accounts for ETCD and MINIO
     """
-    try:        
+    try:
         # currently we use the ETCD root password, so we skip the ETCD user creation.
-        #res = util.check(etcd.create_etcd_user(data['milvus_etcd_username'],data['milvus_etcd_password'],data['milvus_etcd_prefix']),"create_etcd_milvus_user",True)
+        # res = util.check(etcd.create_etcd_user(data['milvus_etcd_username'],data['milvus_etcd_password'],data['milvus_etcd_prefix']),"create_etcd_milvus_user",True)
 
         minioClient = mutil.MinioClient()
         bucket_policy_names = []
         bucket_policy_names.append(f"{data['milvus_bucket_name']}/*")
 
-        res = util.check(minioClient.add_user(data["milvus_s3_username"], data["milvus_s3_password"]),"create_milvus_s3_user",True)
-        res = util.check(minioClient.make_bucket(data["milvus_bucket_name"]),"create_milvus_s3_bucket",res)
-        return util.check(minioClient.assign_rw_bucket_policy_to_user(data["milvus_s3_username"], bucket_policy_names),"assign_milvus_s3_bucket_policy",res)
+        res = util.check(minioClient.add_user(data["milvus_s3_username"], data["milvus_s3_password"]),
+                         "create_milvus_s3_user", True)
+        res = util.check(minioClient.make_bucket(data["milvus_bucket_name"]), "create_milvus_s3_bucket", res)
+        return util.check(minioClient.assign_rw_bucket_policy_to_user(data["milvus_s3_username"], bucket_policy_names),
+                          "assign_milvus_s3_bucket_policy", res)
     except Exception as ex:
-        logging.error("Could not create milvus ETCD and MINIO accounts",ex)
+        logging.error("Could not create milvus ETCD and MINIO accounts", ex)
         return False
-    
+
+
 def create_default_milvus_database(data):
     """
     Creates nuvolaris MILVUS custom resources
     """
-    logging.info("*** configuring MILVUS database for nuvolaris")   
+    logging.info("*** configuring MILVUS database for nuvolaris")
     adminClient = MilvusAdminClient()
-    res = adminClient.setup_user("nuvolaris", data["nuvolaris_password"],"nuvolaris")
-    
-    if(res):
+    res = adminClient.setup_user("nuvolaris", data["nuvolaris_password"], "nuvolaris")
+
+    if (res):
         _annotate_nuv_milvus_metadata(data)
         logging.info("*** configured MILVUS database for nuvolaris")
         return True
-    
+
     return False
 
+
 def _annotate_nuv_milvus_metadata(data):
     """
     annotate nuvolaris configmap with entries for MILVUS connectivity MILVUS_HOST, MILVUS_PORT, MILVUS_TOKEN, MILVUS_DB_NAME
     this is becasue MINIO
-    """ 
+    """
     try:
-        milvus_service =  util.get_service(r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')]}")
-        if(milvus_service):
+        milvus_service = util.get_service(
+            r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')]}")
+        if (milvus_service):
             milvus_host = f"{milvus_service['metadata']['name']}.{milvus_service['metadata']['namespace']}.svc.cluster.local"
             password = data["nuvolaris_password"]
-            
+
             openwhisk.annotate(f"milvus_host={milvus_host}")
             openwhisk.annotate(f"milvus_token=nuvolaris:{password}")
             openwhisk.annotate("milvus_db_name=nuvolaris")
 
             ports = list(milvus_service['spec']['ports'])
             for port in ports:
-                if(port['name']=='milvus'):
-                    openwhisk.annotate(f"milvus_port={port['port']}")                  
+                if (port['name'] == 'milvus'):
+                    openwhisk.annotate(f"milvus_port={port['port']}")
         return None
     except Exception as e:
         logging.error(f"failed to annotate MILVUS for nuvolaris: {e}")
         return None
 
-def _add_milvus_user_metadata(ucfg: UserConfig, user_metadata:UserMetadata):
+
+def _add_milvus_user_metadata(ucfg: UserConfig, user_metadata: UserMetadata):
     """
     adds entries for MILVUS connectivity MILVUS_HOST, MILVUS_PORT, MILVUS_TOKEN, MILVUS_DB_NAME    
-    """ 
+    """
 
     try:
-        milvus_service =  util.get_service(r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')]}")
-        
-        if(milvus_service):
+        milvus_service = util.get_service(
+            r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'nuvolaris-milvus')]}")
+
+        if (milvus_service):
             milvus_host = f"{milvus_service['metadata']['name']}.{milvus_service['metadata']['namespace']}.svc.cluster.local"
             milvus_token = f"{ucfg.get('namespace')}:{ucfg.get('milvus.password')}"
-            user_metadata.add_metadata("MILVUS_HOST",milvus_host)
-            user_metadata.add_metadata("MILVUS_TOKEN",milvus_token)
-            user_metadata.add_metadata("MILVUS_DB_NAME",ucfg.get('milvus.database'))
+            user_metadata.add_metadata("MILVUS_HOST", milvus_host)
+            user_metadata.add_metadata("MILVUS_TOKEN", milvus_token)
+            user_metadata.add_metadata("MILVUS_DB_NAME", ucfg.get('milvus.database'))
 
             ports = list(milvus_service['spec']['ports'])
             for port in ports:
-                if(port['name']=='milvus'):
-                    user_metadata.add_metadata("MILVUS_PORT",port['port'])
+                if (port['name'] == 'milvus'):
+                    user_metadata.add_metadata("MILVUS_PORT", port['port'])
 
         return None
     except Exception as e:
         logging.error(f"failed to build MILVUS metadata for {ucfg.get('namespace')}: {e}")
         return None
 
+
 def create_ow_milvus(ucfg: UserConfig, user_metadata: UserMetadata, owner=None):
     logging.info(f"*** configuring MILVUS database for {ucfg.get('namespace')}")
-    
+
     adminClient = MilvusAdminClient()
     username = ucfg.get("namespace")
     password = ucfg.get("milvus.password")
     database = ucfg.get("milvus.database")
-    res = adminClient.setup_user(username, password,database)
-    
-    if(res):
+    res = adminClient.setup_user(username, password, database)
+
+    if (res):
         _add_milvus_user_metadata(ucfg, user_metadata)
         logging.info(f"*** configured MILVUS database linked to namespace {ucfg.get('namespace')}")
 
+    return res
+
+
 def delete_ow_milvus(ucfg):
     logging.info(f"removing MILVUS database {ucfg.get('namespace')}")
     adminClient = MilvusAdminClient()
-    res = adminClient.remove_user(ucfg.get('namespace'),ucfg.get('milvus.database'))
+    res = adminClient.remove_user(ucfg.get('namespace'), ucfg.get('milvus.database'))
 
     if res:
-        logging.info(f"removed MILVUS database linked to namespace {ucfg.get('namespace')}")                     
+        logging.info(f"removed MILVUS database linked to namespace {ucfg.get('namespace')}")
+
+    return res
+
 
 def delete_by_owner():
     spec = kus.build("milvus")
     res = kube.delete(spec)
-    logging.info(f"delete milvus: {res}")    
+    logging.info(f"delete milvus: {res}")
     return res
 
+
 def delete_by_spec():
-    spec = cfg.get("state.milvus.spec")    
+    spec = cfg.get("state.milvus.spec")
     if spec:
         res = kube.delete(spec)
         logging.info(f"delete milvus: {res}")
         return res
 
-    
 
 def delete(owner=None):
-    if owner:        
+    if owner:
         return delete_by_owner()
     else:
         return delete_by_spec()
 
+
 def patch(status, action, owner=None):
     """
     Called by the operator patcher to create/delete milvus component
     """
     try:
-        logging.info(f"*** handling request to {action} milvus")  
-        if  action == 'create':
+        logging.info(f"*** handling request to {action} milvus")
+        if action == 'create':
             msg = create(owner)
-            operator_util.patch_operator_status(status,'milvus','on')            
+            operator_util.patch_operator_status(status, 'milvus', 'on')
         else:
-            msg = delete(owner)            
-            operator_util.patch_operator_status(status,'milvus','off')
+            msg = delete(owner)
+            operator_util.patch_operator_status(status, 'milvus', 'off')
 
-        logging.info(msg)        
-        logging.info(f"*** handled request to {action} milvus") 
+        logging.info(msg)
+        logging.info(f"*** handled request to {action} milvus")
     except Exception as e:
-        logging.error('*** failed to update milvus: %s' % e)        
-        operator_util.patch_operator_status(status,'milvus','error')
+        logging.error('*** failed to update milvus: %s' % e)
+        operator_util.patch_operator_status(status, 'milvus', 'error')

tests/kind/milvus_standalone_test.ipy

@@ -24,6 +24,9 @@ import nuvolaris.etcd as etcd
 import nuvolaris.milvus_standalone as milvus
 import nuvolaris.minio_deploy as minio
 import nuvolaris.testutil as tu
+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()
 # test
@@ -37,6 +40,50 @@ os.environ['MILVUS_API_HOST']='localhost'
 assert(etcd.create())
 assert(minio.create())
 assert(milvus.create())
+
+# generate some user config
+ucfg = UserConfig({
+    "namespace": "demouser",
+    "password": "s0meComplicat3d",
+    "email": "[email protected]",
+    "milvus": {
+        "password": "s0meComplicat3d",
+        "database": "demouser"
+    }
+})
+# generate some user metadata
+metadata = UserMetadata(ucfg)
+
+# test user creation
+assert(milvus.create_ow_milvus(ucfg, metadata))
+
+# load user metadata
+user_metadata = metadata.get_metadata()
+body = {}
+envs = list(user_metadata['metadata'])
+
+for env in envs:
+    body[env['key']]=env['value']
+
+# test user metadata
+host = body["MILVUS_HOST"]
+token = body["MILVUS_TOKEN"]
+database = body["MILVUS_DB_NAME"]
+port = body["MILVUS_PORT"]
+assert host
+assert token
+assert database
+assert port
+
+# test client
+client = MilvusClient(uri=f'http://localhost:{port}',token=token, db_name=database)
+client.create_collection("testcollection", dimension=128)
+assert("testcollection" in client.list_collections())
+client.drop_collection("testcollection")
+
+# test user deletion
+assert(milvus.delete_ow_milvus(ucfg))
+
 assert(milvus.delete())
 assert(etcd.delete())
 assert(minio.delete())