Refactor Coralogix tests to cloud-only with simplified utilities

conftest.py

@@ -260,7 +260,10 @@ def responses():
         rsps.add_passthru("https://api.ap1.datadoghq.com")
         rsps.add_passthru("https://app.datadoghq.com")
         rsps.add_passthru("https://app.datadoghq.eu")
-        rsps.add_passthru("https://ng-api-http.eu2.coralogix.com")
+        # Allow all Coralogix API calls (query and ingestion endpoints, all regions)
+        rsps.add_passthru(re.compile(r"https://.*\.coralogix\.com"))
+        rsps.add_passthru(re.compile(r"https://.*\.coralogix\.us"))
+        rsps.add_passthru(re.compile(r"https://.*\.coralogix\.in"))
 
         # Allow Elasticsearch/OpenSearch Cloud API calls (various hosting regions)
         rsps.add_passthru(re.compile(r"https://.*\.cloud\.es\.io"))  # Elastic Cloud

docs/data-sources/builtin-toolsets/coralogix-logs.md

@@ -33,7 +33,7 @@ toolsets:
 
 ```
 
-**Note**: Both toolsets use the same API key. The DataPrime toolset supports fields (`CoralogixConfig`): `api_key`, `domain`, `team_hostname`, optional `labels`.
+**Note**: Both toolsets use the same API key. The DataPrime toolset supports fields (`CoralogixConfig`): `api_key`, `domain`, `team_hostname`.
 
 ## Recommended: Customize Coralogix Instructions
 

holmes/plugins/toolsets/coralogix/utils.py

@@ -1,34 +1,15 @@
 import json
 import logging
 from datetime import datetime
-from typing import Any, Dict, List, NamedTuple, Optional
+from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel
 
 
-class FlattenedLog(NamedTuple):
-    timestamp: str
-    log_message: str
-
-
-class CoralogixQueryResult(BaseModel):
-    logs: List[FlattenedLog]
-    http_status: Optional[int]
-    error: Optional[str]
-
-
-class CoralogixLabelsConfig(BaseModel):
-    pod: str = "resource.attributes.k8s.pod.name"
-    namespace: str = "resource.attributes.k8s.namespace.name"
-    log_message: str = "logRecord.body"
-    timestamp: str = "logRecord.attributes.time"
-
-
 class CoralogixConfig(BaseModel):
     team_hostname: str
     domain: str
     api_key: str
-    labels: CoralogixLabelsConfig = CoralogixLabelsConfig()
 
 
 def parse_json_lines(raw_text) -> List[Dict[str, Any]]:
@@ -73,72 +54,3 @@ def normalize_datetime(date_str: Optional[str]) -> str:
         return date_str
 
 
-def extract_field(data_obj: dict[str, Any], field: str):
-    """returns a nested field from a dict
-    e.g. extract_field({"parent": {"child": "value"}}, "parent.child") => value
-    """
-    current_object: Any = data_obj
-    fields = field.split(".")
-
-    for field in fields:
-        if not current_object:
-            return None
-        if isinstance(current_object, dict):
-            current_object = current_object.get(field)
-        else:
-            return None
-
-    return current_object
-
-
-def flatten_structured_log_entries(
-    log_entries: List[Dict[str, Any]],
-    labels_config: CoralogixLabelsConfig,
-) -> List[FlattenedLog]:
-    flattened_logs = []
-    for log_entry in log_entries:
-        try:
-            userData = json.loads(log_entry.get("userData", "{}"))
-            log_message = extract_field(userData, labels_config.log_message)
-            timestamp = extract_field(userData, labels_config.timestamp)
-            if not log_message or not timestamp:
-                log_message = json.dumps(userData)
-            else:
-                flattened_logs.append(
-                    FlattenedLog(timestamp=timestamp, log_message=log_message)
-                )  # Store as tuple for sorting
-
-        except json.JSONDecodeError:
-            logging.error(f"Failed to decode userData JSON: {json.dumps(log_entry)}")
-    return flattened_logs
-
-
-def stringify_flattened_logs(log_entries: List[FlattenedLog]) -> str:
-    formatted_logs = []
-    for entry in log_entries:
-        formatted_logs.append(entry.log_message)
-
-    return "\n".join(formatted_logs) if formatted_logs else "No logs found."
-
-
-def parse_json_objects(
-    json_objects: List[Dict[str, Any]], labels_config: CoralogixLabelsConfig
-) -> List[FlattenedLog]:
-    """Extracts timestamp and log values from parsed JSON objects, sorted in ascending order (oldest first)."""
-    logs: List[FlattenedLog] = []
-
-    for data in json_objects:
-        if isinstance(data, dict) and "result" in data and "results" in data["result"]:
-            logs += flatten_structured_log_entries(
-                log_entries=data["result"]["results"], labels_config=labels_config
-            )
-        elif isinstance(data, dict) and data.get("warning"):
-            logging.info(
-                f"Received the following warning when fetching coralogix logs: {data}"
-            )
-        else:
-            logging.debug(f"Unrecognised partial response from coralogix logs: {data}")
-
-    logs.sort(key=lambda x: x[0])
-
-    return logs