Fetch the most recent 5 logs from the elasticsearch index <index_name> in last n minutes <lookback_minutes>

import requests import json from urllib.parse import urlparse from datetime import datetime # === Config / Inputs === # These are expected to be provided by your runtime/environment: # - ELASTIC_URL_OTEL (env var, via getEnvVar) # - index_name (string) # - log_count (int) # Optionally provide lookback_minutes; otherwise default to 15. try: if not lookback_minutes or not isinstance(lookback_minutes, (int, float)) or lookback_minutes <= 0: print("[INFO] lookback_minutes not set or invalid — defaulting to 15") lookback_minutes = 15 else: print(f"[INFO] Using lookback_minutes={lookback_minutes}") except NameError: print("[INFO] lookback_minutes not defined — defaulting to 15") lookback_minutes = 15 elastic_url = getEnvVar('ELASTIC_URL_OTEL') # Parse URL to determine if SSL should be used parsed_url = urlparse(elastic_url) use_ssl = parsed_url.scheme == 'https' # ---- Build time filter and query ---- print(f"Fetching last {lookback_minutes} minute(s) of logs from index '{index_name}'") time_filter = { "range": { "@timestamp": { "gte": f"now-{lookback_minutes}m", "lte": "now" } } } search_query = { "query": { "bool": { "filter": [ time_filter # restrict to recent window ] } }, "sort": [ {"@timestamp": {"order": "desc"}} ], "size": log_count, "_source": [ "@timestamp", "body", "severity.text", "resource.service.name", "resource.k8s.pod.name", "resource.k8s.namespace.name" ] } try: # Make request to search for logs response = requests.post( f"{elastic_url}/{index_name}/_search", headers={'Content-Type': 'application/json'}, data=json.dumps(search_query), verify=use_ssl, timeout=30 ) if response.status_code == 200: search_results = response.json() # Extract hits and total count hits = search_results.get('hits', {}) total_hits = hits.get('total', {}).get('value', 0) log_entries = hits.get('hits', []) # Process the logs to extract relevant information recent_logs = [] for log_entry in log_entries: source = log_entry.get('_source', {}) # Extract timestamp (handle array format) timestamp = source.get('@timestamp', ['']) if isinstance(timestamp, list) and len(timestamp) > 0: timestamp = timestamp[0] # Extract body (handle array format) body = source.get('body', ['']) if isinstance(body, list) and len(body) > 0: body = body[0] # Extract severity (handle array format) severity = source.get('severity.text', ['']) if isinstance(severity, list) and len(severity) > 0: severity = severity[0] # Extract service name (handle array format) service_name = source.get('resource.service.name', ['']) if isinstance(service_name, list) and len(service_name) > 0: service_name = service_name[0] # Extract pod name (handle array format) pod_name = source.get('resource.k8s.pod.name', ['']) if isinstance(pod_name, list) and len(pod_name) > 0: pod_name = pod_name[0] # Extract namespace (handle array format) namespace = source.get('resource.k8s.namespace.name', ['']) if isinstance(namespace, list) and len(namespace) > 0: namespace = namespace[0] log_info = { 'timestamp': timestamp, 'body': body, 'severity': severity, 'service_name': service_name, 'pod_name': pod_name, 'namespace': namespace, 'index': log_entry.get('_index', ''), 'id': log_entry.get('_id', '') } recent_logs.append(log_info) fetch_successful = True print(f"Successfully fetched {len(recent_logs)} logs from index '{index_name}'") print(f"Total logs available in index (all time): {total_hits}") # Print a summary of the logs for i, log in enumerate(recent_logs, 1): print(f"\nLog {i}:") print(f" Timestamp: {log['timestamp']}") print(f" Service: {log['service_name']}") print(f" Severity: {log['severity']}") print(f" Body: {log['body'][:100]}{'...' if len(log['body']) > 100 else ''}") else: print(f"Error fetching logs: HTTP {response.status_code}") print(f"Response: {response.text}") recent_logs = [] total_hits = 0 fetch_successful = False except Exception as e: print(f"Exception occurred while fetching logs: {str(e)}") recent_logs = [] total_hits = 0 fetch_successful = False print(f"\nOutput parameters:") print(f"recent_logs: {json.dumps(recent_logs, indent=2)}") print(f"total_hits: {total_hits}") print(f"fetch_successful: {fetch_successful}")