✨ Adds t_scheduler used for reliably running code in dynamic-scheduler ⚠️ - #9036
✨ Adds t_scheduler used for reliably running code in dynamic-scheduler ⚠️#9036GitHK wants to merge 39 commits into
t_scheduler used for reliably running code in dynamic-scheduler ⚠️#9036Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #9036 +/- ##
==========================================
- Coverage 87.66% 87.29% -0.37%
==========================================
Files 2124 2067 -57
Lines 83866 82396 -1470
Branches 1569 1569
==========================================
- Hits 73519 71929 -1590
- Misses 9921 10041 +120
Partials 426 426
Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
GitHK
changed the title
t_scheduler used for reliably running code in dynamic-schedulert_scheduler used for reliable code execution, currently attached to dynamic-scheduler
|
pcrespov
left a comment
pcrespov
left a comment
There was a problem hiding this comment.
thx.
At first glance, I like both the Temporal.io functionality and the saga approach. That said, I have a common criticism: the PR feels like it generalizes too early, and you end up patching a generic interface before concrete use cases exist. I would have gone the other way — use Temporal.io as the implementation, build out real use cases, then abstract the repeated parts gradually. Though I may be missing something, since you have more context on the problems you're trying to solve.
|
|
||
| _logger = logging.getLogger(__name__) | ||
|
|
||
| _HEALTHCHECK_TIMEOUT: Final[timedelta] = timedelta(seconds=3) |
There was a problem hiding this comment.
THOUGHT: this is such a standard code. I wonder wether we could create an abstraction and move it to some package. This way we can have a similar pattern to use in all services.
| raise WorkflowAlreadyRegisteredError(name=name) | ||
|
|
||
| self._workflows[name] = workflow_cls | ||
| for act in workflow_cls.get_activities(): |
There was a problem hiding this comment.
why do you need access to the activities (get_temporalio_activities) ? they are inside of workflows.
as far as I understand th full idea is to run workflows, not indiviual activities. Exposing them outside defeats the purpse?
| ) | ||
|
|
||
|
|
||
| class WorkflowEngine: |
There was a problem hiding this comment.
I "love the engine"s ... :-D ... we like abusing same names for everything.
There was a problem hiding this comment.
is the engine basically a higher level client to temporal?
| handle: WorkflowHandle = self._client.get_workflow_handle(workflow_id) | ||
| await handle.signal("resolve", ResolutionSignal(activity_name=activity_name, decision=decision)) | ||
|
|
||
| async def list_running_workflows(self) -> list["RunningWorkflowInfo"]: |
There was a problem hiding this comment.
STYLE: Keep consistency with names, start, cancel ... or start_workflows, cancel_workflows ... but do not mix them
| (must match a key in ``WorkflowStatus.failed_activities``). | ||
| decision: Action to take — ``RETRY`` to re-execute, | ||
| ``SKIP`` to ignore the failure and continue, or | ||
| ``ROLLBACK`` to trigger compensation. |
There was a problem hiding this comment.
I thought ROLLBACK this was automatic ... isn't that the whole idea of the SAGA pattern?
THOGUTH:
If feels like you are creating a lot of abstractions on top of temporalio.workflow but then you open functions to address exceptions.
Do we need to have so many abstraction layers? Could we start with hard-coded worklfows like this example where we already decide how to react? Is it our system so dynamic??
Here 1 and 2 roolback or compensate the activities on error (3) ... and the workflow is predefined.
| from temporalio import activity, workflow | ||
| from temporalio.common import RetryPolicy | ||
|
|
||
| from ..t_scheduler._base_workflow import SagaWorkflow |
There was a problem hiding this comment.
IMO this is wrong. I fyou want to use these you should use the public interface.
You are accessing here protected members of another service module i.e. t_scheduler
In this case, the approach followed in web/server would be
- instead of
t_scheduler.__init__we definet_scheduler.t_scheduler_serviceas the service public api. - other service can use this as
from ..t_scheduler import t_scheduler_service... then all functions from this other service aret_scheduler_service.do_this(...). For models, exceptions or data-types you can import directlyfrom ..t_scheduler.t_scheduler_service import Model1, Exception1etc
|
|
||
|
|
||
| async def t_scheduler_register_workflows_lifespan(app: FastAPI) -> AsyncIterator[State]: | ||
| _register_workflows(get_workflow_registry(app)) |
There was a problem hiding this comment.
so now you just have a workflow to register a heatchekc.
| from models_library.utils.enums import StrAutoEnum | ||
|
|
||
|
|
||
| class WorkflowNames(StrAutoEnum): |
There was a problem hiding this comment.
since you have a registration, i would do this automatically as well...
think in a single registration point for your workflow and produce all these at the same time... otherwise you will have to remember to check all the boxes for every new registration
|
|
||
| """End-to-end integration tests for Temporal workflows. | ||
|
|
||
| These tests exercise the full stack — real Temporal server (via Docker Swarm), |
There was a problem hiding this comment.
Some of these tests are meant to explore how to use temporal. That's fine — it's worth doing, though maybe not critical for everyone. Should we mark them specially so they don't run on every execution? We could define a custom pytest marker and use it to skip them in certain contexts, e.g. on merge to master.
what do you think?
bisgaard-itis
left a comment
bisgaard-itis
left a comment
There was a problem hiding this comment.
This definitely looks very interesting. TBH I am not completely convinced by the idea of introducing yet another distributed task scheduling tool - we already have quite a lot of them laying around (both homemade and not). But I guess there is a good reason. But keep in mind that maintaining those system is also significant effort (ask @giancarloromeo). One thing in particular that I would rethink is how to detect if the workflows are out of sync. I am quite confident we could find a way to automate that completely (see comments)
| @@ -0,0 +1,50 @@ | |||
| #!/bin/bash | |||
| # Posts a one-time PR comment when workflows_signatures.json changes, | |||
| # warning OPS that Temporalio workflows must be shut down before deploying. | |||
There was a problem hiding this comment.
Would it not be possible to handle automate this step completely? Either by always shutting down the temporalio thingy, or by having a health check which checks that the workflows are up to date.
Having a separate CI job for posting a message to a PR seems very cumbersome and error prone to me.
| types: [opened, synchronize] | ||
|
|
||
| jobs: | ||
| ops-temporalio-maintenance-comment: |
There was a problem hiding this comment.
I would try to avoid having this job if possible (see also my other comment)
| - default | ||
| - interactive_services_subnet # for legacy dynamic services | ||
|
|
||
| temporal: |
There was a problem hiding this comment.
Couldn't you add a custom healthcheck to this service which queries the dynamic scheduler to get the workflow signature (or maybe simply a hash of it) and fails if it doesn't match what it expects. That way OPS would not need to be involved
There was a problem hiding this comment.
Do I get it right, that the temporal service is where the different "sagas" are registered?
|
|
||
| If this test fails, run ``make workflows_signatures.json`` in the service root | ||
| to regenerate ``workflows_signatures.json``, then flag the PR for OPS | ||
| to shut down Temporal workflows before deploying. |
There was a problem hiding this comment.
O think an alternative approach to doing this would simply be that you create your own docker image of temporal. It could simply be a copy of the temporal image, but tagged with the version we are using for the simcore services. That way when the ci runs, if the version already deployed doesn't match, it will create a new container. I guess that is actually the simplest solution. This is in the end simply a matter of retagging the official tepmoral image.
| WorkflowNotFoundError: If *workflow_name* is not in the registry. | ||
| temporalio.service.RPCError: If a workflow with the same | ||
| *workflow_id* is already running. | ||
| """ |
There was a problem hiding this comment.
Somehow we have an ability to always introduce a new tool for doing this kind of distributed task scheduling. Couldn't we simply have used celery?
sanderegg
left a comment
sanderegg
left a comment
There was a problem hiding this comment.
Ok this looks interesting. but here are a few comments:
- I would use the
Temporaland notTemporalioto make it clear, you have currently a mixture of them - t_scheduler, I would rename to temporal_client or something similar or do you plan to have a worker as well?
- please add a healthcheck in the docker compose.
| from .basic_types import PortInt | ||
|
|
||
|
|
||
| class TemporalioSettings(BaseCustomSettings): |
There was a problem hiding this comment.
temporal.io is the website and the python SDK right?
why you do not call it just Temporal?
|
|
||
| router = APIRouter() | ||
|
|
||
| _TEMPORALIO_CLIENT_UNHEALTHY_MSG: Final[str] = "Temporalio cannot be reached!" |
There was a problem hiding this comment.
you mix temporal and temporalio. I think it should be probably always just temporal
| async def list_workflows( | ||
| app: Annotated[FastAPI, Depends(get_app)], | ||
| ) -> list[RunningWorkflowInfo]: | ||
| """List all running Temporalio workflows on the scheduler task queue.""" |
There was a problem hiding this comment.
| """List all running Temporalio workflows on the scheduler task queue.""" | |
| """List all running Temporal workflows on the scheduler task queue.""" |
| async def shutdown_workflows( | ||
| app: Annotated[FastAPI, Depends(get_app)], | ||
| ) -> dict[str, int]: | ||
| """Cancel all running Temporalio workflows, triggering saga compensation.""" |
| def __init__(self) -> None: | ||
| self._state: WorkflowState = WorkflowState.RUNNING | ||
|
|
||
| self._pending_decisions: dict[str, Decision] = {} | ||
| self._awaiting_decisions: set[str] = set() | ||
|
|
||
| self._running_activities: set[str] = set() | ||
| self._compensations: list[Compensation] = [] | ||
| self._completed_activities: set[str] = set() | ||
| self._failed_activities: dict[str, str] = {} | ||
| self._compensated_activities: set[str] = set() | ||
| self._failed_compensations: dict[str, str] = {} | ||
| self._skipped_activities: set[str] = set() | ||
|
|
||
| self._steps_total: int = 0 | ||
| self._history: list[dict[str, Any]] = [] | ||
|
|
There was a problem hiding this comment.
would it not be a bit more readable using a dataclass here?
| activity.heartbeat() | ||
| except asyncio.CancelledError: | ||
| task.cancel() | ||
| with contextlib.suppress(asyncio.CancelledError): |
There was a problem hiding this comment.
this is a kind of dangerous construct as this can swallow a higher level cancellation call.
you should probably use cancel_wait_task here
| def __init__(self) -> None: | ||
| self._workflows: dict[str, type[SagaWorkflow]] = {} | ||
| self._activities: list[Callable[..., Coroutine[Any, Any, Any]]] = [] |
There was a problem hiding this comment.
this could also use a dataclass for simplifications
| @activity.defn | ||
| async def healthcheck_activity(_ctx: dict[str, Any]) -> dict[str, Any]: | ||
| _logger.info("Healthcheck activity executed") | ||
| return {"healthcheck": "ok"} | ||
|
|
||
|
|
||
| @activity.defn | ||
| async def undo_healthcheck_activity(_ctx: dict[str, Any]) -> None: | ||
| _logger.info("Healthcheck activity compensated (no-op)") | ||
|
|
There was a problem hiding this comment.
not sure what you are doing here.
is activity.defn implicitely connecting with the temporal platform?
the logger is not really a no-op right? it writes to the stdout stream.
also not sure I understand the undo part of this.
| pytest.param(True, False, True, True, False, id="rabbit_rpc_client_bad"), | ||
| pytest.param(True, True, False, True, False, id="redis_client_bad"), | ||
| pytest.param(True, True, True, False, False, id="docker_api_proxy_bad"), | ||
| pytest.param(True, True, True, True, True, True, id="ok"), |
There was a problem hiding this comment.
maybe using a dataclass here or a named tuple would make it a bit more readable
| POSTGRES_SEEDS: postgres | ||
| TEMPORAL_ADDRESS: temporal:7233 | ||
| BIND_ON_IP: 0.0.0.0 | ||
| networks: |
There was a problem hiding this comment.
healthcheck is missing here. please add one thanks
github-actions
Bot
added
a:infra+ops
|
What do these changes do?
TODOS:
Adds a Temporal.io-based workflow engine (
t_scheduler) to thedynamic-schedulerservice. This provides durable workflow execution (saga pattern) with automatic retries, heartbeats, and state persistence for managing dynamic service lifecycles.Change breakdown by category
Key areas to review
New module:
services/t_scheduler/(~1,000 lines)_base_workflow.py— Saga workflow abstraction (execute/compensate pattern)_engine.py— Workflow engine runner (start/cancel/query workflows)_heartbeat.py— Activity heartbeat interceptor_registry.py— Workflow/activity registration_models.py— Pydantic models for workflow stateNew module:
services/workflows/(~120 lines)Shared package:
packages/settings-library/temporalio.py(44 lines)TemporalioSettings— shared Pydantic settings (TEMPORALIO_HOST,PORT,NAMESPACE,TASK_QUEUE)Integration with existing code (~50 lines across 6 files)
core/events.py— Wires temporal lifespans into app startupcore/settings.py— AddsTemporalioSettingsto app settingsapi/rest/_health.py— Health endpoint checks temporal connectivityapi/rest/_ops.py— Ops endpoint to list/shutdown temporal workflowsapi/rest/_dependencies.py— FastAPI dependency for temporal health checkcli.py— Worker mode argumentDocker / CI (~165 lines)
temporalservice (temporalio/auto-setup:1.29.1) reusing existing PostgreSQLtemporalio/ui:2.36.0)Env / Config (~100 lines)
TEMPORALIO_*variablesrequirements/_base.in—temporalioPython dependencyopenapi.json— New ops endpoints in OpenAPI specImpact on existing services
dynamic-schedulernow requires a running Temporal server at startup (120s retry). Without it, the service exits with code 3.temporalservice uses the existingpostgres— creates separate databases (temporal,temporal_visibility), no conflict withsimcoredb.temporalin the Docker stack whendynamic-schdlris included (CI fix applied totest_computation.py).Related issue/s
How to test
Dev-ops⚠️
requires Add Temporal.io server to simcore stack (required by dynamic-scheduler) osparc-ops-environments#1398 to be done before merging this PR
merge this before merging this PR
Note, for future reference when
temporal workflow shutdown is required: