Pipeline Schedules, Trigger Tokens, Pipeline API, Webhooks, ChatOps, and Event-Driven Automation: Guided Hands-On Workflow and Core Operations

1. Scenario and boundaries

You operate a small external release orchestrator. An event should create a pipeline on project 4242, ref main, then the orchestrator must wait for that exact pipeline to finish before reconciling external state. The lab uses a local mock because the important concepts are request identity, asynchronous status, 429 handling and idempotency—not possession of a real GitLab credential.

2. Preflight and current assumptions

python3 --version
mkdir -p glci-ch33-lab && cd glci-ch33-lab
mkdir -p evidence .local external_state
printf 'root=%s\n' "$PWD"

Current GitLab semantics verified 2026-09-12: trigger and schedule inputs are GA; trigger-token pipelines have source trigger; job-token triggered multi-project pipelines have source pipeline; pipeline schedules run with the owner's permissions; and project webhooks support HMAC signing tokens. Python's standard library is the only executable dependency in this mandatory path.

3. Define the pipeline-side contract before writing the controller

A real target project should reject unexpected sources and constrain behavior through typed inputs. This sample allows four automation sources and rejects everything else:

spec:
  inputs:
    operation:
      type: string
      options: [inspect, reconcile]
      default: inspect
---
workflow:
  rules:
    - if: '$CI_PIPELINE_SOURCE == "schedule"'
    - if: '$CI_PIPELINE_SOURCE == "trigger"'
    - if: '$CI_PIPELINE_SOURCE == "api"'
    - if: '$CI_PIPELINE_SOURCE == "chat"'
    - when: never

automation-entry:
  script:
    - printf 'source=%s pipeline=%s job=%s sha=%s\n' "$CI_PIPELINE_SOURCE" "$CI_PIPELINE_ID" "$CI_JOB_ID" "$CI_COMMIT_SHA"
    - printf 'operation=%s\n' '$[[ inputs.operation ]]'

This is repository/compiled-configuration state. It does not configure an API token, schedule, runner or webhook receiver. Those are different layers.

4. Create the local GitLab-like API

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse
import json, time

HOST = '127.0.0.1'
PORT = 18765
PROJECT = '4242'
TOKEN = 'training-trigger-token'
STATE = Path('.mock-state.json')

def load():
    if not STATE.exists():
        return {'next_id': 3301, 'pipelines': {}, 'rate_limited': {}}
    return json.loads(STATE.read_text(encoding='utf-8'))

def save(s):
    STATE.write_text(json.dumps(s, indent=2, sort_keys=True) + '\n', encoding='utf-8')

def send_json(h, status, payload, headers=None):
    data = json.dumps(payload, sort_keys=True).encode('utf-8')
    h.send_response(status)
    h.send_header('Content-Type', 'application/json')
    h.send_header('Content-Length', str(len(data)))
    for k, v in (headers or {}).items(): h.send_header(k, str(v))
    h.end_headers(); h.wfile.write(data)

class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        print('%s - %s' % (self.address_string(), fmt % args), flush=True)

    def do_POST(self):
        parsed = urlparse(self.path)
        if parsed.path != f'/api/v4/projects/{PROJECT}/trigger/pipeline':
            return send_json(self, 404, {'message':'not found'})
        length = int(self.headers.get('Content-Length','0'))
        form = parse_qs(self.rfile.read(length).decode('utf-8'), keep_blank_values=True)
        if form.get('token',[''])[0] != TOKEN:
            return send_json(self, 401, {'message':'invalid training token'})
        ref = form.get('ref',[''])[0]
        if ref != 'main':
            return send_json(self, 400, {'message':'training server accepts only ref=main'})
        request_key = form.get('inputs[request_key]',[''])[0]
        if not request_key:
            return send_json(self, 400, {'message':'inputs[request_key] is required by this lab policy'})
        s=load()
        if form.get('inputs[simulate_429_once]',['false'])[0].lower() == 'true' and not s['rate_limited'].get(request_key):
            s['rate_limited'][request_key]=True; save(s)
            return send_json(self, 429, {'message':'simulated rate limit'}, {'Retry-After':'1','RateLimit-Remaining':'0'})
        pid=s['next_id']; s['next_id'] += 1
        sha='33'*20
        p={'id':pid,'project_id':int(PROJECT),'ref':ref,'sha':sha,'source':'trigger','status':'pending','polls':0,'request_key':request_key,'created_at':'2026-09-12T20:00:00Z','web_url':f'https://gitlab.example.invalid/training/pipelines/{pid}'}
        s['pipelines'][str(pid)]=p; save(s)
        return send_json(self, 201, p)

    def do_GET(self):
        parsed=urlparse(self.path); s=load()
        if parsed.path == '/debug/state':
            return send_json(self, 200, s)
        prefix=f'/api/v4/projects/{PROJECT}/pipelines/'
        if parsed.path.startswith(prefix):
            pid=parsed.path[len(prefix):]
            if pid not in s['pipelines']:
                return send_json(self,404,{'message':'pipeline not found'})
            p=s['pipelines'][pid]; p['polls'] += 1
            p['status'] = 'running' if p['polls'] == 1 else 'success'
            s['pipelines'][pid]=p; save(s)
            return send_json(self,200,p,{'RateLimit-Limit':'60','RateLimit-Remaining':'59'})
        if parsed.path == f'/api/v4/projects/{PROJECT}/pipelines':
            rows=sorted(s['pipelines'].values(), key=lambda x:x['id'], reverse=True)
            return send_json(self,200,rows,{'X-Page':'1','X-Per-Page':'20','X-Next-Page':''})
        return send_json(self,404,{'message':'not found'})

if __name__ == '__main__':
    print(f'mock GitLab API listening on http://{HOST}:{PORT}', flush=True)
    ThreadingHTTPServer((HOST,PORT), Handler).serve_forever()

Save the preceding Python block as mock_gitlab.py, then start it only inside this disposable lab:

python3 mock_gitlab.py > evidence/mock-server.log 2>&1 &
echo $! > .local/mock-server.pid
sleep 0.3
python3 - <<'PY'
from urllib.request import urlopen
import json
print(json.loads(urlopen('http://127.0.0.1:18765/debug/state').read()))
PY

The mock accepts only project 4242, ref main and the literal training trigger token. It returns a pipeline ID and advances pending → running → success when polled. The special input simulate_429_once exists only to exercise rate-limit handling.

5. Build an external orchestrator with correlation and bounded retry

from pathlib import Path
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import HTTPError
import json, sys, time

BASE='http://127.0.0.1:18765'
PROJECT='4242'
TOKEN='training-trigger-token'
TERMINAL={'success','failed','canceled','skipped','manual'}
JOURNAL=Path('evidence/journal.json')

def request_json(req):
    with urlopen(req, timeout=3) as r:
        return r.status, dict(r.headers.items()), json.loads(r.read().decode('utf-8'))

def load_journal():
    if JOURNAL.exists(): return json.loads(JOURNAL.read_text(encoding='utf-8'))
    return {}

def save_journal(j):
    JOURNAL.parent.mkdir(parents=True, exist_ok=True)
    JOURNAL.write_text(json.dumps(j, indent=2, sort_keys=True)+'\n', encoding='utf-8')

def create(event_id, simulate_429=False):
    form={'token':TOKEN,'ref':'main','inputs[request_key]':event_id,'inputs[operation]':'checkpoint'}
    if simulate_429: form['inputs[simulate_429_once]']='true'
    data=urlencode(form).encode('utf-8')
    url=f'{BASE}/api/v4/projects/{PROJECT}/trigger/pipeline'
    for attempt in range(1,4):
        try:
            status,headers,payload=request_json(Request(url,data=data,method='POST'))
            if status != 201: raise RuntimeError(f'unexpected create status {status}')
            return payload, attempt
        except HTTPError as e:
            body=e.read().decode('utf-8',errors='replace')
            if e.code != 429 or attempt == 3:
                raise RuntimeError(f'create failed HTTP {e.code}: {body}')
            delay=int(e.headers.get('Retry-After','1'))
            print(f'rate_limited attempt={attempt} retry_after={delay}', flush=True)
            time.sleep(min(delay,1))
    raise RuntimeError('unreachable')

def get_pipeline(pid):
    url=f'{BASE}/api/v4/projects/{PROJECT}/pipelines/{pid}'
    return request_json(Request(url,method='GET'))[2]

def main():
    event_id=sys.argv[1] if len(sys.argv)>1 else 'event-33-001'
    simulate='--simulate-429' in sys.argv[2:]
    journal=load_journal()
    if event_id in journal:
        pid=journal[event_id]['pipeline_id']
        print(f'idempotent_reuse event={event_id} pipeline_id={pid}')
    else:
        p,attempts=create(event_id,simulate)
        pid=p['id']; journal[event_id]={'pipeline_id':pid,'source':p['source'],'sha':p['sha'],'ref':p['ref']}
        save_journal(journal)
        print(f'created event={event_id} pipeline_id={pid} attempts={attempts} source={p["source"]}')
    Path('evidence').mkdir(exist_ok=True)
    with Path('evidence/poll.log').open('a',encoding='utf-8') as log:
        for n in range(1,6):
            p=get_pipeline(pid)
            line=f'poll={n} pipeline_id={pid} status={p["status"]} source={p["source"]} sha={p["sha"]}'
            print(line); log.write(line+'\n'); log.flush()
            if p['status'] in TERMINAL:
                if p['status'] != 'success': raise SystemExit(2)
                Path('evidence/final-pipeline.json').write_text(json.dumps(p,indent=2,sort_keys=True)+'\n',encoding='utf-8')
                return
            time.sleep(0.1)
    raise SystemExit('pipeline did not reach a terminal state')

if __name__=='__main__': main()

Save the block as orchestrator.py, then run:

python3 orchestrator.py event-33-001 --simulate-429 | tee evidence/first-run.txt

Expected sequence: one simulated 429, a bounded wait following Retry-After, HTTP creation of one pipeline, then polling of that exact ID until success. The journal is written immediately after the create response so later invocations reuse the correlation.

6. Repeat the same external event

python3 orchestrator.py event-33-001 | tee evidence/repeat-run.txt
python3 - <<'PY'
from urllib.request import urlopen
import json
s=json.loads(urlopen('http://127.0.0.1:18765/debug/state').read())
print('pipeline_count=',len(s['pipelines']))
assert len(s['pipelines']) == 1
PY

The second invocation should print idempotent_reuse. This proves the local controller does not create a second pipeline after it has durable knowledge of the first ID. It does not prove a local JSON journal is sufficient for production concurrency; use transactional durable storage there.

7. Diagnose a wrong-ref request without guessing

from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import HTTPError
form=urlencode({'token':'training-trigger-token','ref':'production','inputs[request_key]':'wrong-ref-001'}).encode()
try:
    urlopen(Request('http://127.0.0.1:18765/api/v4/projects/4242/trigger/pipeline',data=form,method='POST'))
except HTTPError as e:
    print('status=',e.code)
    print('body=',e.read().decode())

Expected: HTTP 400 and an explicit ref error. Preserve it. Do not “fix” the problem by changing the project or selecting whatever default branch happens to exist.

8. Simulate schedule evidence

No cron daemon is required to understand schedule state. Record what a real controller must preserve:

{
  "schedule_id": 33,
  "description": "nightly-inspect",
  "owner": "training-release-bot",
  "ref": "main",
  "inputs": {"operation": "inspect"},
  "expected_pipeline_source": "schedule"
}

In a live GitLab project, inspect the schedule owner before assuming protected-resource access. A manual “Run” of this schedule would execute under the manual actor's permissions.

9. Verify a signed webhook locally

from pathlib import Path
import base64, hashlib, hmac, json, time

secret_path=Path('.local/webhook-signing-token.txt')
raw=hashlib.sha256(b'ch33-training-webhook-key').digest()
secret_path.parent.mkdir(parents=True, exist_ok=True)
secret_path.write_text('whsec_'+base64.b64encode(raw).decode('ascii')+'\n',encoding='utf-8')
body=json.dumps({'object_kind':'pipeline','project':{'id':4242},'object_attributes':{'id':3301,'status':'success'}},separators=(',',':'))
message_id='11111111-2222-3333-4444-555555555555'
timestamp=str(int(time.time()))
message=f'{message_id}.{timestamp}.{body}'.encode('utf-8')
key=base64.b64decode(secret_path.read_text(encoding='utf-8').strip().removeprefix('whsec_'))
sig='v1,'+base64.b64encode(hmac.new(key,message,hashlib.sha256).digest()).decode('ascii')

# Receiver side: verify freshness and constant-time HMAC before parsing/acting.
assert abs(int(time.time())-int(timestamp)) <= 300
expected='v1,'+base64.b64encode(hmac.new(key,message,hashlib.sha256).digest()).decode('ascii')
assert hmac.compare_digest(expected,sig)
print('webhook_verified=true')
print('webhook_id='+message_id)
print('pipeline_id=3301')

Save the block as webhook_demo.py, then run it and protect the local training key file:

python3 webhook_demo.py | tee evidence/webhook-verification.txt
chmod 600 .local/webhook-signing-token.txt

The training signing token is stored locally and never printed. The receiver checks a five-minute freshness window and constant-time HMAC comparison before trusting the payload. Production receivers should also persist the message ID before side effects so webhook retries do not repeat work.

10. Inspect the pipeline collection without assuming it fits one response

from urllib.request import urlopen
import json
u='http://127.0.0.1:18765/api/v4/projects/4242/pipelines'
with urlopen(u) as r:
    rows=json.loads(r.read())
    print('page=',r.headers.get('X-Page'),'per_page=',r.headers.get('X-Per-Page'),'next=',r.headers.get('X-Next-Page'))
    for p in rows: print(p['id'],p['source'],p['status'])

The mock has one page. A real client must follow documented pagination headers/cursors until complete. Never interpret “20 results returned” as “there are only 20 pipelines.”

11. Map each local action to real GitLab state

12. Challenge: choose the correct layer

Your first trigger request returns 201 but the client process crashes before writing its journal. On restart, should you simply resend the POST? Explain which state is missing, what duplicate side effect could occur, and what production design would make the request safely recoverable. The key is not a different GitLab YAML keyword; it is durable orchestration identity plus an idempotent downstream operation.

13. Cleanup

test "$(basename "$PWD")" = glci-ch33-lab
kill "$(cat .local/mock-server.pid)" 2>/dev/null || true
rm -f .local/mock-server.pid .local/webhook-signing-token.txt
cd ..
rm -rf glci-ch33-lab
printf 'cleanup=verified-disposable-lab-removed
'

In a live project, cleanup would target the exact disposable schedule/trigger/webhook IDs created by the lab. Never delete “the newest trigger” or “latest pipeline.”

Knowledge check

14. Summary

You now have a runnable model for external GitLab orchestration without a real credential: guarded creation, exact pipeline correlation, status polling, 429 handling, duplicate suppression, source/ref/SHA evidence and signed-webhook verification. The control principle is stable across real GitLab.com/Self-Managed/Dedicated deployments even when endpoint limits and authorization policies differ.

Official references and version notes

Documentation verification date: 2026-09-12. Current GitLab documentation places pipeline schedules, pipeline trigger tokens, the Pipelines API, project webhooks and ChatOps on Free/Premium/Ultimate unless a narrower feature is explicitly noted. Pipeline inputs for schedules and trigger/API creation are generally available in current GitLab releases. Scheduled pipelines execute with the schedule owner's permissions; a manual run of a schedule uses the permissions of the user who starts that manual run. Trigger-token pipelines report CI_PIPELINE_SOURCE=trigger; pipelines created through the Pipelines API report api; schedules report schedule; ChatOps reports chat; and a job-token call to the trigger endpoint creates a downstream multi-project pipeline with source pipeline. New webhooks can use HMAC-SHA256 signing tokens following Standard Webhooks; GitLab 19.1 documentation recommends signing tokens over the legacy plain X-Gitlab-Token secret. The mandatory labs below are local-only and use Python's standard library; no GitLab token, network account, webhook endpoint or live project is required. The local server intentionally resembles the trigger and pipeline-status endpoints but is not a GitLab emulator. Its rate-limit and lifecycle transitions exist only to make orchestration state observable without external side effects.

• Scheduled pipelines — official reference.
• Pipeline schedules API — official reference.
• Trigger pipelines with the API — official reference.
• Pipeline trigger tokens API — official reference.
• Pipelines API — official reference.
• REST API pagination and rate limits — official reference.
• CI/CD pipeline creation limits — official reference.
• Predefined CI/CD variables — official reference.
• Job rules and CI_PIPELINE_SOURCE values — official reference.
• CI/CD job token — official reference.
• Fine-grained job-token permissions — official reference.
• Webhooks — official reference.
• Webhook events — official reference.
• ChatOps — official reference.
• Access token scopes — official reference.

Current assumptions used in this chapter: Version-sensitive YAML, Runner/executor behavior, APIs, security features, policy controls, tiers, and deprecations must be verified against the exact GitLab, GitLab Runner, tool, and external-system versions used in production. Preserve the exact project/ref/SHA, compiled configuration, pipeline/job IDs, runner/tool versions, and external target evidence used for reproducible work.