Template-level semaphore

[faraguna-ginkgo]

At Ginkgo we had one particular computational protein engineering tool that was limited in the number of parallel executions we could run through it. There was a license server we could ping using a small, CPU-only upstream Task in the DAG, but the server update cadence was too slow to be useful. Our argo-native approach was a template-level semaphore. Importantly, we wanted this semaphore to limit the number of executions of a template across potentially many Workflows. This approach is covered here in the argo docs, but I didn't see any good actual examples in the hera docs (this was kinda close).

Our semaphore

apiVersion: v1
kind: ConfigMap
metadata:
  name: workflow-semaphore-config
  namespace: foo
data:
  tool-semaphore: "18"

Registering using e.g. kubectl apply -f workflow-semaphore-config.yaml

Then used with WorkflowTemplates like:

import randomname
from hera.workflows import (
    WorkflowTemplate,
    script,
    Parameter,
    Steps,
    Step,
    Workflow,
)
from hera.workflows.models import (
    WorkflowTemplateRef,
    Synchronization,
    SemaphoreRef,
    ConfigMapKeySelector,
    TemplateRef,
)
from hera.exceptions import AlreadyExists

@script(
    synchronization=Synchronization(
        semaphore=SemaphoreRef(
            config_map_key_ref=ConfigMapKeySelector(
                name="workflow-semaphore-config",
                key="tool-semaphore",
            )
        )
    ),
)
def limited_sleep(message: str):
    import time

    time.sleep(10)
    print(message)


with WorkflowTemplate(
    name="limited-sleep",
    entrypoint="limited-sleep",
    arguments=[
        Parameter(
            name="message",
            default="Hello, world!",
            value="Hello, world!",
        )
    ],
) as wt:
    limited_sleep(
        arguments={
            "message": "{{inputs.parameters.message}}",
        },
    )

try:
    wt.create()
except AlreadyExists:
    wt.update()

with WorkflowTemplate(
    name="parallel-limited-sleep",
    entrypoint="limited-sleep-steps",
) as wt:
    limited_sleep_ref = TemplateRef(
        name="limited-sleep",
        template="limited-sleep",
    )
    with Steps(name="limited-sleep-steps") as steps:
        Step(
            name="limited-sleep",
            template_ref=limited_sleep_ref,
            with_items=list(range(20)),
            arguments={
                "message": "{{item}}",
            },
        )
try:
    wt.create()
except AlreadyExists:
    wt.update()

wt_ref = WorkflowTemplateRef(name="parallel-limited-sleep")
for i in range(2):
    w = Workflow(
        name=f"limited-sleep-{randomname.get_name()}",
        workflow_template_ref=wt_ref,
    )
    w.create()
    print(w.name)

Please excuse any anti-patterns ;)

[faraguna-ginkgo]

This also shows the barebones of how we approached composability / modularity with WorkflowTemplate objects. I ended up really liking this "write once, use whenever" approach that gave users an easy time launching user-facing, meta worflows.