- # Juju/k8s cluster
- k8s_jb_id = None
- try:
- k8s_jb_id, uninstall_sw = await self.juju_k8scluster.init_env(k8s_credentials)
- db_k8scluster_update["_admin.juju-bundle.id"] = k8s_jb_id
- db_k8scluster_update["_admin.juju-bundle.created"] = uninstall_sw
- except Exception as e:
- error_text_list.append("Failing init juju-bundle: {}".format(e))
- db_k8scluster_update["_admin.juju-bundle.error_msg"] = str(e)
- if isinstance(e, N2VCException):
- self.logger.error(logging_text + "Failing init juju-bundle: {}".format(e))
- else:
- self.logger.error(logging_text + "Failing init juju-bundle: {}".format(e), exc_info=True)
-
- step = "Getting the list of repos"
- if k8s_hc_id:
- self.logger.debug(logging_text + step)
- task_list = []
- db_k8srepo_list = self.db.get_list("k8srepos", {"type": "helm-chart"})
- for repo in db_k8srepo_list:
- step = "Adding repo {} to cluster: {}".format(repo["name"], k8s_hc_id)
- self.logger.debug(logging_text + step)
- task = asyncio.ensure_future(self.helm_k8scluster.repo_add(cluster_uuid=k8s_hc_id,
- name=repo["name"], url=repo["url"],
- repo_type="chart"))
- task_list.append(task)
- repo_k8scluster_list = deep_get(repo, ("_admin", "cluster-inserted")) or []
- repo_k8scluster_list.append(k8s_hc_id)
- self.update_db_2("k8srepos", repo["_id"], {"_admin.cluster-inserted": repo_k8scluster_list})
-
- if task_list:
- self.logger.debug(logging_text + 'Waiting for terminate tasks of repo_add')
- done, pending = await asyncio.wait(task_list, timeout=3600)
- if pending:
- self.logger.error(logging_text + 'There are pending tasks: {}'.format(pending))
-
- # mark as an error if both helm-chart and juju-bundle have been failed
- if k8s_hc_id or k8s_jb_id:
- db_k8scluster_update["_admin.operationalState"] = "ENABLED"
+ error_text_list = []
+ tasks_name_ok = []
+ reached_timeout = False
+ now = time()
+
+ while pending_tasks:
+ _timeout = max(1, self.timeout_create - (time() - now)) # ensure not negative with max
+ step = "Waiting for k8scluster init tasks"
+ done, pending_tasks = await asyncio.wait(pending_tasks, timeout=_timeout,
+ return_when=asyncio.FIRST_COMPLETED)
+ if not done:
+ # timeout. Set timeout is reached and process pending as if they hase been finished
+ done = pending_tasks
+ pending_tasks = None
+ reached_timeout = True
+ for task in done:
+ task_name = task2name[task]
+ if reached_timeout:
+ exc = "Timeout"
+ elif task.cancelled():
+ exc = "Cancelled"
+ else:
+ exc = task.exception()
+
+ if exc:
+ error_text_list.append("Failing init {}: {}".format(task_name, exc))
+ db_k8scluster_update["_admin.{}.error_msg".format(task_name)] = str(exc)
+ db_k8scluster_update["_admin.{}.id".format(task_name)] = None
+ db_k8scluster_update["_admin.{}.operationalState".format(task_name)] = "ERROR"
+ self.logger.error(logging_text + "{} init fail: {}".format(task_name, exc),
+ exc_info=not isinstance(exc, (N2VCException, str)))
+ else:
+ k8s_id, uninstall_sw = task.result()
+ tasks_name_ok.append(task_name)
+ self.logger.debug(logging_text + "{} init success. id={} created={}".format(
+ task_name, k8s_id, uninstall_sw))
+ db_k8scluster_update["_admin.{}.error_msg".format(task_name)] = None
+ db_k8scluster_update["_admin.{}.id".format(task_name)] = k8s_id
+ db_k8scluster_update["_admin.{}.created".format(task_name)] = uninstall_sw
+ db_k8scluster_update["_admin.{}.operationalState".format(task_name)] = "ENABLED"
+ # update database
+ step = "Updating database for " + task_name
+ self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
+ if tasks_name_ok:
+ operation_details = "ready for " + ", ".join(tasks_name_ok)
+ operation_state = "COMPLETED"
+ db_k8scluster_update["_admin.operationalState"] = "ENABLED" if not error_text_list else "DEGRADED"
+ operation_details += "; " + ";".join(error_text_list)