+ async def _labels_dict(self, db_dict, kdu_instance):
+ # get the network service registry
+ ns_id = db_dict["filter"]["_id"]
+ try:
+ db_nsr = self.db.get_one("nsrs", {"_id": ns_id})
+ except Exception as e:
+ print("nsr {} not found: {}".format(ns_id, e))
+ nsd_id = db_nsr["nsd"]["_id"]
+
+ # get the kdu registry
+ for index, kdu in enumerate(db_nsr["_admin"]["deployed"]["K8s"]):
+ if kdu["kdu-instance"] == kdu_instance:
+ db_kdur = kdu
+ break
+ else:
+ # No kdur found, could be the case of an EE chart
+ return {}
+
+ kdu_name = db_kdur["kdu-name"]
+ member_vnf_index = db_kdur["member-vnf-index"]
+ # get the vnf registry
+ try:
+ db_vnfr = self.db.get_one(
+ "vnfrs",
+ {"nsr-id-ref": ns_id, "member-vnf-index-ref": member_vnf_index},
+ )
+ except Exception as e:
+ print("vnfr {} not found: {}".format(member_vnf_index, e))
+
+ vnf_id = db_vnfr["_id"]
+ vnfd_id = db_vnfr["vnfd-id"]
+
+ return {
+ "managed-by": "osm.etsi.org",
+ "osm.etsi.org/ns-id": ns_id,
+ "osm.etsi.org/nsd-id": nsd_id,
+ "osm.etsi.org/vnf-id": vnf_id,
+ "osm.etsi.org/vnfd-id": vnfd_id,
+ "osm.etsi.org/kdu-id": kdu_instance,
+ "osm.etsi.org/kdu-name": kdu_name,
+ }
+
+ async def _contains_labels(self, kdu_instance, namespace, kube_config, env):
+ command = "env KUBECONFIG={} {} get manifest {} --namespace={}".format(
+ kube_config,
+ self._helm_command,
+ quote(kdu_instance),
+ quote(namespace),
+ )
+ output, rc = await self._local_async_exec(
+ command=command, raise_exception_on_error=False, env=env
+ )
+ manifests = yaml.safe_load_all(output)
+ for manifest in manifests:
+ # Check if the manifest has metadata and labels
+ if (
+ manifest is not None
+ and "metadata" in manifest
+ and "labels" in manifest["metadata"]
+ ):
+ labels = {
+ "managed-by",
+ "osm.etsi.org/kdu-id",
+ "osm.etsi.org/kdu-name",
+ "osm.etsi.org/ns-id",
+ "osm.etsi.org/nsd-id",
+ "osm.etsi.org/vnf-id",
+ "osm.etsi.org/vnfd-id",
+ }
+ if labels.issubset(manifest["metadata"]["labels"].keys()):
+ return True
+ return False
+