diff --git a/ceph_devstack/__init__.py b/ceph_devstack/__init__.py
@@ -104,6 +104,14 @@ def parse_args(args: List[str]) -> argparse.Namespace:
         "container",
         help="The container to wait for",
     )
+    parser_log = subparsers.add_parser("logs", help="Dump teuthology logs")
+    parser_log.add_argument("-r", "--run-name", type=str, default=None)
+    parser_log.add_argument("-j", "--job-id", type=str, default=None)
+    parser_log.add_argument(
+        "--locate",
+        action=argparse.BooleanOptionalAction,
+        help="Display log file path instead of contents",
+    )
     return parser.parse_args(args)
 
 

diff --git a/ceph_devstack/cli.py b/ceph_devstack/cli.py
@@ -40,6 +40,10 @@ async def run():
             return
         elif args.command == "wait":
             return await obj.wait(container_name=args.container)
+        elif args.command == "logs":
+            return await obj.logs(
+                run_name=args.run_name, job_id=args.job_id, locate=args.locate
+            )
         else:
             await obj.apply(args.command)
             return 0

diff --git a/ceph_devstack/resources/ceph/__init__.py b/ceph_devstack/resources/ceph/__init__.py
@@ -23,6 +23,8 @@
     LoopControlDeviceWriteable,
     SELinuxModule,
 )
+from ceph_devstack.resources.ceph.utils import get_most_recent_run, get_job_id
+from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
 
 
 class SSHKeyPair(Secret):
@@ -226,3 +228,39 @@ async def wait(self, container_name: str):
                     return await object.wait()
         logger.error(f"Could not find container {container_name}")
         return 1
+
+    async def logs(
+        self, run_name: str = None, job_id: str = None, locate: bool = False
+    ):
+        try:
+            log_file = self.get_log_file(run_name, job_id)
+        except FileNotFoundError:
+            logger.error("No log file found")
+        except TooManyJobsFound as e:
+            msg = "Found too many jobs ({jobs}) for target run. Please pick a job id with -j option.".format(
+                jobs=", ".join(e.jobs)
+            )
+            logger.error(msg)
+        else:
+            if locate:
+                print(log_file)
+            else:
+                buffer_size = 8 * 1024
+                with open(log_file) as f:
+                    while chunk := f.read(buffer_size):
+                        print(chunk, end="")
+
+    def get_log_file(self, run_name: str = None, job_id: str = None):
+        archive_dir = Teuthology().archive_dir.expanduser()
+
+        if not run_name:
+            run_name = get_most_recent_run(os.listdir(archive_dir))
+        run_dir = archive_dir.joinpath(run_name)
+
+        if not job_id:
+            job_id = get_job_id(os.listdir(run_dir))
+
+        log_file = run_dir.joinpath(job_id, "teuthology.log")
+        if not log_file.exists():
+            raise FileNotFoundError
+        return log_file
diff --git a/ceph_devstack/resources/ceph/exceptions.py b/ceph_devstack/resources/ceph/exceptions.py
@@ -0,0 +1,3 @@
+class TooManyJobsFound(Exception):
+    def __init__(self, jobs: list[str]):
+        self.jobs = jobs
diff --git a/ceph_devstack/resources/ceph/utils.py b/ceph_devstack/resources/ceph/utils.py
@@ -0,0 +1,44 @@
+import re
+from datetime import datetime
+
+from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
+
+RUN_DIRNAME_PATTERN = re.compile(
+    r"^(?P<username>^[a-z_]([a-z0-9_-]{0,31}|[a-z0-9_-]{0,30}))-(?P<timestamp>\d{4}-\d{2}-\d{2}_\d{2}:\d{2}:\d{2})"
+)
+
+
+def get_logtimestamp(dirname: str) -> datetime:
+    match_ = RUN_DIRNAME_PATTERN.search(dirname)
+    return datetime.strptime(match_.group("timestamp"), "%Y-%m-%d_%H:%M:%S")
+
+
+def get_most_recent_run(runs: list[str]) -> str:
+    try:
+        run_name = next(
+            iter(
+                sorted(
+                    (
+                        dirname
+                        for dirname in runs
+                        if RUN_DIRNAME_PATTERN.search(dirname)
+                    ),
+                    key=lambda dirname: get_logtimestamp(dirname),
+                    reverse=True,
+                )
+            )
+        )
+        return run_name
+    except StopIteration:
+        raise FileNotFoundError
+
+
+def get_job_id(jobs: list[str]):
+    job_dir_pattern = re.compile(r"^\d+$")
+    dirs = [d for d in jobs if job_dir_pattern.match(d)]
+
+    if len(dirs) == 0:
+        raise FileNotFoundError
+    elif len(dirs) > 1:
+        raise TooManyJobsFound(dirs)
+    return dirs[0]
diff --git a/ceph_devstack/resources/test/test_devstack.py b/ceph_devstack/resources/test/test_devstack.py
@@ -0,0 +1,182 @@
+import os
+import io
+import contextlib
+import random as rd
+from datetime import datetime, timedelta
+import secrets
+import string
+
+import pytest
+
+from ceph_devstack import config
+from ceph_devstack.resources.ceph.utils import (
+    get_logtimestamp,
+    get_most_recent_run,
+    get_job_id,
+)
+from ceph_devstack.resources.ceph.exceptions import TooManyJobsFound
+from ceph_devstack.resources.ceph import CephDevStack
+
+
+class TestDevStack:
+    def test_get_logtimestamp(self):
+        dirname = "root-2025-03-20_18:34:43-orch:cephadm:smoke-small-main-distro-default-testnode"
+        assert get_logtimestamp(dirname) == datetime(2025, 3, 20, 18, 34, 43)
+
+    def test_get_most_recent_run_returns_most_recent_run(self):
+        runs = [
+            "root-2024-02-07_12:23:43-orch:cephadm:smoke-small-devlop-distro-smithi-testnode",
+            "root-2025-02-20_11:23:43-orch:cephadm:smoke-small-devlop-distro-smithi-testnode",
+            "root-2025-03-20_18:34:43-orch:cephadm:smoke-small-main-distro-default-testnode",
+            "root-2025-01-18_18:34:43-orch:cephadm:smoke-small-main-distro-default-testnode",
+        ]
+        assert (
+            get_most_recent_run(runs)
+            == "root-2025-03-20_18:34:43-orch:cephadm:smoke-small-main-distro-default-testnode"
+        )
+
+    def test_get_job_id_returns_job_on_unique_job(self):
+        jobs = ["97"]
+        assert get_job_id(jobs) == "97"
+
+    def test_get_job_id_throws_filenotfound_on_missing_job(self):
+        jobs = []
+        with pytest.raises(FileNotFoundError):
+            get_job_id(jobs)
+
+    def test_get_job_id_throws_toomanyjobsfound_on_more_than_one_job(self):
+        jobs = ["1", "2"]
+        with pytest.raises(TooManyJobsFound) as exc:
+            get_job_id(jobs)
+        assert exc.value.jobs == jobs
+
+    async def test_logs_command_display_log_file_of_latest_run(
+        self, tmp_path, create_log_file
+    ):
+        data_dir = str(tmp_path)
+        config["data_dir"] = data_dir
+        f = io.StringIO()
+        content = "custom log content"
+        now = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+        forty_days_ago = (datetime.now() - timedelta(days=40)).strftime(
+            "%Y-%m-%d_%H:%M:%S"
+        )
+
+        create_log_file(data_dir, timestamp=now, content=content)
+        create_log_file(data_dir, timestamp=forty_days_ago)
+
+        with contextlib.redirect_stdout(f):
+            devstack = CephDevStack()
+            await devstack.logs()
+        assert content in f.getvalue()
+
+    async def test_logs_display_roughly_contents_of_log_file(
+        self, tmp_path, create_log_file
+    ):
+        data_dir = str(tmp_path)
+        config["data_dir"] = data_dir
+        f = io.StringIO()
+        content = "".join(
+            secrets.choice(string.ascii_letters + string.digits)
+            for _ in range(6 * 8 * 1024)
+        )
+        now = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+        create_log_file(data_dir, timestamp=now, content=content)
+
+        with contextlib.redirect_stdout(f):
+            devstack = CephDevStack()
+            await devstack.logs()
+        assert content == f.getvalue()
+
+    async def test_logs_command_display_log_file_of_given_job_id(
+        self, tmp_path, create_log_file
+    ):
+        data_dir = str(tmp_path)
+        config["data_dir"] = data_dir
+        f = io.StringIO()
+        content = "custom log message"
+        now = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+
+        create_log_file(
+            data_dir,
+            timestamp=now,
+            test_type="ceph",
+            job_id="1",
+            content="another log",
+        )
+        create_log_file(
+            data_dir, timestamp=now, test_type="ceph", job_id="2", content=content
+        )
+
+        with contextlib.redirect_stdout(f):
+            devstack = CephDevStack()
+            await devstack.logs(job_id="2")
+        assert content in f.getvalue()
+
+    async def test_logs_display_content_of_provided_run_name(
+        self, tmp_path, create_log_file
+    ):
+        data_dir = str(tmp_path)
+        config["data_dir"] = data_dir
+        f = io.StringIO()
+        content = "custom content"
+        now = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+        three_days_ago = (datetime.now() - timedelta(days=3)).strftime(
+            "%Y-%m-%d_%H:%M:%S"
+        )
+
+        create_log_file(
+            data_dir,
+            timestamp=now,
+        )
+        run_name = create_log_file(
+            data_dir,
+            timestamp=three_days_ago,
+            content=content,
+        ).split("/")[-3]
+
+        with contextlib.redirect_stdout(f):
+            devstack = CephDevStack()
+            await devstack.logs(run_name=run_name)
+        assert content in f.getvalue()
+
+    async def test_logs_locate_display_file_path_instead_of_config(
+        self, tmp_path, create_log_file
+    ):
+        data_dir = str(tmp_path)
+
+        config["data_dir"] = data_dir
+        f = io.StringIO()
+        log_file = create_log_file(data_dir)
+        with contextlib.redirect_stdout(f):
+            devstack = CephDevStack()
+            await devstack.logs(locate=True)
+        assert log_file in f.getvalue()
+
+    @pytest.fixture(scope="class")
+    def create_log_file(self):
+        def _create_log_file(data_dir: str, **kwargs):
+            parts = {
+                "timestamp": (
+                    datetime.now() - timedelta(days=rd.randint(1, 100))
+                ).strftime("%Y-%m-%d_%H:%M:%S"),
+                "test_type": rd.choice(["ceph", "rgw", "rbd", "mds"]),
+                "job_id": rd.randint(1, 100),
+                "content": "some log data",
+                **kwargs,
+            }
+            timestamp = parts["timestamp"]
+            test_type = parts["test_type"]
+            job_id = parts["job_id"]
+            content = parts["content"]
+
+            run_name = f"root-{timestamp}-orch:cephadm:{test_type}-small-main-distro-default-testnode"
+            log_dir = f"{data_dir}/archive/{run_name}/{job_id}"
+
+            os.makedirs(log_dir, exist_ok=True)
+            log_file = f"{log_dir}/teuthology.log"
+            with open(log_file, "w") as f:
+                f.write(content)
+            return log_file
+
+        return _create_log_file