if self._stable_repo_url == "None":
self._stable_repo_url = None
+ # Lock to avoid concurrent execution of helm commands
+ self.cmd_lock = asyncio.Lock()
+
def _get_namespace(self, cluster_uuid: str) -> str:
"""
Obtains the namespace used by the cluster with the uuid passed by argument
# version
kdu_model, version = self._split_version(kdu_model)
- repo = self._split_repo(kdu_model)
+ _, repo = self._split_repo(kdu_model)
if repo:
- self.repo_update(cluster_id, repo)
+ await self.repo_update(cluster_id, repo)
command = self._get_install_command(
kdu_model,
namespace=namespace,
db_dict=db_dict,
operation="install",
- run_once=False,
)
)
namespace=namespace,
db_dict=db_dict,
operation="install",
- run_once=True,
- check_every=0,
)
if rc != 0:
# version
kdu_model, version = self._split_version(kdu_model)
- repo = self._split_repo(kdu_model)
+ _, repo = self._split_repo(kdu_model)
if repo:
- self.repo_update(cluster_uuid, repo)
+ await self.repo_update(cluster_uuid, repo)
command = self._get_upgrade_command(
kdu_model,
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="upgrade",
- run_once=False,
)
)
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="upgrade",
- run_once=True,
- check_every=0,
)
if rc != 0:
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="scale",
- run_once=False,
)
)
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="scale",
- run_once=True,
- check_every=0,
)
if rc != 0:
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="rollback",
- run_once=False,
)
)
namespace=instance_info["namespace"],
db_dict=db_dict,
operation="rollback",
- run_once=True,
- check_every=0,
)
if rc != 0:
environ.update(env)
try:
- process = await asyncio.create_subprocess_exec(
- *command,
- stdout=asyncio.subprocess.PIPE,
- stderr=asyncio.subprocess.PIPE,
- env=environ,
- )
+ async with self.cmd_lock:
+ process = await asyncio.create_subprocess_exec(
+ *command,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ env=environ,
+ )
- # wait for command terminate
- stdout, stderr = await process.communicate()
+ # wait for command terminate
+ stdout, stderr = await process.communicate()
- return_code = process.returncode
+ return_code = process.returncode
output = ""
if stdout:
return output, return_code
except asyncio.CancelledError:
+ # first, kill the process if it is still running
+ if process.returncode is None:
+ process.kill()
raise
except K8sException:
raise
environ.update(env)
try:
- read, write = os.pipe()
- await asyncio.create_subprocess_exec(*command1, stdout=write, env=environ)
- os.close(write)
- process_2 = await asyncio.create_subprocess_exec(
- *command2, stdin=read, stdout=asyncio.subprocess.PIPE, env=environ
- )
- os.close(read)
- stdout, stderr = await process_2.communicate()
+ async with self.cmd_lock:
+ read, write = os.pipe()
+ process_1 = await asyncio.create_subprocess_exec(
+ *command1, stdout=write, env=environ
+ )
+ os.close(write)
+ process_2 = await asyncio.create_subprocess_exec(
+ *command2, stdin=read, stdout=asyncio.subprocess.PIPE, env=environ
+ )
+ os.close(read)
+ stdout, stderr = await process_2.communicate()
- return_code = process_2.returncode
+ return_code = process_2.returncode
output = ""
if stdout:
return output, return_code
except asyncio.CancelledError:
+ # first, kill the processes if they are still running
+ for process in (process_1, process_2):
+ if process.returncode is None:
+ process.kill()
raise
except K8sException:
raise
if repo_url:
repo_str = " --repo {}".format(repo_url)
- idx = kdu_model.find("/")
- if idx >= 0:
- idx += 1
- kdu_model = kdu_model[idx:]
+ # Obtain the Chart's name and store it in the var kdu_model
+ kdu_model, _ = self._split_repo(kdu_model=kdu_model)
kdu_model, version = self._split_version(kdu_model)
if version:
version_str = ""
full_command = self._get_inspect_command(
- inspect_command, kdu_model, repo_str, version_str
+ show_command=inspect_command,
+ kdu_model=kdu_model,
+ repo_str=repo_str,
+ version=version_str,
)
- output, _rc = await self._local_async_exec(command=full_command)
+ output, _ = await self._local_async_exec(command=full_command)
return output
operation: str,
kdu_instance: str,
namespace: str = None,
- check_every: float = 10,
db_dict: dict = None,
- run_once: bool = False,
- ):
- while True:
- try:
- await asyncio.sleep(check_every)
- detailed_status = await self._status_kdu(
- cluster_id=cluster_id,
- kdu_instance=kdu_instance,
- yaml_format=False,
- namespace=namespace,
- )
- status = detailed_status.get("info").get("description")
- self.log.debug("KDU {} STATUS: {}.".format(kdu_instance, status))
- # write status to db
- result = await self.write_app_status_to_db(
- db_dict=db_dict,
- status=str(status),
- detailed_status=str(detailed_status),
- operation=operation,
- )
- if not result:
- self.log.info("Error writing in database. Task exiting...")
- return
- except asyncio.CancelledError:
- self.log.debug("Task cancelled")
- return
- except Exception as e:
- self.log.debug(
- "_store_status exception: {}".format(str(e)), exc_info=True
- )
- pass
- finally:
- if run_once:
- return
+ ) -> None:
+ """
+ Obtains the status of the KDU instance based on Helm Charts, and stores it in the database.
+
+ :param cluster_id (str): the cluster where the KDU instance is deployed
+ :param operation (str): The operation related to the status to be updated (for instance, "install" or "upgrade")
+ :param kdu_instance (str): The KDU instance in relation to which the status is obtained
+ :param namespace (str): The Kubernetes namespace where the KDU instance was deployed. Defaults to None
+ :param db_dict (dict): A dictionary with the database necessary information. It shall contain the
+ values for the keys:
+ - "collection": The Mongo DB collection to write to
+ - "filter": The query filter to use in the update process
+ - "path": The dot separated keys which targets the object to be updated
+ Defaults to None.
+ """
+
+ try:
+ detailed_status = await self._status_kdu(
+ cluster_id=cluster_id,
+ kdu_instance=kdu_instance,
+ yaml_format=False,
+ namespace=namespace,
+ )
+
+ status = detailed_status.get("info").get("description")
+ self.log.debug(f"Status for KDU {kdu_instance} obtained: {status}.")
+
+ # write status to db
+ result = await self.write_app_status_to_db(
+ db_dict=db_dict,
+ status=str(status),
+ detailed_status=str(detailed_status),
+ operation=operation,
+ )
+
+ if not result:
+ self.log.info("Error writing in database. Task exiting...")
+
+ except asyncio.CancelledError as e:
+ self.log.warning(
+ f"Exception in method {self._store_status.__name__} (task cancelled): {e}"
+ )
+ except Exception as e:
+ self.log.warning(f"Exception in method {self._store_status.__name__}: {e}")
# params for use in -f file
# returns values file option and filename (in order to delete it at the end)
kdu_model = parts[0]
return kdu_model, version
- async def _split_repo(self, kdu_model: str) -> str:
+ def _split_repo(self, kdu_model: str) -> (str, str):
+ """Obtain the Helm Chart's repository and Chart's names from the KDU model
+
+ Args:
+ kdu_model (str): Associated KDU model
+
+ Returns:
+ (str, str): Tuple with the Chart name in index 0, and the repo name
+ in index 2; if there was a problem finding them, return None
+ for both
+ """
+
+ chart_name = None
repo_name = None
+
idx = kdu_model.find("/")
if idx >= 0:
+ chart_name = kdu_model[idx + 1 :]
repo_name = kdu_model[:idx]
- return repo_name
+
+ return chart_name, repo_name
async def _find_repo(self, kdu_model: str, cluster_uuid: str) -> str:
"""Obtain the Helm repository for an Helm Chart
str: the repository URL; if Helm Chart is a local one, the function returns None
"""
+ _, repo_name = self._split_repo(kdu_model=kdu_model)
+
repo_url = None
- idx = kdu_model.find("/")
- if idx >= 0:
- repo_name = kdu_model[:idx]
+ if repo_name:
# Find repository link
local_repo_list = await self.repo_list(cluster_uuid)
for repo in local_repo_list:
- repo_url = repo["url"] if repo["name"] == repo_name else None
+ if repo["name"] == repo_name:
+ repo_url = repo["url"]
+ break # it is not necessary to continue the loop if the repo link was found...
+
return repo_url
projects / osm / N2VC.git / blobdiff