ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-03-21 20:59:10 +00:00
Code Activity

[PR #11462/ce7cb4e9 backport][stable-12] New module icinga2_downtime (#11532)

Browse source 
New module icinga2_downtime (#11462)

* feat: Icinga 2 downtime module added allowing to schedule and remove downtimes through its REST API.



* ensure compatibility with ModuleTestCase

feat: errors raised from MH now contain the changed flag
ref: move module exit out of the decorated run method



* revised module

ref: module refactored using StateModuleHelper now
ref: suggested changes by reviewer added



* revert change regarding changed flag in MH



* refactoring and set changed flag explicitly on error



* Check whether there was a state change on module failure removed.



* ref: test cases migrated to the new feature that allows passing through exceptions



* Update plugins/module_utils/icinga2.py



* Update plugins/module_utils/icinga2.py



* Update plugins/modules/icinga2_downtime.py



* ref: make module helper private



* fix: ensure that all non-null values are added to the request otherwise a `false` value is dropped



* ref: module description extended with the note that check mode is not supported



* Update plugins/modules/icinga2_downtime.py



* fix: documentation updated



* ref: documentation updated
ref: doc fragment added



* Update plugins/doc_fragments/icinga2_api.py



* ref: doc fragment renamed to `_icinga2_api.py`



* ref: maintainer to doc fragment in BOTMETA.yml added



* Update plugins/modules/icinga2_downtime.py



* Update plugins/modules/icinga2_downtime.py



* Update plugins/modules/icinga2_downtime.py



---------





(cherry picked from commit ce7cb4e914)

Signed-off-by: Fiehe Christoph  <c.fiehe@eurodata.de>
Co-authored-by: Christoph Fiehe <cfiehe@users.noreply.github.com>
Co-authored-by: Fiehe Christoph <c.fiehe@eurodata.de>
Co-authored-by: Felix Fontein <felix@fontein.de>
Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
This commit is contained in:
patchback[bot] 2026-02-23 06:17:51 +01:00 • committed by GitHub
parent 8c429ac69d
commit 4b6cd41512
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 696 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.github/BOTMETA.yml vendored
View file

@ -136,6 +136,8 @@ files:
$doc_fragments/hwc.py:
labels: hwc
maintainers: $team_huawei
$doc_fragments/_icinga2_api.py:
maintainers: cfiehe
$doc_fragments/nomad.py:
maintainers: chris93111 apecnascimento
$doc_fragments/pipx.py:
@ -365,6 +367,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:
@ -716,6 +720,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:

30
plugins/doc_fragments/_icinga2_api.py Normal file
View file

@ -0,0 +1,30 @@
# 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: 2026 Christoph Fiehe <christoph.fiehe@gmail.com>
# Note that this doc fragment is **PRIVATE** to the collection. It can have breaking changes at any time.
# Do not use this from other collections or standalone plugins/modules!
from __future__ import annotations
class ModuleDocFragment:
# Use together with ansible.builtin.url and icinga2_argument_spec from
# ansible_collections.community.general.plugins.module_utils._icinga2
DOCUMENTATION = r"""
options:
url:
description:
- URL of the Icinga 2 REST API.
type: str
required: true
ca_path:
description:
- CA certificates bundle to use to verify the Icinga 2 server certificate.
type: path
timeout:
description:
- How long to wait for the server to send data before giving up.
type: int
default: 10
"""

127
plugins/module_utils/_icinga2.py Normal file
View file

@ -0,0 +1,127 @@
# 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: 2026 Christoph Fiehe <christoph.fiehe@gmail.com>
# Note that this module util is **PRIVATE** to the collection. It can have breaking changes at any time.
# Do not use this from other collections or standalone plugins/modules!
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 = f"{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 is not None:
data["filter_vars"] = filter_vars
if fixed is not None:
data["fixed"] = fixed
if all_services is not None:
data["all_services"] = all_services
if trigger_name is not None:
data["trigger_name"] = trigger_name
if child_options is not None:
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 = f"{self.base_path}/remove-downtime"
data: dict[str, t.Any] = {"type": object_type}
if name is not None:
data[object_type.lower()] = name
if filter is not None:
data["filter"] = filter
if filter_vars is not None:
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

309
plugins/modules/icinga2_downtime.py Normal file
View file

@ -0,0 +1,309 @@
#!/usr/bin/python
#
# 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: 2026 Christoph Fiehe <christoph.fiehe@gmail.com>
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
details:
- In case of a complex filter expression, it may become very complex to decide
whether downtime creation or removal will succeed and trigger a change.
diff_mode:
support: none
options:
all_services:
description:
- Whether downtimes should be created 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 the downtime is fixed or flexible.
- If omitted, Icinga 2 creates a fixed downtime by default.
type: bool
name:
description:
- Name of the downtime object.
- 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 object 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:
- community.general._icinga2_api
- community.general.attributes
- ansible.builtin.url
"""
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
community.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(results) 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 downtime object.
returned: if a downtime was scheduled successfully
type: int
sample: 28911
name:
description: Name of the downtime object.
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 json
from contextlib import suppress
from ansible_collections.community.general.plugins.module_utils._icinga2 import (
Icinga2Client,
icinga2_argument_spec,
)
from ansible_collections.community.general.plugins.module_utils.module_helper import StateModuleHelper
class Icinga2Downtime(StateModuleHelper):
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 = dict(
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"]],
)
def __init_module__(self) -> None:
self.client = Icinga2Client(
module=self.module, # type:ignore[arg-type]
url=self.vars.url,
ca_path=self.vars.ca_path,
timeout=self.vars.timeout,
)
def state_present(self) -> None:
duration = self.vars.duration
end_time = self.vars.end_time
start_time = self.vars.start_time
if end_time <= start_time:
self.do_raise(msg="The end time must be later than the start time.")
if duration is None:
duration = end_time - start_time
response, info = self.client.actions.schedule_downtime(
all_services=self.vars.all_services,
author=self.vars.author,
child_options=self.vars.child_options,
comment=self.vars.comment,
duration=duration,
end_time=end_time,
filter_vars=self.vars.filter_vars,
filter=self.vars.filter,
fixed=self.vars.fixed,
object_type=self.vars.object_type,
start_time=start_time,
trigger_name=self.vars.trigger_name,
)
status_code = info["status"]
if 200 <= status_code <= 299:
self.vars.set("results", json.loads(response.read())["results"], output=True)
self.vars.msg = "Successfully scheduled downtime."
self.changed = True
elif status_code >= 400:
with suppress(KeyError, ValueError):
self.vars.set("error", json.loads(info["body"])) # type:ignore[arg-type]
self.do_raise(msg="Unable to schedule downtime.")
def state_absent(self) -> None:
response, info = self.client.actions.remove_downtime(
filter_vars=self.vars.filter_vars,
filter=self.vars.filter,
name=self.vars.name,
object_type=self.vars.object_type,
)
status_code = info["status"]
if 200 <= status_code <= 299:
self.vars.set("results", json.loads(response.read())["results"], output=True)
self.vars.msg = "Successfully removed downtime."
self.changed = True
elif status_code == 404:
self.vars.msg = "No matching downtime object found."
elif status_code >= 400:
with suppress(KeyError, ValueError):
self.vars.set("error", json.loads(info["body"])) # type:ignore[arg-type]
self.do_raise(msg="Unable to remove downtime.")
def main():
Icinga2Downtime.execute()
if __name__ == "__main__":
main()

224
tests/unit/plugins/modules/test_icinga2_downtime.py Normal file
View file

@ -0,0 +1,224 @@
# 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.module_utils.mh.deco import no_handle_exceptions
from ansible_collections.community.general.plugins.modules import icinga2_downtime
class TestIcinga2Downtime(ModuleTestCase):
def setUp(self):
super().setUp()
self.module = icinga2_downtime
@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 no_handle_exceptions(AnsibleExitJson, AnsibleFailJson):
with self.assertRaises(AnsibleExitJson) as result:
self.module.main()
self.assertFalse(result.exception.args[0]["failed"])
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": json.dumps({"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 no_handle_exceptions(AnsibleExitJson, AnsibleFailJson):
with self.assertRaises(AnsibleFailJson) as result:
self.module.main()
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 no_handle_exceptions(AnsibleExitJson, AnsibleFailJson):
with self.assertRaises(AnsibleExitJson) as result:
self.module.main()
self.assertFalse(result.exception.args[0]["failed"])
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": json.dumps({"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 no_handle_exceptions(AnsibleExitJson, AnsibleFailJson):
with self.assertRaises(AnsibleExitJson) as result:
self.module.main()
self.assertFalse(result.exception.args[0]["failed"])
self.assertFalse(result.exception.args[0]["changed"])
remove_downtime_mock.assert_called_once_with(
filter=None,
filter_vars=None,
name=module_args["name"],
object_type=module_args["object_type"],
)