auto-closed

auto-acknowledge.py

@@ -0,0 +1,75 @@
+import os
+import sys
+import logging
+from zabbix_utils import ZabbixAPI
+
+stdout_handler = logging.StreamHandler(stream=sys.stdout)
+handlers = [stdout_handler]
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s",
+    handlers=handlers,
+)
+logger = logging.getLogger(__name__)
+
+# Configuration
+ZABBIX_URL = os.getenv("ZABBIX_URL", "https://nm.vn.backone.cloud/api_jsonrpc.php")
+API_TOKEN = os.getenv(
+    "API_TOKEN", "dac06ab72f7c64392ff23adc021a61fd27397b1790406d97bd883c6d4ae7b628"
+)
+TRIGGER_DESCRIPTION_FILTER = os.getenv(
+    "TRIGGER_DESCRIPTION_FILTER", "Interface eth3: High bandwidth usage"
+)  # The exact or partial name/description of the trigger
+ACK_MESSAGE = os.getenv(
+    "ACK_MESSAGE", "Automatically acknowledged via zabbix-utils script."
+)
+
+# Initialize the ZabbixAPI object and log in using the token
+try:
+    api = ZabbixAPI(url=ZABBIX_URL)
+    # The login method accepts the 'token' parameter for token-based authentication
+    api.login(token=API_TOKEN)
+    logging.info("Authentication successful!")
+
+except Exception as e:
+    logging.error(f"Authentication failed or an error occurred: {e}")
+    sys.exit(1)
+
+
+def auto_acknowledge(description_filter=TRIGGER_DESCRIPTION_FILTER):
+    try:
+        # 1. Search for problem events matching the trigger description
+        # The 'trigger' filter is used to search the trigger name/description
+        events = api.event.get(
+            output="extend",
+            # Only get events that are currently a problem and not yet acknowledged
+            filter={"value": 1, "acknowledged": 0},
+            search={"name": description_filter},
+            # select_triggers=["triggerid", "description"],
+        )
+
+        print(events)
+        # sys.exit(0)
+
+        if not events:
+            print(
+                f"No unacknowledged problem events found matching '{TRIGGER_DESCRIPTION_FILTER}'."
+            )
+            return
+
+        event_ids = [event["eventid"] for event in events]
+
+        # 2. Acknowledge the found events
+        acknowledged_events = api.event.acknowledge(
+            eventids=event_ids, message=ACK_MESSAGE, action=1
+        )
+
+        print(f"Acknowledged event IDs: {acknowledged_events['eventids']}")
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+if __name__ == "__main__":
+    auto_acknowledge()

auto-close.py

@@ -0,0 +1,77 @@
+import os
+import sys
+import logging
+from zabbix_utils import ZabbixAPI
+
+stdout_handler = logging.StreamHandler(stream=sys.stdout)
+handlers = [stdout_handler]
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s",
+    handlers=handlers,
+)
+logger = logging.getLogger(__name__)
+
+# Configuration
+ZABBIX_URL = os.getenv("ZABBIX_URL", "https://nm.vn.backone.cloud/api_jsonrpc.php")
+API_TOKEN = os.getenv(
+    "API_TOKEN", "dac06ab72f7c64392ff23adc021a61fd27397b1790406d97bd883c6d4ae7b628"
+)
+TRIGGER_DESCRIPTION_FILTER = os.getenv(
+    "TRIGGER_DESCRIPTION_FILTER", "Interface dsa: High error rate"
+)  # The exact or partial name/description of the trigger
+ACK_MESSAGE = os.getenv("ACK_MESSAGE", "Automatically closed via zabbix-utils script.")
+
+# Initialize the ZabbixAPI object and log in using the token
+try:
+    api = ZabbixAPI(url=ZABBIX_URL)
+    # The login method accepts the 'token' parameter for token-based authentication
+    api.login(token=API_TOKEN)
+    logging.info("Authentication successful!")
+
+except Exception as e:
+    logging.error(f"Authentication failed or an error occurred: {e}")
+    sys.exit(1)
+
+
+def auto_acknowledge(description_filter=TRIGGER_DESCRIPTION_FILTER):
+    try:
+        # 1. Search for problem events matching the trigger description
+        # The 'trigger' filter is used to search the trigger name/description
+        events = api.event.get(
+            output="extend",
+            # Only get events that are currently a problem and not yet acknowledged
+            filter={
+                "value": 1,
+                "acknowledged": 0,
+            },
+            # trigger_severities=[2, 3, 4, 5],
+            search={"name": description_filter},
+            # select_triggers=["triggerid", "description"],
+        )
+
+        print(events)
+        sys.exit(0)
+
+        if not events:
+            print(
+                f"No unacknowledged problem events found matching '{TRIGGER_DESCRIPTION_FILTER}'."
+            )
+            return
+
+        event_ids = [event["eventid"] for event in events]
+
+        # 2. Acknowledge the found events
+        acknowledged_events = api.event.acknowledge(
+            eventids=event_ids, message=ACK_MESSAGE, action=1
+        )
+
+        print(f"Acknowledged and closed event IDs: {acknowledged_events['eventids']}")
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+if __name__ == "__main__":
+    auto_acknowledge()

disable_host_by_trigger.py

@@ -31,6 +31,7 @@ try:
 
 except Exception as e:
     logging.error(f"Authentication failed or an error occurred: {e}")
+    sys.exit(1)
 
 
 def get_enabled_hosts_with_triggers():