Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

dvc-10427 Fix Celery Monitor Shutdowns #142

Merged
merged 5 commits into from
Oct 3, 2024
Merged

dvc-10427 Fix Celery Monitor Shutdowns #142

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f15acd7
dvc-10427 Add 2" delay to expirations check
nablabits Sep 10, 2024
9ac90e9
Add a tip to run a single test to the documentation
nablabits Sep 10, 2024
d56d993
Exclude pyenv virtual environments
nablabits Sep 10, 2024
307c98b
dvc-10427 Send the shutdown message to the appropriate worker.
nablabits Sep 25, 2024
667b431
Revert "dvc-10427 Add 2" delay to expirations check"
nablabits Oct 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 2 additions & 5 deletions src/dvc_task/app/filesystem.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ def _delete_expired(
now: float,
cache: Dict[str, str],
include_tickets: bool = False,
) -> None:
shcheklein marked this conversation as resolved.
Show resolved Hide resolved
):
assert isinstance(msg.properties, dict)
properties = cast(Dict[str, Any], msg.properties)
delivery_info: Dict[str, str] = properties.get("delivery_info", {})
Expand All @@ -255,10 +255,7 @@ def _delete_expired(
pass

queues = set(exclude) if exclude else set()

# Kombu will create expiration timestamps 1" in the future, let's make sure we
# pick them.
now = datetime.now().timestamp() + 2 # noqa: DTZ005
now = datetime.now().timestamp() # noqa: DTZ005
for msg in self._iter_data_folder():
_delete_expired(msg, queues, now, self._queued_msg_path_cache)
for msg in self._iter_processed_folder():
Expand Down
18 changes: 1 addition & 17 deletions tests/app/test_filesystem.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,6 @@
"""Filesystem app tests."""

import json
from copy import deepcopy
from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, Optional

import pytest
Expand Down Expand Up @@ -158,23 +156,11 @@ def test_purge(tmp_dir: TmpDir):
def test_gc(tmp_dir: TmpDir):
"""Expired messages and processed tickets should be removed."""
app = FSApp(wdir=str(tmp_dir), mkdir=True)

now = datetime.now().timestamp() # noqa: DTZ005
expirations = []
for delay in (1, 3):
msg = deepcopy(EXPIRED_MSG)
msg["headers"]["expires"] = int(now + delay)
expirations.append(msg)

one_sec_expiration, three_secs_expiration = expirations

tmp_dir.gen(
{
"broker": {
"in": {
"expired.msg": json.dumps(EXPIRED_MSG), # way in the past
"future_expired.msg": json.dumps(one_sec_expiration),
"future_not_yet_expired.msg": json.dumps(three_secs_expiration),
"expired.msg": json.dumps(EXPIRED_MSG),
"unexpired.msg": json.dumps(TEST_MSG),
"ticket.msg": json.dumps(TICKET_MSG),
},
Expand All @@ -189,8 +175,6 @@ def test_gc(tmp_dir: TmpDir):

app._gc()
assert not (tmp_dir / "broker" / "in" / "expired.msg").exists()
assert not (tmp_dir / "broker" / "in" / "future_expired.msg").exists()
assert (tmp_dir / "broker" / "in" / "future_not_yet_expired.msg").exists()
assert (tmp_dir / "broker" / "in" / "unexpired.msg").exists()
assert (tmp_dir / "broker" / "in" / "ticket.msg").exists()
assert not (tmp_dir / "broker" / "processed" / "expired.msg").exists()
Expand Down
Loading