diff --git a/tests/integration_tests/modules/test_keys_to_console.py b/tests/integration_tests/modules/test_keys_to_console.py
index b8e74f57ba8..2f53fd7f4df 100644
--- a/tests/integration_tests/modules/test_keys_to_console.py
+++ b/tests/integration_tests/modules/test_keys_to_console.py
@@ -12,7 +12,7 @@
 from tests.integration_tests.util import (
     HAS_CONSOLE_LOG,
     get_console_log,
-    get_syslog_or_console,
+    get_journal_syslog,
 )
 
 BLACKLIST_USER_DATA = """\
@@ -52,16 +52,14 @@ class TestKeysToConsoleBlacklist:
 
     @pytest.mark.parametrize("key_type", ["ECDSA"])
     def test_excluded_keys(self, class_client, key_type):
-        assert "({})".format(key_type) not in get_syslog_or_console(
-            class_client
-        )
+        assert "({})".format(key_type) not in get_journal_syslog(class_client)
 
     # retry decorator here because it can take some time to be reflected
-    # in syslog
+    # in the journal
     @retry(tries=60, delay=1)
     @pytest.mark.parametrize("key_type", ["ED25519", "RSA"])
     def test_included_keys(self, class_client, key_type):
-        assert "({})".format(key_type) in get_syslog_or_console(class_client)
+        assert "({})".format(key_type) in get_journal_syslog(class_client)
 
 
 @pytest.mark.user_data(BLACKLIST_ALL_KEYS_USER_DATA)
@@ -75,12 +73,12 @@ class TestAllKeysToConsoleBlacklist:
     """
 
     def test_header_excluded(self, class_client):
-        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_journal_syslog(
             class_client
         )
 
     def test_footer_excluded(self, class_client):
-        assert "END SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+        assert "END SSH HOST KEY FINGERPRINTS" not in get_journal_syslog(
             class_client
         )
 
@@ -95,17 +93,15 @@ class TestKeysToConsoleDisabled:
 
     @pytest.mark.parametrize("key_type", ["ECDSA", "ED25519", "RSA"])
     def test_keys_excluded(self, class_client, key_type):
-        assert "({})".format(key_type) not in get_syslog_or_console(
-            class_client
-        )
+        assert "({})".format(key_type) not in get_journal_syslog(class_client)
 
     def test_header_excluded(self, class_client):
-        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_journal_syslog(
             class_client
         )
 
     def test_footer_excluded(self, class_client):
-        assert "END SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+        assert "END SSH HOST KEY FINGERPRINTS" not in get_journal_syslog(
             class_client
         )
 
diff --git a/tests/integration_tests/modules/test_ssh_auth_key_fingerprints.py b/tests/integration_tests/modules/test_ssh_auth_key_fingerprints.py
index d55cba91e70..8cc68ae7880 100644
--- a/tests/integration_tests/modules/test_ssh_auth_key_fingerprints.py
+++ b/tests/integration_tests/modules/test_ssh_auth_key_fingerprints.py
@@ -19,7 +19,7 @@
     OS_IMAGE_TYPE,
     PLATFORM,
 )
-from tests.integration_tests.util import HAS_CONSOLE_LOG, get_syslog_or_console
+from tests.integration_tests.util import HAS_CONSOLE_LOG, get_journal_syslog
 
 USER_DATA_SSH_AUTHKEY_DISABLE = """\
 #cloud-config
@@ -55,10 +55,10 @@ def test_ssh_authkey_fingerprints_disable(self, client):
         reason=f"No console_log available for minimal images on {PLATFORM}",
     )
     def test_ssh_authkey_fingerprints_enable(self, client):
-        syslog_output = get_syslog_or_console(client)
-        assert re.search(r"256 SHA256:.*(ECDSA)", syslog_output) is not None
-        assert re.search(r"256 SHA256:.*(ED25519)", syslog_output) is not None
-        assert re.search(r"2048 SHA256:.*(RSA)", syslog_output) is None
+        log_output = get_journal_syslog(client)
+        assert re.search(r"256 SHA256:.*(ECDSA)", log_output) is not None
+        assert re.search(r"256 SHA256:.*(ED25519)", log_output) is not None
+        assert re.search(r"2048 SHA256:.*(RSA)", log_output) is None
 
 
 @pytest.mark.user_data(
diff --git a/tests/integration_tests/util.py b/tests/integration_tests/util.py
index 6e86b677552..69b3778622f 100644
--- a/tests/integration_tests/util.py
+++ b/tests/integration_tests/util.py
@@ -16,10 +16,7 @@
 
 from cloudinit.subp import subp
 from tests.integration_tests.decorators import retry
-from tests.integration_tests.integration_settings import (
-    OS_IMAGE_TYPE,
-    PLATFORM,
-)
+from tests.integration_tests.integration_settings import PLATFORM
 from tests.integration_tests.releases import CURRENT_RELEASE, NOBLE
 
 LOG = logging.getLogger("integration_testing.util")
@@ -578,13 +575,15 @@ def get_console_log(client: "IntegrationInstance"):
     return console_log
 
 
-@retry(tries=5, delay=1)  # Retry on get_console_log failures
-def get_syslog_or_console(client: "IntegrationInstance") -> str:
-    """minimal OS_IMAGE_TYPE does not contain rsyslog"""
-    if OS_IMAGE_TYPE == "minimal" and HAS_CONSOLE_LOG:
-        return get_console_log(client)
-    else:
-        return client.read_from_file("/var/log/syslog")
+@retry(tries=5, delay=1)  # Retry on transient journalctl failures
+def get_journal_syslog(client: "IntegrationInstance") -> str:
+    """Syslog events are categorized _TRANSPORT=syslog from systemd v205."""
+    # Prefer syslog transport categorized messages over presence of
+    # /var/log/syslog as systemd v255 introduced systemd-executor
+    # which sandboxes unit processes resulting in direct writes to
+    # /dev/console being logged directly to journal binary instead
+    # of mirrored as rsyslog events.
+    return client.execute(["journalctl", "_TRANSPORT=syslog", "-b", "0"])
 
 
 @lru_cache()