You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tooling.apache.org by "alitheg (via GitHub)" <gi...@apache.org> on 2026/01/15 11:34:41 UTC
[PR] Possible recurrent jobs implementation (tooling-trusted-releases)
alitheg opened a new pull request, #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529
Add the ability to schedule future jobs. Worker claim updated to not claim future tasks. Metadata and workflow status tasks can optionally schedule their next iteration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
Re: [PR] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg merged PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694761703
##########
atr/tasks/metadata.py:
##########
@@ -45,6 +50,15 @@ async def update(args: Update) -> results.Results | None:
log.info(
f"Metadata update completed successfully: added {added_count}, updated {updated_count}",
)
+
+ # Schedule next update
+ if args.next_schedule and args.next_schedule > 0:
Review Comment:
Sure, will update to seconds - seems reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694783375
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
+ await asyncio.sleep(30)
+ try:
+ await tasks.clear_scheduled()
+ metadata = await tasks.metadata_update(asf_uid="system", schedule_next=True)
Review Comment:
For these two - is there any real benefit to asking the metadata_update method to add a random delay? The first is already delayed by 60s (changed from 30) from server start - we could just add another delay before the workflow status and then they'd be both delayed, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694786284
##########
atr/tasks/gha.py:
##########
@@ -56,6 +61,11 @@ class DistributionWorkflow(schema.Strict):
name: str = schema.description("Name of the run")
+class WorkflowStatusCheck(schema.Strict):
+ run_id: int | None = schema.description("Run ID")
Review Comment:
I added it for future use - to allow a specific task to be refreshed. But I suppose if we're running this every few seconds it wouldn't matter anyway...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694758014
##########
atr/worker.py:
##########
@@ -114,7 +114,12 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any]] | No
# Get the ID of the oldest queued task
oldest_queued_task = (
sqlmodel.select(sql.Task.id)
- .where(sql.Task.status == task.QUEUED)
+ .where(
+ sqlmodel.and_(
+ sql.Task.status == task.QUEUED,
+ sql.Task.added <= datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=2),
Review Comment:
It's a good question. As I wrote it I'm sure I had a scenario in mind where with tasks scheduling themselves we wouldn't necessarily want a potentially instant claim on them. But ever since (now but also when I committed it) I can't think of how that could be a problem
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
Re: [PR] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#issuecomment-3754299916
@sbp I realised we already have sufficient plumbing for scheduling, since the worker runs on a loop and claims tasks.
There are some potential improvements I can think of to this - such as adding a "scheduled" column to Task instead of using "added" - but I didn't want to spend too long on it as a first pass. Let me know what you think of the concept and we can refine or throw it away!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694731710
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
+ await asyncio.sleep(30)
Review Comment:
If I run this as a task as it did before - I assume it should still check at shutdown in case it's somehow still running and stop it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "alitheg (via GitHub)" <gi...@apache.org>.
alitheg commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694752015
##########
atr/tasks/__init__.py:
##########
@@ -259,6 +295,33 @@ async def tar_gz_checks(asf_uid: str, release: sql.Release, revision: str, path:
return tasks
+async def workflow_update(
+ asf_uid: str,
+ caller_data: db.Session | None = None,
+ schedule: datetime.datetime | None = None,
+ schedule_next: bool = False,
+) -> sql.Task:
+ """Queue a workflow status update task."""
+ args = gha.WorkflowStatusCheck(next_schedule=0, run_id=0)
+ if schedule_next:
+ args.next_schedule = 2
+ async with db.ensure_session(caller_data) as data:
+ task = sql.Task(
+ status=sql.TaskStatus.QUEUED,
+ task_type=sql.TaskType.WORKFLOW_STATUS,
+ task_args=args.model_dump(),
+ asf_uid=asf_uid,
Review Comment:
In a commit a little while back I added support for tasks to be passed their own task ID, if it's present in the function signature. We could do the same for asf_uid so that it's passed in if they want it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org
[GH] Possible recurrent jobs implementation (tooling-trusted-releases)
Posted by "sbp (via GitHub)" <gi...@apache.org>.
sbp commented on code in PR #529:
URL: https://github.com/apache/tooling-trusted-releases/pull/529#discussion_r2694609299
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
+ await asyncio.sleep(30)
Review Comment:
Because `_register_recurrent_tasks` is no longer run as a task, the server steps in to the `await` here and it blocks the server startup. The fix is to run this as a task, as before.
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
Review Comment:
This is out of sync with the code that it's commenting, which has changed to 30s.
##########
atr/tasks/__init__.py:
##########
@@ -57,6 +60,27 @@ async def asc_checks(asf_uid: str, release: sql.Release, revision: str, signatur
return tasks
+async def clear_scheduled(caller_data: db.Session | None = None):
+ """Clear all future scheduled tasks of the given types."""
+ async with db.ensure_session(caller_data) as data:
+ via = sql.validate_instrumented_attribute
+ now = datetime.datetime.now(datetime.UTC)
+
+ delete_stmt = sqlmodel.delete(sql.Task).where(
+ via(sql.Task.task_type).in_(
+ [
+ sql.TaskType.METADATA_UPDATE,
+ sql.TaskType.WORKFLOW_STATUS,
+ ]
+ ),
+ via(sql.Task.status) == sql.TaskStatus.QUEUED,
+ via(sql.Task.added) > now,
Review Comment:
Perhaps we should remove this and clear all such tasks? There could be a little gap where a task is 1s old and hence not yet claimed (because of the 2s buffer), and therefore would not be removed either.
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
+ await asyncio.sleep(30)
+ try:
+ await tasks.clear_scheduled()
+ metadata = await tasks.metadata_update(asf_uid="system", schedule_next=True)
Review Comment:
This is scheduled to run straight away, but for a long time I've been wondering whether I should have added a random wait or not, probably between 60 and 600 seconds or so. Now that we have multiple tasks set to start straight away, I think we probably should do this to prevent thundering herd.
##########
atr/tasks/metadata.py:
##########
@@ -45,6 +50,15 @@ async def update(args: Update) -> results.Results | None:
log.info(
f"Metadata update completed successfully: added {added_count}, updated {updated_count}",
)
+
+ # Schedule next update
+ if args.next_schedule and args.next_schedule > 0:
Review Comment:
`(args.next_schedule > 0)`
##########
atr/server.py:
##########
@@ -463,20 +464,36 @@ async def _initialise_test_environment() -> None:
await data.commit()
-async def _metadata_update_scheduler() -> None:
- """Periodically schedule remote metadata updates."""
- # Wait one minute to allow the server to start
- await asyncio.sleep(60)
+#
+# async def _metadata_update_scheduler() -> None:
+# """Periodically schedule remote metadata updates."""
+# # Wait one minute to allow the server to start
+# await asyncio.sleep(60)
+#
+# while True:
+# try:
+# task = await tasks.metadata_update(asf_uid="system")
+# log.info(f"Scheduled remote metadata update with ID {task.id}")
+# except Exception as e:
+# log.exception(f"Failed to schedule remote metadata update: {e!s}")
+#
+# # Schedule next update in 24 hours
+# await asyncio.sleep(86400)
- while True:
- try:
- task = await tasks.metadata_update(asf_uid="system")
- log.info(f"Scheduled remote metadata update with ID {task.id}")
- except Exception as e:
- log.exception(f"Failed to schedule remote metadata update: {e!s}")
- # Schedule next update in 24 hours
- await asyncio.sleep(86400)
+async def _register_recurrent_tasks() -> None:
+ """Schedule recurring tasks"""
+ # Wait one minute to allow the server to start
+ await asyncio.sleep(30)
+ try:
+ await tasks.clear_scheduled()
+ metadata = await tasks.metadata_update(asf_uid="system", schedule_next=True)
+ log.info(f"Scheduled remote metadata update with ID {metadata.id}")
+ workflow = await tasks.workflow_update(asf_uid="system", schedule_next=True)
Review Comment:
Must prevent thundering herd here too.
##########
atr/tasks/metadata.py:
##########
@@ -45,6 +50,15 @@ async def update(args: Update) -> results.Results | None:
log.info(
f"Metadata update completed successfully: added {added_count}, updated {updated_count}",
)
+
+ # Schedule next update
+ if args.next_schedule and args.next_schedule > 0:
Review Comment:
Also I'd probably call this `next_schedule_minutes` if it's in minutes, but even better would be to use seconds. Everything is measured in seconds these days. I mean, these seconds.
##########
atr/worker.py:
##########
@@ -114,7 +114,12 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any]] | No
# Get the ID of the oldest queued task
oldest_queued_task = (
sqlmodel.select(sql.Task.id)
- .where(sql.Task.status == task.QUEUED)
+ .where(
+ sqlmodel.and_(
+ sql.Task.status == task.QUEUED,
+ sql.Task.added <= datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=2),
Review Comment:
Why `datetime.timedelta(seconds=2)`? What does this little buffer period do?
##########
atr/tasks/gha.py:
##########
@@ -192,6 +274,15 @@ async def _request_and_retry(
return None
+async def _schedule_next(args: WorkflowStatusCheck):
+ if args.next_schedule:
+ next_schedule = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=args.next_schedule)
+ await tasks.workflow_update("system", schedule=next_schedule, schedule_next=True)
Review Comment:
We could change `system` here if the ASF UID were part of the args to the task itself, like it is for the metadata update task.
##########
atr/worker.py:
##########
@@ -114,7 +114,12 @@ async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any]] | No
# Get the ID of the oldest queued task
oldest_queued_task = (
sqlmodel.select(sql.Task.id)
- .where(sql.Task.status == task.QUEUED)
+ .where(
+ sqlmodel.and_(
+ sql.Task.status == task.QUEUED,
+ sql.Task.added <= datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=2),
Review Comment:
`(datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=2))`
##########
atr/tasks/gha.py:
##########
@@ -56,6 +61,11 @@ class DistributionWorkflow(schema.Strict):
name: str = schema.description("Name of the run")
+class WorkflowStatusCheck(schema.Strict):
+ run_id: int | None = schema.description("Run ID")
Review Comment:
Is `run_id` still used?
##########
atr/tasks/__init__.py:
##########
@@ -259,6 +295,33 @@ async def tar_gz_checks(asf_uid: str, release: sql.Release, revision: str, path:
return tasks
+async def workflow_update(
+ asf_uid: str,
+ caller_data: db.Session | None = None,
+ schedule: datetime.datetime | None = None,
+ schedule_next: bool = False,
+) -> sql.Task:
+ """Queue a workflow status update task."""
+ args = gha.WorkflowStatusCheck(next_schedule=0, run_id=0)
+ if schedule_next:
+ args.next_schedule = 2
+ async with db.ensure_session(caller_data) as data:
+ task = sql.Task(
+ status=sql.TaskStatus.QUEUED,
+ task_type=sql.TaskType.WORKFLOW_STATUS,
+ task_args=args.model_dump(),
+ asf_uid=asf_uid,
Review Comment:
Maybe the ASF UID should go in the args to the task itself? The metadata update task does it that way, and then when it reschedules itself it still knows the ASF UID it was started with. Currently this is always `system`, but if we allow it to be stopped and started by an admin that would change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tooling.apache.org
For additional commands, e-mail: dev-help@tooling.apache.org