From 99de0c15e9c47807eb060d76b9ccfcf497aedbc5 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sun, 1 Nov 2026 10:13:49 +0100 Subject: [PATCH] fix(sys-dns-cloudflare-records): bound each record job and retry only transient failures A filesystem stall killed a Cloudflare record job before async_wrapper saw its start notice (Broken pipe), and the shared reap masked it. Each record job now has a request timeout and an attempt deadline; judge.yml reaps every job of an attempt and async_retry_split sorts the failures: hangs, timeouts, connection errors, 429 and 5xx get up to cloudflare_attempts tries, anything else fails at once and names its record. An answer that says the record already stands counts as done. Co-Authored-By: Claude Opus 5.5 --- plugins/filter/async_failures.py | 55 ++++++++ .../defaults/main.yml | 16 +++ .../tasks/apply.yml | 79 +++++++++++ .../tasks/attempt.yml | 15 +++ .../tasks/judge.yml | 34 +++++ .../sys-dns-cloudflare-records/tasks/main.yml | 103 +++------------ .../plugins/filter/test_async_retry_split.py | 125 ++++++++++++++++++ 7 files changed, 341 insertions(+), 86 deletions(-) create mode 100644 roles/sys-dns-cloudflare-records/tasks/apply.yml create mode 100644 roles/sys-dns-cloudflare-records/tasks/attempt.yml create mode 100644 roles/sys-dns-cloudflare-records/tasks/judge.yml create mode 100644 tests/unit/python/plugins/filter/test_async_retry_split.py diff --git a/plugins/filter/async_failures.py b/plugins/filter/async_failures.py index 1f1828ad42..e2fa37316d 100644 --- a/plugins/filter/async_failures.py +++ b/plugins/filter/async_failures.py @@ -18,6 +18,7 @@ phrasing and does not have to know which key carries it. from __future__ import annotations +import re from typing import Any _UNFINISHED = "job did not finish" @@ -105,8 +106,62 @@ def async_failures(results: Any, tolerate: Any) -> list[str]: return out +def _original(result: dict) -> Any: + """The loop value a result belongs to, reaped or polled in place. + + Args: + result: one reaped ``async_status`` result, or one result of a looped + task that polled its own job. + """ + fired = result.get("item") + if isinstance(fired, dict) and "ansible_loop_var" in fired: + return fired.get(fired["ansible_loop_var"]) + return fired + + +def async_retry_split( + results: Any, tolerate: Any, transient: Any +) -> dict[str, list[dict]]: + """Sort failed jobs into those worth another attempt and those to stop for. + + A job that never finished always counts as transient: a hung request, or a + job that died before its module ran and left its job file at "started". + Any other failure is transient only when what it said matches ``transient``. + + Args: + results: reaped ``async_status`` results, or the results of a looped + task that polled its own jobs. + tolerate: substrings that mark an answer as an accepted no-op. + transient: regular expressions on what a failed job said that mark the + failure as worth retrying. + + Returns: + ``{"retry": [...], "failed": [...]}``, each entry + ``{"item": , "error": }``. + """ + tolerated = [str(t) for t in (tolerate or [])] + patterns = [re.compile(str(p)) for p in (transient or [])] + split: dict[str, list[dict]] = {"retry": [], "failed": []} + for result in results or []: + if not isinstance(result, dict) or not _is_failure(result): + continue + said = _said(result) + if said and any(phrase in said for phrase in tolerated): + continue + unreadable = _SUPPRESSED in result + unfinished = not unreadable and int(result.get("finished") or 0) != 1 + retry = unfinished or any(p.search(said) for p in patterns) + if not said: + said = _UNREACHABLE if unreadable else _UNFINISHED + split["retry" if retry else "failed"].append( + {"item": _original(result), "error": said} + ) + return split + + class FilterModule: def filters(self) -> dict[str, Any]: return { "async_failures": async_failures, + "async_retry_split": async_retry_split, } diff --git a/roles/sys-dns-cloudflare-records/defaults/main.yml b/roles/sys-dns-cloudflare-records/defaults/main.yml index 42a6c65206..2617d00ba6 100644 --- a/roles/sys-dns-cloudflare-records/defaults/main.yml +++ b/roles/sys-dns-cloudflare-records/defaults/main.yml @@ -2,3 +2,19 @@ cloudflare_no_log: "{{ MASK_CREDENTIALS_IN_LOGS | bool }}" cloudflare_records: [] + +cloudflare_request_timeout: 20 # Seconds one Cloudflare HTTP socket operation may stall before the module call fails +cloudflare_attempt_timeout: 60 # Seconds one record job may run before async_wrapper kills it as hung +cloudflare_attempts: 3 # Attempts per record; only hangs, timeouts, connection errors, 429 and 5xx earn another +cloudflare_retry_delay: 10 # Seconds between two attempts +cloudflare_collect_retries: "{{ ((cloudflare_attempt_timeout | int) / 5) | round(0, 'ceil') | int + 2 }}" # Cap on 5 s polls per job; the attempt's collect deadline ends them sooner +cloudflare_tolerate: # Substrings of an answer that mean the record already stands as wanted + - "An identical record already exists" + - "code: 81058" +cloudflare_transient: # Regexes on a failed job's message that earn another attempt; anything else fails at once + - "unexpected HTTP code (-1|5[0-9][0-9])" + - "Status: 429" + - "Timeout exceeded" + - "did not complete within the requested time" + - "timed out" + - "could not find job" diff --git a/roles/sys-dns-cloudflare-records/tasks/apply.yml b/roles/sys-dns-cloudflare-records/tasks/apply.yml new file mode 100644 index 0000000000..2a27374098 --- /dev/null +++ b/roles/sys-dns-cloudflare-records/tasks/apply.yml @@ -0,0 +1,79 @@ +--- +- name: ⏸️ Wait before applying the hung or transiently failed records again + ansible.builtin.wait_for: + timeout: "{{ cloudflare_retry_delay | int }}" + when: cloudflare_attempt | int > 1 + +- name: 📌 Apply A/AAAA + community.general.cloudflare_dns: + api_token: "{{ lookup('api', 'cloudflare.api_token') }}" + zone: "{{ item.zone }}" + type: "{{ item.type }}" + name: "{{ item.name }}" + content: "{{ item.content }}" + proxied: "{{ item.proxied | default(false) }}" + ttl: "{{ item.ttl | default(1) }}" + state: "{{ item.state | default('present') }}" + solo: "{{ item.solo | default(false) }}" + timeout: "{{ cloudflare_request_timeout | int }}" + loop: "{{ _cf_pending | selectattr('type','in',['A','AAAA']) | list }}" + loop_control: { label: "{{ item.type }} {{ item.name }} -> {{ item.content }}" } + ignore_errors: true + # rationale: async; judge.yml reaps and judges every job of this attempt before the next attempt or the rest of the play + async: "{{ cloudflare_attempt_timeout | int }}" + poll: "{{ ASYNC_ENABLED | bool | ternary(0, 1) }}" + no_log: "{{ cloudflare_no_log | bool }}" + register: _cf_a + changed_when: false + +- name: 📌 Apply CNAME/MX/TXT + community.general.cloudflare_dns: + api_token: "{{ lookup('api', 'cloudflare.api_token') }}" + zone: "{{ item.zone }}" + type: "{{ item.type }}" + name: "{{ item.name }}" + value: "{{ item.value }}" + ttl: "{{ item.ttl | default(1) }}" + priority: "{{ (item.type == 'MX') | ternary(item.priority | default(10), omit) }}" + state: "{{ item.state | default('present') }}" + solo: "{{ item.solo | default(false) }}" + timeout: "{{ cloudflare_request_timeout | int }}" + loop: "{{ _cf_pending | selectattr('type','in',['CNAME','MX','TXT']) | list }}" + loop_control: { label: "{{ item.type }} {{ item.name }} -> {{ item.value }}" } + ignore_errors: true + # rationale: async; judge.yml reaps and judges every job of this attempt before the next attempt or the rest of the play + async: "{{ cloudflare_attempt_timeout | int }}" + poll: "{{ ASYNC_ENABLED | bool | ternary(0, 1) }}" + no_log: "{{ cloudflare_no_log | bool }}" + register: _cf_cname + changed_when: false + +- name: 📌 Apply SRV + community.general.cloudflare_dns: + api_token: "{{ lookup('api', 'cloudflare.api_token') }}" + zone: "{{ item.zone }}" + type: SRV + service: "{{ item.service }}" + proto: "{{ item.proto }}" + name: "{{ item.name }}" + priority: "{{ item.priority }}" + weight: "{{ item.weight }}" + port: "{{ item.port }}" + value: "{{ item.value }}" + ttl: "{{ item.ttl | default(1) }}" + state: "{{ item.state | default('present') }}" + solo: "{{ item.solo | default(false) }}" + timeout: "{{ cloudflare_request_timeout | int }}" + loop: "{{ _cf_pending | selectattr('type','equalto','SRV') | list }}" + loop_control: { label: "SRV {{ item.service }}.{{ item.proto }} {{ item.name }} -> {{ item.value }}:{{ item.port }}" } + ignore_errors: true + # rationale: async; judge.yml reaps and judges the jobs of every SRV record that opted out of best effort + async: "{{ cloudflare_attempt_timeout | int }}" + poll: "{{ ASYNC_ENABLED | bool | ternary(0, 1) }}" + no_log: "{{ cloudflare_no_log | bool }}" + register: _cf_srv + changed_when: false + +- name: 🕒 Stamp the deadline by which every job of this attempt has finished or been killed + ansible.builtin.set_fact: + _cf_collect_deadline: "{{ now(utc=true).timestamp() + (cloudflare_attempt_timeout | int) + 10 }}" diff --git a/roles/sys-dns-cloudflare-records/tasks/attempt.yml b/roles/sys-dns-cloudflare-records/tasks/attempt.yml new file mode 100644 index 0000000000..bf455d2ff6 --- /dev/null +++ b/roles/sys-dns-cloudflare-records/tasks/attempt.yml @@ -0,0 +1,15 @@ +--- +- name: "🔁 Cloudflare records, attempt {{ cloudflare_attempt }}" + when: _cf_pending | length > 0 + block: + - name: 📌 Apply the pending Cloudflare records + ansible.builtin.include_tasks: + file: apply.yml + + - name: ⚖️ Judge the outcome of every pending Cloudflare record + ansible.builtin.include_tasks: + file: judge.yml + vars: + _cf_judged: >- + {{ _cf_a.results + _cf_cname.results + + (_cf_srv.results | selectattr('item.ignore_errors', 'defined') | rejectattr('item.ignore_errors') | list) }} diff --git a/roles/sys-dns-cloudflare-records/tasks/judge.yml b/roles/sys-dns-cloudflare-records/tasks/judge.yml new file mode 100644 index 0000000000..b996b516a8 --- /dev/null +++ b/roles/sys-dns-cloudflare-records/tasks/judge.yml @@ -0,0 +1,34 @@ +--- +- name: ⏳ Collect the Cloudflare record jobs of this attempt + ansible.builtin.async_status: + jid: "{{ item.ansible_job_id }}" + loop: "{{ _cf_judged | selectattr('ansible_job_id', 'defined') | list }}" + loop_control: + label: "{{ item.item.type }} {{ item.item.name }}" + register: _cf_reaped + until: _cf_reaped.finished or (now(utc=true).timestamp() > (_cf_collect_deadline | float)) + retries: "{{ cloudflare_collect_retries | int }}" + delay: 5 + changed_when: false + ignore_errors: true + when: ASYNC_ENABLED | bool + +- name: 🧮 Split this attempt's failures into records to retry and records to stop for + ansible.builtin.set_fact: + _cf_split: >- + {{ ((_cf_reaped.results + (_cf_judged | rejectattr('ansible_job_id', 'defined') | list)) + if ASYNC_ENABLED | bool else _cf_judged) + | async_retry_split(cloudflare_tolerate, cloudflare_transient) }} + +- name: 🛑 Fail at once for every Cloudflare record a retry cannot fix + ansible.builtin.fail: + msg: >- + Cloudflare records: {{ _cf_split.failed | length }} record(s) failed for good + on attempt {{ cloudflare_attempt }}: + {% for f in _cf_split.failed %}{{ f.item.type }} {{ f.item.name }} ({{ f.item.zone }}): {{ f.error }}{{ ' || ' if not loop.last }}{% endfor %} + when: _cf_split.failed | length > 0 + +- name: 🗂️ Queue the hung and transiently failed records for the next attempt + ansible.builtin.set_fact: + _cf_retry: "{{ _cf_split.retry }}" + _cf_pending: "{{ _cf_split.retry | map(attribute='item') | list }}" diff --git a/roles/sys-dns-cloudflare-records/tasks/main.yml b/roles/sys-dns-cloudflare-records/tasks/main.yml index 266c536859..adbb622520 100644 --- a/roles/sys-dns-cloudflare-records/tasks/main.yml +++ b/roles/sys-dns-cloudflare-records/tasks/main.yml @@ -7,94 +7,25 @@ no_log: "{{ cloudflare_no_log | bool }}" when: MODE_ASSERT | bool -- name: 📌 Apply A/AAAA - community.general.cloudflare_dns: - api_token: "{{ lookup('api', 'cloudflare.api_token') }}" - zone: "{{ item.zone }}" - type: "{{ item.type }}" - name: "{{ item.name }}" - content: "{{ item.content }}" - proxied: "{{ item.proxied | default(false) }}" - ttl: "{{ item.ttl | default(1) }}" - state: "{{ item.state | default('present') }}" - solo: "{{ item.solo | default(false) }}" - loop: "{{ cloudflare_records | selectattr('type','in',['A','AAAA']) | list }}" - loop_control: { label: "{{ item.type }} {{ item.name }} -> {{ item.content }}" } - delegate_to: localhost - become: false - # rationale: async; the jobs run on the controller and are reaped there below, before the rest of the play - async: "{{ ASYNC_TIME }}" - poll: "{{ ASYNC_POLL }}" - no_log: "{{ cloudflare_no_log | bool }}" - register: _cf_a - changed_when: false - failed_when: - - _cf_a.failed - - _cf_a.msg is not search('An identical record already exists|81058') +- name: 🗂️ Queue every Cloudflare record for the first attempt + ansible.builtin.set_fact: + _cf_pending: "{{ cloudflare_records }}" + _cf_retry: [] -- name: 📌 Apply CNAME/MX/TXT - community.general.cloudflare_dns: - api_token: "{{ lookup('api', 'cloudflare.api_token') }}" - zone: "{{ item.zone }}" - type: "{{ item.type }}" - name: "{{ item.name }}" - value: "{{ item.value }}" - ttl: "{{ item.ttl | default(1) }}" - priority: "{{ (item.type == 'MX') | ternary(item.priority | default(10), omit) }}" - state: "{{ item.state | default('present') }}" - solo: "{{ item.solo | default(false) }}" - loop: "{{ cloudflare_records | selectattr('type','in',['CNAME','MX','TXT']) | list }}" - loop_control: { label: "{{ item.type }} {{ item.name }} -> {{ item.value }}" } - delegate_to: localhost - become: false - # rationale: async; the jobs run on the controller and are reaped there below, before the rest of the play - async: "{{ ASYNC_TIME }}" - poll: "{{ ASYNC_POLL }}" - no_log: "{{ cloudflare_no_log | bool }}" - register: _cf_cname - changed_when: false - failed_when: - - _cf_cname.failed - - _cf_cname.msg is not search('An identical record already exists|81058') - -- name: 📌 Apply SRV - community.general.cloudflare_dns: - api_token: "{{ lookup('api', 'cloudflare.api_token') }}" - zone: "{{ item.zone }}" - type: SRV - service: "{{ item.service }}" - proto: "{{ item.proto }}" - name: "{{ item.name }}" - priority: "{{ item.priority }}" - weight: "{{ item.weight }}" - port: "{{ item.port }}" - value: "{{ item.value }}" - ttl: "{{ item.ttl | default(1) }}" - state: "{{ item.state | default('present') }}" - solo: "{{ item.solo | default(false) }}" - loop: "{{ cloudflare_records | selectattr('type','equalto','SRV') | list }}" - loop_control: { label: "SRV {{ item.service }}.{{ item.proto }} {{ item.name }} -> {{ item.value }}:{{ item.port }}" } - ignore_errors: "{{ item.ignore_errors | default(true) }}" - delegate_to: localhost - become: false - # rationale: async; the jobs run on the controller, those of records that must not fail are reaped below - async: "{{ ASYNC_TIME }}" - poll: "{{ ASYNC_POLL }}" - no_log: "{{ cloudflare_no_log | bool }}" - register: _cf_srv - changed_when: false - -- name: ⏳ Reap the Cloudflare record jobs on the controller +- name: 🔁 Apply the Cloudflare records, retrying only hung and transiently failed ones ansible.builtin.include_tasks: - file: utils/async/reap.yml + file: attempt.yml apply: delegate_to: localhost become: false - vars: - async_reap_jobs: >- - {{ _cf_a.results + _cf_cname.results - + (_cf_srv.results | selectattr('item.ignore_errors', 'defined') | rejectattr('item.ignore_errors') | list) }} - async_reap_label: "Cloudflare records" - async_reap_tolerate: - - "An identical record already exists" - - "81058" + loop: "{{ range(1, (cloudflare_attempts | int) + 1) | list }}" + loop_control: + loop_var: cloudflare_attempt + +- name: 🛑 Fail for every Cloudflare record still failing after its last attempt + ansible.builtin.fail: + msg: >- + Cloudflare records: {{ _cf_retry | length }} record(s) still failed after + {{ cloudflare_attempts }} attempt(s) of at most {{ cloudflare_attempt_timeout }} s each: + {% for r in _cf_retry %}{{ r.item.type }} {{ r.item.name }} ({{ r.item.zone }}): {{ r.error }}{{ ' || ' if not loop.last }}{% endfor %} + when: _cf_retry | length > 0 diff --git a/tests/unit/python/plugins/filter/test_async_retry_split.py b/tests/unit/python/plugins/filter/test_async_retry_split.py new file mode 100644 index 0000000000..5e8803a52a --- /dev/null +++ b/tests/unit/python/plugins/filter/test_async_retry_split.py @@ -0,0 +1,125 @@ +import unittest +from pathlib import Path + +import yaml + +from plugins.filter.async_failures import async_retry_split + +_DEFAULTS = ( + Path(__file__).resolve().parents[5] + / "roles" + / "sys-dns-cloudflare-records" + / "defaults" + / "main.yml" +) +_ROLE = yaml.safe_load(_DEFAULTS.read_text(encoding="utf-8")) +_TOLERATE = _ROLE["cloudflare_tolerate"] +_TRANSIENT = _ROLE["cloudflare_transient"] + +_ERP = {"zone": "example.org", "type": "AAAA", "name": "*.erp", "content": "::1"} + + +def _reaped(**job): + fired = {"ansible_loop_var": "item", "item": _ERP, "ansible_job_id": "j1.1"} + return {"ansible_loop_var": "item", "item": fired, "ansible_job_id": "j1.1", **job} + + +def _split(*results): + return async_retry_split(list(results), _TOLERATE, _TRANSIENT) + + +class TestAsyncRetrySplit(unittest.TestCase): + def test_a_job_that_never_finished_is_retried_and_names_its_record(self): + split = _split(_reaped(started=True, finished=False)) + self.assertEqual([], split["failed"]) + self.assertEqual( + [{"item": _ERP, "error": "job did not finish"}], split["retry"] + ) + + def test_an_attempt_killed_on_its_timeout_is_retried(self): + split = _split(_reaped(finished=True, failed=True, msg="Timeout exceeded")) + self.assertEqual(["Timeout exceeded"], [r["error"] for r in split["retry"]]) + + def test_connection_errors_rate_limits_and_5xx_are_retried(self): + messages = [ + "Failed API call /zones?name=example.org; got unexpected HTTP code -1: Connection failure: timed out", + "Failed API call /zones/z/dns_records; got unexpected HTTP code 502: Bad Gateway", + "API client is rate limited; Status: 429; Method: GET: Call: /zones", + "async task did not complete within the requested time - 60s", + ] + split = _split(*(_reaped(finished=True, failed=True, msg=m) for m in messages)) + self.assertEqual([], split["failed"]) + self.assertEqual(messages, [r["error"] for r in split["retry"]]) + + def test_a_4xx_or_auth_error_fails_at_once_without_a_retry(self): + messages = [ + "API bad request; Status: 400; Method: POST: Call: /zones/z/dns_records; Error details: code: 9005, error: Content for AAAA record is invalid.; ", + "API user does not have permission; Status: 401; Method: GET: Call: /zones", + "API request not authenticated; Status: 403; Method: GET: Call: /zones", + "Failed API call /zones/z/dns_records; got unexpected HTTP code 404: Not Found", + ] + split = _split(*(_reaped(finished=True, failed=True, msg=m) for m in messages)) + self.assertEqual([], split["retry"]) + self.assertEqual( + [m.strip() for m in messages], [f["error"] for f in split["failed"]] + ) + self.assertTrue(all(f["item"] is _ERP for f in split["failed"])) + + def test_an_identical_record_is_success_not_a_retry(self): + split = _split( + _reaped( + finished=True, + failed=True, + msg="API bad request; Status: 400; Error details: code: 81058, error: An identical record already exists.; ", + ) + ) + self.assertEqual({"retry": [], "failed": []}, split) + + def test_a_zone_id_containing_81058_never_passes_a_4xx_as_success(self): + msg = "API bad request; Status: 400; Method: POST: Call: /zones/a81058b/dns_records; Error details: code: 9005, error: bad" + split = _split(_reaped(finished=True, failed=True, msg=msg)) + self.assertEqual([msg], [f["error"] for f in split["failed"]]) + + def test_a_job_file_not_yet_written_is_retried(self): + split = _split(_reaped(finished=True, failed=True, msg="could not find job")) + self.assertEqual(["could not find job"], [r["error"] for r in split["retry"]]) + + def test_an_unreadable_job_fails_and_is_not_retried(self): + split = _split(_reaped(finished=True, failed_when_suppressed_exception="x")) + self.assertEqual([], split["retry"]) + self.assertEqual(1, len(split["failed"])) + + def test_a_result_polled_in_place_names_its_record_too(self): + polled = { + "ansible_loop_var": "item", + "item": _ERP, + "ansible_job_id": "j2.2", + "finished": True, + "failed": True, + "msg": "Timeout exceeded", + } + self.assertEqual(_ERP, _split(polled)["retry"][0]["item"]) + + def test_finished_jobs_and_empty_input_need_nothing(self): + self.assertEqual( + {"retry": [], "failed": []}, _split(_reaped(finished=True, changed=False)) + ) + self.assertEqual( + {"retry": [], "failed": []}, async_retry_split(None, None, None) + ) + + def test_a_hang_on_every_attempt_ends_with_the_record_named(self): + pending = [_ERP] + last = [] + for _attempt in range(int(_ROLE["cloudflare_attempts"])): + if not pending: + break + split = _split(*(_reaped(started=True, finished=False) for _ in pending)) + self.assertEqual([], split["failed"]) + last = split["retry"] + pending = [r["item"] for r in last] + self.assertEqual([{"item": _ERP, "error": "job did not finish"}], last) + + +if __name__ == "__main__": + unittest.main()