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

Jump to bottom

Fix execution time checking by keeping lastUpdatedTime in db #1573

Merged
merged 3 commits into from Mar 4, 2024
Merged

Fix execution time checking by keeping lastUpdatedTime in db #1573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fb0a6c7
update lastUpdatedTime even if already processed for same interval
ikreymer Mar 3, 2024
2cb853b
store lastUpdateTime in db
ikreymer Mar 3, 2024
5bf4217
combine set_crawl_exec_last_update into inc_crawl_exec_time
ikreymer Mar 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
7 changes: 7 additions & 0 deletions backend/btrixcloud/crawls.py
View file
Open in desktop
Expand Up @@ -461,6 +461,13 @@ async def inc_crawl_exec_time(self, crawl_id, exec_time, last_updated_time):
},
)

async def get_crawl_exec_last_update_time(self, crawl_id):
"""get crawl last updated time"""
res = await self.crawls.find_one(
{"_id": crawl_id, "type": "crawl"}, projection=["_lut"]
)
return res and res.get("_lut")

async def get_crawl_state(self, crawl_id):
"""return current crawl state of a crawl"""
res = await self.crawls.find_one(
Expand Down
24 changes: 10 additions & 14 deletions backend/btrixcloud/operator.py
View file
Open in desktop
Expand Up @@ -515,7 +515,9 @@ async def sync_crawls(self, data: MCSyncData):

else:
status.scale = crawl.scale
status.lastUpdatedTime = to_k8s_date(dt_now())
now = dt_now()
await self.crawl_ops.inc_crawl_exec_time(crawl_id, 0, now)
status.lastUpdatedTime = to_k8s_date(now)

children = self._load_redis(params, status, data.children)

Expand Down Expand Up @@ -1107,12 +1109,15 @@ async def increment_pod_exec_time(
"""inc exec time tracking"""
now = dt_now()

if not status.lastUpdatedTime:
update_start_time = await self.crawl_ops.get_crawl_exec_last_update_time(
crawl_id
)

if not update_start_time:
await self.crawl_ops.inc_crawl_exec_time(crawl_id, 0, now)
status.lastUpdatedTime = to_k8s_date(now)
return

update_start_time = from_k8s_date(status.lastUpdatedTime)

reason = None
update_duration = (now - update_start_time).total_seconds()

Expand Down Expand Up @@ -1186,16 +1191,6 @@ async def increment_pod_exec_time(
max_duration = max(duration, max_duration)

if exec_time:
if not await self.crawl_ops.inc_crawl_exec_time(
crawl_id, exec_time, status.lastUpdatedTime
):
# if lastUpdatedTime is same as previous, something is wrong, don't update!
print(
"Already updated for lastUpdatedTime, skipping execTime update!",
flush=True,
)
return

await self.org_ops.inc_org_time_stats(oid, exec_time, True)
status.crawlExecTime += exec_time
status.elapsedCrawlTime += max_duration
Expand All @@ -1205,6 +1200,7 @@ async def increment_pod_exec_time(
flush=True,
)

await self.crawl_ops.inc_crawl_exec_time(crawl_id, exec_time, now)
status.lastUpdatedTime = to_k8s_date(now)

def should_mark_waiting(self, state, started):
Expand Down