From d6cf7d63e308557bcb6a6053c170bafa2b35864e Mon Sep 17 00:00:00 2001 From: Fiehe Christoph Date: Mon, 2 Feb 2026 17:20:02 +0100 Subject: [PATCH] feat: Icinga 2 downtime module added allowing to schedule and remove downtimes through its REST API. Signed-off-by: Fiehe Christoph --- .github/BOTMETA.yml | 4 + plugins/module_utils/icinga2.py | 125 ++++++ plugins/modules/icinga2_downtime.py | 355 ++++++++++++++++++ .../plugins/modules/test_icinga2_downtime.py | 221 +++++++++++ 4 files changed, 705 insertions(+) create mode 100644 plugins/module_utils/icinga2.py create mode 100644 plugins/modules/icinga2_downtime.py create mode 100644 tests/unit/plugins/modules/test_icinga2_downtime.py diff --git a/.github/BOTMETA.yml b/.github/BOTMETA.yml index 9066eaa4f5..f573a8dcb6 100644 --- a/.github/BOTMETA.yml +++ b/.github/BOTMETA.yml @@ -362,6 +362,8 @@ files: keywords: cloud huawei hwc labels: huawei hwc_utils networking maintainers: $team_huawei + $module_utils/icinga2.py: + maintainers: cfiehe $module_utils/identity/keycloak/keycloak.py: maintainers: $team_keycloak $module_utils/identity/keycloak/keycloak_clientsecret.py: @@ -713,6 +715,8 @@ files: maintainers: $team_huawei huaweicloud $modules/ibm_sa_: maintainers: tzure + $modules/icinga2_downtime.py: + maintainers: cfiehe $modules/icinga2_feature.py: maintainers: nerzhul $modules/icinga2_host.py: diff --git a/plugins/module_utils/icinga2.py b/plugins/module_utils/icinga2.py new file mode 100644 index 0000000000..3f218a5001 --- /dev/null +++ b/plugins/module_utils/icinga2.py @@ -0,0 +1,125 @@ +# Copyright (c) Contributors to the Ansible project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileCopyrightText: Ansible Project + +from __future__ import annotations + +import json +import typing as t + +from ansible.module_utils.common.text.converters import to_bytes +from ansible.module_utils.urls import fetch_url, url_argument_spec + +if t.TYPE_CHECKING: + from http.client import HTTPResponse + from urllib.error import HTTPError + + from ansible.module_utils.basic import AnsibleModule + + +class Icinga2Client: + def __init__( + self, + module: AnsibleModule, + url: str, + ca_path: str | None = None, + timeout: int | float | None = None, + ) -> None: + self.module = module + self.url = url.rstrip("/") + self.ca_path = ca_path + self.timeout = timeout + self.actions = Actions(client=self) + + def send_request( + self, method: str, path: str, data: dict[str, t.Any] | None = None + ) -> tuple[HTTPResponse | HTTPError, dict[str, t.Any]]: + url = f"{self.url}/{path}" + headers = { + "X-HTTP-Method-Override": method.upper(), + "Accept": "application/json", + } + return fetch_url( + module=self.module, + url=url, + ca_path=self.ca_path, + data=to_bytes(json.dumps(data)), + headers=headers, + timeout=self.timeout, + ) + + +class Actions: + base_path = "v1/actions" + + def __init__(self, client: Icinga2Client) -> None: + self.client = client + + def schedule_downtime( + self, + object_type: str, + filter: str, + author: str, + comment: str, + start_time: int, + end_time: int, + duration: int, + filter_vars: dict[str, t.Any] | None = None, + fixed: bool | None = None, + all_services: bool | None = None, + trigger_name: str | None = None, + child_options: str | None = None, + ) -> tuple[HTTPResponse | HTTPError, dict[str, t.Any]]: + path = "{}/{}".format(self.base_path, "schedule-downtime") + + data: dict[str, t.Any] = { + "type": object_type, + "filter": filter, + "author": author, + "comment": comment, + "start_time": start_time, + "end_time": end_time, + "duration": duration, + } + if filter_vars: + data["filter_vars"] = filter_vars + if fixed: + data["fixed"] = fixed + if all_services: + data["all_services"] = all_services + if trigger_name: + data["trigger_name"] = trigger_name + if child_options: + data["child_options"] = child_options + + return self.client.send_request(method="POST", path=path, data=data) + + def remove_downtime( + self, + object_type: str, + name: str | None = None, + filter: str | None = None, + filter_vars: dict[str, t.Any] | None = None, + ) -> tuple[HTTPResponse | HTTPError, dict[str, t.Any]]: + path = "{}/{}".format(self.base_path, "remove-downtime") + + data: dict[str, t.Any] = {"type": object_type} + if name: + data[object_type.lower()] = name + if filter: + data["filter"] = filter + if filter_vars: + data["filter_vars"] = filter_vars + + return self.client.send_request(method="POST", path=path, data=data) + + +def icinga2_argument_spec() -> dict[str, t.Any]: + argument_spec = url_argument_spec() + argument_spec.update( + url=dict(type="str", required=True), + ca_path=dict(type="path"), + timeout=dict(type="int", default=10), + ) + return argument_spec diff --git a/plugins/modules/icinga2_downtime.py b/plugins/modules/icinga2_downtime.py new file mode 100644 index 0000000000..b4c31cc3a0 --- /dev/null +++ b/plugins/modules/icinga2_downtime.py @@ -0,0 +1,355 @@ +#!/usr/bin/python + +# Copyright (c) Contributors to the Ansible project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileCopyrightText: Ansible Project + +from __future__ import annotations + +DOCUMENTATION = r""" +module: icinga2_downtime +short_description: Manages Icinga 2 downtimes +version_added: "12.4.0" +description: + - Manages downtimes in Icinga 2 through its REST API. + - Options as described at U(https://icinga.com/docs/icinga-2/latest/doc/12-icinga2-api/#schedule-downtime). +author: + - Christoph Fiehe (@cfiehe) +attributes: + check_mode: + support: none + diff_mode: + support: none +options: + url: + description: + - URL of the Icinga 2 API. + type: str + required: true + ca_path: + description: + - CA certificates bundle to use to verify the Icinga 2 server certificate. + type: path + validate_certs: + description: + - If V(false), SSL certificates will not be validated. + - This should only be used on personally controlled sites using self-signed certificates. + type: bool + default: true + timeout: + description: + - How long to wait for the server to send data before giving up. + type: int + default: 10 + all_services: + description: + - Whether downtimes should be set for all services of the matched host objects. + - If omitted, Icinga 2 does not create downtimes for all services of the matched host objects by default. + type: bool + author: + description: + - Name of the author. + type: str + default: "Ansible" + comment: + description: + - A descriptive comment. + type: str + default: Downtime scheduled by Ansible + child_options: + description: + - Schedule child downtimes. + type: str + choices: ["DowntimeNoChildren", "DowntimeTriggeredChildren", "DowntimeNonTriggeredChildren"] + duration: + description: + - Duration of the downtime. + - Required in case of a flexible downtime. + type: int + end_time: + description: + - End time of the downtime as unix timestamp. + type: int + filter_vars: + description: + - Variable names and values used in the filter expression. + type: dict + filter: + description: + - Filter expression limiting the objects to operate on. + type: str + fixed: + description: + - Whether this downtime is fixed or flexible. + - If omitted, Icinga 2 creates a fixed downtime by default. + type: bool + name: + description: + - Name of the downtime. + - This option has no effect for states other than V(absent). + type: str + object_type: + description: + - Use V(Host) for a host downtime and V(Service) for a service downtime. + - Use V(Downtime) and give the name of the downtime you want to remove. + type: str + choices: ["Service", "Host", "Downtime"] + default: Host + start_time: + description: + - Start time of the downtime as unix timestamp. + type: int + state: + description: + - State of the downtime. + type: str + choices: ["present", "absent"] + default: present + trigger_name: + description: + - Name of the downtime trigger. + type: str +extends_documentation_fragment: + - ansible.builtin.url + - community.general.attributes +""" + +EXAMPLES = r""" +- name: Schedule a host downtime + community.general.icinga2_downtime: + url: "https://icinga2.example.com:5665" + url_username: icingadmin + url_password: secret + state: present + author: Ansible + comment: Scheduled downtime for test purposes. + all_services: true + start_time: "{{ downtime_start_time }}" + end_time: "{{ downtime_end_time }}" + duration: "{{ downtime_duration }}" + fixed: true + object_type: Host + filter: |- + host.name=="host.example.com" + delegate_to: localhost + register: icinga2_downtime_response + vars: + downtime_start_time: "{{ ansible_date_time['epoch'] | int }}" + downtime_end_time: "{{ downtime_start_time | int + 3600 }}" + downtime_duration: "{{ downtime_end_time | int - downtime_start_time | int }}" + +- name: Remove scheduled host downtime + edloc.general.icinga2_downtime: + url: "https://icinga2.example.com:5665" + url_username: icingadmin + url_password: secret + state: absent + author: Ansible + object_type: Downtime + name: "{{ icinga2_downtime_response.results[0].name }}" + delegate_to: localhost + when: icinga2_downtime_response.results | default([]) | length > 0 +""" + +RETURN = r""" +# Returns the results of downtime scheduling as a list of JSON dictionaries from the Icinga 2 API under the C(downtimes) key. +# Refer to https://icinga.com/docs/icinga-2/latest/doc/12-icinga2-api/#schedule-downtime for more details. +results: + description: Results of downtime scheduling or removal + type: list + returned: success + elements: dict + contains: + code: + description: Success or error code of downtime scheduling. + returned: always + type: int + sample: 200 + legacy_id: + description: Legacy id of the scheduled downtime. + returned: if a downtime was scheduled successfully + type: int + sample: 28911 + name: + description: Name of the scheduled downtime. + returned: if a downtime was scheduled successfully + type: str + sample: host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51 + status: + description: Human-readable message describing the result of downtime scheduling. + returned: always + type: str + sample: Successfully scheduled downtime 'host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51' for object 'host.example.com'. + sample: + [ + { + "code": 200, + "legacy_id": 28911, + "name": "host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51", + "status": "Successfully scheduled downtime 'host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51' for object 'host.example.com'.", + } + ] +error: + description: Error message as JSON dictionary returned from the Icinga 2 API. + type: dict + returned: if downtime scheduling or removal did not succeed. + sample: + { + "error": 404, + "status": "No objects found." + } +""" + +import typing as t + +from ansible.module_utils.basic import AnsibleModule + +from ansible_collections.community.general.plugins.module_utils.icinga2 import ( + Icinga2Client, + icinga2_argument_spec, +) + + +def main() -> None: + argument_spec = icinga2_argument_spec() + argument_spec.update( + all_services=dict(type="bool"), + author=dict(type="str", default="Ansible"), + comment=dict(type="str", default="Downtime scheduled by Ansible"), + child_options=dict( + type="str", + choices=[ + "DowntimeNoChildren", + "DowntimeTriggeredChildren", + "DowntimeNonTriggeredChildren", + ], + ), + duration=dict(type="int"), + end_time=dict(type="int"), + filter_vars=dict(type="dict"), + filter=dict(type="str"), + fixed=dict(type="bool"), + name=dict(type="str"), + object_type=dict(type="str", choices=["Service", "Host", "Downtime"], default="Host"), + start_time=dict(type="int"), + state=dict(type="str", choices=["present", "absent"], default="present"), + trigger_name=dict(type="str"), + ) + + module = AnsibleModule( + argument_spec=argument_spec, + supports_check_mode=False, + required_if=( + ("state", "present", ["comment", "start_time", "end_time", "filter"]), + ("fixed", False, ["duration"]), + ), + required_one_of=[["filter", "name"]], + ) + + client = Icinga2Client( + module=module, url=module.params["url"], ca_path=module.params["ca_path"], timeout=module.params.get("timeout") + ) + + if module.params["state"] == "present": + schedule_downtime(module, client) + elif module.params["state"] == "absent": + remove_downtime(module, client) + + +def schedule_downtime(module: AnsibleModule, client: Icinga2Client) -> None: + duration = module.params.get("duration") + end_time = module.params.get("end_time") + start_time = module.params.get("start_time") + + if end_time <= start_time: + module.fail_json(msg="The end time must be later than the start time.") + + if duration is None: + duration = end_time - start_time + + response, info = client.actions.schedule_downtime( + all_services=module.params.get("all_services"), + author=module.params.get("author"), + child_options=module.params.get("child_options"), + comment=module.params.get("comment"), + duration=duration, + end_time=end_time, + filter_vars=module.params.get("filter_vars"), + filter=module.params.get("filter"), + fixed=module.params.get("fixed"), + object_type=module.params.get("object_type"), + start_time=start_time, + trigger_name=module.params.get("trigger_name"), + ) + + status_code = info["status"] + result: dict[str, t.Any] = { + "changed": False, + "failed": False, + } + + if 200 <= status_code <= 299: + result["changed"] = True + msg = "Successfully scheduled downtime." + try: + result["results"] = module.from_json(response.read())["results"] + except (ValueError, KeyError): + # As a precaution, catch key and value error in case of a malformed response message. + msg += "\nWarning: Malformed response received from server. Skipping content." + + result["msg"] = msg + module.exit_json(**result) # type:ignore[arg-type] + else: + result["failed"] = True + result["msg"] = "Unable to schedule downtime." + if status_code >= 400: + try: + result["error"] = module.from_json(info.get("body")) + except ValueError: + pass + module.fail_json(**result) # type:ignore[arg-type] + + +def remove_downtime(module: AnsibleModule, client: Icinga2Client) -> None: + response, info = client.actions.remove_downtime( + filter_vars=module.params.get("filter_vars"), + filter=module.params.get("filter"), + name=module.params.get("name"), + object_type=module.params.get("object_type"), + ) + + status_code = info["status"] + result: dict[str, t.Any] = { + "changed": False, + "failed": False, + } + + if 200 <= status_code <= 299: + result["changed"] = True + msg = "Successfully removed downtime." + try: + result["results"] = module.from_json(response.read())["results"] + except (ValueError, KeyError): + # As a precaution, catch key and value error in case of a malformed response message. + msg += "\nWarning: Malformed response received from server. Skipping content." + + result["msg"] = msg + module.exit_json(**result) # type:ignore[arg-type] + else: + if status_code == 404: + result["msg"] = "No matching downtime found." + module.exit_json(**result) # type:ignore[arg-type] + else: + if status_code >= 400: + try: + result["error"] = module.from_json(info.get("body")) + except ValueError: + pass + result["failed"] = True + result["msg"] = "Unable to remove downtime." + module.fail_json(**result) # type:ignore[arg-type] + + +if __name__ == "__main__": + main() diff --git a/tests/unit/plugins/modules/test_icinga2_downtime.py b/tests/unit/plugins/modules/test_icinga2_downtime.py new file mode 100644 index 0000000000..8631c2a30c --- /dev/null +++ b/tests/unit/plugins/modules/test_icinga2_downtime.py @@ -0,0 +1,221 @@ +# Copyright (c) Ansible project +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch +from urllib.error import HTTPError + +from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import ( + AnsibleExitJson, + AnsibleFailJson, + ModuleTestCase, + set_module_args, +) + +from ansible_collections.community.general.plugins.modules import icinga2_downtime + + +class TestIcinga2Downtime(ModuleTestCase): + def setUp(self): + super().setUp() + self.module = icinga2_downtime + + def tearDown(self): + super().tearDown() + + @patch("ansible_collections.community.general.plugins.modules.icinga2_downtime.Icinga2Client") + def test_schedule_downtime_successfully(self, client_mock): + module_args = { + "url": "http://icinga2.example.com:5665", + "url_username": "icingaadmin", + "url_password": "secret", + "author": "Ansible", + "comment": "This is a test comment.", + "state": "present", + "start_time": 1769954400, + "end_time": 1769958000, + "duration": 3600, + "fixed": True, + "object_type": "Host", + "filter": 'host.name=="host.example.com"', + } + with set_module_args(module_args): + info = { + "content-type": "application/json", + "server": "Icinga/r2.15.1-1", + "status": 200, + "url": "https://icinga2.example.com:5665/v1/actions/schedule-downtime", + } + response = { + "results": [ + { + "code": 200, + "legacy_id": 28911, + "name": "host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51", + "status": "Successfully scheduled downtime 'host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51' for object 'host.example.com'.", + } + ] + } + response_read_mock = MagicMock(return_value=json.dumps(response)) + response_mock = MagicMock(read=response_read_mock) + schedule_downtime_mock = MagicMock(return_value=(response_mock, info)) + actions_mock = MagicMock(schedule_downtime=schedule_downtime_mock) + client_mock.return_value = MagicMock(actions=actions_mock) + + with self.assertRaises(AnsibleExitJson) as result: + self.module.main() + + self.assertTrue(result.exception.args[0]["changed"]) + self.assertEqual(result.exception.args[0]["results"], response["results"]) + schedule_downtime_mock.assert_called_once_with( + all_services=None, + author=module_args["author"], + child_options=None, + comment=module_args["comment"], + duration=module_args["duration"], + end_time=module_args["end_time"], + filter=module_args["filter"], + filter_vars=None, + fixed=module_args["fixed"], + object_type=module_args["object_type"], + start_time=module_args["start_time"], + trigger_name=None, + ) + + @patch("ansible_collections.community.general.plugins.modules.icinga2_downtime.Icinga2Client") + def test_schedule_downtime_failed(self, client_mock): + module_args = { + "url": "http://icinga2.example.com:5665", + "url_username": "icingaadmin", + "url_password": "secret", + "author": "Ansible", + "comment": "This is a test comment.", + "state": "present", + "start_time": 1769954400, + "end_time": 1769958000, + "duration": 3600, + "fixed": True, + "object_type": "Host", + "filter": 'host.name=="unknown.example.com"', + } + with set_module_args(module_args): + info = { + "body": '{"error":404,"status":"No objects found."}', + "content-length": "42", + "content-type": "application/json", + "msg": "HTTP Error 404: Not Found", + "server": "Icinga/r2.15.1-1", + "status": 404, + "url": "https://icinga2.example.com:5665/v1/actions/remove-downtime", + } + response = HTTPError(url=info["url"], code=404, msg=info["msg"], hdrs={}, fp=None) + schedule_downtime_mock = MagicMock(return_value=(response, info)) + actions_mock = MagicMock(schedule_downtime=schedule_downtime_mock) + client_mock.return_value = MagicMock(actions=actions_mock) + + with self.assertRaises(AnsibleFailJson) as result: + self.module.main() + + self.assertFalse(result.exception.args[0]["changed"]) + self.assertTrue(result.exception.args[0]["failed"]) + self.assertEqual( + result.exception.args[0]["error"], + {"error": 404, "status": "No objects found."}, + ) + schedule_downtime_mock.assert_called_once_with( + all_services=None, + author=module_args["author"], + child_options=None, + comment=module_args["comment"], + duration=module_args["duration"], + end_time=module_args["end_time"], + filter=module_args["filter"], + filter_vars=None, + fixed=module_args["fixed"], + object_type=module_args["object_type"], + start_time=module_args["start_time"], + trigger_name=None, + ) + + @patch("ansible_collections.community.general.plugins.modules.icinga2_downtime.Icinga2Client") + def test_remove_existing_downtime(self, client_mock): + module_args = { + "url": "http://icinga2.example.com:5665", + "url_username": "icingaadmin", + "url_password": "secret", + "state": "absent", + "name": "host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51", + "object_type": "Downtime", + } + with set_module_args(module_args): + info = { + "content-type": "application/json", + "server": "Icinga/r2.15.1-1", + "status": 200, + "url": "https://icinga2.example.com:5665/v1/actions/remove-downtime", + } + response = { + "results": [ + { + "code": 200, + "status": "Successfully removed downtime 'host.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51' and 0 child downtimes.", + } + ] + } + response_read_mock = MagicMock(return_value=json.dumps(response)) + response_mock = MagicMock(read=response_read_mock) + remove_downtime_mock = MagicMock(return_value=(response_mock, info)) + actions_mock = MagicMock(remove_downtime=remove_downtime_mock) + client_mock.return_value = MagicMock(actions=actions_mock) + + with self.assertRaises(AnsibleExitJson) as result: + self.module.main() + + self.assertTrue(result.exception.args[0]["changed"]) + self.assertEqual(result.exception.args[0]["results"], response["results"]) + remove_downtime_mock.assert_called_once_with( + filter=None, + filter_vars=None, + name=module_args["name"], + object_type=module_args["object_type"], + ) + + @patch("ansible_collections.community.general.plugins.modules.icinga2_downtime.Icinga2Client") + def test_remove_non_existing_downtime(self, client_mock): + module_args = { + "url": "http://icinga2.example.com:5665", + "url_username": "icingaadmin", + "url_password": "secret", + "state": "absent", + "name": "unknown.example.com!e19c705a-54c2-49c5-8014-70ff624f9e51", + "object_type": "Downtime", + } + with set_module_args(module_args): + info = { + "body": '{"error":404,"status":"No objects found."}', + "content-length": "42", + "content-type": "application/json", + "msg": "HTTP Error 404: Not Found", + "server": "Icinga/r2.15.1-1", + "status": 404, + "url": "https://icinga2.example.com:5665/v1/actions/remove-downtime", + } + response = HTTPError(url=info["url"], code=404, msg=info["msg"], hdrs={}, fp=None) + remove_downtime_mock = MagicMock(return_value=(response, info)) + actions_mock = MagicMock(remove_downtime=remove_downtime_mock) + client_mock.return_value = MagicMock(actions=actions_mock) + + with self.assertRaises(AnsibleExitJson) as result: + self.module.main() + + self.assertFalse(result.exception.args[0]["changed"]) + self.assertFalse(result.exception.args[0]["failed"]) + remove_downtime_mock.assert_called_once_with( + filter=None, + filter_vars=None, + name=module_args["name"], + object_type=module_args["object_type"], + )