Add capacity values for the DMS K8S profile

Get the attribute "capacity" value from the following Kubernetes command
kubectl describe node controller-0

This commit will add the value for the capacity of the DMS K8S profile.

Test Case:
The watcher service can store the correct value in the database.

Issue-ID: INF-477

Change-Id: Iec0e7e49e96848a5bfc57757372f69377b505717
Signed-off-by: Zhang Rong(Jon) <[email protected]>

o2ims/adapter/clients/ocloud_client.py

@@ -14,6 +14,7 @@
 
 # client talking to Stx standalone
 
+import base64
 import uuid
 import json
 from typing import List
@@ -22,6 +23,7 @@
 from cgtsclient.client import get_client as get_stx_client
 from cgtsclient.exc import EndpointException
 from dcmanagerclient.api.client import client as get_dc_client
+from kubernetes import client as k8sclient, config as k8sconfig
 
 from o2common.config import config
 from o2common.service.client.base_client import BaseClient
@@ -249,6 +251,24 @@ def getSubcloudClient(self, subcloud_id):
 
         return config_client
 
+    def getK8sClient(self, k8scluster):
+        def _b64_encode_str(msg: str, encode: str = 'utf-8') -> str:
+            msg_bytes = msg.encode('utf-8')
+            base64_bytes = base64.b64encode(msg_bytes)
+            base64_msg = base64_bytes.decode('utf-8')
+            return base64_msg
+
+        conf_dict = config.gen_k8s_config_dict(
+            k8scluster.cluster_api_endpoint,
+            _b64_encode_str(k8scluster.cluster_ca_cert),
+            k8scluster.admin_user,
+            _b64_encode_str(k8scluster.admin_client_cert),
+            _b64_encode_str(k8scluster.admin_client_key),
+        )
+        k8sconfig.load_kube_config_from_dict(conf_dict)
+        v1 = k8sclient.CoreV1Api()
+        return v1
+
     def setStxClient(self, resource_pool_id):
         systems = self.stxclient.isystem.list()
         if resource_pool_id == systems[0].uuid:
@@ -365,15 +385,59 @@ def _checkLabelExistOnCluster(self, client, label_to_check,
                 return True
         return False
 
-    def _setK8sCapabilities(self, client, k8scluster):
+    def _getK8sNodes(self, k8sclient):
+        return k8sclient.list_node()
+
+    def _getK8sNodeDetail(self, k8sclient, node_name):
+        return k8sclient.read_node(name=node_name)
+
+    def _getK8sCapabilities(self, k8s_client):
+        k8s_capabilities = {}
+        nodes = self._getK8sNodes(k8s_client)
+        for node in nodes.items:
+            logger.debug(f'k8s node {node.metadata.name} allocatable: '
+                         f'{node.status.allocatable}')
+            for allocatable in node.status.allocatable:
+                if allocatable.startswith('intel.com/pci_sriov_net_'):
+                    k8s_capabilities[f'{node.metadata.name}_sriov'] = True
+                if allocatable == 'windriver.com/isolcpus':
+                    k8s_capabilities[f'{node.metadata.name}_isolcpus'] = True
+        return k8s_capabilities
+
+    def _setK8sCapabilities(self, k8scluster, client, k8s_client):
         capabilities = {}
         label_OS_2chk = {'key': 'OS', 'value': 'low_latency'}
         if self._checkLabelExistOnCluster(client, label_OS_2chk, True):
             logger.debug("low latency host inside of the k8s cluster")
             capabilities[label_OS_2chk['key']] = label_OS_2chk['value']
+
+        # Add Kubernetes capabilities
+        k8s_capabilities = self._getK8sCapabilities(k8s_client)
+        capabilities.update(k8s_capabilities)
+
         setattr(k8scluster, 'capabilities', json.dumps(capabilities))
         return k8scluster
 
+    def _getK8sCapacity(self, k8s_client):
+        k8s_capacity = {}
+        nodes = self._getK8sNodes(k8s_client)
+        for node in nodes.items:
+            logger.debug(f'k8s node {node.metadata.name} capacity: '
+                         f'{node.status.capacity}')
+            for key, value in node.status.capacity.items():
+                k8s_capacity[f'{node.metadata.name}_{key}'] = value
+        return k8s_capacity
+
+    def _setK8sCapacity(self, k8scluster, client, k8s_client):
+        capacity = {}
+
+        # Add Kubernetes capacity
+        k8s_capacity = self._getK8sCapacity(k8s_client)
+        capacity.update(k8s_capacity)
+
+        setattr(k8scluster, 'capacity', json.dumps(capacity))
+        return k8scluster
+
     def getLabelList(self, **filters) -> List[ocloudModel.StxGenericModel]:
         hostid = filters.get('hostid', None)
         assert (hostid is not None), 'missing hostid to query label list'
@@ -386,7 +450,11 @@ def getK8sList(self, **filters) -> List[ocloudModel.StxGenericModel]:
         def process_cluster(client, cluster):
             setattr(cluster, 'cloud_name', systems[0].name)
             setattr(cluster, 'cloud_uuid', systems[0].uuid)
-            cluster = self._setK8sCapabilities(client, cluster)
+
+            k8s_client = self.getK8sClient(cluster)
+            cluster = self._setK8sCapabilities(cluster, client, k8s_client)
+            cluster = self._setK8sCapacity(cluster, client, k8s_client)
+
             logger.debug('k8sresources cluster_api_endpoint: ' +
                          str(cluster.cluster_api_endpoint))
             return ocloudModel.StxGenericModel(ResourceTypeEnum.DMS,
@@ -436,7 +504,10 @@ def getK8sDetail(self, name) -> ocloudModel.StxGenericModel:
         def process_k8s_cluster(client, k8s_cluster, cloud_name, cloud_uuid):
             setattr(k8s_cluster, 'cloud_name', cloud_name)
             setattr(k8s_cluster, 'cloud_uuid', cloud_uuid)
-            k8s_cluster = self._setK8sCapabilities(client, k8s_cluster)
+
+            k8s_client = self.getK8sClient(k8s_cluster)
+            cluster = self._setK8sCapabilities(k8s_cluster, client, k8s_client)
+            cluster = self._setK8sCapacity(cluster, client, k8s_client)
             return k8s_cluster
 
         systems = self.stxclient.isystem.list()

o2ims/adapter/orm.py

@@ -135,8 +135,8 @@
     Column("description", String(255)),
     Column("serviceUri", String(255)),
     Column("supportedLocations", String(255)),
-    Column("capabilities", String(255)),
-    Column("capacity", String(255)),
+    Column("capabilities", Text),
+    Column("capacity", Text),
     Column("profile", Text())
     # Column("extensions", String(1024))
 )

o2ims/service/auditor/dms_handler.py

@@ -82,7 +82,7 @@ def create_by(stxobj: StxGenericModel, parentid: str) -> DeploymentManager:
     # logger.info(content)
     supportedLocations = ''
     capabilities = content['capabilities']
-    capacity = ''
+    capacity = content['capacity']
     dmsendpoint = content['cluster_api_endpoint']
     profile = _convert_content(content)
     localmodel = DeploymentManager(
@@ -113,6 +113,7 @@ def update_by(target: DeploymentManager, stxobj: StxGenericModel,
     target.hash = stxobj.hash
     target.oCloudId = parentid
     target.capabilities = content['capabilities']
+    target.capacity = content['capacity']
     target.version_number = target.version_number + 1
     target.profile = _convert_content(content)
 

o2ims/views/ocloud_dto.py

@@ -369,9 +369,17 @@ class DeploymentManagerDTO:
 
     capacity = api_ims_inventory_v1.model(
         "DeploymentManagerCapacity", {
-            'OS': fields.String(
-                example='low_latency',
-                description='Show the OS capacity of ' +
+            'cpu': fields.String(
+                example='32',
+                description='Show the cpu capacity of ' +
+                'the Deployment Manager'),
+            'hugepages-2Mi': fields.String(
+                example='2048',
+                description='Show the 2Mi hugepages capacity of ' +
+                'the Deployment Manager'),
+            'hugepages-1Gi': fields.String(
+                example='2048',
+                description='Show the 1Gi hugepages capacity of ' +
                 'the Deployment Manager'),
         })
 

requirements.txt

@@ -27,3 +27,5 @@ gunicorn
 
 # Import JWT to support OAuth2
 pyjwt==2.6.0
+
+kubernetes>=30.0.0