diff --git a/src/deadline_worker_agent/metrics.py b/src/deadline_worker_agent/metrics.py
new file mode 100644
index 00000000..7ee660b6
--- /dev/null
+++ b/src/deadline_worker_agent/metrics.py
@@ -0,0 +1,103 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+from __future__ import annotations
+
+from logging import Logger
+from threading import Timer
+from typing import Any
+
+import os
+import psutil
+
+
+class HostMetricsLogger:
+    """Context manager that regularly logs host metrics"""
+
+    logger: Logger
+    interval_s: float
+    _timer: Timer | None
+    _prev_network: Any | None
+
+    def __init__(self, logger: Logger, interval_s: float) -> None:
+        assert interval_s > 0, "interval_s must be a positive number"
+        self._timer = None
+        self._prev_network = None
+        self.logger = logger
+        self.interval_s = interval_s
+
+    def __enter__(self) -> HostMetricsLogger:
+        self.log_metrics()
+        return self
+
+    def __exit__(self, type, value, traceback) -> None:
+        if self._timer:
+            self._timer.cancel()
+            self._timer = None
+
+    def log_metrics(self):
+        """
+        Queries information about the host machine and logs the information as a space-delimited
+        line of the form: <label> <value> ...
+        """
+        memory = psutil.virtual_memory()
+        swap = psutil.swap_memory()
+        disk = psutil.disk_usage(os.sep)
+
+        # On Windows it may be necessary to issue diskperf -y command from cmd.exe first in order to enable IO counters
+        disk_counters = psutil.disk_io_counters(nowrap=True)
+        if disk_counters is None:
+            disk_read = disk_write = "NOT_AVAILABLE"
+        elif not (hasattr(disk_counters, "read_bytes") and hasattr(disk_counters, "write_bytes")):
+            # TODO: Support disk speed on NetBSD and OpenBSD
+            disk_read = disk_write = "NOT_SUPPORTED"
+        else:
+            disk_read = str(round(disk_counters.read_bytes / self.interval_s))
+            disk_write = str(round(disk_counters.write_bytes / self.interval_s))
+
+        # We need to poll network IO to get rate
+        network = psutil.net_io_counters(nowrap=True)
+        if network is None:
+            network_sent = network_recv = "NOT_AVAILABLE"
+        else:
+            if self._prev_network:
+                network_sent_bps = round(
+                    (network.bytes_sent - self._prev_network.bytes_sent) / self.interval_s
+                )
+                network_recv_bps = round(
+                    (network.bytes_recv - self._prev_network.bytes_recv) / self.interval_s
+                )
+            else:
+                network_sent_bps = network_recv_bps = 0
+            network_sent = str(network_sent_bps)
+            network_recv = str(network_recv_bps)
+        self._prev_network = network
+
+        stats = {
+            "cpu-usage-percent": str(psutil.cpu_percent()),
+            "memory-total-bytes": str(memory.total),
+            "memory-used-bytes": str(memory.total - memory.available),
+            "memory-used-percent": str(memory.percent),
+            "swap-used-bytes": str(swap.used),
+            "total-disk-bytes": str(disk.total),
+            "total-disk-used-bytes": str(disk.used),
+            "total-disk-used-percent": str(round(disk.used / disk.total, ndigits=1)),
+            "user-disk-available-bytes": str(disk.free),
+            "network-sent-bytes-per-second": network_sent,
+            "network-recv-bytes-per-second": network_recv,
+            "disk-read-bytes-per-second": disk_read,
+            "disk-write-bytes-per-second": disk_write,
+        }
+
+        # Output as space-delimited "key value" pairs for consumption by Cloudwatch to use as metrics
+        self.logger.info(" ".join(" ".join(kvp) for kvp in stats.items()))
+        self._set_timer()
+
+    def _set_timer(self) -> None:
+        """
+        Sets the timer to log the host metrics at a regular interval.
+
+        Args:
+            interval_s (float): The interval in seconds to print the host metrics at.
+        """
+        self._timer = Timer(self.interval_s, self.log_metrics)
+        self._timer.start()
diff --git a/src/deadline_worker_agent/startup/cli_args.py b/src/deadline_worker_agent/startup/cli_args.py
index 17b9055c..2cb49d2c 100644
--- a/src/deadline_worker_agent/startup/cli_args.py
+++ b/src/deadline_worker_agent/startup/cli_args.py
@@ -22,6 +22,8 @@ class ParsedCommandLineArguments(Namespace):
     persistence_dir: Path | None = None
     # TODO: Remove when deprecating --no-allow-instance-profile
     no_allow_instance_profile: bool | None = None
+    host_metrics_logging: bool | None = None
+    host_metrics_logging_interval_seconds: float | None = None
 
 
 def get_argument_parser() -> ArgumentParser:
@@ -108,6 +110,20 @@ def get_argument_parser() -> ArgumentParser:
         dest="allow_instance_profile",
         default=None,
     )
+    parser.add_argument(
+        "--host-metrics-logging-interval-seconds",
+        help="The interval between host metrics log messages. Default is 60.",
+        default=None,
+        type=float,
+    )
+    parser.add_argument(
+        "--no-host-metrics-logging",
+        help="Turn off host metrics logging. Default is on.",
+        action="store_const",
+        dest="host_metrics_logging",
+        const=False,
+        default=None,
+    )
     parser.add_argument(
         "--verbose",
         "-v",
diff --git a/src/deadline_worker_agent/startup/config.py b/src/deadline_worker_agent/startup/config.py
index 124e23c4..4b7c90d7 100644
--- a/src/deadline_worker_agent/startup/config.py
+++ b/src/deadline_worker_agent/startup/config.py
@@ -65,6 +65,10 @@ class Configuration:
     """Path to the directory where the Worker Agent writes its logs."""
     local_session_logs: bool
     """Whether to write session logs to the local filesystem"""
+    host_metrics_logging: bool
+    """Whether host metrics logging is enabled"""
+    host_metrics_logging_interval_seconds: float
+    """The interval in seconds between host metrics logs"""
 
     # Used to optimize the memory allocation and attribute lookup speed. Tells python to not create a dict
     # for the attributes.
@@ -83,6 +87,8 @@ class Configuration:
         "worker_state_file",
         "worker_logs_dir",
         "local_session_logs",
+        "host_metrics_logging",
+        "host_metrics_logging_interval_seconds",
     )
 
     def __init__(
@@ -116,6 +122,12 @@ def __init__(
             settings_kwargs["worker_persistence_dir"] = parsed_cli_args.persistence_dir.absolute()
         if parsed_cli_args.local_session_logs is not None:
             settings_kwargs["local_session_logs"] = parsed_cli_args.local_session_logs
+        if parsed_cli_args.host_metrics_logging is not None:
+            settings_kwargs["host_metrics_logging"] = parsed_cli_args.host_metrics_logging
+        if parsed_cli_args.host_metrics_logging_interval_seconds is not None:
+            settings_kwargs[
+                "host_metrics_logging_interval_seconds"
+            ] = parsed_cli_args.host_metrics_logging_interval_seconds
 
         settings = WorkerSettings(**settings_kwargs)
 
@@ -143,6 +155,8 @@ def __init__(
         self.capabilities = settings.capabilities
         self.worker_logs_dir = settings.worker_logs_dir
         self.local_session_logs = settings.local_session_logs
+        self.host_metrics_logging = settings.host_metrics_logging
+        self.host_metrics_logging_interval_seconds = settings.host_metrics_logging_interval_seconds
 
         self._validate()
 
@@ -164,6 +178,11 @@ def _validate(self) -> None:
         if self.impersonation.inactive and self.impersonation.posix_job_user:
             raise ConfigurationError("Cannot specify an impersonation user with impersonation off")
 
+        if self.host_metrics_logging_interval_seconds <= 0:
+            raise ConfigurationError(
+                f"Host metrics logging interval must be a positive number, but got: {repr(self.host_metrics_logging_interval_seconds)}"
+            )
+
     def log(self, logger: Optional[_logging.Logger] = None, level: int = _logging.DEBUG) -> None:
         """Emit logs that represent the effective Configuration.
 
diff --git a/src/deadline_worker_agent/startup/config_file.py b/src/deadline_worker_agent/startup/config_file.py
index 3ca74a39..99c2df69 100644
--- a/src/deadline_worker_agent/startup/config_file.py
+++ b/src/deadline_worker_agent/startup/config_file.py
@@ -39,6 +39,8 @@ class LoggingConfigSection(BaseModel):
     verbose: Optional[bool] = None
     worker_logs_dir: Optional[Path] = None
     local_session_logs: Optional[bool] = None
+    host_metrics_logging: Optional[bool] = None
+    host_metrics_logging_interval_seconds: Optional[float] = None
 
 
 class OsConfigSection(BaseModel):
@@ -107,6 +109,12 @@ def as_settings(
             output_settings["worker_logs_dir"] = self.logging.worker_logs_dir
         if self.logging.local_session_logs is not None:
             output_settings["local_session_logs"] = self.logging.local_session_logs
+        if self.logging.host_metrics_logging is not None:
+            output_settings["host_metrics_logging"] = self.logging.host_metrics_logging
+        if self.logging.host_metrics_logging_interval_seconds is not None:
+            output_settings[
+                "host_metrics_logging_interval_seconds"
+            ] = self.logging.host_metrics_logging_interval_seconds
         if self.os.shutdown_on_stop is not None:
             output_settings["no_shutdown"] = self.os.shutdown_on_stop
         if self.os.impersonation is not None:
diff --git a/src/deadline_worker_agent/startup/entrypoint.py b/src/deadline_worker_agent/startup/entrypoint.py
index 8936bc93..d0bcb79c 100644
--- a/src/deadline_worker_agent/startup/entrypoint.py
+++ b/src/deadline_worker_agent/startup/entrypoint.py
@@ -160,6 +160,8 @@ def filter(self, record: logging.LogRecord) -> bool:
                 cleanup_session_user_processes=config.cleanup_session_user_processes,
                 worker_persistence_dir=config.worker_persistence_dir,
                 worker_logs_dir=config.worker_logs_dir if config.local_session_logs else None,
+                host_metrics_logging=config.host_metrics_logging,
+                host_metrics_logging_interval_seconds=config.host_metrics_logging_interval_seconds,
             )
             try:
                 worker_sessions.run()
@@ -336,7 +338,7 @@ def _log_agent_info() -> None:
     _logger.info(f"Platform: {sys.platform}")
 
 
-def _get_gpu_count() -> int:
+def _get_gpu_count(*, verbose: bool = True) -> int:
     """
     Get the number of GPUs available on the machine.
 
@@ -350,17 +352,20 @@ def _get_gpu_count() -> int:
             ["nvidia-smi", "--query-gpu=count", "--format=csv,noheader"]
         )
     except FileNotFoundError:
-        _logger.warning("Could not detect GPU count, nvidia-smi not found")
+        if verbose:
+            _logger.warning("Could not detect GPU count, nvidia-smi not found")
         return 0
     except subprocess.CalledProcessError:
-        _logger.warning("Could not detect GPU count, error running nvidia-smi")
+        if verbose:
+            _logger.warning("Could not detect GPU count, error running nvidia-smi")
         return 0
     else:
-        _logger.info("Number of GPUs: %s", output.decode().strip())
+        if verbose:
+            _logger.info("Number of GPUs: %s", output.decode().strip())
         return int(output.decode().strip())
 
 
-def _get_gpu_memory() -> int:
+def _get_gpu_memory(*, verbose: bool = True) -> int:
     """
     Get the total GPU memory available on the machine.
 
@@ -374,11 +379,14 @@ def _get_gpu_memory() -> int:
             ["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader"]
         )
     except FileNotFoundError:
-        _logger.warning("Could not detect GPU memory, nvidia-smi not found")
+        if verbose:
+            _logger.warning("Could not detect GPU memory, nvidia-smi not found")
         return 0
     except subprocess.CalledProcessError:
-        _logger.warning("Could not detect GPU memory, error running nvidia-smi")
+        if verbose:
+            _logger.warning("Could not detect GPU memory, error running nvidia-smi")
         return 0
     else:
-        _logger.info("Total GPU Memory: %s", output.decode().strip())
+        if verbose:
+            _logger.info("Total GPU Memory: %s", output.decode().strip())
         return int(output.decode().strip().replace("MiB", ""))
diff --git a/src/deadline_worker_agent/startup/settings.py b/src/deadline_worker_agent/startup/settings.py
index 98b5087b..8032060e 100644
--- a/src/deadline_worker_agent/startup/settings.py
+++ b/src/deadline_worker_agent/startup/settings.py
@@ -64,6 +64,10 @@ class WorkerSettings(BaseSettings):
         The path to the directory where the Worker Agent persists its state.
     local_session_logs : bool
         Whether to write session logs to the local filesystem
+    host_metrics_logging : bool
+        Whether to log host metrics
+    host_metrics_logging_interval_seconds : float
+        The interval between host metrics log messages
     """
 
     farm_id: str = Field(regex=r"^farm-[a-z0-9]{32}$")
@@ -83,6 +87,8 @@ class WorkerSettings(BaseSettings):
     worker_logs_dir: Path = DEFAULT_POSIX_WORKER_LOGS_DIR
     worker_persistence_dir: Path = DEFAULT_POSIX_WORKER_PERSISTENCE_DIR
     local_session_logs: bool = True
+    host_metrics_logging: bool = True
+    host_metrics_logging_interval_seconds: float = 60
 
     class Config:
         fields = {
@@ -101,6 +107,10 @@ class Config:
             "worker_logs_dir": {"env": "DEADLINE_WORKER_LOGS_DIR"},
             "worker_persistence_dir": {"env": "DEADLINE_WORKER_PERSISTENCE_DIR"},
             "local_session_logs": {"env": "DEADLINE_WORKER_LOCAL_SESSION_LOGS"},
+            "host_metrics_logging": {"env": "DEADLINE_WORKER_HOST_METRICS_LOGGING"},
+            "host_metrics_logging_interval_seconds": {
+                "env": "DEADLINE_WORKER_HOST_METRICS_LOGGING_INTERVAL_SECONDS"
+            },
         }
 
         @classmethod
diff --git a/src/deadline_worker_agent/worker.py b/src/deadline_worker_agent/worker.py
index acfd3b32..00388580 100644
--- a/src/deadline_worker_agent/worker.py
+++ b/src/deadline_worker_agent/worker.py
@@ -6,6 +6,7 @@
 import signal
 import sys
 import traceback
+from contextlib import nullcontext
 from concurrent.futures import Executor, Future, ThreadPoolExecutor, wait
 from datetime import datetime, timedelta, timezone
 from logging import getLogger
@@ -19,6 +20,7 @@
 
 from .boto import DeadlineClient
 from .errors import ServiceShutdown
+from .metrics import HostMetricsLogger
 from .scheduler import WorkerScheduler
 from .sessions import Session
 from .startup.config import ImpersonationOverrides
@@ -68,6 +70,7 @@ class Worker:
     _logs_client: boto3.client
     _boto_session: WorkerBoto3Session
     _worker_persistence_dir: Path
+    _host_metrics_logger: HostMetricsLogger | None = None
 
     def __init__(
         self,
@@ -83,6 +86,8 @@ def __init__(
         cleanup_session_user_processes: bool,
         worker_persistence_dir: Path,
         worker_logs_dir: Path | None,
+        host_metrics_logging: bool,
+        host_metrics_logging_interval_seconds: float | None,
     ) -> None:
         self._deadline_client = deadline_client
         self._s3_client = s3_client
@@ -106,6 +111,14 @@ def __init__(
         self._boto_session = boto_session
         self._worker_persistence_dir = worker_persistence_dir
 
+        if host_metrics_logging:
+            assert (
+                host_metrics_logging_interval_seconds is not None
+            ), "host_metrics_logging_interval_seconds is required if host metrics logging is enabled"
+            self._host_metrics_logger = HostMetricsLogger(
+                logger=logger, interval_s=host_metrics_logging_interval_seconds
+            )
+
         signal.signal(signal.SIGTERM, self._signal_handler)
         signal.signal(signal.SIGINT, self._signal_handler)
         # TODO: Remove this once WA is stable or put behind a debug flag
@@ -162,10 +175,14 @@ def run(self) -> None:
         """Runs the main Worker loop for processing sessions."""
 
         monitor_ec2_shutdown: Future[WorkerShutdown | None] | None = None
-        with self._executor, AwsCredentialsRefresher(
-            identifier="Worker Agent",
-            session=self._boto_session,
-            failure_callback=self._aws_credentials_refresh_failure,
+        with (
+            self._executor,
+            AwsCredentialsRefresher(
+                identifier="Worker Agent",
+                session=self._boto_session,
+                failure_callback=self._aws_credentials_refresh_failure,
+            ),
+            self._host_metrics_logger or nullcontext(),
         ):
             scheduler_future = self._executor.submit(self._scheduler.run)
             futures: list[Future[Any]] = [
diff --git a/test/unit/startup/test_config.py b/test/unit/startup/test_config.py
index cd3a775d..6ddb7fc4 100644
--- a/test/unit/startup/test_config.py
+++ b/test/unit/startup/test_config.py
@@ -35,6 +35,8 @@ def mock_worker_settings_cls() -> Generator[MagicMock, None, None]:
         "worker_logs_dir": Path("/var/log/amazon/deadline"),
         "worker_persistence_dir": Path("/var/lib/deadline"),
         "local_session_logs": None,
+        "host_metrics_logging": True,
+        "host_metrics_logging_interval_seconds": 10,
     }
 
     class FakeWorkerSettings:
@@ -799,10 +801,9 @@ def test_uses_worker_settings(
         mock_worker_settings: MagicMock = mock_worker_settings_cls.return_value
         mock_worker_settings.posix_job_user = posix_job_user_setting
 
-        # with patch.object(
-        #     config_mod.Configuration, "_get_user_and_group_from_posix_job_user",
-        #     return_value=posix_job_user.split(":") if posix_job_user else None,
-        # ) as mock:
+        # Needed because MagicMock does not support gt/lt comparison
+        mock_worker_settings.host_metrics_logging_interval_seconds = 10
+
         # WHEN
         config = config_mod.Configuration(parsed_cli_args=parsed_args)
 
@@ -833,6 +834,11 @@ def test_uses_worker_settings(
             config.worker_credentials_dir
             is mock_worker_settings.worker_persistence_dir / "credentials"
         )
+        assert config.host_metrics_logging is mock_worker_settings.host_metrics_logging
+        assert (
+            config.host_metrics_logging_interval_seconds
+            is mock_worker_settings.host_metrics_logging_interval_seconds
+        )
 
 
 class TestLog:
diff --git a/test/unit/startup/test_config_file.py b/test/unit/startup/test_config_file.py
index 76a6a932..c8aab5f8 100644
--- a/test/unit/startup/test_config_file.py
+++ b/test/unit/startup/test_config_file.py
@@ -608,6 +608,7 @@ def when() -> ConfigFile:
 verbose = true
 worker_logs_dir = "/var/log/amazon/deadline"
 local_session_logs = false
+host_metrics_logging_interval_seconds = 1
 
 [os]
 impersonation = false
@@ -679,6 +680,7 @@ def test_config_load_full_toml(
         assert config.logging.verbose is True
         assert config.logging.worker_logs_dir == Path("/var/log/amazon/deadline")
         assert config.logging.local_session_logs is False
+        assert config.logging.host_metrics_logging_interval_seconds == 1
 
         assert config.os.impersonation is False
         assert config.os.posix_job_user == "user:group"
diff --git a/test/unit/startup/test_entrypoint.py b/test/unit/startup/test_entrypoint.py
index df7dc597..dc4e2479 100644
--- a/test/unit/startup/test_entrypoint.py
+++ b/test/unit/startup/test_entrypoint.py
@@ -126,6 +126,8 @@ def configuration() -> MagicMock:
     config.fleet_id = "fleet-456"
     config.profile = None
     config.sessions = True
+    # Required because MagicMock does not support int comparison
+    config.host_metrics_logging_interval_seconds = 10
     return config
 
 
@@ -277,6 +279,8 @@ def test_log_configuration(
         _config_mock.load().worker_credentials_dir = Path(tempdir) / "credentials"
         _config_mock.load().worker_logs_dir.mkdir()
         _config_mock.load().worker_credentials_dir.mkdir()
+        # Required because MagicMock does not support int comparison
+        _config_mock.load().host_metrics_logging_interval_seconds = 10
 
         # Mock logging.getLogger
         root_logger = MagicMock()
@@ -595,6 +599,8 @@ def test_passes_worker_logs_dir(
         cleanup_session_user_processes=ANY,
         worker_persistence_dir=ANY,
         worker_logs_dir=tmp_path,
+        host_metrics_logging=ANY,
+        host_metrics_logging_interval_seconds=ANY,
     )
 
 
diff --git a/test/unit/startup/test_settings.py b/test/unit/startup/test_settings.py
index 3bff5847..1f4c8e57 100644
--- a/test/unit/startup/test_settings.py
+++ b/test/unit/startup/test_settings.py
@@ -122,6 +122,20 @@ class FieldTestCaseParams(NamedTuple):
         expected_default=True,
         expected_default_factory_return_value=None,
     ),
+    FieldTestCaseParams(
+        field_name="host_metrics_logging",
+        expected_type=bool,
+        expected_required=False,
+        expected_default=True,
+        expected_default_factory_return_value=None,
+    ),
+    FieldTestCaseParams(
+        field_name="host_metrics_logging_interval_seconds",
+        expected_type=float,
+        expected_required=False,
+        expected_default=60,
+        expected_default_factory_return_value=None,
+    ),
 ]
 
 
diff --git a/test/unit/test_metrics.py b/test/unit/test_metrics.py
new file mode 100644
index 00000000..5e9e865d
--- /dev/null
+++ b/test/unit/test_metrics.py
@@ -0,0 +1,321 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+from __future__ import annotations
+from collections import namedtuple
+
+import logging
+from typing import Any, Generator
+from unittest.mock import MagicMock, patch
+
+import pytest
+import re
+
+from deadline_worker_agent.metrics import HostMetricsLogger
+import deadline_worker_agent.metrics as metrics_mod
+
+
+class TestHostMetricsLogger:
+    BYTES_PATTERN = r"[0-9]+(?:\.[0-9]+)?"
+    PERCENT_PATTERN = r"[0-9]{1,3}(?:\.[0-9]+)?"
+
+    @pytest.fixture
+    def logger(self) -> MagicMock:
+        return MagicMock()
+
+    @pytest.fixture
+    def host_metrics_logger(self, logger: MagicMock) -> HostMetricsLogger:
+        return HostMetricsLogger(logger=logger, interval_s=1)
+
+    def test_enter(self, host_metrics_logger: HostMetricsLogger):
+        # GIVEN
+        with patch.object(host_metrics_logger, "_set_timer") as mock_set_timer:
+            # WHEN
+            with host_metrics_logger:
+                # THEN
+                mock_set_timer.assert_called_once()
+
+    @pytest.mark.parametrize("timer_exists", [True, False])
+    def test_exit(
+        self,
+        timer_exists: bool,
+        host_metrics_logger: HostMetricsLogger,
+    ):
+        # GIVEN
+        timer = MagicMock()
+
+        # WHEN
+        with host_metrics_logger:
+            if timer_exists:
+                host_metrics_logger._timer = timer
+
+        # THEN
+        if timer_exists:
+            timer.cancel.assert_called_once()
+            assert host_metrics_logger._timer is None
+        else:
+            timer.cancel.assert_not_called()
+
+    def test_set_timer(self, host_metrics_logger: HostMetricsLogger):
+        # GIVEN
+        with patch.object(metrics_mod, "Timer") as mock_timer_cls:
+            # WHEN
+            host_metrics_logger._set_timer()
+
+        # THEN
+        mock_timer_cls.assert_called_once_with(
+            host_metrics_logger.interval_s, host_metrics_logger.log_metrics
+        )
+        mock_timer_cls.return_value.start.assert_called_once()
+
+    def test_log_metrics_sets_timer(
+        self,
+        host_metrics_logger: HostMetricsLogger,
+    ):
+        # GIVEN
+        with (
+            patch.object(metrics_mod, "psutil"),
+            patch.object(host_metrics_logger, "_set_timer") as mock_set_timer,
+        ):
+            # WHEN
+            host_metrics_logger.log_metrics()
+
+        # THEN
+        mock_set_timer.assert_called_once()
+
+    class TestLogMetrics:
+        @pytest.fixture(autouse=True)
+        def mock_timer(self) -> Generator[MagicMock, None, None]:
+            # We don't want to actually create/start a timer
+            with patch.object(metrics_mod, "Timer") as mock:
+                yield mock
+
+        @pytest.fixture
+        def virtual_memory(self) -> tuple:
+            vm = namedtuple("vm", ["total", "available", "percent", "used", "free"])
+            return vm(40, 10, 25, 30, 10)
+
+        @pytest.fixture
+        def swap_memory(self) -> tuple:
+            sm = namedtuple("sm", ["total", "used", "free", "percent", "sin", "sout"])
+            return sm(20, 10, 10, 50, 0, 0)
+
+        @pytest.fixture
+        def disk_usage(self) -> tuple:
+            du = namedtuple("du", ["total", "used", "free", "percent"])
+            return du(100, 25, 75, 25)
+
+        @pytest.fixture
+        def cpu_percent(self) -> int:
+            return 10
+
+        @pytest.fixture
+        def net_io_counters(self) -> tuple:
+            nioc = namedtuple(
+                "nioc",
+                [
+                    "bytes_sent",
+                    "bytes_recv",
+                    "packets_sent",
+                    "packets_recv",
+                    "errin",
+                    "errout",
+                    "dropin",
+                    "dropout",
+                ],
+            )
+            return nioc(123, 321, 100, 300, 2, 3, 1, 0)
+
+        @pytest.fixture
+        def disk_io_counters(self) -> tuple:
+            dioc = namedtuple(
+                "dioc",
+                [
+                    "read_count",
+                    "write_count",
+                    "read_bytes",
+                    "write_bytes",
+                    "read_time",
+                    "write_time",
+                ],
+            )
+            return dioc(123, 321, 123123, 321321, 100, 200)
+
+        @pytest.fixture
+        def mock_psutil(
+            self,
+            virtual_memory: tuple,
+            swap_memory: tuple,
+            disk_usage: tuple,
+            cpu_percent: int,
+            net_io_counters: tuple,
+            disk_io_counters: tuple,
+        ) -> Generator[MagicMock, None, None]:
+            with patch.object(metrics_mod, "psutil") as mock:
+                mock.virtual_memory.return_value = virtual_memory
+                mock.swap_memory.return_value = swap_memory
+                mock.disk_usage.return_value = disk_usage
+                mock.cpu_percent.return_value = cpu_percent
+                mock.net_io_counters.return_value = net_io_counters
+                mock.disk_io_counters.return_value = disk_io_counters
+
+                yield mock
+
+        @pytest.fixture
+        def log_metrics(
+            self,
+            host_metrics_logger: HostMetricsLogger,
+            mock_psutil: MagicMock,
+        ) -> None:
+            with patch.object(host_metrics_logger, "_set_timer"):
+                host_metrics_logger.log_metrics()
+
+        @pytest.fixture
+        def log_line(self, logger: MagicMock, log_metrics: None) -> str:
+            return get_first_and_only_call_arg(logger.info)
+
+        def test_logs_cpu(self, log_line: str):
+            # THEN
+            assert re.search(
+                rf"cpu-usage-percent {TestHostMetricsLogger.PERCENT_PATTERN}", log_line
+            )
+
+        def test_logs_memory(self, log_line: str):
+            # THEN
+            assert re.search(rf"memory-total-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line)
+            assert re.search(rf"memory-used-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line)
+            assert re.search(
+                rf"memory-used-percent {TestHostMetricsLogger.PERCENT_PATTERN}", log_line
+            )
+
+        def test_logs_swap(self, log_line: str):
+            # THEN
+            assert re.search(rf"swap-used-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line)
+
+        def test_logs_disk(self, log_line: str):
+            # THEN
+            assert re.search(rf"total-disk-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line)
+            assert re.search(
+                rf"total-disk-used-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line
+            )
+            assert re.search(
+                rf"total-disk-used-percent {TestHostMetricsLogger.PERCENT_PATTERN}", log_line
+            )
+            assert re.search(
+                rf"user-disk-available-bytes {TestHostMetricsLogger.BYTES_PATTERN}", log_line
+            )
+
+        def test_logs_disk_rate(self, log_line: str):
+            # THEN
+            assert re.search(
+                rf"disk-read-bytes-per-second {TestHostMetricsLogger.BYTES_PATTERN}", log_line
+            )
+            assert re.search(
+                rf"disk-write-bytes-per-second {TestHostMetricsLogger.BYTES_PATTERN}", log_line
+            )
+
+        def test_logs_network_rate(self, log_line: str):
+            # THEN
+            assert re.search(
+                rf"network-sent-bytes-per-second {TestHostMetricsLogger.BYTES_PATTERN}",
+                log_line,
+            )
+            assert re.search(
+                rf"network-recv-bytes-per-second {TestHostMetricsLogger.BYTES_PATTERN}",
+                log_line,
+            )
+
+        def test_disk_rate_not_available(
+            self,
+            mock_psutil: MagicMock,
+            host_metrics_logger: HostMetricsLogger,
+            logger: MagicMock,
+        ):
+            # GIVEN
+            mock_psutil.disk_io_counters.reset_mock()
+            mock_psutil.disk_io_counters.return_value = None
+
+            # WHEN
+            host_metrics_logger.log_metrics()
+
+            # THEN
+            log_line = get_first_and_only_call_arg(logger.info)
+            assert re.search(r"disk-read-bytes-per-second NOT_AVAILABLE", log_line)
+            assert re.search(r"disk-write-bytes-per-second NOT_AVAILABLE", log_line)
+
+        def test_disk_rate_not_supported(
+            self,
+            mock_psutil: MagicMock,
+            host_metrics_logger: HostMetricsLogger,
+            logger: MagicMock,
+        ):
+            # GIVEN
+            mock_psutil.disk_io_counters.return_value = tuple()
+
+            # WHEN
+            host_metrics_logger.log_metrics()
+
+            # THEN
+            log_line = get_first_and_only_call_arg(logger.info)
+            assert re.search(r"disk-read-bytes-per-second NOT_SUPPORTED", log_line)
+            assert re.search(r"disk-write-bytes-per-second NOT_SUPPORTED", log_line)
+
+        def test_network_rate_not_available(
+            self,
+            mock_psutil: MagicMock,
+            host_metrics_logger: HostMetricsLogger,
+            logger: MagicMock,
+        ):
+            # GIVEN
+            mock_psutil.net_io_counters.return_value = None
+
+            # WHEN
+            host_metrics_logger.log_metrics()
+
+            # THEN
+            log_line = get_first_and_only_call_arg(logger.info)
+            assert re.search(r"network-sent-bytes-per-second NOT_AVAILABLE", log_line)
+            assert re.search(r"network-recv-bytes-per-second NOT_AVAILABLE", log_line)
+
+    def test_log_metrics_correct_encoding(self, caplog: pytest.LogCaptureFixture) -> None:
+        # GIVEN
+        DECIMAL_NUMBER_PATTERN = r"\d+(?:\.\d+)?"
+        EXPECTED_LOG_MESSAGE_PATTERN = " ".join(
+            # fmt: off
+            [
+                "cpu-usage-percent", DECIMAL_NUMBER_PATTERN,
+                "memory-total-bytes", DECIMAL_NUMBER_PATTERN,
+                "memory-used-bytes", DECIMAL_NUMBER_PATTERN,
+                "memory-used-percent", DECIMAL_NUMBER_PATTERN,
+                "swap-used-bytes", DECIMAL_NUMBER_PATTERN,
+                "total-disk-bytes", DECIMAL_NUMBER_PATTERN,
+                "total-disk-used-bytes", DECIMAL_NUMBER_PATTERN,
+                "total-disk-used-percent", DECIMAL_NUMBER_PATTERN,
+                "user-disk-available-bytes", DECIMAL_NUMBER_PATTERN,
+                "network-sent-bytes-per-second", rf"(?:{DECIMAL_NUMBER_PATTERN}|NOT_AVAILABLE)",
+                "network-recv-bytes-per-second", rf"(?:{DECIMAL_NUMBER_PATTERN}|NOT_AVAILABLE)",
+                "disk-read-bytes-per-second", rf"(?:{DECIMAL_NUMBER_PATTERN}|NOT_AVAILABLE|NOT_SUPPORTED)",
+                "disk-write-bytes-per-second", rf"(?:{DECIMAL_NUMBER_PATTERN}|NOT_AVAILABLE|NOT_SUPPORTED)",
+            ]
+            # fmt: on
+        )
+        logger = logging.getLogger(__name__)
+        caplog.set_level(0, logger.name)
+        host_metrics_logger = HostMetricsLogger(logger=logger, interval_s=1)
+
+        # WHEN
+        with (
+            # We don't want to actually create/start a timer
+            patch.object(metrics_mod, "Timer"),
+        ):
+            host_metrics_logger.log_metrics()
+
+        # THEN
+        assert len(caplog.messages) == 1
+        assert re.match(EXPECTED_LOG_MESSAGE_PATTERN, caplog.messages[0])
+
+
+def get_first_and_only_call_arg(mock: MagicMock) -> Any:
+    assert len(mock.mock_calls) == 1
+    mock_call = mock.mock_calls[0]
+    assert len(mock_call.args) == 1
+    return mock_call.args[0]
diff --git a/test/unit/test_worker.py b/test/unit/test_worker.py
index a9a43441..3c2fa921 100644
--- a/test/unit/test_worker.py
+++ b/test/unit/test_worker.py
@@ -69,6 +69,8 @@ def worker(
         cleanup_session_user_processes=True,
         worker_persistence_dir=Path("/var/lib/deadline"),
         worker_logs_dir=worker_logs_dir,
+        host_metrics_logging=True,
+        host_metrics_logging_interval_seconds=60,
     )