New to Kestra?
Use blueprints to kickstart your first workflows.
IANA Time Zone Database Release Impact Watcher
Detect new IANA tzdata releases with an HTTP trigger, diff zone rules, match your time zone inventory and alert owning teams before DST or offset changes.
Categories
CoreInfrastructure
id: iana-tzdata-release-impact-watcher
namespace: company.team
description: |
Notices a new IANA time zone database release, diffs its data against the previous release, matches the changed zones against your inventory and alerts each owning team with the date and days left.
inputs:
- id: inventory
type: JSON
displayName: Time zone inventory
description: JSON list of {"zone", "owning_team", "systems"}. SAMPLE DATA,
replace it with your own. It includes a backward link (Canada/Central), a
renamed zone (Europe/Kiev), a zone merged in 2022 (Europe/Amsterdam) and a
typo (Europe/Kyiw).
defaults: |
[
{"zone": "America/Winnipeg", "owning_team": "payroll", "systems": ["shift-scheduler", "timesheet-api"]},
{"zone": "Canada/Central", "owning_team": "billing", "systems": ["legacy-invoicing"]},
{"zone": "America/Edmonton", "owning_team": "field-ops", "systems": ["dispatch"]},
{"zone": "America/Inuvik", "owning_team": "field-ops", "systems": ["dispatch"]},
{"zone": "America/Vancouver", "owning_team": "payroll", "systems": ["shift-scheduler"]},
{"zone": "Africa/Casablanca", "owning_team": "customer-support", "systems": ["callback-scheduler"]},
{"zone": "Europe/Dublin", "owning_team": "finance", "systems": ["ledger-close"]},
{"zone": "Europe/Kiev", "owning_team": "mobile", "systems": ["push-scheduler"]},
{"zone": "Europe/Amsterdam", "owning_team": "data-platform", "systems": ["etl-cron"]},
{"zone": "America/New_York", "owning_team": "billing", "systems": ["invoice-run"]},
{"zone": "Asia/Tokyo", "owning_team": "mobile", "systems": ["push-scheduler"]},
{"zone": "Europe/Kyiw", "owning_team": "mobile", "systems": ["push-scheduler"]}
]
- id: replay_version
type: STRING
displayName: Replay one release
required: false
validator: ^([0-9]{4}[a-z])?$
description: Reprocess one release (for example 2026e or 2022b) against its
predecessor. Ignores and does not move the stored version.
- id: mode
type: SELECT
displayName: Mode
values: [ watch, replay_failed ]
defaults: watch
description: watch = process every release published since the stored version.
replay_failed = retry only the releases in the dead-letter list.
- id: news_url
type: STRING
displayName: NEWS file
defaults: https://data.iana.org/time-zones/tzdb/NEWS
description: The release index. https://ftp.iana.org/tz/tzdb/NEWS is the
uncached IANA origin.
- id: tarball_mirror
type: STRING
displayName: Release tarball mirror
defaults: https://data.iana.org/time-zones/releases
description: Folder that holds tzdataVERSION.tar.gz, for example an internal
artifact mirror. https://ftp.iana.org/tz/releases has the same files.
- id: horizon_days
type: INT
displayName: Look-ahead (days)
defaults: 730
min: 30
max: 3650
description: How far ahead future changes are compared.
- id: max_releases
type: INT
displayName: Max releases per run
defaults: 6
min: 1
max: 20
description: Upper bound for a catch-up after a long pause. The newest releases
are kept.
variables:
input_problem: >-
{{ inputs.inventory | jq('if type != "array" or length == 0 or length > 5000
then "inventory must be a JSON list of 1 to 5000 rows" elif any(.[]; type !=
"object") then "every inventory row must be a JSON object" else "" end') |
first }}
failed_task: >-
{{ errorLogs() | jq('[.[] | select(.taskId != null and (.taskId |
test("^(slack_|log_failure)") | not))] | (last // {}) | .taskId //
"unknown"') | first }}
failure_message: >-
{{ errorLogs() | jq('[.[] | select(.taskId != null and (.taskId |
test("^(slack_|log_failure)") | not))] as $e | ($e | last | .taskId) as $t |
[$e[] | select(.taskId == $t) | .message][0] // "see the logs"') | first |
abbreviate(300) }}
execution_link: "{{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace
}}/{{ flow.id }}/{{ execution.id }}"
checks:
- when: "{{ render(vars.input_problem) is empty }}"
message: Invalid inventory. Send a JSON list of 1 to 5000 objects with zone,
owning_team and systems.
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ inputs.mode == 'watch' or (inputs.replay_version ?? '') is empty }}"
message: Use either replay_version or mode replay_failed, not both.
style: ERROR
behavior: BLOCK_EXECUTION
concurrency:
limit: 1
behavior: QUEUE
sla:
- id: run_deadline
type: MAX_DURATION
duration: PT30M
behavior: FAIL
labels:
sla: breached
triggers:
- id: new_tzdb_release
type: io.kestra.plugin.core.http.Trigger
description: Polls the 6-byte version file on IANA's uncached origin every 30
minutes and starts a run only when it differs from the last processed
version in KV. The CDN copy on data.iana.org can be up to a day old.
uri: https://ftp.iana.org/tz/tzdb/version
interval: PT30M
responseCondition: "{{ response.statusCode == 200 and (response.body | trim |
length) <= 8 and (response.body | trim) != (kv('tzwatch_last_version',
errorOnMissing=false) ?? '') }}"
tasks:
- id: validate_inputs
type: io.kestra.plugin.core.execution.Fail
description: Same rule as the flow check, repeated so that trigger-created
executions fail fast too.
condition: "{{ render(vars.input_problem) is not empty }}"
errorMessage: "Invalid inventory: {{ render(vars.input_problem) }}"
- id: dead_letters
type: io.kestra.plugin.core.kv.GetKeys
prefix: tzwatch_dlq_
- id: fetch_news
type: io.kestra.plugin.core.http.Download
description: Hard fail after retries. Without NEWS there is no release list, so
nothing can be planned. The query string makes the CDN fetch a fresh copy
once per new version.
uri: "{{ inputs.news_url }}?v={{ (trigger.body ?? inputs.replay_version ??
'manual') | trim }}"
headers:
User-Agent: kestra-blueprint-tzdata-watcher/1.0
timeout: PT1M
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 4
- id: plan
type: io.kestra.plugin.graalvm.js.Eval
description: Reads the release headers in NEWS and decides which releases to
process, each paired with its predecessor. Runs in the worker JVM, no
container.
outputs: [ items, count, latest, stored, note ]
script: |
(function () {
const news = {{ read(outputs.fetch_news.uri) | toJson }};
const rel = [];
const re = /^Release ([0-9]{4}[a-z]) - ([0-9]{4}-[0-9]{2}-[0-9]{2})/gm;
let m;
while ((m = re.exec(news)) !== null) rel.push({ version: m[1], released: m[2] });
if (rel.length < 20) throw new Error('NEWS lists only ' + rel.length + ' releases; the mirror returned something else');
rel.forEach((r, i) => { r.previous = i + 1 < rel.length ? rel[i + 1].version : null; });
const latest = rel[0].version, stored = {{ (kv('tzwatch_last_version', errorOnMissing=false) ?? '') | toJson }};
const polled = {{ ((trigger.body ?? '') | trim) | toJson }}, replay = {{ (inputs.replay_version ?? '') | toJson }};
const dlq = {{ outputs.dead_letters.keys | toJson }}.map((k) => k.replace('tzwatch_dlq_', ''));
let picked = [], note;
if (replay) {
picked = rel.filter((r) => r.version === replay && r.previous);
if (!picked.length) throw new Error('replay_version ' + replay + ' is not a release in NEWS with a predecessor (latest is ' + latest + ')');
note = 'REPLAY of ' + replay + ' against ' + picked[0].previous + '; the stored version is not changed';
} else if ({{ inputs.mode | toJson }} === 'replay_failed') {
picked = rel.filter((r) => dlq.indexOf(r.version) >= 0 && r.previous).reverse();
note = 'REPLAY of ' + picked.length + ' dead-lettered releases: ' + (dlq.join(', ') || 'none');
} else if (polled && polled !== latest) {
note = 'WAIT: the version file says ' + polled + ' but NEWS starts with ' + latest + '. The next poll retries.';
} else if (!stored || rel.findIndex((r) => r.version === stored) < 0) {
picked = [rel[0]];
note = 'BASELINE: ' + (stored ? 'stored version ' + stored + ' is unknown' : 'no stored version') + ', analysing only the latest release ' + latest;
} else {
picked = rel.slice(0, rel.findIndex((r) => r.version === stored)).reverse();
note = picked.length ? picked.length + ' new release(s) since ' + stored : 'UP TO DATE: ' + latest + ' was already processed';
}
const max = {{ inputs.max_releases }}, inv = {{ inputs.inventory | toJson }};
if (picked.length > max) { note += '; only the newest ' + max + ' are processed'; picked = picked.slice(-max); }
const common = { mirror: {{ inputs.tarball_mirror | toJson }}, horizon_days: {{ inputs.horizon_days }},
now: {{ execution.startDate | date("yyyy-MM-dd'T'HH:mm:ss'Z'", timeZone='UTC') | toJson }},
inventory: typeof inv === 'string' ? JSON.parse(inv) : inv }; // trigger-created runs pass the JSON default as text
return { items: JSON.stringify(picked.map((r) => Object.assign({}, r, common))), count: picked.length, latest: latest, stored: stored, note: note };
})()
- id: has_work
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.plan.outputs.count > 0 }}"
then:
- id: analyze_releases
type: io.kestra.plugin.core.flow.Loop
description: One iteration per release, two at a time. Each iteration is a
try/catch, so one bad release cannot stop the others.
values: "{{ outputs.plan.outputs.items }}"
concurrencyLimit: 2
outputs:
- id: result
type: JSON
value: "{{ {'version': fromJson(item.value).version, 'impacts':
outputs.diff_release.outputFiles['impacts.jsonl'] ?? '',
'summary': outputs.diff_release.vars.summary ?? {}} | toJson }}"
tasks:
- id: release_try
type: io.kestra.plugin.core.flow.Sequential
description: allowFailure turns a failed release into a WARNING iteration; the
local errors branch records it in the dead-letter list.
allowFailure: true
tasks:
- id: fetch_both
type: io.kestra.plugin.core.flow.Dag
description: The two downloads have no dependency on each other, so they run in
parallel; the diff waits for both.
tasks:
- task:
id: fetch_previous
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(item.value).mirror }}/tzdata{{ fromJson(item.value).previous
}}.tar.gz"
headers:
User-Agent: kestra-blueprint-tzdata-watcher/1.0
timeout: PT2M
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 4
- task:
id: fetch_release
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(item.value).mirror }}/tzdata{{ fromJson(item.value).version
}}.tar.gz"
headers:
User-Agent: kestra-blueprint-tzdata-watcher/1.0
timeout: PT2M
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 4
- task:
id: diff_release
type: io.kestra.plugin.scripts.python.Script
description: |
Standard library only. Reads both tarballs in memory, builds every zone's UTC offset and abbreviation timeline (a small zic), compares old and new, classifies each changed name and matches the inventory. It also cuts this release's section out of NEWS.
No retry: the inputs are immutable release files, so a failure here is a parsing problem and goes to the dead-letter list.
timeout: PT5M
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.13-slim
inputFiles:
previous.tar.gz: "{{ outputs.fetch_previous.uri }}"
release.tar.gz: "{{ outputs.fetch_release.uri }}"
inventory.json: "{{ fromJson(item.value).inventory | toJson }}"
outputFiles:
- impacts.jsonl
env:
VERSION: "{{ fromJson(item.value).version }}"
PREVIOUS: "{{ fromJson(item.value).previous }}"
RELEASED: "{{ fromJson(item.value).released }}"
NOW: "{{ fromJson(item.value).now }}"
HORIZON_DAYS: "{{ fromJson(item.value).horizon_days }}"
script: |
import bisect, calendar, datetime as dt, json, os, re, sys, tarfile
VERSION, PREVIOUS = os.environ["VERSION"], os.environ["PREVIOUS"]
NOW = int(dt.datetime.fromisoformat(os.environ["NOW"].replace("Z", "+00:00")).timestamp())
END = NOW + int(os.environ["HORIZON_DAYS"]) * 86400
YMAX, EPOCH, NEG = dt.datetime.fromtimestamp(END, dt.timezone.utc).year + 1, dt.date(1970, 1, 1).toordinal(), -2 ** 40
MAIN = "africa antarctica asia australasia europe northamerica southamerica etcetera factory backward pacificnew systemv".split()
MON = {m: i + 1 for i, m in enumerate("jan feb mar apr may jun jul aug sep oct nov dec".split())}
WDAY = {d: i for i, d in enumerate("mon tue wed thu fri sat sun".split())}
ALERT = {"FUTURE", "IN_EFFECT", "RENAMED", "REMOVED"}
stamp = lambda t, f: (dt.datetime(1970, 1, 1) + dt.timedelta(seconds=t)).strftime(f)
def fail(code, msg):
print(code + ": " + msg, file=sys.stderr)
sys.exit(3)
def read_tar(path): # members are read in memory, never extracted, so archive paths cannot escape
try:
with tarfile.open(path, "r:gz") as tf:
return {m.name.lstrip("./"): tf.extractfile(m).read().decode("utf-8", "replace")
for m in tf.getmembers() if m.isfile() and "/" not in m.name.lstrip("./")}
except (tarfile.TarError, OSError, EOFError) as e:
fail("BAD_TARBALL", "%s is not a readable tzdata tarball (%s)" % (path, e))
def parse(files, names):
rules, zones, links, cur = {}, {}, {}, None
for name in names:
for raw in files.get(name, "").splitlines():
f = raw.split("#", 1)[0].split()
if not f:
continue
if raw[0].isspace() and cur:
zones[cur].append(f)
else:
kw, cur = f[0].lower(), None
if kw in ("rule", "r"):
rules.setdefault(f[1], []).append(f[2:])
elif kw in ("zone", "z"):
cur, zones[f[1]] = f[1], [f[2:]]
elif kw in ("link", "l"):
links[f[2]] = f[1]
if cur and len(zones[cur][-1]) <= 3:
cur = None # a line without UNTIL ends the zone
return rules, zones, links
def secs(s):
s = s.rstrip("wsugzd")
if s in ("-", ""):
return 0
p = [float(x) for x in s.lstrip("+-").split(":")] + [0, 0]
return (-1 if s.startswith("-") else 1) * int(p[0] * 3600 + p[1] * 60 + p[2])
def day_ord(y, m, on):
if on.startswith("last"):
o = dt.date(y, m, calendar.monthrange(y, m)[1]).toordinal()
while (o - 1) % 7 != WDAY[on[4:7].lower()]:
o -= 1
return o
for op, step in ((">=", 1), ("<=", -1)):
if op in on:
wd, n = on.split(op)
o = dt.date(y, m, 1).toordinal() + int(n) - 1
while (o - 1) % 7 != WDAY[wd[:3].lower()]:
o += step
return o
return dt.date(y, m, 1).toordinal() + int(on) - 1
local_secs = lambda y, m=1, on="1", t="0": (day_ord(y, m, on) - EPOCH) * 86400 + secs(t)
suffix = lambda t: {"s": "s", "u": "u", "g": "u", "z": "u"}.get(t[-1:], "w")
year_of = lambda t: dt.date.fromordinal(max(1, min(t // 86400 + EPOCH, dt.date.max.toordinal()))).year
def zfmt(off):
h, r = divmod(abs(off), 3600)
return ("-" if off < 0 else "+") + "%02d" % h + ("%02d" % (r // 60) if r else "") + ("%02d" % (r % 60) if r % 60 else "")
def abbr(fmt, letter, save, off):
if "%z" in fmt:
return fmt.replace("%z", zfmt(off))
return fmt.split("/")[1 if save else 0] if "/" in fmt else fmt.replace("%s", "" if letter == "-" else letter)
def timeline(zone, rules):
"""(utc, utoff, abbr) state changes: a small zic. Checked against zic output for every 2026e zone, 1970-2030."""
out, start, prev_std = [], NEG, None
def push(t, off, ab):
if not out or out[-1][1:] != (off, ab):
out.append((t, off, ab))
for f in zone:
std, rname, fmt, until = secs(f[0]), f[1], f[2], f[3:]
ul = local_secs(int(until[0]), MON[until[1][:3].lower()] if len(until) > 1 else 1,
until[2] if len(until) > 2 else "1", until[3] if len(until) > 3 else "0") if until else None
us = suffix(until[3]) if len(until) > 3 else "w"
until_utc = lambda save: None if ul is None else ul - (0 if us == "u" else std) - (save if us == "w" else 0)
if rname == "-" or re.match(r"^-?[0-9]", rname):
save = 0 if rname == "-" else secs(rname)
push(start, std + save, abbr(fmt, "-", save, std + save))
else:
rs = rules.get(rname, [])
std_letters = [r[7] for r in rs if secs(r[6]) == 0]
save, letter, ev = 0, std_letters[0] if std_letters else "-", []
y0, y1 = (year_of(start) - 1 if start > NEG else 1800), min(int(until[0]) if until else YMAX, YMAX)
for frm, to, _typ, inm, on, at, sv, lt in rs:
fy = -9999 if frm.lower().startswith("mi") else int(frm)
ty = fy if to.lower().startswith("o") else (9999 if to.lower().startswith("ma") else int(to))
ev += [(local_secs(y, MON[inm[:3].lower()], on, at), suffix(at), secs(sv), lt) for y in range(max(fy, y0), min(ty, y1) + 1)]
ev.sort(key=lambda e: e[0])
started = start == NEG
if started:
push(start, std, abbr(fmt, letter, 0, std))
for loc, suf, sv, lt in ev:
conv = lambda wall, st=std: loc - {"u": 0, "s": st}.get(suf, wall)
if not started: # at a line start the clocks still show the previous line's offsets
if conv(out[-1][1] if out else std + save, std if prev_std is None else prev_std) <= start:
save, letter = sv, lt
continue
started = True
if conv(std + save) <= start:
save, letter = sv, lt
push(start, std + save, abbr(fmt, letter, save, std + save))
continue
push(start, std + save, abbr(fmt, letter, save, std + save))
t, u = conv(std + save), until_utc(save)
if u is not None and t >= u:
break
save, letter = sv, lt
push(t, std + save, abbr(fmt, letter, save, std + save))
if not started:
push(start, std + save, abbr(fmt, letter, save, std + save))
start, prev_std = (until_utc(save) if until else None), std
if start is None or start > END:
break
return out
def at(states, t):
i = bisect.bisect_right([s[0] for s in states], t) - 1
return states[i][1:] if i >= 0 else None
describe = lambda s: "%s UTC%s" % (s[1], zfmt(s[0])) if s else "n/a"
def body(name, zones, links, rules):
z = links.get(name, name)
return (z, json.dumps([zones[z], [rules[r] for r in sorted({l[1] for l in zones[z] if l[1] in rules})]])) if z in zones else (None, None)
def compare(a, b):
first = future = run = now_run = None
for t in sorted({s[0] for s in a} | {s[0] for s in b} | {NOW}):
if t > END:
break
differs = at(a, t) != at(b, t)
run = (run if run is not None else t) if differs else None
if differs:
first = first if first is not None else t
future = future if future is not None or t < NOW else t
now_run = run if t == NOW else now_run
if first is None:
return {"kind": "NO_EFFECT"}
kind, t = ("IN_EFFECT", now_run) if now_run is not None else (("FUTURE", future) if future is not None else ("PAST", first))
r = {"kind": kind, "t": t if t > NEG else None, "before": describe(at(a, t)), "after": describe(at(b, t))}
r["local"] = stamp(t + at(b, t)[0], "%Y-%m-%d %H:%M") if t > NEG else "earliest (LMT) entry"
return r
old_f, new_f = read_tar("previous.tar.gz"), read_tar("release.tar.gz")
if (new_f.get("version", "").strip() or VERSION) != VERSION:
fail("WRONG_VERSION", "release.tar.gz says %s, expected %s" % (new_f["version"].strip(), VERSION))
(o_rules, o_zones, o_links), (n_rules, n_zones, n_links) = parse(old_f, MAIN), parse(new_f, MAIN)
if len(n_zones) < 300:
fail("PARSE_ERROR", "only %d zones parsed from tzdata%s" % (len(n_zones), VERSION))
o_back, n_back = parse(old_f, ["backzone"])[1], parse(new_f, ["backzone"])[1]
# NEWS: this release's section (or the top one when the header is wrong), headings, 'Briefly', zone ids and dates.
news = new_f.get("NEWS", "")
m = re.search(r"^Release %s - .*?(?=^Release |\Z)" % re.escape(VERSION), news, re.S | re.M)
top = re.search(r"^Release (\S+) - .*?(?=^Release |\Z)", news, re.S | re.M)
section = m.group(0) if m else (top.group(0) if top else "")
warn = [] if m else ["NEWS in tzdata%s.tar.gz has no 'Release %s' header; used its top section (headed %s)" % (VERSION, VERSION, top.group(1) if top else "nothing")]
headings = re.findall(r"^ (Changes to [^\n]+|Other [^\n]+)$", section, re.M)
brief = re.search(r"^ Briefly:\n(.*?)\n\n", section, re.S | re.M)
brief = " ".join(brief.group(1).split()) if brief else ""
fut = re.search(r"^ Changes to (?:past and )?future time ?stamps\n(.*?)(?=^ \S|\Z)", section, re.S | re.M)
fut = fut.group(1) if fut else ""
news_zones = sorted(set(re.findall(r"\b[A-Z][A-Za-z_]+/[A-Za-z_]+(?:/[A-Za-z_]+)?\b", fut)))
news_says = "FUTURE" if fut else ("PAST_ONLY" if any("past" in h for h in headings) else "CODE_OR_DOCS_ONLY")
# Every name in the database whose resolved data changed, not only the inventory.
results, cache = {}, {}
for name in sorted(set(o_zones) | set(o_links) | set(n_zones) | set(n_links)):
(oz, ob), (nz, nb) = body(name, o_zones, o_links, o_rules), body(name, n_zones, n_links, n_rules)
r = None
if nz is None:
r = {"kind": "REMOVED" if oz else "UNKNOWN", "detail": "no longer in the database"}
elif oz is None:
r = {"kind": "NEW_NAME", "detail": "introduced in this release; hosts with older tzdata do not know it"}
else:
if ob != nb:
r = dict(cache.setdefault((oz, nz), compare(timeline(o_zones[oz], o_rules), timeline(n_zones[nz], n_rules))))
if name in o_zones and name in n_links: # a zone became a backward-compatibility link
r = r if r and r["kind"] in ("FUTURE", "IN_EFFECT") else dict(r or {}, kind="RENAMED" if nz not in o_zones else "MERGED")
r["detail"] = "now a link to " + nz + ("; pre-1970 data moved to backzone" if r["kind"] == "MERGED" else "")
if (r is None or r["kind"] == "NO_EFFECT") and name in n_back and o_back.get(name) != n_back[name]:
r = {"kind": "BACKZONE", "detail": "only its backzone entry changed (pre-1970, used only with PACKRATDATA=backzone)"}
if r and r["kind"] != "UNKNOWN":
results[name] = dict(r, canonical=nz or oz)
inventory, rows, counts = json.load(open("inventory.json", encoding="utf-8")), [], {}
known = set(n_zones) | set(n_links) | set(o_zones) | set(o_links)
for i, row in enumerate(inventory, 1):
row = row if isinstance(row, dict) else {}
zone, team = (str(row.get(k) or "").strip() for k in ("zone", "owning_team"))
systems = row.get("systems", [])
systems = ", ".join(map(str, systems)) if isinstance(systems, list) else str(systems)
if not zone or not team:
r = {"kind": "INVALID", "detail": "inventory row %d needs zone and owning_team" % i}
elif zone not in known:
r = {"kind": "INVALID", "detail": "inventory row %d: %s is not a tz database name" % (i, zone)}
elif zone not in results:
continue
else:
r = results[zone]
if zone in n_links and r["kind"] in ("FUTURE", "IN_EFFECT", "PAST") and not r.get("detail"):
r = dict(r, detail="via link %s -> %s, prefer the canonical name" % (zone, r["canonical"]))
t = r.get("t")
days = t // 86400 - NOW // 86400 if t and r["kind"] in ("FUTURE", "IN_EFFECT") else None
rows.append({"release": VERSION, "previous": PREVIOUS, "released_on": os.environ.get("RELEASED", ""),
"owning_team": team or "unassigned", "zone": zone or "(missing)", "systems": systems,
"canonical": r.get("canonical", ""), "kind": r["kind"], "alert": r["kind"] in ALERT,
"effective_utc": stamp(t, "%Y-%m-%dT%H:%MZ") if t else "", "effective_local": r.get("local", ""),
"days_until": days, "ledger_ttl": "P%dD" % (max(days or 0, 0) + 30),
"ledger_key": "tzwatch_sent_" + re.sub(r"[^a-z0-9]+", "_", "%s_%s_%s_%s" % (team, zone, VERSION, r["kind"]), flags=re.I).lower(),
"before": r.get("before", ""), "after": r.get("after", ""), "detail": r.get("detail", ""), "news": brief[:300]})
counts[r["kind"]] = counts.get(r["kind"], 0) + 1
with open("impacts.jsonl", "w", encoding="utf-8") as out:
out.writelines(json.dumps(rec, ensure_ascii=False) + "\n" for rec in rows)
db = {}
for n, r in sorted(results.items()):
if n in n_zones or n in o_zones:
db.setdefault(r["kind"], []).append(n + (" " + r["local"] if r["kind"] in ("FUTURE", "IN_EFFECT") else ""))
db = {k: v[:10] + (["+%d more" % (len(v) - 10)] if len(v) > 10 else []) for k, v in db.items()}
future = [n for n, r in results.items() if r["kind"] in ("FUTURE", "IN_EFFECT")]
if fut and not future:
warn.append("NEWS announces future changes but the data diff found none within %s days" % os.environ["HORIZON_DAYS"])
if future and not fut:
warn.append("the data diff found future changes that NEWS does not announce: " + ", ".join(sorted(future)[:5]))
summary = {"release": VERSION, "previous": PREVIOUS, "news_says": news_says, "future_impact": bool(future), "news_brief": brief[:400], "news_headings": headings,
"news_zones": news_zones, "news_dates": sorted(set(re.findall(r"\b20[0-9]{2}-[01][0-9]-[0-3][0-9]\b", fut))),
"database_changes": db, "names_compared": len(set(n_zones) | set(n_links)), "inventory_rows": len(inventory),
"inventory_hits": counts, "alertable": sum(r["alert"] for r in rows), "warnings": warn,
"not_alerted": ["%s %s: %s %s" % (r["owning_team"], r["zone"], r["kind"], r["detail"] or r["effective_local"]) for r in rows if not r["alert"]][:20]}
print("::" + json.dumps({"outputs": {"summary": summary}}) + "::")
dependsOn: [ fetch_previous, fetch_release ]
- id: clear_dead_letter
type: io.kestra.plugin.core.kv.Delete
description: A successful analysis heals an earlier dead-letter entry for this
release.
key: "tzwatch_dlq_{{ fromJson(item.value).version }}"
errorOnMissing: false
errors:
- id: record_dead_letter
type: io.kestra.plugin.core.kv.Set
description: One key per release. The reason comes from which step has no
output; attempts survive replays.
key: "tzwatch_dlq_{{ fromJson(item.value).version }}"
kvType: JSON
value: >-
{{ {
'version': fromJson(item.value).version,
'previous': fromJson(item.value).previous,
'reason': (outputs.fetch_previous.uri ?? '') is empty ? 'UNREACHABLE tzdata' ~ fromJson(item.value).previous ~ '.tar.gz' : ((outputs.fetch_release.uri ?? '') is empty ? 'UNREACHABLE tzdata' ~ fromJson(item.value).version ~ '.tar.gz' : 'ANALYSIS_FAILED (see diff_release logs)'),
'mirror': fromJson(item.value).mirror,
'attempts': ((kv('tzwatch_dlq_' ~ fromJson(item.value).version, errorOnMissing=false) ?? {}).attempts ?? 0) + 1,
'last_failed_at': now() | date("yyyy-MM-dd'T'HH:mm:ss'Z'", timeZone='UTC'),
'execution_id': execution.id
} | toJson }}
- id: require_one_release
type: io.kestra.plugin.core.execution.Fail
description: Partial results are fine, none are not. If every release failed
(for example the mirror is down), fail without moving the stored
version, so the trigger fires again on the next poll.
condition: "{{ (outputs.analyze_releases.outputs | jq('[.[].outputs.result |
select(.impacts != \"\")] | length') | first) == 0 }}"
errorMessage: "None of the {{ outputs.plan.outputs.count }} releases could be
analysed:{% for r in (outputs.analyze_releases.outputs |
jq('[.[].outputs.result.version] | sort') | first) %} {{ r }} = {{
(kv('tzwatch_dlq_' ~ r, errorOnMissing=false) ?? {}).reason ??
'unknown' }};{% endfor %} they are in the dead-letter list. Fix the
cause, then run mode=replay_failed."
- id: merge_impacts
type: io.kestra.plugin.core.storage.Concat
description: Oldest release first, because DeduplicateItems keeps the last
record per key.
files: "{{ outputs.analyze_releases.outputs | jq('[.[].outputs.result |
select(.impacts != \"\")] | sort_by(.version) | [.[].impacts]') |
first | toJson }}"
extension: .jsonl
- id: dedupe_impacts
type: io.kestra.plugin.core.storage.DeduplicateItems
description: One record per team, zone and alert class. When several releases
touched the same zone, the newest wins, and an inventory typo reported
by every release appears once.
from: "{{ outputs.merge_impacts.uri }}"
expr: "{{ owning_team }}|{{ zone }}|{{ alert }}"
- id: impact_report
type: io.kestra.plugin.serdes.csv.IonToCsv
description: Every matched record, alerted or not, as the flow output.
from: "{{ outputs.dedupe_impacts.uri }}"
- id: alertable_impacts
type: io.kestra.plugin.core.storage.FilterItems
description: Keeps FUTURE, IN_EFFECT, RENAMED and REMOVED. Past-only,
backzone-only, no-effect and invalid rows are dropped; log_report
lists them, nobody is paged.
from: "{{ outputs.dedupe_impacts.uri }}"
filterCondition: "{{ alert }}"
filterType: INCLUDE
errorOrNullBehavior: EXCLUDE
- id: unsent_impacts
type: io.kestra.plugin.core.storage.FilterItems
description: Drops records whose alert ledger key exists. This makes re-runs,
overlapping catch-ups and replays idempotent for the teams.
from: "{{ outputs.alertable_impacts.uri }}"
filterCondition: "{{ kv(ledger_key, errorOnMissing=false) is not null }}"
filterType: EXCLUDE
- id: notify_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.unsent_impacts.processedItemsTotal -
outputs.unsent_impacts.droppedItemsTotal > 0 }}"
then:
- id: unsent_json
type: io.kestra.plugin.serdes.json.IonToJson
from: "{{ outputs.unsent_impacts.uri }}"
newLine: true
- id: split_by_team
type: io.kestra.plugin.core.storage.Split
description: One file per owning_team, grouped by the first capture group of the
regex. No database needed.
from: "{{ outputs.unsent_json.uri }}"
regexPattern: '"owning_team":"([^"]+)"'
- id: notify_teams
type: io.kestra.plugin.core.flow.Loop
values: "{{ outputs.split_by_team.uris }}"
outputs:
- id: notice
type: JSON
value: "{{ {'team': outputs.team_message.outputs.team, 'text':
outputs.team_message.outputs.text} | toJson }}"
tasks:
- id: team_message
type: io.kestra.plugin.graalvm.js.Eval
outputs: [ team, text, ledger ]
script: |
(function () {
const rows = {{ read(item.value) | toJson }}.split('\n').filter((l) => l.trim()).map((l) => JSON.parse(l));
rows.sort((a, b) => (a.effective_utc || '9999').localeCompare(b.effective_utc || '9999'));
const when = (r) => r.days_until >= 0 ? 'in ' + r.days_until + ' days' : 'IN EFFECT for ' + -r.days_until + ' days';
const lines = rows.map((r) => r.days_until === null
? '- ' + r.zone + ' (' + r.systems + '): ' + r.kind + ' in tzdata ' + r.release + ', ' + r.detail + '. Update your configuration.'
: '- ' + r.zone + ' (' + r.systems + '): ' + r.before + ' -> ' + r.after + ' from ' + r.effective_local + ' local, ' + when(r) + ' (tzdata ' + r.release + (r.detail ? '; ' + r.detail : '') + ')');
return { team: rows[0].owning_team, ledger: JSON.stringify(rows.map((r) => ({ key: r.ledger_key, ttl: r.ledger_ttl }))),
text: 'Time zone changes for ' + rows[0].owning_team + ':\n' + lines.join('\n') + '\nNEWS: ' + rows[rows.length - 1].news };
})()
- id: log_team_alert
type: io.kestra.plugin.core.log.Log
level: WARN
message: "{{ outputs.team_message.outputs.text }}"
- id: slack_team_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: allowFailure, because the WARN log above (and log_report) is the
alert of record when Slack is not configured.
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "{{ outputs.team_message.outputs.text }}"
- id: remember_alerts
type: io.kestra.plugin.core.flow.Loop
values: "{{ outputs.team_message.outputs.ledger }}"
tasks:
- id: ledger_entry
type: io.kestra.plugin.core.kv.Set
description: One key per team, zone, release and kind. The TTL ends 30 days
after the change takes effect, so the ledger cleans itself
up.
key: "{{ fromJson(item.value).key }}"
value: "{{ execution.id }}"
ttl: "{{ fromJson(item.value).ttl }}"
- id: log_report
type: io.kestra.plugin.core.log.Log
message: |
{{ outputs.plan.outputs.note }}
{% for r in (outputs.analyze_releases.outputs | jq('[.[].outputs.result] | sort_by(.version)') | first) %}{% if r.impacts == '' %}tzdata {{ r.version }}: FAILED, dead-lettered
{% else %}tzdata {{ r.version }} (vs {{ r.summary.previous }}): NEWS says {{ r.summary.news_says }}{{ r.summary.future_impact ? '' : ' (no future impact anywhere, nothing to alert)' }}; database changes {{ r.summary.database_changes | toJson }}; inventory {{ r.summary.inventory_hits | toJson }}{% if r.summary.warnings | length > 0 %}; WARNINGS {{ r.summary.warnings | toJson }}{% endif %}
{% endif %}{% endfor %}Records after dedupe: {{ outputs.dedupe_impacts.numKeys }}; logged only: {{ outputs.alertable_impacts.droppedItemsTotal }}; alertable: {{ outputs.unsent_impacts.processedItemsTotal }}, of which already alerted earlier: {{ outputs.unsent_impacts.droppedItemsTotal }}
{% for r in (outputs.analyze_releases.outputs | jq('[.[].outputs.result.summary.not_alerted // [] | .[]] | unique') | first) %}Logged, not alerted: {{ r }}
{% endfor %}{% for n in (outputs.notify_teams.outputs ?? []) %}ALERTED {{ n.outputs.notice.text }}
{% endfor %}
Report: {{ outputs.impact_report.uri }}
- id: advance_pointer
type: io.kestra.plugin.core.kv.Set
description: Only watch runs move the pointer. Failed releases stay in the
dead-letter list, so the trigger stops firing but nothing is lost.
runIf: "{{ inputs.mode == 'watch' and (inputs.replay_version ?? '') is empty }}"
key: tzwatch_last_version
value: "{{ outputs.plan.outputs.latest }}"
- id: flag_partial_result
type: io.kestra.plugin.core.execution.Fail
description: allowFailure turns a partial result into a WARNING execution, so it
stands out and afterExecution reports it. The dead letters are already
recorded.
allowFailure: true
condition: "{{ (outputs.analyze_releases.outputs | jq('[.[].outputs.result |
select(.impacts == \"\")] | length') | first) > 0 }}"
errorMessage: "Some releases failed and are in the dead-letter list: {{
outputs.analyze_releases.outputs | jq('[.[].outputs.result |
select(.impacts == \"\") | .version] | join(\", \")') | first }}. Run
mode=replay_failed."
else:
- id: nothing_to_do
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.plan.outputs.note }}"
errors:
- id: log_failure
type: io.kestra.plugin.core.log.Log
level: ERROR
message: "tzdata watcher {{ (labels.sla ?? '') == 'breached' ? 'was stopped by
the run_deadline SLA' : 'failed in task ' ~ render(vars.failed_task) ~ ':
' ~ render(vars.failure_message) }} {{ render(vars.execution_link) }}"
- id: failure_mute
type: io.kestra.plugin.core.kv.Get
key: tzwatch_failure_muted
errorOnMissing: false
- id: alert_unless_muted
type: io.kestra.plugin.core.flow.If
description: The trigger retries every 30 minutes during an outage. A KV key
with a 6 hour TTL keeps that to one alert per outage.
condition: "{{ (outputs.failure_mute.value ?? '') is empty }}"
then:
- id: slack_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "tzdata watcher failed in task {{ render(vars.failed_task) }}: {{
render(vars.failure_message) }}
{{ render(vars.execution_link) }}"
- id: mute_failures
type: io.kestra.plugin.core.kv.Set
key: tzwatch_failure_muted
value: "{{ execution.id }}"
ttl: PT6H
finally:
- id: run_summary
type: io.kestra.plugin.core.log.Log
description: Runs after success, failure and the errors branch.
message: "Run summary: {{ outputs.plan.outputs.note ?? 'no plan (failed before
planning)' }}; stored version {{ kv('tzwatch_last_version',
errorOnMissing=false) ?? 'none' }}; dead letters {{
outputs.dead_letters.keys | toJson }} before this run"
afterExecution:
- id: report_bad_end
type: io.kestra.plugin.core.flow.If
description: |
Covers what errors does not: WARNING (dead-lettered releases), KILLED, CANCELLED and SLA breaches. FAILED runs are alerted, and muted, by errors.
An explicit allowlist, because on Kestra 2.0.5 afterExecution can be evaluated at the first failed attempt of a retried task, while the state is RETRYING.
condition: "{{ execution.state == 'WARNING' or execution.state == 'KILLED' or
execution.state == 'CANCELLED' or (labels.sla ?? '') == 'breached' }}"
then:
- id: slack_outcome
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "tzdata watcher ended {{ execution.state }}{{ (labels.sla ?? '') ==
'breached' ? ' (stopped by the PT30M SLA)' : '' }}. Dead letters: {{
(outputs.analyze_releases.outputs ?? []) | jq('[.[].outputs.result |
select(.impacts == \"\") | .version] | join(\", \")') | first }} {{
render(vars.execution_link) }}"
outputs:
- id: impact_report
type: STRING
description: Internal-storage URI of the CSV with every matched record.
value: "{{ outputs.impact_report.uri ?? '' }}"
- id: plan
type: STRING
value: "{{ outputs.plan.outputs.note ?? '' }}"
Governments change clocks at short notice: in 2026 British Columbia, Alberta, the Northwest Territories and Manitoba dropped DST and Morocco moved to permanent +00. Schedulers, billing cut-offs and calendars break when hosts keep old time zone data. This flow notices a new IANA tz database release soon after it is published, finds which of your zones changed, and tells each owning team the new offset, the date and the days left. The default inventory is SAMPLE DATA (made-up teams and systems). The version file, NEWS and release tarballs come live from IANA.
How it works
new_tzdb_release(HTTP trigger) polls the 6-byteversionfile every 30 minutes and starts a run only when it differs fromtzwatch_last_versionin KV.fetch_newsdownloads NEWS.plan(GraalVM JS) reads its release headers and picks the work: every release since the stored version (at mostmax_releases), only the latest on the first run (BASELINE),replay_version, or the dead-letter list (mode: replay_failed). If the version file is ahead of NEWS, it waits for the next poll.analyze_releases(Loop, two at a time) runs one try/catch per release. Inside it, thefetch_bothDag downloadstzdataPREV.tar.gzandtzdataVER.tar.gzin parallel. Thendiff_release(Python, standard library) reads both tarballs in memory, builds each zone's offset and abbreviation timeline, and compares old and new up tohorizon_daysahead.- Each changed name gets a class: FUTURE, IN_EFFECT (already differs today), PAST (history only), NO_EFFECT (source edits only), RENAMED (now a link to a new name, like Europe/Kiev to Europe/Kyiv in 2022b), MERGED (folded into another zone, pre-1970 data moved to backzone), BACKZONE, REMOVED or NEW_NAME. Links such as Canada/Central resolve to their target. The release's NEWS section is extracted and checked against the diff.
- The inventory is matched. A bad row (unknown zone, missing team) becomes an INVALID record; the batch carries on.
merge_impactsjoins the release files.dedupe_impactskeeps the newest record per team and zone,alertable_impactskeeps FUTURE, IN_EFFECT, RENAMED and REMOVED,unsent_impactsdrops records already in the alert ledger, andsplit_by_teamwrites one file per team.notify_teamslogs a WARN and posts to Slack for each team, then writes one ledger key per record.advance_pointerstores the latest version, in watch mode only.
What this blueprint teaches
HTTP trigger instead of Schedule plus If
A Schedule starts an execution on every tick and an If inside finds nothing to do, so a quiet month leaves thousands of empty runs. io.kestra.plugin.core.http.Trigger evaluates responseCondition in the scheduler and creates an execution only when the version differs from KV, never while its last run is still going. The pointer moves only after a successful run, so the next poll retries a failed one. It polls ftp.iana.org because the CDN copy on data.iana.org can be a day old.
Storage tasks on record files, no database
DeduplicateItems keeps the last record per expr, so order matters: merge_impacts concatenates the oldest release first. FilterItems filters on typed fields ({{ alert }}) and can call kv() per item, which is how unsent_impacts makes re-runs idempotent. Split with regexPattern groups lines by team. JSON lines are valid Ion, so no conversion is needed before the filters.
Dag inside a Loop
fetch_both declares dependsOn instead of nesting Parallel and Sequential: both downloads start together and the diff waits for both.
Per-release try/catch, dead letters and replay
release_try has allowFailure and its own errors branch. A failed release becomes a WARNING iteration and record_dead_letter stores tzwatch_dlq_VERSION (reason, mirror, attempts) while the others carry on. require_one_release fails the run when nothing succeeded; flag_partial_result marks a partial run WARNING. mode: replay_failed retries the list and clear_dead_letter heals it.
errors, finally, afterExecution, sla, checks and KV TTL
errors logs and posts the failed task and execution link; failure_mute, a KV key with a 6 hour ttl, limits an outage to one alert while the trigger retries. finally writes the run summary. afterExecution covers WARNING, KILLED and SLA breaches. run_deadline fails a run after 30 minutes and labels it. checks reject a malformed inventory or conflicting replay inputs before the execution exists. Ledger keys expire 30 days after their change takes effect.
What you get
- One message per team with zone, systems, old and new offset, local time and days left, plus a CSV of every matched record (
impact_report). - No duplicate pages: re-runs, overlapping catch-ups and replays skip ledgered records.
- Releases that only change past timestamps, code or docs are logged, not alerted.
- Backoff retries and timeouts on every download, dead letters with reasons, one outage alert per 6 hours.
Who it's for
- Platform and SRE teams who own schedulers, cron fleets and base images.
- Billing, payroll and calendar teams whose cut-offs depend on local time.
Why orchestrate this with Kestra
- The trigger, state, retries, alert mute and ledger are declared, not coded.
- Each release has its own logs and failure record, and replay is one input.
Prerequisites
- Kestra 2.0 with the Python (Docker runner,
python:3.13-slim), GraalVM, serdes and Slack plugins. - Outbound HTTPS to
ftp.iana.organddata.iana.org, or to your own mirror. No API key.
Secrets
SLACK_WEBHOOK_URL: Slack incoming webhook for team and failure alerts (optional; WARN logs are the alert of record).
Quick start
- Replace
inventorywith your zones, owning teams and systems. - Execute once. With no stored version the run is a BASELINE that analyses only the latest release.
- To see a catch-up, set KV
tzwatch_last_versionto an older release such as2026b; the next poll processes every release since. - Try
replay_version: 2022bfor RENAMED and MERGED records, or2025cfor a past-only release. - Add
SLACK_WEBHOOK_URL.
How to extend
- Route each team to its own channel by adding a secret name to the inventory.
- Load the inventory from a CMDB or from
TZsettings in your repositories. - Open a ticket per FUTURE record and close it when hosts report the new
tzdata.
Links
Tasks
Orchestrate with Kestra
Orchestrate Slack with KestraCreate a Blueprint
Go to the developer platformMore Infrastructure Blueprints
More Core Blueprints
