You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
122 lines
5.4 KiB
122 lines
5.4 KiB
# Copyright (C) 2005-2025 Splunk Inc. All Rights Reserved.
|
|
|
|
from migration_utility.itsi_migration_utility_handler import ItsiMigrationUtilityHandler
|
|
from migration_utility.migration_utility_manifest import \
|
|
precheck_failure_severity, dict_for_mapping_category
|
|
from migration_utility.constants import MODES
|
|
|
|
from ITOA.setup_logging import getLogger
|
|
import sys
|
|
import uuid
|
|
from splunk.clilib.bundle_paths import make_splunkhome_path
|
|
sys.path.append(make_splunkhome_path(["etc", "apps", "SA-ITOA", "lib"]))
|
|
|
|
logger = getLogger(
|
|
logger_file="itsi_migration_utility.log",
|
|
logger_name="itsi_migration_utility_prechecker",
|
|
)
|
|
|
|
|
|
class ItsiMigrationUtilityInterface():
|
|
PRE_CHECK = 'precheck'
|
|
AUTO_REMEDIATION = 'auto_remediation'
|
|
|
|
def __init__(self, sessionkey, transaction_id=None):
|
|
self.sessionkey = sessionkey
|
|
# if transaction_id is not provided, then create a new transaction_id
|
|
if transaction_id:
|
|
self.transaction_id = transaction_id
|
|
else:
|
|
self.transaction_id = uuid.uuid4().hex
|
|
|
|
self.itsi_muh_object = ItsiMigrationUtilityHandler(
|
|
self.sessionkey, self.transaction_id)
|
|
self.precheck_failure_severity = precheck_failure_severity
|
|
self.dict_for_mapping_category = dict_for_mapping_category
|
|
|
|
def get_details(self, operation_mode=PRE_CHECK):
|
|
prechecks_ids = self.itsi_muh_object.dict_for_mapping_precheck_ids.values()
|
|
prechecks_details_response = {}
|
|
for precheck_id in prechecks_ids:
|
|
if operation_mode == self.PRE_CHECK:
|
|
prechecks_details_response[precheck_id] = self.get_precheck_details(precheck_id)
|
|
elif operation_mode == self.AUTO_REMEDIATION:
|
|
prechecks_details_response[precheck_id] = self.get_auto_remediation_details(precheck_id)
|
|
return prechecks_details_response
|
|
|
|
# Following method needs to be called to fetch the description and resolution of precheck failures
|
|
def get_precheck_details(self, precheck_id):
|
|
description_details, resolution_details, documentation_details = self.itsi_muh_object.get_precheck_details(
|
|
self.itsi_muh_object, precheck_id)
|
|
|
|
# getting the only key of the documentation_details dict as there will always be a
|
|
# single key-value pair
|
|
|
|
# key of the documentation_details dict is label that is to be displayed
|
|
# value of this label is the documentaion link
|
|
documentation_label = list(documentation_details.keys())[0]
|
|
documentation_link = documentation_details[documentation_label]
|
|
|
|
# fetching precheck severity and category
|
|
precheck_severity = self.precheck_failure_severity[precheck_id]
|
|
precheck_category = self.dict_for_mapping_category[precheck_id]
|
|
blocks_upgrade = False
|
|
auto_remediation = True
|
|
|
|
# if severity is Major, then it will block the upgrade else will not
|
|
if precheck_severity == 'Major':
|
|
blocks_upgrade = True
|
|
|
|
# if remediation is unavailable, auto_remediation will be set to false
|
|
if self.get_auto_remediation_details(precheck_id)["remediation_details"] == 'N/A':
|
|
auto_remediation = False
|
|
|
|
precheck_details = {
|
|
"severity": precheck_severity,
|
|
"blocks_upgrade": blocks_upgrade,
|
|
"category": precheck_category,
|
|
"description": description_details,
|
|
"resolution": resolution_details,
|
|
"documentation_label": documentation_label,
|
|
"documentation_link": documentation_link,
|
|
"auto_remediation": auto_remediation
|
|
}
|
|
|
|
return precheck_details
|
|
|
|
# Following method needs to be called to execute operations of prechecks
|
|
def execute(self, operation_mode, precheck_ids=None):
|
|
# calling the execute() method of itsi_migration_utility_handler.py to trigger
|
|
# the execution of operations of prechecks
|
|
failed_precheck_count = 0
|
|
message = "PRECHECK"
|
|
if operation_mode == MODES["PRECHECK"]:
|
|
logger.info(f'[transaction_id={self.transaction_id}] '
|
|
f'[operation_mode={operation_mode}] '
|
|
'Executing prechecks to validate ITSI Kvstore data.')
|
|
elif operation_mode == MODES["AUTO_REMEDIATION"]:
|
|
logger.info(f'[transaction_id={self.transaction_id}] '
|
|
f'[operation_mode={operation_mode}] '
|
|
'Executing auto-remediation to resolve the precheck failures.')
|
|
try:
|
|
result, message, failed_precheck_count = self.itsi_muh_object.execute(self.itsi_muh_object, operation_mode, precheck_ids)
|
|
return result, message, failed_precheck_count
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
logger.error(f'[transaction_id={self.transaction_id}] '
|
|
f'[operation_mode={operation_mode}] '
|
|
'Error while executing job.')
|
|
return False, str(e), failed_precheck_count
|
|
|
|
# Following method needs to be called to fetch the description and resolution of precheck failures
|
|
def get_auto_remediation_details(self, precheck_id):
|
|
remediation_details, category_fixed = self.itsi_muh_object.get_auto_remediation_details(
|
|
self.itsi_muh_object, precheck_id)
|
|
|
|
precheck_details = {
|
|
"remediation_category": category_fixed,
|
|
"remediation_details": remediation_details
|
|
}
|
|
|
|
return precheck_details
|