Example use cases
These are illustrations of what a workflow definition can express, not shipped features. Each one uses the built-in gitlab and policy modules and the predeclared Starlark functions that exist today. The caller that starts the workflow passes the ids and a token as arguments to main; the token arrives as a sensitive value.
Policy-gated approval by a person
The workflow asks the policy module for a verdict on a requested deployment. If the verdict is “approval required”, it creates a channel, hands it to GitLab with the request, and waits up to three days for the answer, holding no resources meanwhile. Then it deploys, or fails.
load("module:policy", "evaluate", "REQUIRE_APPROVAL", "ALLOW")
load("module:gitlab", "call_api", "post_value")
def main(w, project_id, environment, token):
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
}
decision = gather(evaluate(
trigger = "com.gitlab.deploy.requested",
resource = "projects/%d/environments/%s" % (project_id, environment),
))
if decision["verdict"] == REQUIRE_APPROVAL:
reply = channel()
gather(post_value(
"/api/v4/projects/%d/deploy_approvals" % project_id,
value = {"environment": environment, "reply": reply},
headers = headers,
))
if gather(reply, timeout = 3 * 24 * time.hour) != "approved":
fail("deployment not approved")
elif decision["verdict"] != ALLOW:
fail("deployment denied by policy")
status, _, _, err = gather(call_api(
"POST",
"/api/v4/projects/%d/deployments" % project_id,
headers = headers,
body = json.encode({"environment": environment}),
))
return statusThe /deploy_approvals endpoint is illustrative. post_value posts the value together with a channel token, and the receiving side is whatever calls SendToWorkflowChannel with that channel token and the workflow token; the value it sends is what gather(reply) returns. The workflow’s result is the HTTP status of the deployment call. Today evaluate returns ALLOW until a policy source is wired. The Policy enforcement guide walks through this definition step by step.
Progressive deployment: poll until healthy, then promote
Wait until the staging deployment reports success, then create the production deployment. poll re-invokes an action on an interval until a CEL check over its result passes, without growing the history per attempt. derived_action decodes the response body where the action runs, so the check sees a plain string.
load("module:gitlab", "call_api")
def _deployment_status(invoke, *args, **kwargs):
status, _, body, _ = invoke(*args, **kwargs)
if status != 200:
return "unknown"
return json.decode(str(body))["status"]
deployment_status = derived_action(call_api, transform = _deployment_status)
def main(w, project_id, deployment_id, token):
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
}
path = "/api/v4/projects/%d/deployments/%d" % (project_id, deployment_id)
outcome = gather(poll(
action = deployment_status,
check = "ret == 'success' || ret == 'failed'",
args = ["GET", path],
kwargs = {"headers": headers},
interval = 5 * time.minute,
timeout = 2 * time.hour,
timeout_value = "timed out",
))
if outcome != "success":
fail("staging deployment %s" % outcome)
gather(call_api(
"POST",
"/api/v4/projects/%d/deployments" % project_id,
headers = headers,
body = json.encode({"environment": "production"}),
))Artifact lifecycle on a schedule of sleeps
Retention is a wait followed by an API call: keep a package for two weeks, then delete it if it still exists. Promotion has the same shape with a different call, and because a sleeping workflow costs nothing, one workflow per artifact is affordable.
load("module:gitlab", "call_api")
def main(w, project_id, package_id, token):
headers = {"Authorization": "Bearer " + token}
path = "/api/v4/projects/%d/packages/%d" % (project_id, package_id)
sleep(14 * 24 * time.hour)
status, _, _, _ = gather(call_api("GET", path, headers = headers))
if status == 200:
gather(call_api("DELETE", path, headers = headers))Cross-project coordination
Wait until an upstream project publishes a release, then open a tracking issue in a downstream project, using one token that can see both. Here poll checks the HTTP status directly: ret is the (status, headers, body, error) tuple call_api returns.
load("module:gitlab", "call_api")
def main(w, upstream_id, downstream_id, tag, token):
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
}
path = "/api/v4/projects/%d/releases/%s" % (upstream_id, tag)
released = gather(poll(
action = call_api,
check = "ret[0] == 200",
args = ["GET", path],
kwargs = {"headers": headers},
interval = time.hour,
timeout = 7 * 24 * time.hour,
))
if released == None:
fail("release %s never appeared" % tag)
gather(call_api(
"POST",
"/api/v4/projects/%d/issues" % downstream_id,
headers = headers,
body = json.encode({"title": "Adopt upstream release %s" % tag}),
))Long-running reminders and escalations
Sleep, check, act, repeat: remind after two days if an issue is still open, escalate with a label after five more. Each round is a durable wait followed by two recorded actions; closing the issue in between ends the workflow.
load("module:gitlab", "call_api")
def main(w, project_id, issue_iid, token):
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
}
path = "/api/v4/projects/%d/issues/%d" % (project_id, issue_iid)
for hours, label in [(48, "reminder-sent"), (120, "escalated")]:
sleep(hours * time.hour)
status, _, body, _ = gather(call_api("GET", path, headers = headers))
if status != 200 or json.decode(str(body))["state"] != "opened":
return "resolved"
gather(call_api(
"PUT",
path,
headers = headers,
body = json.encode({"add_labels": label}),
))
return "escalated"Building blocks
| Building block | Today |
|---|---|
call_api and post_value from the gitlab module | Available |
sleep and timer | Available |
poll and derived_action | Available |
Channels: channel(), gather, select and SendToWorkflowChannel | Available |
| Child workflows | Available |
evaluate from the policy module | Partial
returns ALLOW until a policy source is wired |