Create index.py
taylor_socfortress committed
Jul 10, 2023 at 16:45 UTC
511378730bd168e94b1ac250a8fddb8268005cb0
1 file changed
+79
backend/app/services/WazuhIndexer/index.py
new
+79
@@ -0,0 +1,79 @@
1
+from typing import Dict
2
+import requests
3
+from elasticsearch7 import Elasticsearch
4
+from loguru import logger
5
+from app.services.WazuhIndexer.universal import UniversalService
6
+
7
+
8
+class IndexService:
9
+ """
10
+ A service class that encapsulates the logic for pulling indices from the Wazuh-Indexer.
11
+ """
12
+
13
+ def __init__(self):
14
+ self._collect_wazuhindexer_details()
15
+ self._initialize_es_client()
16
+
17
+ def _collect_wazuhindexer_details(self):
18
+ self.connector_url, self.connector_username, self.connector_password = UniversalService().collect_wazuhindexer_details("Wazuh-Indexer")
19
+
20
+ def _initialize_es_client(self):
21
+ self.es = Elasticsearch(
22
+ [self.connector_url],
23
+ http_auth=(self.connector_username, self.connector_password),
24
+ verify_certs=False,
25
+ timeout=15,
26
+ max_retries=10,
27
+ retry_on_timeout=False,
28
+ )
29
+
30
+ def _are_details_collected(self) -> bool:
31
+ return all([self.connector_url, self.connector_username, self.connector_password])
32
+
33
+ def collect_indices_summary(self) -> Dict[str, object]:
34
+ """
35
+ Collects summary information for each index from the Wazuh-Indexer.
36
+
37
+ Returns:
38
+ dict: A dictionary containing the success status, a message, and potentially the indices.
39
+ """
40
+ if not self._are_details_collected():
41
+ return {"message": "Failed to collect Wazuh-Indexer details", "success": False}
42
+
43
+ index_summary = self._collect_indices()
44
+ if not index_summary["success"]:
45
+ return index_summary
46
+
47
+ summary = self._format_indices_summary(index_summary["indices"])
48
+
49
+ return {
50
+ "message": "Successfully collected indices summary",
51
+ "success": True,
52
+ "indices": summary,
53
+ }
54
+
55
+ def _format_indices_summary(self, indices: Dict[str, object]) -> Dict[str, object]:
56
+ return [
57
+ {
58
+ "index": index["index"],
59
+ "health": index["health"],
60
+ "docs_count": index["docs.count"],
61
+ "store_size": index["store.size"],
62
+ "replica_count": index["rep"],
63
+ }
64
+ for index in indices
65
+ ]
66
+
67
+ def _collect_indices(self) -> Dict[str, object]:
68
+ """
69
+ Collects the indices from the Wazuh-Indexer.
70
+
71
+ Returns:
72
+ dict: A dictionary containing the success status, a message and potentially the indices.
73
+ """
74
+ try:
75
+ indices = self.es.cat.indices(format="json")
76
+ return {"message": "Successfully collected indices", "success": True, "indices": indices}
77
+ except Exception as e:
78
+ logger.error(e)
79
+ return {"message": "Failed to collect indices", "success": False}