mirror of
https://github.com/ansible-collections/hetzner.hcloud.git
synced 2026-02-04 08:01:49 +00:00
7ac361a9cc
##### SUMMARY
The format of TXT records must consist of one or many quoted strings of
255 characters.
Use this function to format TXT record that must match the format
required by the API:
```yml
- name: Create a SPF record
hetzner.hcloud.zone_rrset:
zone: example.com
name: "@"
type: "TXT"
records:
- value: "{{ 'v=spf1 include:_spf.example.net ~all' | hetzner.hcloud.text_record }}"
state: present
```
##### ISSUE TYPE
- Feature Pull Request
##### COMPONENT NAME
zone_rrset
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from ansible.errors import AnsibleFilterError
|
|
from ansible.module_utils.common.text.converters import to_native
|
|
|
|
|
|
# pylint: disable=unused-argument
|
|
def load_balancer_status(load_balancer: dict, *args, **kwargs) -> Literal["unknown", "unhealthy", "healthy"]:
|
|
"""
|
|
Return the status of a Load Balancer based on its targets.
|
|
"""
|
|
|
|
def targets_status(targets: list) -> Literal["unknown", "unhealthy", "healthy"]:
|
|
result = "healthy"
|
|
|
|
for target in targets:
|
|
# Label selector targets have child targets that must be checked
|
|
if target["type"] == "label_selector":
|
|
status = targets_status(target["targets"])
|
|
if status == "unhealthy":
|
|
return "unhealthy"
|
|
|
|
if status in (None, "unknown"):
|
|
result = "unknown"
|
|
|
|
continue
|
|
|
|
# Report missing health status as unknown
|
|
if not target.get("health_status"):
|
|
return "unknown"
|
|
|
|
for health_status in target.get("health_status"):
|
|
status = health_status.get("status")
|
|
if status == "unhealthy":
|
|
return "unhealthy"
|
|
|
|
if status in (None, "unknown"):
|
|
result = "unknown"
|
|
|
|
return result
|
|
|
|
try:
|
|
return targets_status(load_balancer["targets"])
|
|
except Exception as exc:
|
|
raise AnsibleFilterError(f"load_balancer_status - {to_native(exc)}", orig_exc=exc) from exc
|
|
|
|
|
|
# pylint: disable=unused-argument
|
|
def txt_record(record: str, *args, **kwargs) -> str:
|
|
"""
|
|
Return the status of a Load Balancer based on its targets.
|
|
"""
|
|
try:
|
|
record = record.replace('"', '\\"')
|
|
|
|
parts = []
|
|
for start in range(0, len(record), 255):
|
|
end = min(start + 255, len(record))
|
|
parts.append('"' + record[start:end] + '"')
|
|
record = " ".join(parts)
|
|
|
|
return record
|
|
except Exception as exc:
|
|
raise AnsibleFilterError(f"txt_record - {to_native(exc)}", orig_exc=exc) from exc
|
|
|
|
|
|
class FilterModule:
|
|
"""
|
|
Hetzner Cloud filters.
|
|
"""
|
|
|
|
def filters(self):
|
|
return {
|
|
"load_balancer_status": load_balancer_status,
|
|
"txt_record": txt_record,
|
|
}
|