Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Extract granular log handler from SOTA and fix mocking in unit tests #565

Open
wants to merge 3 commits into
base: develop
Choose a base branch
from
+173 −146
Open

Extract granular log handler from SOTA and fix mocking in unit tests #565

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7df0b43
pull out GranularLogHandler from SOTA into its own file
gblewis1 Oct 5, 2024
5bd345a
mock file operations in granular_log_handler
gblewis1 Oct 5, 2024
df1d899
update fail-under back to 81
gblewis1 Oct 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions inbm/dispatcher-agent/dispatcher/sota/granular_log_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""
Copyright (C) 2017-2024 Intel Corporation
SPDX-License-Identifier: Apache-2.0
"""

import logging
import os
import threading

from inbm_common_lib.utility import remove_file, get_os_version
from inbm_lib.detect_os import detect_os, LinuxDistType
from inbm_lib.constants import OTA_PENDING, FAIL, OTA_SUCCESS, ROLLBACK, GRANULAR_LOG_FILE

from ..update_logger import UpdateLogger

logger = logging.getLogger(__name__)

class GranularLogHandler:
def __init__(self) -> None:
self._granular_lock = threading.Lock()

def save_granular_log(self, update_logger: UpdateLogger, check_package: bool = True) -> None:
"""Save the granular log.
In Ubuntu, it saves the package level information.
In TiberOS, it saves the detail of the SOTA update.

@param check_package: True if you want to check the package's status and version and record them in Ubuntu.
"""
log = {}
current_os = detect_os()
# TODO: Remove Mariner when confirmed that TiberOS is in use
with self._granular_lock:
if current_os == LinuxDistType.tiber.name or current_os == LinuxDistType.Mariner.name:
# Delete the previous log if exist.
if os.path.exists(GRANULAR_LOG_FILE):
remove_file(GRANULAR_LOG_FILE)

if update_logger.detail_status == FAIL or update_logger.detail_status == ROLLBACK:
log = {
"StatusDetail.Status": update_logger.detail_status,
"FailureReason": update_logger.error
}
elif update_logger.detail_status == OTA_SUCCESS or update_logger.detail_status == OTA_PENDING:
log = {
"StatusDetail.Status": update_logger.detail_status,
"Version": get_os_version()
}
# In TiberOS, no package level information needed.
update_logger.save_granular_log_file(log=log, check_package=False)
else:
update_logger.save_granular_log_file(check_package=check_package)
51 changes: 10 additions & 41 deletions inbm/dispatcher-agent/dispatcher/sota/sota.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import threading
from typing import Any, List, Optional, Union, Mapping

from dispatcher.sota.granular_log_handler import GranularLogHandler
from inbm_common_lib.exceptions import UrlSecurityException
from inbm_common_lib.utility import canonicalize_uri, remove_file, get_os_version
from inbm_common_lib.request_message_constants import SOTA_FAILURE
Expand All @@ -24,7 +25,7 @@
from .constants import SUCCESS, SOTA_STATE, SOTA_CACHE, PROCEED_WITHOUT_ROLLBACK_DEFAULT
from .downloader import Downloader
from .log_helper import get_log_destination
from .os_factory import ISotaOs, SotaOsFactory, TiberOSBasedSotaOs
from .os_factory import ISotaOs, SotaOsFactory
from .os_updater import OsUpdater
from .rebooter import Rebooter
from .setup_helper import SetupHelper
Expand All @@ -37,7 +38,6 @@

logger = logging.getLogger(__name__)


class SOTAUtil: # FIXME intermediate step in refactor
def check_diagnostic_disk(self,
estimated_size: Union[float, int],
Expand Down Expand Up @@ -114,7 +114,7 @@ def __init__(self,
self.sota_mode = parsed_manifest['sota_mode']
self._update_logger = update_logger
self._dispatcher_broker = dispatcher_broker
self._granular_lock = threading.Lock()
self._granular_log_handler = GranularLogHandler()

try:
manifest_package_list = parsed_manifest['package_list']
Expand Down Expand Up @@ -260,7 +260,7 @@ def execute(self, proceed_without_rollback: bool, skip_sleeps: bool = False) ->
self._update_logger.update_log(FAIL)
self._update_logger.detail_status = ROLLBACK
self._update_logger.error = "Critical service failure."
self.save_granular_log(check_package=False)
self._granular_log_handler.save_granular_log(update_logger=self._update_logger, check_package=False)
snapshot.revert(rebooter, time_to_wait_before_reboot)
elif self.sota_state == 'diagnostic_system_healthy':
try:
Expand All @@ -270,15 +270,15 @@ def execute(self, proceed_without_rollback: bool, skip_sleeps: bool = False) ->
self._dispatcher_broker.send_result(msg)
snapshot.commit()
self._update_logger.detail_status = OTA_SUCCESS
self.save_granular_log(check_package=False)
self._granular_log_handler.save_granular_log(update_logger=self._update_logger, check_package=False)
except SotaError as e:
msg = "FAILED INSTALL: System has not been properly updated; reverting."
logger.debug(str(e))
self._dispatcher_broker.send_result(msg)
self._update_logger.update_log(FAIL)
self._update_logger.detail_status = ROLLBACK
self._update_logger.error = f"{msg}. Error: {e}"
self.save_granular_log(check_package=False)
self._granular_log_handler.save_granular_log(update_logger=self._update_logger, check_package=False)
snapshot.revert(rebooter, time_to_wait_before_reboot)
else:
self.execute_from_manifest(setup_helper=setup_helper,
Expand Down Expand Up @@ -393,13 +393,13 @@ def execute_from_manifest(self,
# mode here to record the successful SOTA with current os version.
# TODO: Remove Mariner when confirmed that TiberOS is in use
elif detect_os() == LinuxDistType.tiber.name or detect_os() == LinuxDistType.Mariner.name:
self.save_granular_log()
self._granular_log_handler.save_granular_log(update_logger=self._update_logger)

# The download-only mode only downloads the packages without installing them.
# Since there is no installation, there will be no changes in the package status or version.
# The apt history.log also doesn't record any changes. Therefore we can skip saving granular log.
elif self.sota_mode != 'download-only':
self.save_granular_log()
self._granular_log_handler.save_granular_log(update_logger=self._update_logger)


if (self.sota_mode == 'download-only') or (not self._is_reboot_device()):
Expand All @@ -418,9 +418,9 @@ def execute_from_manifest(self,
# mode because we want to record the artifact download failure.
# TODO: Remove Mariner when confirmed that TiberOS is in use
if detect_os() == LinuxDistType.tiber.name or detect_os() == LinuxDistType.Mariner.name:
self.save_granular_log()
self._granular_log_handler.save_granular_log(update_logger=self._update_logger)
elif self.sota_mode != 'download-only':
self.save_granular_log()
self._granular_log_handler.save_granular_log(update_logger=self._update_logger)
self._dispatcher_broker.telemetry(SOTA_FAILURE)
self._dispatcher_broker.send_result(SOTA_FAILURE)
raise SotaError(SOTA_FAILURE)
Expand All @@ -440,37 +440,6 @@ def _is_ota_no_update_available(self, cmd_list: List) -> bool:
return True
return False

def save_granular_log(self, check_package: bool = True) -> None:
"""Save the granular log.
In Ubuntu, it saves the package level information.
In TiberOS, it saves the detail of the SOTA update.

@param check_package: True if you want to check the package's status and version and record them in Ubuntu.
"""
log = {}
current_os = detect_os()
# TODO: Remove Mariner when confirmed that TiberOS is in use
with self._granular_lock:
if current_os == LinuxDistType.tiber.name or current_os == LinuxDistType.Mariner.name:
# Delete the previous log if exist.
if os.path.exists(GRANULAR_LOG_FILE):
remove_file(GRANULAR_LOG_FILE)

if self._update_logger.detail_status == FAIL or self._update_logger.detail_status == ROLLBACK:
log = {
"StatusDetail.Status": self._update_logger.detail_status,
"FailureReason": self._update_logger.error
}
elif self._update_logger.detail_status == OTA_SUCCESS or self._update_logger.detail_status == OTA_PENDING:
log = {
"StatusDetail.Status": self._update_logger.detail_status,
"Version": get_os_version()
}
# In TiberOS, no package level information needed.
self._update_logger.save_granular_log_file(log=log, check_package=False)
else:
self._update_logger.save_granular_log_file(check_package=check_package)

def check(self) -> None:
"""Perform manifest checking before SOTA"""
logger.debug("")
Expand Down
109 changes: 109 additions & 0 deletions inbm/dispatcher-agent/tests/unit/sota/test_granular_log_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
"""
Copyright (C) 2017-2024 Intel Corporation
SPDX-License-Identifier: Apache-2.0
"""

import testtools
from unittest.mock import patch, mock_open

from dispatcher.sota.granular_log_handler import GranularLogHandler
from dispatcher.update_logger import UpdateLogger
from inbm_lib.constants import OTA_SUCCESS, OTA_PENDING, FAIL, ROLLBACK

class TestGranularLogHandler(testtools.TestCase):
@patch('dispatcher.sota.granular_log_handler.remove_file')
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('dispatcher.sota.granular_log_handler.get_os_version', return_value='2.0.20240802.0213')
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_version, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = OTA_SUCCESS

with patch('builtins.open', mock_open()) as m_open:
GranularLogHandler().save_granular_log(update_logger=update_logger, check_package=False)

expected_content = {
"UpdateLog": [
{
"StatusDetail.Status": OTA_SUCCESS,
"Version": '2.0.20240802.0213'
}
]
}

mock_dump.assert_called_with(expected_content, m_open(), indent=4)


@patch('dispatcher.sota.granular_log_handler.remove_file')
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('dispatcher.sota.granular_log_handler.get_os_version', return_value='2.0.20240802.0213')
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_os_version, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = OTA_PENDING

with patch('builtins.open', mock_open()) as m_open:
GranularLogHandler().save_granular_log(update_logger=update_logger, check_package=False)

expected_content = {
"UpdateLog": [
{
"StatusDetail.Status": OTA_PENDING,
"Version": '2.0.20240802.0213'
}
]
}

mock_dump.assert_called_with(expected_content, m_open(), indent=4)

@patch('dispatcher.sota.granular_log_handler.remove_file')
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_fail_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = FAIL
update_logger.error = 'Error getting artifact size from https://registry-rs.internal.ledgepark.intel.com/v2/one-intel-edge/tiberos/manifests/latest using token'

with patch('builtins.open', mock_open()) as m_open:
GranularLogHandler().save_granular_log(update_logger=update_logger, check_package=False)

expected_content = {
"UpdateLog": [
{
"StatusDetail.Status": FAIL,
"FailureReason": 'Error getting artifact size from https://registry-rs.internal.ledgepark.intel.com/v2/one-intel-edge/tiberos/manifests/latest using token'
}
]
}

mock_dump.assert_called_with(expected_content, m_open(), indent=4)


@patch('dispatcher.sota.granular_log_handler.remove_file')
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_rollback_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = ROLLBACK
update_logger.error = 'FAILED INSTALL: System has not been properly updated; reverting..'
with patch('builtins.open', mock_open()) as m_open:
GranularLogHandler().save_granular_log(update_logger=update_logger, check_package=False)

expected_content = {
"UpdateLog": [
{
"StatusDetail.Status": ROLLBACK,
"FailureReason": 'FAILED INSTALL: System has not been properly updated; reverting..'
}
]
}

mock_dump.assert_called_with(expected_content, m_open(), indent=4)
Loading