Add test case for k8s cleanup container

tests/container_upgrade/cleanup_testcases.json

@@ -0,0 +1,5 @@
+{
+    "testcases": [
+        "kubesonic/test_k8s_cleanup.py"
+    ]
+}

tests/container_upgrade/container_upgrade_helper.py

@@ -28,6 +28,7 @@
     "docker-auditd-watchdog": "auditd_watchdog",
     "docker-sonic-bmp": "bmp",
     "docker-bmp-watchdog": "bmp_watchdog",
+    "kubesonic-cleanup": "k8s_cleanup",
 }
 
 existing_service_list = [

tests/container_upgrade/parameters.json

@@ -19,5 +19,8 @@
     },
     "docker-bmp-watchdog": {
         "parameters": "--pid=host --net=host -v /etc/localtime:/etc/localtime:ro -v /etc/sonic:/etc/sonic:ro"
+    },
+    "kubesonic-cleanup": {
+        "parameters": "--privileged --pid=host --net=host"
     }
 }

tests/kubesonic/test_k8s_cleanup.py

@@ -0,0 +1,75 @@
+import logging
+import pytest
+
+from tests.common.plugins.loganalyzer.loganalyzer import LogAnalyzer
+
+logger = logging.getLogger(__name__)
+
+pytestmark = [
+    pytest.mark.topology('any'),
+    pytest.mark.disable_loganalyzer
+]
+
+CLEANUP_CONTAINER_NAME = "k8s_cleanup"
+
+
+def test_k8s_cleanup(duthosts, rand_one_dut_hostname, request):
+    """
+    Test the cleanup of kubesonic containers
+    """
+    # Check if the test is called from container upgrade test
+    log_file = request.config.getoption("--log-file", default=None)
+    logger.info(f"Kubesonic cleanup test log_file: {log_file}")
+    if not (log_file and "container_upgrade" in log_file):
+        pytest.skip("Skipping test as this test is not called from container upgrade test")
+
+    duthost = duthosts[rand_one_dut_hostname]
+    status_cmd = r"docker inspect {} --format \{{\{{.State.Running\}}\}}".format(CLEANUP_CONTAINER_NAME)
+    cleanup_container_status = duthost.shell(status_cmd, module_ignore_errors=True)
+
+    if "No such object" in cleanup_container_status["stderr"]:
+        pytest.fail("Skipping test as the kubesonic cleanup container is not present")
+    elif cleanup_container_status["stdout"] != "true":
+        pytest.fail("Kubesonic cleanup container is not running")
+    else:
+        logger.info("Kubesonic cleanup container is running")
+
+    # Check if the watchdog script exited code is 0 or not
+    exec_watchdog_cmd = f"docker exec {CLEANUP_CONTAINER_NAME} /watchdog.sh"
+    exec_watchdog_status = duthost.shell(exec_watchdog_cmd, module_ignore_errors=True)
+    if exec_watchdog_status["rc"] != 0:
+        pytest.fail("Kubesonic watchdog script exited with non-zero code: {}".format(
+            exec_watchdog_status["rc"]))
+    else:
+        logger.info("Kubesonic watchdog script executed successfully")
+
+    # Check if the syslog contains the expected message
+    loganalyzer = LogAnalyzer(ansible_host=duthost, marker_prefix="kubesonic_cleanup_test")
+    loganalyzer.expect_regex = [r".*kubesonic-image-cleanup: root-overlay size status\(total available\): (\d+) (\d+)"]
+    loganalyzer.match_regex = [r"(?i).*kubesonic-image-cleanup:.*fail.*"]
+    marker = loganalyzer.init()
+    try:
+        exec_cleanup_script_cmd = f"docker exec {CLEANUP_CONTAINER_NAME} /image_cleanup.sh"
+        exec_cleanup_script_status = duthost.shell(exec_cleanup_script_cmd, module_ignore_errors=True)
+        if exec_cleanup_script_status["rc"] != 0:
+            pytest.fail("Kubesonic cleanup script exited with non-zero code: {}".format(
+                exec_cleanup_script_status["rc"]))
+        else:
+            logger.info("Kubesonic cleanup script executed successfully")
+
+        exec_report_disk_size_cmd = f"docker exec {CLEANUP_CONTAINER_NAME} /report_disk_size.sh"
+        exec_report_disk_size_status = duthost.shell(exec_report_disk_size_cmd, module_ignore_errors=True)
+        if exec_report_disk_size_status["rc"] != 0:
+            pytest.fail("Kubesonic report disk size script exited with non-zero code: {}".format(
+                exec_report_disk_size_status["rc"]))
+        else:
+            logger.info("Kubesonic report disk size script executed successfully")
+
+        loganalyzer_summary = loganalyzer.analyze(marker, fail=False)
+        if loganalyzer_summary["total"]["match"] != 0:
+            pytest.fail("Kubesonic cleanup container error log is found")
+        if loganalyzer_summary["total"]["expected_match"] == 0:
+            pytest.fail("Kubesonic cleanup container disk size log is not found")
+
+    except Exception as e:
+        pytest.fail(f"Failed to check syslog for kubesonic cleanup test: {e}")