From d5a6ef34618a6bfa18e29b7af131ade2209ea095 Mon Sep 17 00:00:00 2001 From: irl Date: Sat, 20 Dec 2025 17:14:16 +0000 Subject: [PATCH] feat(cloudns): prototype module for handling monitoring checks --- plugins/modules/cloudns_monitor.py | 241 +++++++++++++++++++++++++++++ 1 file changed, 241 insertions(+) create mode 100644 plugins/modules/cloudns_monitor.py diff --git a/plugins/modules/cloudns_monitor.py b/plugins/modules/cloudns_monitor.py new file mode 100644 index 0000000..6761a4c --- /dev/null +++ b/plugins/modules/cloudns_monitor.py @@ -0,0 +1,241 @@ +#!/usr/bin/python + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = r''' +--- +module: cloudns_monitor + +short_description: Configure a ClouDNS monitoring check + +description: Configure a ClouDNS monitoring check + +options: + name: + description: The name of the ClouDNS monitoring check. + required: true + type: str + +author: + - @irl +''' + +EXAMPLES = r''' +TODO +''' + +RETURN = r''' +TODO +''' + +import requests +from ansible.module_utils.basic import AnsibleModule + +CHECK_TYPE = { + 'ping': '17', + 'web': '18', +} + +def get_existing_check(auth_id: str, auth_password: str, name: str) -> dict | None: + url = "https://api.cloudns.net/monitoring/get-records.json" + params = { + 'auth-id': auth_id, + 'auth-password': auth_password, + 'search': name, + 'rows-per-page': 50, + 'page': 1, + } + r = requests.get(url, params=params) + checks = r.json() + for check_id in checks: + if checks[check_id]['name'] == name: + return checks[check_id] + return None + +def update_existing_check(auth_id: str, auth_password: str, check_id: str, + check_type: str, host: str, ip: str, http_status_code: str, + name: str, path: str, port: int) -> dict | None: + url = "https://api.cloudns.net/monitoring/update.json" + params = { + 'auth-id': auth_id, + 'auth-password': auth_password, + 'id': check_id, + 'check_type': CHECK_TYPE[check_type], + 'host': host, + 'http_status_code': http_status_code, + 'ip': ip, + 'name': name, + 'port': port, + 'path': path, + 'status_change_checks': 5, + 'check_period': 60, + 'timeout': 5, + } + r = requests.get(url, params=params) + return r.json() + +def create_check(auth_id: str, auth_password: str, + check_type: str, host: str, ip: str, http_status_code: str, + name: str, path: str, port: int) -> dict | None: + url = "https://api.cloudns.net/monitoring/create.json" + params = { + 'auth-id': auth_id, + 'auth-password': auth_password, + 'check_type': CHECK_TYPE[check_type], + 'host': host, + 'http_status_code': http_status_code, + 'ip': ip, + 'name': name, + 'port': port, + 'path': path, + 'status_change_checks': 5, + 'check_period': 60, + 'timeout': 5, + } + r = requests.get(url, params=params) + return r.json() + +def get_existing_notifications(auth_id: str, auth_password: str, check_id: str) -> dict | None: + url = "https://api.cloudns.net/monitoring/list-notifications.json" + params = { + 'auth-id': auth_id, + 'auth-password': auth_password, + 'id': check_id, + 'rows-per-page': 50, + 'page': 1, + } + r = requests.get(url, params=params) + return r.json() + +def run_module(): + module_args = dict( + auth_id=dict(type='str', required=True), + auth_password=dict(type='str', no_log=True, required=True), + check_type=dict(type='str', default="web"), + emails=dict(type='list', elements='str', required=True), + host=dict(type='str', required=True), + http_status_code=dict(type='str', default=200), + ip=dict(type='str', required=True), + name=dict(type='str', required=True), + path=dict(type='str', default=''), + port=dict(type='int', default=443), + ) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True + ) + + result = dict( + changed=False, + existing_check=get_existing_check( + module.params['auth_id'], + module.params['auth_password'], + module.params['name'] + ), + ) + + if module.check_mode: + module.exit_json(**result) + + if not result['existing_check']: + result['changed'] = True + create_check(auth_id=module.params['auth_id'], + auth_password=module.params['auth_password'], + check_type=module.params['check_type'], + host=module.params['host'], + http_status_code=module.params['http_status_code'], + ip=module.params['ip'], + name=module.params['name'], + path=module.params['path'], + port=module.params['port'], + ) + else: + if result['existing_check']['check_type'] != CHECK_TYPE[module.params['check_type']]: + result['changed'] = True + if result['existing_check']['host'] != module.params['host']: + result['changed'] = True + if result['existing_check']['http_status_code'] != module.params['http_status_code']: + result['changed'] = True + if result['existing_check']['ip'] != module.params['ip']: + result['changed'] = True + if result['existing_check']['path'] != module.params['path']: + result['changed'] = True + if int(result['existing_check']['port']) != module.params['port']: + result['changed'] = True + + if result['changed']: + result['update_response'] = update_existing_check( + auth_id=module.params['auth_id'], + auth_password=module.params['auth_password'], + check_id=result['existing_check']['id'], + check_type=module.params['check_type'], + host=module.params['host'], + http_status_code=module.params['http_status_code'], + ip=module.params['ip'], + name=module.params['name'], + path=module.params['path'], + port=module.params['port'], + ) + + result['current_check'] = get_existing_check( + module.params['auth_id'], module.params['auth_password'], module.params['name']) + + emails_to_add = set(module.params['emails']) + notifications_to_remove = set() + + result['existing_notifications'] = get_existing_notifications( + auth_id=module.params['auth_id'], + auth_password=module.params['auth_password'], + check_id=result['current_check']['id'], + ) + + for notification in result['existing_notifications']: + if notification['type'] == "mail": + if notification['value'] in emails_to_add: + emails_to_add.remove(notification['value']) + else: + notifications_to_remove.add(notification['notification_id']) + + if emails_to_add or notifications_to_remove: + result['changed'] = True + + for email in emails_to_add: + url = "https://api.cloudns.net/monitoring/add-notification.json" + params = { + 'auth-id': module.params['auth_id'], + 'auth-password': module.params['auth_password'], + 'id': result['current_check']['id'], + 'type': 'mail', + 'value': email, + } + r = requests.get(url, params=params) + r.raise_for_status() + + for email in notifications_to_remove: + url = "https://api.cloudns.net/monitoring/delete-notifications.json" + params = { + 'auth-id': module.params['auth_id'], + 'auth-password': module.params['auth_password'], + 'id': result['current_check']['id'], + 'notification-id': email, + } + r = requests.get(url, params=params) + r.raise_for_status() + + result['current_notifications'] = get_existing_notifications( + auth_id=module.params['auth_id'], + auth_password=module.params['auth_password'], + check_id=result['current_check']['id'], + ) + + module.exit_json(**result) + + +def main(): + run_module() + + +if __name__ == '__main__': + main()