installers/charm/osm-update-db-operator/src/db_upgrade.py

   1 # Copyright 2022 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 # not use this file except in compliance with the License. You may obtain
   5 # a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 # License for the specific language governing permissions and limitations
  13 # under the License.
  14
  15 """Upgrade DB charm module."""
  16
  17 import json
  18 import logging
  19
  20 from pymongo import MongoClient
  21 from uuid import uuid4
  22
  23 logger = logging.getLogger(__name__)
  24
  25
  26 class MongoUpgrade1214:
  27     """Upgrade MongoDB Database from OSM v12 to v14."""
  28
  29     @staticmethod
  30     def gather_vnfr_healing_alerts(vnfr, vnfd):
  31         alerts = []
  32         nsr_id = vnfr["nsr-id-ref"]
  33         df = vnfd.get("df", [{}])[0]
  34         # Checking for auto-healing configuration
  35         if "healing-aspect" in df:
  36             healing_aspects = df["healing-aspect"]
  37             for healing in healing_aspects:
  38                 for healing_policy in healing.get("healing-policy", ()):
  39                     vdu_id = healing_policy["vdu-id"]
  40                     vdur = next(
  41                         (
  42                             vdur
  43                             for vdur in vnfr["vdur"]
  44                             if vdu_id == vdur["vdu-id-ref"]
  45                         ),
  46                         {},
  47                     )
  48                     if not vdur:
  49                         continue
  50                     metric_name = "vm_status"
  51                     vdu_name = vdur.get("name")
  52                     vnf_member_index = vnfr["member-vnf-index-ref"]
  53                     uuid = str(uuid4())
  54                     name = f"healing_{uuid}"
  55                     action = healing_policy
  56                     # action_on_recovery = healing.get("action-on-recovery")
  57                     # cooldown_time = healing.get("cooldown-time")
  58                     # day1 = healing.get("day1")
  59                     alert = {
  60                         "uuid": uuid,
  61                         "name": name,
  62                         "metric": metric_name,
  63                         "tags": {
  64                             "ns_id": nsr_id,
  65                             "vnf_member_index": vnf_member_index,
  66                             "vdu_name": vdu_name,
  67                         },
  68                         "alarm_status": "ok",
  69                         "action_type": "healing",
  70                         "action": action,
  71                     }
  72                     alerts.append(alert)
  73         return alerts
  74
  75     @staticmethod
  76     def gather_vnfr_scaling_alerts(vnfr, vnfd):
  77         alerts = []
  78         nsr_id = vnfr["nsr-id-ref"]
  79         df = vnfd.get("df", [{}])[0]
  80         # Checking for auto-scaling configuration
  81         if "scaling-aspect" in df:
  82             rel_operation_types = {
  83                 "GE": ">=",
  84                 "LE": "<=",
  85                 "GT": ">",
  86                 "LT": "<",
  87                 "EQ": "==",
  88                 "NE": "!=",
  89             }
  90             scaling_aspects = df["scaling-aspect"]
  91             all_vnfd_monitoring_params = {}
  92             for ivld in vnfd.get("int-virtual-link-desc", ()):
  93                 for mp in ivld.get("monitoring-parameters", ()):
  94                     all_vnfd_monitoring_params[mp.get("id")] = mp
  95             for vdu in vnfd.get("vdu", ()):
  96                 for mp in vdu.get("monitoring-parameter", ()):
  97                     all_vnfd_monitoring_params[mp.get("id")] = mp
  98             for df in vnfd.get("df", ()):
  99                 for mp in df.get("monitoring-parameter", ()):
 100                     all_vnfd_monitoring_params[mp.get("id")] = mp
 101             for scaling_aspect in scaling_aspects:
 102                 scaling_group_name = scaling_aspect.get("name", "")
 103                 # Get monitored VDUs
 104                 all_monitored_vdus = set()
 105                 for delta in scaling_aspect.get(
 106                     "aspect-delta-details", {}
 107                 ).get("deltas", ()):
 108                     for vdu_delta in delta.get("vdu-delta", ()):
 109                         all_monitored_vdus.add(vdu_delta.get("id"))
 110                 monitored_vdurs = list(
 111                     filter(
 112                         lambda vdur: vdur["vdu-id-ref"]
 113                         in all_monitored_vdus,
 114                         vnfr["vdur"],
 115                     )
 116                 )
 117                 if not monitored_vdurs:
 118                     logger.error("Scaling criteria is referring to a vnf-monitoring-param that does not contain a reference to a vdu or vnf metric")
 119                     continue
 120                 for scaling_policy in scaling_aspect.get(
 121                     "scaling-policy", ()
 122                 ):
 123                     if scaling_policy["scaling-type"] != "automatic":
 124                         continue
 125                     threshold_time = scaling_policy.get(
 126                         "threshold-time", "1"
 127                     )
 128                     cooldown_time = scaling_policy.get("cooldown-time", "0")
 129                     for scaling_criteria in scaling_policy["scaling-criteria"]:
 130                         monitoring_param_ref = scaling_criteria.get(
 131                             "vnf-monitoring-param-ref"
 132                         )
 133                         vnf_monitoring_param = all_vnfd_monitoring_params[
 134                             monitoring_param_ref
 135                         ]
 136                         for vdur in monitored_vdurs:
 137                             vdu_id = vdur["vdu-id-ref"]
 138                             metric_name = vnf_monitoring_param.get("performance-metric")
 139                             vdu_name = vdur["name"]
 140                             vnf_member_index = vnfr["member-vnf-index-ref"]
 141                             scalein_threshold = scaling_criteria.get("scale-in-threshold")
 142                             # Looking for min/max-number-of-instances
 143                             instances_min_number = 1
 144                             instances_max_number = 1
 145                             vdu_profile = df["vdu-profile"]
 146                             if vdu_profile:
 147                                 profile = next(
 148                                     item
 149                                     for item in vdu_profile
 150                                     if item["id"] == vdu_id
 151                                 )
 152                                 instances_min_number = profile.get("min-number-of-instances", 1)
 153                                 instances_max_number = profile.get("max-number-of-instances", 1)
 154
 155                             if scalein_threshold:
 156                                 uuid = str(uuid4())
 157                                 name = f"scalein_{uuid}"
 158                                 operation = scaling_criteria["scale-in-relational-operation"]
 159                                 rel_operator = rel_operation_types.get(operation, "<=")
 160                                 metric_selector = f'{metric_name}{{ns_id="{nsr_id}", vnf_member_index="{vnf_member_index}", vdu_id="{vdu_id}"}}'
 161                                 expression = f"(count ({metric_selector}) > {instances_min_number}) and (avg({metric_selector}) {rel_operator} {scalein_threshold})"
 162                                 labels = {
 163                                     "ns_id": nsr_id,
 164                                     "vnf_member_index": vnf_member_index,
 165                                     "vdu_id": vdu_id,
 166                                 }
 167                                 prom_cfg = {
 168                                     "alert": name,
 169                                     "expr": expression,
 170                                     "for": str(threshold_time) + "m",
 171                                     "labels": labels,
 172                                 }
 173                                 action = scaling_policy
 174                                 action = {
 175                                     "scaling-group": scaling_group_name,
 176                                     "cooldown-time": cooldown_time,
 177                                 }
 178                                 alert = {
 179                                     "uuid": uuid,
 180                                     "name": name,
 181                                     "metric": metric_name,
 182                                     "tags": {
 183                                         "ns_id": nsr_id,
 184                                         "vnf_member_index": vnf_member_index,
 185                                         "vdu_id": vdu_id,
 186                                     },
 187                                     "alarm_status": "ok",
 188                                     "action_type": "scale_in",
 189                                     "action": action,
 190                                     "prometheus_config": prom_cfg,
 191                                 }
 192                                 alerts.append(alert)
 193
 194                             scaleout_threshold = scaling_criteria.get("scale-out-threshold")
 195                             if scaleout_threshold:
 196                                 uuid = str(uuid4())
 197                                 name = f"scaleout_{uuid}"
 198                                 operation = scaling_criteria["scale-out-relational-operation"]
 199                                 rel_operator = rel_operation_types.get(operation, "<=")
 200                                 metric_selector = f'{metric_name}{{ns_id="{nsr_id}", vnf_member_index="{vnf_member_index}", vdu_id="{vdu_id}"}}'
 201                                 expression = f"(count ({metric_selector}) < {instances_max_number}) and (avg({metric_selector}) {rel_operator} {scaleout_threshold})"
 202                                 labels = {
 203                                     "ns_id": nsr_id,
 204                                     "vnf_member_index": vnf_member_index,
 205                                     "vdu_id": vdu_id,
 206                                 }
 207                                 prom_cfg = {
 208                                     "alert": name,
 209                                     "expr": expression,
 210                                     "for": str(threshold_time) + "m",
 211                                     "labels": labels,
 212                                 }
 213                                 action = scaling_policy
 214                                 action = {
 215                                     "scaling-group": scaling_group_name,
 216                                     "cooldown-time": cooldown_time,
 217                                 }
 218                                 alert = {
 219                                     "uuid": uuid,
 220                                     "name": name,
 221                                     "metric": metric_name,
 222                                     "tags": {
 223                                         "ns_id": nsr_id,
 224                                         "vnf_member_index": vnf_member_index,
 225                                         "vdu_id": vdu_id,
 226                                     },
 227                                     "alarm_status": "ok",
 228                                     "action_type": "scale_out",
 229                                     "action": action,
 230                                     "prometheus_config": prom_cfg,
 231                                 }
 232                                 alerts.append(alert)
 233         return alerts
 234
 235     @staticmethod
 236     def _migrate_alerts(osm_db):
 237         """Create new alerts collection.
 238         """
 239         if "alerts" in osm_db.list_collection_names():
 240             return
 241         logger.info("Entering in MongoUpgrade1214._migrate_alerts function")
 242
 243         # Get vnfds from MongoDB
 244         logger.info("Reading VNF descriptors:")
 245         vnfds = osm_db["vnfds"]
 246         db_vnfds = []
 247         for vnfd in vnfds.find():
 248             logger.info(f'  {vnfd["_id"]}: {vnfd["description"]}')
 249             db_vnfds.append(vnfd)
 250
 251         # Get vnfrs from MongoDB
 252         logger.info("Reading VNFRs")
 253         vnfrs = osm_db["vnfrs"]
 254
 255         # Gather healing and scaling alerts for each vnfr
 256         healing_alerts = []
 257         scaling_alerts = []
 258         for vnfr in vnfrs.find():
 259             logger.info(f'  vnfr {vnfr["_id"]}')
 260             vnfd = next((sub for sub in db_vnfds if sub["_id"] == vnfr["vnfd-id"]), None)
 261             healing_alerts.extend(MongoUpgrade1214.gather_vnfr_healing_alerts(vnfr, vnfd))
 262             scaling_alerts.extend(MongoUpgrade1214.gather_vnfr_scaling_alerts(vnfr, vnfd))
 263
 264         # Add new alerts in MongoDB
 265         alerts = osm_db["alerts"]
 266         for alert in healing_alerts:
 267             logger.info(f"Storing healing alert in MongoDB: {alert}")
 268             alerts.insert_one(alert)
 269         for alert in scaling_alerts:
 270             logger.info(f"Storing scaling alert in MongoDB: {alert}")
 271             alerts.insert_one(alert)
 272
 273         # Delete old alarms collections
 274         logger.info("Deleting alarms and alarms_action collections")
 275         alarms = osm_db["alarms"]
 276         alarms.drop()
 277         alarms_action = osm_db["alarms_action"]
 278         alarms_action.drop()
 279
 280
 281     @staticmethod
 282     def upgrade(mongo_uri):
 283         """Upgrade alerts in MongoDB."""
 284         logger.info("Entering in MongoUpgrade1214.upgrade function")
 285         myclient = MongoClient(mongo_uri)
 286         osm_db = myclient["osm"]
 287         MongoUpgrade1214._migrate_alerts(osm_db)
 288
 289
 290 class MongoUpgrade1012:
 291     """Upgrade MongoDB Database from OSM v10 to v12."""
 292
 293     @staticmethod
 294     def _remove_namespace_from_k8s(nsrs, nsr):
 295         namespace = "kube-system:"
 296         if nsr["_admin"].get("deployed"):
 297             k8s_list = []
 298             for k8s in nsr["_admin"]["deployed"].get("K8s"):
 299                 if k8s.get("k8scluster-uuid"):
 300                     k8s["k8scluster-uuid"] = k8s["k8scluster-uuid"].replace(namespace, "", 1)
 301                 k8s_list.append(k8s)
 302             myquery = {"_id": nsr["_id"]}
 303             nsrs.update_one(myquery, {"$set": {"_admin.deployed.K8s": k8s_list}})
 304
 305     @staticmethod
 306     def _update_nsr(osm_db):
 307         """Update nsr.
 308
 309         Add vim_message = None if it does not exist.
 310         Remove "namespace:" from k8scluster-uuid.
 311         """
 312         if "nsrs" not in osm_db.list_collection_names():
 313             return
 314         logger.info("Entering in MongoUpgrade1012._update_nsr function")
 315
 316         nsrs = osm_db["nsrs"]
 317         for nsr in nsrs.find():
 318             logger.debug(f"Updating {nsr['_id']} nsr")
 319             for key, values in nsr.items():
 320                 if isinstance(values, list):
 321                     item_list = []
 322                     for value in values:
 323                         if isinstance(value, dict) and value.get("vim_info"):
 324                             index = list(value["vim_info"].keys())[0]
 325                             if not value["vim_info"][index].get("vim_message"):
 326                                 value["vim_info"][index]["vim_message"] = None
 327                             item_list.append(value)
 328                     myquery = {"_id": nsr["_id"]}
 329                     nsrs.update_one(myquery, {"$set": {key: item_list}})
 330             MongoUpgrade1012._remove_namespace_from_k8s(nsrs, nsr)
 331
 332     @staticmethod
 333     def _update_vnfr(osm_db):
 334         """Update vnfr.
 335
 336         Add vim_message to vdur if it does not exist.
 337         Copy content of interfaces into interfaces_backup.
 338         """
 339         if "vnfrs" not in osm_db.list_collection_names():
 340             return
 341         logger.info("Entering in MongoUpgrade1012._update_vnfr function")
 342         mycol = osm_db["vnfrs"]
 343         for vnfr in mycol.find():
 344             logger.debug(f"Updating {vnfr['_id']} vnfr")
 345             vdur_list = []
 346             for vdur in vnfr["vdur"]:
 347                 if vdur.get("vim_info"):
 348                     index = list(vdur["vim_info"].keys())[0]
 349                     if not vdur["vim_info"][index].get("vim_message"):
 350                         vdur["vim_info"][index]["vim_message"] = None
 351                     if vdur["vim_info"][index].get(
 352                         "interfaces", "Not found"
 353                     ) != "Not found" and not vdur["vim_info"][index].get("interfaces_backup"):
 354                         vdur["vim_info"][index]["interfaces_backup"] = vdur["vim_info"][index][
 355                             "interfaces"
 356                         ]
 357                 vdur_list.append(vdur)
 358             myquery = {"_id": vnfr["_id"]}
 359             mycol.update_one(myquery, {"$set": {"vdur": vdur_list}})
 360
 361     @staticmethod
 362     def _update_k8scluster(osm_db):
 363         """Remove namespace from helm-chart and helm-chart-v3 id."""
 364         if "k8sclusters" not in osm_db.list_collection_names():
 365             return
 366         logger.info("Entering in MongoUpgrade1012._update_k8scluster function")
 367         namespace = "kube-system:"
 368         k8sclusters = osm_db["k8sclusters"]
 369         for k8scluster in k8sclusters.find():
 370             if k8scluster["_admin"].get("helm-chart") and k8scluster["_admin"]["helm-chart"].get(
 371                 "id"
 372             ):
 373                 if k8scluster["_admin"]["helm-chart"]["id"].startswith(namespace):
 374                     k8scluster["_admin"]["helm-chart"]["id"] = k8scluster["_admin"]["helm-chart"][
 375                         "id"
 376                     ].replace(namespace, "", 1)
 377             if k8scluster["_admin"].get("helm-chart-v3") and k8scluster["_admin"][
 378                 "helm-chart-v3"
 379             ].get("id"):
 380                 if k8scluster["_admin"]["helm-chart-v3"]["id"].startswith(namespace):
 381                     k8scluster["_admin"]["helm-chart-v3"]["id"] = k8scluster["_admin"][
 382                         "helm-chart-v3"
 383                     ]["id"].replace(namespace, "", 1)
 384             myquery = {"_id": k8scluster["_id"]}
 385             k8sclusters.update_one(myquery, {"$set": k8scluster})
 386
 387     @staticmethod
 388     def upgrade(mongo_uri):
 389         """Upgrade nsr, vnfr and k8scluster in DB."""
 390         logger.info("Entering in MongoUpgrade1012.upgrade function")
 391         myclient = MongoClient(mongo_uri)
 392         osm_db = myclient["osm"]
 393         MongoUpgrade1012._update_nsr(osm_db)
 394         MongoUpgrade1012._update_vnfr(osm_db)
 395         MongoUpgrade1012._update_k8scluster(osm_db)
 396
 397
 398 class MongoUpgrade910:
 399     """Upgrade MongoDB Database from OSM v9 to v10."""
 400
 401     @staticmethod
 402     def upgrade(mongo_uri):
 403         """Add parameter alarm status = OK if not found in alarms collection."""
 404         myclient = MongoClient(mongo_uri)
 405         osm_db = myclient["osm"]
 406         collist = osm_db.list_collection_names()
 407
 408         if "alarms" in collist:
 409             mycol = osm_db["alarms"]
 410             for x in mycol.find():
 411                 if not x.get("alarm_status"):
 412                     myquery = {"_id": x["_id"]}
 413                     mycol.update_one(myquery, {"$set": {"alarm_status": "ok"}})
 414
 415
 416 class MongoPatch1837:
 417     """Patch Bug 1837 on MongoDB."""
 418
 419     @staticmethod
 420     def _update_nslcmops_params(osm_db):
 421         """Updates the nslcmops collection to change the additional params to a string."""
 422         logger.info("Entering in MongoPatch1837._update_nslcmops_params function")
 423         if "nslcmops" in osm_db.list_collection_names():
 424             nslcmops = osm_db["nslcmops"]
 425             for nslcmop in nslcmops.find():
 426                 if nslcmop.get("operationParams"):
 427                     if nslcmop["operationParams"].get("additionalParamsForVnf") and isinstance(
 428                         nslcmop["operationParams"].get("additionalParamsForVnf"), list
 429                     ):
 430                         string_param = json.dumps(
 431                             nslcmop["operationParams"]["additionalParamsForVnf"]
 432                         )
 433                         myquery = {"_id": nslcmop["_id"]}
 434                         nslcmops.update_one(
 435                             myquery,
 436                             {
 437                                 "$set": {
 438                                     "operationParams": {"additionalParamsForVnf": string_param}
 439                                 }
 440                             },
 441                         )
 442                     elif nslcmop["operationParams"].get("primitive_params") and isinstance(
 443                         nslcmop["operationParams"].get("primitive_params"), dict
 444                     ):
 445                         string_param = json.dumps(nslcmop["operationParams"]["primitive_params"])
 446                         myquery = {"_id": nslcmop["_id"]}
 447                         nslcmops.update_one(
 448                             myquery,
 449                             {"$set": {"operationParams": {"primitive_params": string_param}}},
 450                         )
 451
 452     @staticmethod
 453     def _update_vnfrs_params(osm_db):
 454         """Updates the vnfrs collection to change the additional params to a string."""
 455         logger.info("Entering in MongoPatch1837._update_vnfrs_params function")
 456         if "vnfrs" in osm_db.list_collection_names():
 457             mycol = osm_db["vnfrs"]
 458             for vnfr in mycol.find():
 459                 if vnfr.get("kdur"):
 460                     kdur_list = []
 461                     for kdur in vnfr["kdur"]:
 462                         if kdur.get("additionalParams") and not isinstance(
 463                             kdur["additionalParams"], str
 464                         ):
 465                             kdur["additionalParams"] = json.dumps(kdur["additionalParams"])
 466                         kdur_list.append(kdur)
 467                     myquery = {"_id": vnfr["_id"]}
 468                     mycol.update_one(
 469                         myquery,
 470                         {"$set": {"kdur": kdur_list}},
 471                     )
 472                     vnfr["kdur"] = kdur_list
 473
 474     @staticmethod
 475     def patch(mongo_uri):
 476         """Updates the database to change the additional params from dict to a string."""
 477         logger.info("Entering in MongoPatch1837.patch function")
 478         myclient = MongoClient(mongo_uri)
 479         osm_db = myclient["osm"]
 480         MongoPatch1837._update_nslcmops_params(osm_db)
 481         MongoPatch1837._update_vnfrs_params(osm_db)
 482
 483
 484 MONGODB_UPGRADE_FUNCTIONS = {
 485     "9": {"10": [MongoUpgrade910.upgrade]},
 486     "10": {"12": [MongoUpgrade1012.upgrade]},
 487     "12": {"14": [MongoUpgrade1214.upgrade]},
 488 }
 489 MYSQL_UPGRADE_FUNCTIONS = {}
 490 BUG_FIXES = {
 491     1837: MongoPatch1837.patch,
 492 }
 493
 494
 495 class MongoUpgrade:
 496     """Upgrade MongoDB Database."""
 497
 498     def __init__(self, mongo_uri):
 499         self.mongo_uri = mongo_uri
 500
 501     def upgrade(self, current, target):
 502         """Validates the upgrading path and upgrades the DB."""
 503         self._validate_upgrade(current, target)
 504         for function in MONGODB_UPGRADE_FUNCTIONS.get(current)[target]:
 505             function(self.mongo_uri)
 506
 507     def _validate_upgrade(self, current, target):
 508         """Check if the upgrade path chosen is possible."""
 509         logger.info("Validating the upgrade path")
 510         if current not in MONGODB_UPGRADE_FUNCTIONS:
 511             raise Exception(f"cannot upgrade from {current} version.")
 512         if target not in MONGODB_UPGRADE_FUNCTIONS[current]:
 513             raise Exception(f"cannot upgrade from version {current} to {target}.")
 514
 515     def apply_patch(self, bug_number: int) -> None:
 516         """Checks the bug-number and applies the fix in the database."""
 517         if bug_number not in BUG_FIXES:
 518             raise Exception(f"There is no patch for bug {bug_number}")
 519         patch_function = BUG_FIXES[bug_number]
 520         patch_function(self.mongo_uri)
 521
 522
 523 class MysqlUpgrade:
 524     """Upgrade Mysql Database."""
 525
 526     def __init__(self, mysql_uri):
 527         self.mysql_uri = mysql_uri
 528
 529     def upgrade(self, current, target):
 530         """Validates the upgrading path and upgrades the DB."""
 531         self._validate_upgrade(current, target)
 532         for function in MYSQL_UPGRADE_FUNCTIONS[current][target]:
 533             function(self.mysql_uri)
 534
 535     def _validate_upgrade(self, current, target):
 536         """Check if the upgrade path chosen is possible."""
 537         logger.info("Validating the upgrade path")
 538         if current not in MYSQL_UPGRADE_FUNCTIONS:
 539             raise Exception(f"cannot upgrade from {current} version.")
 540         if target not in MYSQL_UPGRADE_FUNCTIONS[current]:
 541             raise Exception(f"cannot upgrade from version {current} to {target}.")