| 1 | from fastapi import HTTPException |
| 2 | from loguru import logger |
| 3 | |
| 4 | from app.connectors.graylog.schema.monitoring import GraylogEventNotificationsResponse |
| 5 | from app.connectors.graylog.schema.monitoring import GraylogMessages |
| 6 | from app.connectors.graylog.schema.monitoring import GraylogMessagesResponse |
| 7 | from app.connectors.graylog.schema.monitoring import GraylogMetricsResponse |
| 8 | from app.connectors.graylog.schema.monitoring import GraylogThroughputMetrics |
| 9 | from app.connectors.graylog.schema.monitoring import GraylogThroughputMetricsCollection |
| 10 | from app.connectors.graylog.schema.monitoring import GraylogUncommittedJournalEntries |
| 11 | from app.connectors.graylog.utils.universal import send_get_request |
| 12 | |
| 13 | |
| 14 | async def get_messages(page_number: int) -> GraylogMessagesResponse: |
| 15 | """Get messages from Graylog. |
| 16 | |
| 17 | Args: |
| 18 | page_number (int): The page number of messages to retrieve. |
| 19 | |
| 20 | Returns: |
| 21 | GraylogMessagesResponse: The response object containing the retrieved messages. |
| 22 | |
| 23 | Raises: |
| 24 | HTTPException: If there is an error collecting the messages. |
| 25 | """ |
| 26 | logger.info("Getting messages from Graylog") |
| 27 | params = {"page": page_number} |
| 28 | messages_collected = await send_get_request( |
| 29 | endpoint="/api/system/messages", |
| 30 | params=params, |
| 31 | ) |
| 32 | try: |
| 33 | if messages_collected["success"]: |
| 34 | graylog_messages_list = [] |
| 35 | for message in messages_collected["data"]["messages"]: |
| 36 | graylog_message = GraylogMessages( |
| 37 | caller=message["caller"], |
| 38 | content=message["content"], |
| 39 | node_id=message["node_id"], |
| 40 | timestamp=message["timestamp"], |
| 41 | ) |
| 42 | graylog_messages_list.append(graylog_message) |
| 43 | return GraylogMessagesResponse( |
| 44 | graylog_messages=graylog_messages_list, |
| 45 | success=True, |
| 46 | message="Messages collected successfully", |
| 47 | total_messages=messages_collected["data"]["total"], |
| 48 | ) |
| 49 | |
| 50 | except KeyError as e: |
| 51 | logger.error(f"Failed to collect messages key: {e}") |
| 52 | raise HTTPException( |
| 53 | status_code=500, |
| 54 | detail=f"Failed to collect messages key: {e}", |
| 55 | ) |
| 56 | except Exception as e: |
| 57 | logger.error(f"Failed to collect messages: {e}") |
| 58 | raise HTTPException(status_code=500, detail=f"Failed to collect messages: {e}") |
| 59 | return GraylogMessagesResponse( |
| 60 | graylog_messages=[], |
| 61 | success=False, |
| 62 | message="Failed to collect messages", |
| 63 | ) |
| 64 | |
| 65 | |
| 66 | async def fetch_metrics_from_graylog() -> dict: |
| 67 | """ |
| 68 | Fetches metrics from Graylog. |
| 69 | |
| 70 | Returns: |
| 71 | A dictionary containing the fetched metrics. |
| 72 | """ |
| 73 | return await send_get_request(endpoint="/api/system/metrics") |
| 74 | |
| 75 | |
| 76 | async def fetch_uncommitted_journal_entries() -> dict: |
| 77 | """ |
| 78 | Fetches uncommitted journal entries from the Graylog system. |
| 79 | |
| 80 | Returns: |
| 81 | dict: A dictionary containing the uncommitted journal entries. |
| 82 | """ |
| 83 | return await send_get_request(endpoint="/api/system/journal") |
| 84 | |
| 85 | |
| 86 | def merge_metrics_data(throughput_metrics_collected: dict) -> dict: |
| 87 | """ |
| 88 | Merge the throughput metrics and input/output metrics into a single dictionary. |
| 89 | |
| 90 | Args: |
| 91 | throughput_metrics_collected (dict): The collected metrics data. |
| 92 | |
| 93 | Returns: |
| 94 | dict: The merged metrics data. |
| 95 | """ |
| 96 | throughput_metrics = throughput_metrics_collected["data"]["gauges"] |
| 97 | input_output_metrics = throughput_metrics_collected["data"]["counters"] |
| 98 | return {**throughput_metrics, **input_output_metrics} |
| 99 | |
| 100 | |
| 101 | def filter_and_create_throughput_metrics(merged_metrics: dict) -> list: |
| 102 | """ |
| 103 | Filters the merged metrics based on the model fields and creates a list of GraylogThroughputMetrics objects. |
| 104 | |
| 105 | Args: |
| 106 | merged_metrics (dict): A dictionary containing merged metrics. |
| 107 | |
| 108 | Returns: |
| 109 | list: A list of GraylogThroughputMetrics objects. |
| 110 | """ |
| 111 | model_fields = [field_info.alias for field_info in GraylogThroughputMetricsCollection.__fields__.values()] |
| 112 | throughput_metrics_list = [ |
| 113 | GraylogThroughputMetrics(metric=metric_name, value=metric_data.get("value", 0)) |
| 114 | for metric_name, metric_data in merged_metrics.items() |
| 115 | if metric_name in model_fields |
| 116 | ] |
| 117 | return throughput_metrics_list |
| 118 | |
| 119 | |
| 120 | async def get_metrics() -> GraylogMetricsResponse: |
| 121 | """ |
| 122 | Retrieves metrics from Graylog. |
| 123 | |
| 124 | Returns: |
| 125 | GraylogMetricsResponse: The response object containing the collected metrics. |
| 126 | """ |
| 127 | logger.info("Getting metrics from Graylog") |
| 128 | throughput_metrics_collected = await fetch_metrics_from_graylog() |
| 129 | uncommitted_journal_entries_collected = await fetch_uncommitted_journal_entries() |
| 130 | try: |
| 131 | if throughput_metrics_collected["success"] and uncommitted_journal_entries_collected["success"]: |
| 132 | merged_metrics = merge_metrics_data(throughput_metrics_collected) |
| 133 | throughput_metrics_list = filter_and_create_throughput_metrics( |
| 134 | merged_metrics, |
| 135 | ) |
| 136 | |
| 137 | uncommitted_journal_entries = GraylogUncommittedJournalEntries( |
| 138 | uncommitted_journal_entries=uncommitted_journal_entries_collected["data"]["uncommitted_journal_entries"], |
| 139 | ) |
| 140 | |
| 141 | return GraylogMetricsResponse( |
| 142 | throughput_metrics=throughput_metrics_list, |
| 143 | uncommitted_journal_entries=uncommitted_journal_entries.uncommitted_journal_entries, |
| 144 | success=True, |
| 145 | message="Metrics collected successfully", |
| 146 | ) |
| 147 | except KeyError as e: |
| 148 | raise HTTPException( |
| 149 | status_code=500, |
| 150 | detail=f"Failed to collect metrics key: {e}", |
| 151 | ) |
| 152 | except Exception as e: |
| 153 | raise HTTPException(status_code=500, detail=f"Failed to collect metrics: {e}") |
| 154 | |
| 155 | return GraylogMetricsResponse( |
| 156 | throughput_metrics=[], |
| 157 | uncommitted_journal_entries=0, |
| 158 | success=False, |
| 159 | message="Failed to collect metrics", |
| 160 | ) |
| 161 | |
| 162 | |
| 163 | async def get_event_notifications() -> GraylogEventNotificationsResponse: |
| 164 | """ |
| 165 | Retrieves event notifications from Graylog. |
| 166 | |
| 167 | Returns: |
| 168 | GraylogEventNotificationsResponse: The response object containing the collected event notifications. |
| 169 | """ |
| 170 | logger.info("Getting event notifications from Graylog") |
| 171 | event_notifications_collected = await send_get_request( |
| 172 | endpoint="/api/events/notifications", |
| 173 | ) |
| 174 | try: |
| 175 | if event_notifications_collected["success"]: |
| 176 | return GraylogEventNotificationsResponse( |
| 177 | event_notifications=event_notifications_collected["data"], |
| 178 | success=True, |
| 179 | message="Event notifications collected successfully", |
| 180 | ) |
| 181 | except KeyError as e: |
| 182 | raise HTTPException( |
| 183 | status_code=500, |
| 184 | detail=f"Failed to collect event notifications key: {e}", |
| 185 | ) |
| 186 | except Exception as e: |
| 187 | raise HTTPException( |
| 188 | status_code=500, |
| 189 | detail=f"Failed to collect event notifications: {e}", |
| 190 | ) |
| 191 | |
| 192 | return GraylogEventNotificationsResponse( |
| 193 | event_notifications=[], |
| 194 | success=False, |
| 195 | message="Failed to collect event notifications", |
| 196 | ) |