acc-ltd/EKSLOGS
0
1import boto32import json3from datetime import datetime, timedelta4from datetime import datetime, timedelta, timezone5from collections import defaultdict6import os7import subprocess8# from dotenv import load_dotenv9import logging10 11# Basic configuration12logging.basicConfig(13 level=logging.INFO, # Options: DEBUG, INFO, WARNING, ERROR, CRITICAL14 format='%(asctime)s - %(levelname)s - %(message)s',15 handlers=[16 logging.StreamHandler() # Logs to stdout (needed for Docker)17 ]18)19# Load .env file20# load_dotenv()21 22# Save filtered logs to files23# for stream_name, logs in logs_by_stream.items():24# if logs: # only write if logs exist25# safe_name = stream_name.replace("/", "_").replace(":", "_")26# file_path = os.path.join(output_dir, f"{safe_name}.json")27# with open(file_path, 'w') as f:28# json.dump(logs, f, indent=2)29 30# Save filtered logs to files grouped by pod_id31 32 33 34class EKSCLoudWatch:35 def __init__(self):36 # Load AWS credentials from config.json37 # with open('config.json') as f:38 # config = json.load(f)39 40 aws_access_key = os.getenv('aws_access_key_id')41 aws_secret_key = os.getenv('aws_secret_access_key')42 region = os.getenv('region')43 44 45 # Initialize session and logs client46 session = boto3.Session(47 aws_access_key_id=aws_access_key,48 aws_secret_access_key=aws_secret_key,49 region_name=region50 )51 self.client = session.client('logs')52 53 54 def sendError(self,logs_by_pod):55 webhook_url = "https://acc-ltd-acc-ai-ops-dashboard-apis.hf.space/api/v1/webhooks/aws/sns"56 topic_arn = "arn:aws:sns:ap-south-1:1234567812:ec2-health-alerts"57 subject = "EKS POD apllication Failed"58 output_dir = "log_streams_output"59 for pod_id, pod_data in logs_by_pod.items():60 if pod_data["logs"]:61 safe_name = pod_id.replace("/", "_").replace(":", "_")62 file_path = os.path.join(output_dir, f"{safe_name}.json")63 # with open(file_path, 'w') as f:64 # json.dump(pod_data, f, indent=2)65 66 67 68 print(f"\n✅ Only logs containing 'Error' have been saved to '{output_dir}'\n")69 70 for pod_id, pod_data in logs_by_pod.items():71 if pod_data["logs"]:72 # Serialize the pod_data as a string for the Message field73 message_content = json.dumps(pod_data)74 75 # Construct the full SNS-style JSON body76 sns_payload = {77 "Type": "Notification",78 "MessageId": "test-message-id-1234",79 "TopicArn": topic_arn,80 "Subject": subject,81 "Message": message_content,82 "Timestamp": "2025-07-23T05:45:00.000Z"83 }84 85 # Convert to JSON string86 payload_str = json.dumps(sns_payload)87 88 # Run curl using subprocess89 curl_command = [90 "curl", "--location", webhook_url,91 "--header", "Content-Type: application/json",92 "--data-raw", payload_str93 ]94 95 print(f"Sending alert for pod: {pod_id}")96 result = subprocess.run(curl_command, capture_output=True, text=True)97 98 try:99 response_json = json.loads(result.stdout.strip())100 status = response_json.get("status")101 incident_id = response_json.get("incident_id")102 103 logging.info(f"✅ Status: {status}")104 logging.info(f"🆔 Incident ID: {incident_id}")105 106 except json.JSONDecodeError:107 print("❌ Failed to decode response as JSON:")108 print(result.stdout)109 def fetchLogs(self,eksname):110 # Log group name111 #log_group = "/aws/containerinsights/atlas-api-manager-eks-cluster/application"112 log_group = f"/aws/containerinsights/{eksname}/application"113 114 # Today's UTC time range115 now = datetime.now(timezone.utc)116 start = now117 end = now + timedelta(minutes=30)118 119 # Convert to epoch milliseconds120 start_ms = int(start.timestamp() * 1000)121 end_ms = int(end.timestamp() * 1000)122 123 # Create output directory124 output_dir = "log_streams_output"125 #os.makedirs(output_dir, exist_ok=True)126 127 # Group error logs by log stream128 logs_by_stream = defaultdict(list)129 130 # Paginate through log events131 paginator = self.client.get_paginator('filter_log_events')132 pages = paginator.paginate(133 logGroupName=log_group,134 startTime=start_ms,135 endTime=end_ms136 )137 logs_by_pod = {} # dict of pod_id -> full object with metadata and logs138 last_pod_id = None139 current_log_accumulator = ""140 pod_metadata = None141 for page in pages:142 for event in page.get('events', []):143 message = event['message']144 145 # Filter messages containing "error"146 if "error" not in message.lower():147 continue148 149 try:150 parsed = json.loads(message)151 pod_info = parsed.get("kubernetes", {})152 pod_id = pod_info.get("pod_id")153 log = parsed.get("log", "").strip()154 155 if not pod_id or not isinstance(log, str):156 continue157 158 # If new pod encountered, store previous logs and reset159 if pod_id != last_pod_id:160 if last_pod_id and current_log_accumulator:161 logs_by_pod[last_pod_id]["logs"].append(current_log_accumulator.strip())162 current_log_accumulator = log163 last_pod_id = pod_id164 165 # If new, initialize metadata166 if pod_id not in logs_by_pod:167 logs_by_pod[pod_id] = {168 "eksname":eksname,169 "pod_id": pod_id,170 "pod_name": pod_info.get("pod_name"),171 "namespace_name": pod_info.get("namespace_name"),172 "host": pod_info.get("host"),173 "container_name": pod_info.get("container_name"),174 "docker_id": pod_info.get("docker_id"),175 "container_hash": pod_info.get("container_hash"),176 "container_image": pod_info.get("container_image"),177 "logs": []178 }179 else:180 current_log_accumulator += "\n" + log181 182 except Exception as e:183 continue # skip malformed messages184 185 # Save the last accumulated block186 if last_pod_id and current_log_accumulator:187 logs_by_pod[last_pod_id]["logs"].append(current_log_accumulator.strip())188 189 190 self.sendError(logs_by_pod)191 # ✅ Print results (or save to file)192 for pod_id, logs in logs_by_pod.items():193 print(f"\n--- Logs for Pod: {pod_id} ---\n")194 for block in logs:195 print(block)196 print("-" * 40)197 198 199objEks=EKSCLoudWatch()200ListofEks=['atlas-api-manager-eks-cluster']201for name in ListofEks:202 objEks.fetchLogs(name)203