diff --git a/sentry_sdk/integrations/arq.py b/sentry_sdk/integrations/arq.py
index 9d6497f03a..f26ebe3b13 100644
--- a/sentry_sdk/integrations/arq.py
+++ b/sentry_sdk/integrations/arq.py
@@ -1,7 +1,10 @@
 import sys
+import time
+from datetime import timedelta, timezone
 
 import sentry_sdk
 from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS
+from sentry_sdk.crons import monitor
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.scope import should_send_default_pii
@@ -15,6 +18,8 @@
     ensure_integration_enabled,
     event_from_exception,
     has_data_collection_enabled,
+    logger,
+    match_regex_list,
     parse_version,
     reraise,
 )
@@ -22,6 +27,7 @@
 try:
     import arq.worker
     from arq.connections import ArqRedis
+    from arq.typing import WEEKDAYS
     from arq.version import VERSION as ARQ_VERSION
     from arq.worker import JobExecutionFailed, Retry, RetryJob, Worker
 except ImportError:
@@ -30,14 +36,14 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from typing import Any, Dict, Optional, Union
+    from typing import Any, Dict, List, Optional, Union
 
     from arq.cron import CronJob
     from arq.jobs import Job
     from arq.typing import WorkerCoroutine
     from arq.worker import Function
 
-    from sentry_sdk._types import Event, EventProcessor, ExcInfo, Hint
+    from sentry_sdk._types import Event, EventProcessor, ExcInfo, Hint, MonitorConfig
 
 ARQ_CONTROL_FLOW_EXCEPTIONS = (JobExecutionFailed, Retry, RetryJob)
 
@@ -46,6 +52,14 @@ class ArqIntegration(Integration):
     identifier = "arq"
     origin = f"auto.queue.{identifier}"
 
+    def __init__(
+        self,
+        monitor_cron_jobs: bool = False,
+        exclude_cron_jobs: "Optional[List[str]]" = None,
+    ) -> None:
+        self.monitor_cron_jobs = monitor_cron_jobs
+        self.exclude_cron_jobs = exclude_cron_jobs
+
     @staticmethod
     def setup_once() -> None:
         try:
@@ -278,7 +292,15 @@ def _sentry_create_worker(*args: "Any", **kwargs: "Any") -> "Worker":
                 _get_arq_cron_job(cron_job) for cron_job in kwargs.get("cron_jobs", [])
             ]
 
-        return old_create_worker(*args, **kwargs)
+        worker = old_create_worker(*args, **kwargs)
+
+        integration = sentry_sdk.get_client().get_integration(ArqIntegration)
+        if integration is not None and integration.monitor_cron_jobs:
+            for cron_job in worker.cron_jobs:
+                with capture_internal_exceptions():
+                    _monitor_cron_job(integration, cron_job, worker)
+
+        return worker
 
     arq.worker.create_worker = _sentry_create_worker
 
@@ -294,3 +316,114 @@ def _get_arq_cron_job(cron_job: "CronJob") -> "CronJob":
     cron_job.coroutine = _wrap_coroutine(cron_job.name, cron_job.coroutine)
 
     return cron_job
+
+
+def _monitor_cron_job(
+    integration: "ArqIntegration", cron_job: "CronJob", worker: "Worker"
+) -> None:
+    # Each create_worker call wraps the coroutine again.
+    if getattr(cron_job, "_sentry_monitored", False):
+        return
+    cron_job._sentry_monitored = True
+
+    # arq names cron jobs "cron:<function name>" by default.
+    monitor_slug = cron_job.name
+    if monitor_slug.startswith("cron:"):
+        monitor_slug = monitor_slug[len("cron:") :]
+    if match_regex_list(monitor_slug, integration.exclude_cron_jobs):
+        return
+
+    monitor_config = _get_monitor_config(cron_job, worker)
+    if monitor_config is None:
+        logger.debug(
+            "arq cron job %r can't be expressed as a Sentry Crons schedule. "
+            "Sending check-ins without a monitor config.",
+            cron_job.name,
+        )
+
+    coroutine = cron_job.coroutine
+
+    async def _sentry_cron_coroutine(
+        ctx: "Dict[Any, Any]", *args: "Any", **kwargs: "Any"
+    ) -> "Any":
+        with monitor(monitor_slug=monitor_slug, monitor_config=monitor_config):
+            return await coroutine(ctx, *args, **kwargs)
+
+    cron_job.coroutine = _sentry_cron_coroutine
+
+
+def _get_crontab_field(value: "Any", offset: int = 0) -> "Optional[str]":
+    if value is None:
+        return "*"
+    if isinstance(value, int) and not isinstance(value, bool):
+        value = [value]
+    if not isinstance(value, (set, list, tuple)) or not value:
+        return None
+    if not all(isinstance(v, int) and not isinstance(v, bool) for v in value):
+        return None
+    if offset:
+        value = {(v + offset) % 7 for v in value}
+    return ",".join(str(v) for v in sorted(value))
+
+
+def _get_crontab(cron_job: "CronJob") -> "Optional[str]":
+    # Sentry crontabs have minute resolution.
+    if cron_job.second not in (0, {0}, [0], (0,)):
+        return None
+
+    weekday = cron_job.weekday
+    if isinstance(weekday, str):
+        weekday = WEEKDAYS.index(weekday.lower())
+
+    # arq requires both to match, cron requires either.
+    if cron_job.day is not None and weekday is not None:
+        return None
+
+    fields = [
+        _get_crontab_field(cron_job.minute),
+        _get_crontab_field(cron_job.hour),
+        _get_crontab_field(cron_job.day),
+        _get_crontab_field(cron_job.month),
+        # arq counts weekdays from Monday = 0, cron from Sunday = 0.
+        _get_crontab_field(weekday, offset=1),
+    ]
+    if None in fields:
+        return None
+    return " ".join(fields)  # type: ignore[arg-type]
+
+
+def _get_timezone_name(worker: "Worker") -> "Optional[str]":
+    tz = getattr(worker, "timezone", None)
+    if tz is None:
+        # Before arq 0.25, cron jobs ran in naive local time.
+        if time.timezone == 0 and (not time.daylight or time.altzone == 0):
+            return "UTC"
+        return None
+
+    # zoneinfo and pytz
+    name = getattr(tz, "key", None) or getattr(tz, "zone", None)
+    if isinstance(name, str):
+        return name
+
+    # arq defaults to the system's fixed UTC offset at startup.
+    if isinstance(tz, timezone):
+        offset = tz.utcoffset(None)
+        if offset % timedelta(hours=1):
+            return None
+        hours = offset // timedelta(hours=1)
+        return "Etc/GMT%+d" % -hours if hours else "UTC"
+
+    return None
+
+
+def _get_monitor_config(
+    cron_job: "CronJob", worker: "Worker"
+) -> "Optional[MonitorConfig]":
+    crontab = _get_crontab(cron_job)
+    timezone_name = _get_timezone_name(worker)
+    if crontab is None or timezone_name is None:
+        return None
+    return {
+        "schedule": {"type": "crontab", "value": crontab},
+        "timezone": timezone_name,
+    }
diff --git a/tests/integrations/arq/test_arq_crons.py b/tests/integrations/arq/test_arq_crons.py
new file mode 100644
index 0000000000..0ef47a2ac4
--- /dev/null
+++ b/tests/integrations/arq/test_arq_crons.py
@@ -0,0 +1,252 @@
+import time
+from datetime import timedelta, timezone
+
+import arq.worker
+import pytest
+from arq import cron
+from arq.connections import ArqRedis
+from fakeredis.aioredis import FakeRedis
+
+from sentry_sdk.integrations.arq import ArqIntegration, _get_monitor_config
+
+try:
+    from zoneinfo import ZoneInfo
+except ImportError:
+    ZoneInfo = None
+
+
+@pytest.fixture(autouse=True)
+def patch_fakeredis_info_command():
+    from fakeredis._fakesocket import FakeSocket
+
+    if not hasattr(FakeSocket, "info"):
+        from fakeredis._commands import command
+        from fakeredis._helpers import SimpleString
+
+        @command((SimpleString,), name="info")
+        def info(self, section):
+            return section
+
+        FakeSocket.info = info
+
+
+def _check_ins(envelopes):
+    return [
+        item.payload.json
+        for envelope in envelopes
+        for item in envelope.items
+        if item.headers.get("type") == "check_in"
+    ]
+
+
+async def ok_job(ctx):
+    pass
+
+
+async def failing_job(ctx):
+    raise ValueError("oops")
+
+
+async def _run_cron_jobs(*cron_jobs, **settings):
+    pool = ArqRedis(pool_or_conn=FakeRedis().connection_pool)
+
+    class WorkerSettings:
+        redis_pool = pool
+        burst = True
+        poll_delay = 0
+
+    WorkerSettings.cron_jobs = list(cron_jobs)
+    for key, value in settings.items():
+        setattr(WorkerSettings, key, value)
+
+    worker = arq.worker.create_worker(WorkerSettings)
+    await worker.main()
+    await worker.close()
+
+
+@pytest.mark.asyncio
+async def test_cron_job_ok(sentry_init, capture_envelopes, monkeypatch):
+    # arq < 0.25 runs cron jobs in local time.
+    monkeypatch.setattr(time, "timezone", 0)
+    monkeypatch.setattr(time, "daylight", 0)
+    sentry_init(integrations=[ArqIntegration(monitor_cron_jobs=True)])
+    envelopes = capture_envelopes()
+
+    await _run_cron_jobs(
+        cron(ok_job, minute={0, 30}, run_at_startup=True), timezone=timezone.utc
+    )
+
+    in_progress, ok = _check_ins(envelopes)
+    assert in_progress["monitor_slug"] == "ok_job"
+    assert in_progress["status"] == "in_progress"
+    assert in_progress["monitor_config"] == {
+        "schedule": {"type": "crontab", "value": "0,30 * * * *"},
+        "timezone": "UTC",
+    }
+    assert ok["monitor_slug"] == "ok_job"
+    assert ok["status"] == "ok"
+    assert ok["check_in_id"] == in_progress["check_in_id"]
+    assert ok["duration"] >= 0
+
+
+@pytest.mark.asyncio
+async def test_cron_job_error(sentry_init, capture_envelopes):
+    sentry_init(integrations=[ArqIntegration(monitor_cron_jobs=True)])
+    envelopes = capture_envelopes()
+
+    await _run_cron_jobs(
+        cron(failing_job, name="nightly", hour=3, minute=0, run_at_startup=True),
+        timezone=timezone.utc,
+    )
+
+    check_ins = _check_ins(envelopes)
+    assert [c["status"] for c in check_ins] == ["in_progress", "error"]
+    assert {c["monitor_slug"] for c in check_ins} == {"nightly"}
+
+    events = [envelope.get_event() for envelope in envelopes]
+    (error,) = [e for e in events if e is not None and e.get("exception")]
+    assert error["exception"]["values"][0]["type"] == "ValueError"
+
+
+@pytest.mark.asyncio
+async def test_cron_job_without_config(sentry_init, capture_envelopes):
+    sentry_init(integrations=[ArqIntegration(monitor_cron_jobs=True)])
+    envelopes = capture_envelopes()
+
+    # Every 10 seconds can't be expressed as a crontab.
+    await _run_cron_jobs(
+        cron(ok_job, second={0, 10, 20, 30, 40, 50}, run_at_startup=True),
+        timezone=timezone.utc,
+    )
+
+    # A scheduled run can follow the startup run.
+    check_ins = _check_ins(envelopes)
+    assert [c["status"] for c in check_ins[:2]] == ["in_progress", "ok"]
+    assert all("monitor_config" not in c for c in check_ins)
+
+
+@pytest.mark.asyncio
+async def test_cron_jobs_not_monitored_by_default(sentry_init, capture_envelopes):
+    sentry_init(integrations=[ArqIntegration()])
+    envelopes = capture_envelopes()
+
+    await _run_cron_jobs(cron(ok_job, run_at_startup=True))
+
+    assert _check_ins(envelopes) == []
+
+
+@pytest.mark.asyncio
+async def test_exclude_cron_jobs(sentry_init, capture_envelopes):
+    sentry_init(
+        integrations=[
+            ArqIntegration(monitor_cron_jobs=True, exclude_cron_jobs=["ok_.*"])
+        ]
+    )
+    envelopes = capture_envelopes()
+
+    await _run_cron_jobs(
+        cron(ok_job, run_at_startup=True),
+        cron(ok_job, name="other", run_at_startup=True),
+        timezone=timezone.utc,
+    )
+
+    assert {c["monitor_slug"] for c in _check_ins(envelopes)} == {"other"}
+
+
+@pytest.mark.asyncio
+async def test_cron_job_monitored_once(sentry_init, capture_envelopes):
+    sentry_init(integrations=[ArqIntegration(monitor_cron_jobs=True)])
+    envelopes = capture_envelopes()
+
+    cron_job = cron(ok_job, run_at_startup=True)
+    await _run_cron_jobs(cron_job, timezone=timezone.utc)
+    cron_job.next_run = None
+    await _run_cron_jobs(cron_job, timezone=timezone.utc)
+
+    statuses = [c["status"] for c in _check_ins(envelopes)]
+    assert statuses[:4] == ["in_progress", "ok", "in_progress", "ok"]
+
+
+class _Worker:
+    def __init__(self, timezone):
+        self.timezone = timezone
+
+
+class _OldWorker:
+    pass
+
+
+@pytest.mark.parametrize(
+    "kwargs, crontab",
+    [
+        ({}, "* * * * *"),
+        ({"minute": 15}, "15 * * * *"),
+        ({"hour": {9, 17}, "minute": 0}, "0 9,17 * * *"),
+        ({"day": 1, "hour": 0, "minute": 0}, "0 0 1 * *"),
+        ({"month": {1, 7}, "day": 1, "hour": 0, "minute": 0}, "0 0 1 1,7 *"),
+        # arq: Monday = 0, cron: Sunday = 0
+        ({"weekday": 0, "hour": 8, "minute": 0}, "0 8 * * 1"),
+        ({"weekday": {5, 6}, "hour": 8, "minute": 0}, "0 8 * * 0,6"),
+        ({"weekday": "sun", "hour": 8, "minute": 0}, "0 8 * * 0"),
+        ({"weekday": "fri", "minute": 0, "second": {0}}, "0 * * * 5"),
+    ],
+)
+def test_get_monitor_config(kwargs, crontab):
+    assert _get_monitor_config(cron(ok_job, **kwargs), _Worker(timezone.utc)) == {
+        "schedule": {"type": "crontab", "value": crontab},
+        "timezone": "UTC",
+    }
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"second": None},
+        {"second": 30},
+        {"second": {0, 30}},
+        # arq requires both to match, cron either.
+        {"day": 1, "weekday": 0},
+    ],
+)
+def test_get_monitor_config_not_representable(kwargs):
+    assert _get_monitor_config(cron(ok_job, **kwargs), _Worker(timezone.utc)) is None
+
+
+@pytest.mark.parametrize(
+    "tz, name",
+    [
+        (timezone.utc, "UTC"),
+        (timezone(timedelta(hours=2)), "Etc/GMT-2"),
+        (timezone(timedelta(hours=-7)), "Etc/GMT+7"),
+        (timezone(timedelta(hours=5, minutes=30)), None),
+    ],
+)
+def test_get_monitor_config_timezone(tz, name):
+    config = _get_monitor_config(cron(ok_job, minute=0), _Worker(tz))
+    if name is None:
+        assert config is None
+    else:
+        assert config["timezone"] == name
+
+
+@pytest.mark.skipif(ZoneInfo is None, reason="zoneinfo not available")
+def test_get_monitor_config_zoneinfo():
+    config = _get_monitor_config(
+        cron(ok_job, minute=0), _Worker(ZoneInfo("Europe/Vienna"))
+    )
+    assert config["timezone"] == "Europe/Vienna"
+
+
+@pytest.mark.parametrize(
+    "tz_offset, daylight, name",
+    [(0, 0, "UTC"), (0, 1, None), (18000, 1, None)],
+)
+def test_get_monitor_config_local_time(monkeypatch, tz_offset, daylight, name):
+    monkeypatch.setattr(time, "timezone", tz_offset)
+    monkeypatch.setattr(time, "altzone", tz_offset - 3600)
+    monkeypatch.setattr(time, "daylight", daylight)
+    config = _get_monitor_config(cron(ok_job, minute=0), _OldWorker())
+    if name is None:
+        assert config is None
+    else:
+        assert config["timezone"] == name