WIP: refactoring - almost done

Author: Yuriy Bezsonov

infrastructure/scripts/setup/monitoring-jvm.sh

@@ -83,19 +83,22 @@ FOLDER_RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" \
   -d "{\"title\": \"$FOLDER_NAME\"}" \
   "$GRAFANA_URL/api/folders")
 
+FOLDER_UID=$(echo "$FOLDER_RESPONSE" | jq -r '.uid // empty')
 FOLDER_ID=$(echo "$FOLDER_RESPONSE" | jq -r '.id // empty')
-if [[ -z "$FOLDER_ID" ]]; then
+if [[ -z "$FOLDER_UID" ]]; then
   # Try to get existing folder
-  EXISTING_FOLDER=$(curl -s -u "$GRAFANA_USER:$GRAFANA_PASSWORD" "$GRAFANA_URL/api/folders" | jq -r ".[] | select(.title == \"$FOLDER_NAME\") | .id // empty")
+  EXISTING_FOLDER=$(curl -s -u "$GRAFANA_USER:$GRAFANA_PASSWORD" "$GRAFANA_URL/api/folders" | jq -r ".[] | select(.title == \"$FOLDER_NAME\")")
   if [[ -n "$EXISTING_FOLDER" ]]; then
-    FOLDER_ID="$EXISTING_FOLDER"
-    log "📁 Using existing folder: $FOLDER_ID"
+    FOLDER_UID=$(echo "$EXISTING_FOLDER" | jq -r '.uid')
+    FOLDER_ID=$(echo "$EXISTING_FOLDER" | jq -r '.id')
+    log "📁 Using existing folder: $FOLDER_UID"
   else
+    FOLDER_UID=""
     FOLDER_ID=0
-    log "⚠️ Using General folder (ID: 0)"
+    log "⚠️ Using General folder"
   fi
 else
-  log "✅ Folder created: $FOLDER_ID"
+  log "✅ Folder created: $FOLDER_UID"
 fi
 
 log "📊 Creating JVM dashboard..."
@@ -178,104 +181,101 @@ WEBHOOK_USER="grafana-alerts"
 SECRET_VALUE=$(aws secretsmanager get-secret-value --secret-id "$SECRET_NAME" --query 'SecretString' --output text)
 WEBHOOK_PASSWORD=$(echo "$SECRET_VALUE" | jq -r '.password')
 
-CONTACT_RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" \
-  -u "$GRAFANA_USER:$GRAFANA_PASSWORD" \
-  -d "{
-    \"name\": \"$CONTACT_POINT_NAME\",
-    \"type\": \"webhook\",
-    \"settings\": {
-      \"url\": \"$LAMBDA_URL\",
-      \"httpMethod\": \"POST\",
-      \"username\": \"$WEBHOOK_USER\",
-      \"password\": \"$WEBHOOK_PASSWORD\",
-      \"authorization_scheme\": \"basic\"
-    },
-    \"disableResolveMessage\": false
-  }" \
-  "$GRAFANA_URL/api/v1/provisioning/contact-points")
+# Check if contact point already exists
+EXISTING_CONTACT=$(curl -s -u "$GRAFANA_USER:$GRAFANA_PASSWORD" "$GRAFANA_URL/api/v1/provisioning/contact-points" | jq -r ".[] | select(.name == \"$CONTACT_POINT_NAME\") | .name // empty")
 
-log "🚨 Creating alert rule with classic conditions..."
-ALERT_RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" \
-  -u "$GRAFANA_USER:$GRAFANA_PASSWORD" \
-  -d "{
-    \"title\": \"$ALERT_TITLE\",
-    \"condition\": \"B\",
-    \"data\": [
-      {
-        \"refId\": \"A\",
-        \"relativeTimeRange\": {\"from\": 600, \"to\": 0},
-        \"datasourceUid\": \"promds\",
-        \"model\": {
-          \"expr\": \"sum(jvm_threads_live_threads{job=~\\\"kubernetes-pods|ecs-unicorn-store-spring\\\"}) by (task_pod_id, cluster_type, cluster, container_name, namespace, container_ip)\",
-          \"instant\": true,
-          \"refId\": \"A\"
-        }
+if [[ -z "$EXISTING_CONTACT" ]]; then
+  CONTACT_RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" \
+    -u "$GRAFANA_USER:$GRAFANA_PASSWORD" \
+    -d "{
+      \"name\": \"$CONTACT_POINT_NAME\",
+      \"type\": \"webhook\",
+      \"settings\": {
+        \"url\": \"$LAMBDA_URL\",
+        \"httpMethod\": \"POST\",
+        \"username\": \"$WEBHOOK_USER\",
+        \"password\": \"$WEBHOOK_PASSWORD\",
+        \"authorization_scheme\": \"basic\"
       },
-      {
-        \"refId\": \"B\",
-        \"relativeTimeRange\": {\"from\": 0, \"to\": 0},
-        \"datasourceUid\": \"-100\",
-        \"model\": {
-          \"conditions\": [
-            {
-              \"evaluator\": {\"params\": [$THREAD_THRESHOLD], \"type\": \"gt\"},
-              \"operator\": {\"type\": \"and\"},
-              \"query\": {\"params\": [\"A\"]},
-              \"reducer\": {\"params\": [], \"type\": \"last\"},
-              \"type\": \"query\"
-            }
-          ],
-          \"refId\": \"B\",
-          \"type\": \"classic_conditions\"
-        }
+      \"disableResolveMessage\": false
+    }" \
+    "$GRAFANA_URL/api/v1/provisioning/contact-points")
+
+  # Check if contact point creation was successful
+  if echo "$CONTACT_RESPONSE" | jq -e '.name' > /dev/null 2>&1; then
+    log "✅ Contact point created"
+  else
+    log "❌ Contact point creation failed:"
+    echo "$CONTACT_RESPONSE" | jq .
+  fi
+else
+  log "✅ Contact point already exists"
+fi
+
+log "🚨 Creating alert rule with proper label preservation..."
+# Note: Using raw metrics without sum() and by() to preserve all original labels
+# This ensures both 'pod' (for EKS) and 'task_pod_id' (for ECS) labels are available
+# The Lambda function will process ALL metrics in the valueString, handling multiple
+# containers (EKS pods + ECS tasks) in a single alert when they exceed the threshold
+ALERT_PAYLOAD="{
+  \"title\": \"$ALERT_TITLE\",
+  \"condition\": \"B\",
+  \"data\": [
+    {
+      \"refId\": \"A\",
+      \"relativeTimeRange\": {\"from\": 600, \"to\": 0},
+      \"datasourceUid\": \"promds\",
+      \"model\": {
+        \"expr\": \"jvm_threads_live_threads{job=~\\\"kubernetes-pods|ecs-unicorn-store-spring\\\"}\",
+        \"instant\": true,
+        \"refId\": \"A\"
       }
-    ],
-    \"intervalSeconds\": 60,
-    \"noDataState\": \"NoData\",
-    \"execErrState\": \"Alerting\",
-    \"for\": \"1m\",
-    \"annotations\": {
-      \"summary\": \"High JVM Threads\",
-      \"description\": \"High number of JVM threads detected. Triggering Lambda thread dump.\",
-      \"webhookUrl\": \"$LAMBDA_URL\"
     },
-    \"labels\": {
-      \"severity\": \"critical\",
-      \"alertname\": \"High JVM Threads\",
-      \"cluster\": \"{{ \$labels.cluster }}\",
-      \"cluster_type\": \"{{ \$labels.cluster_type }}\",
-      \"container_name\": \"{{ \$labels.container_name }}\",
-      \"namespace\": \"{{ \$labels.namespace }}\",
-      \"task_pod_id\": \"{{ \$labels.task_pod_id }}\",
-      \"container_ip\": \"{{ \$labels.container_ip }}\"
+    {
+      \"refId\": \"B\",
+      \"relativeTimeRange\": {\"from\": 0, \"to\": 0},
+      \"datasourceUid\": \"-100\",
+      \"model\": {
+        \"conditions\": [
+          {
+            \"evaluator\": {\"params\": [$THREAD_THRESHOLD], \"type\": \"gt\"},
+            \"operator\": {\"type\": \"and\"},
+            \"query\": {\"params\": [\"A\"]},
+            \"reducer\": {\"params\": [], \"type\": \"last\"},
+            \"type\": \"query\"
+          }
+        ],
+        \"refId\": \"B\",
+        \"type\": \"classic_conditions\"
+      }
     }
-  }" \
-  "$GRAFANA_URL/api/v1/provisioning/alert-rules")
+  ],
+  \"intervalSeconds\": 60,
+  \"noDataState\": \"NoData\",
+  \"execErrState\": \"Alerting\",
+  \"for\": \"1m\",
+  \"annotations\": {
+    \"summary\": \"High JVM Threads\",
+    \"description\": \"High number of JVM threads detected. Triggering Lambda thread dump.\",
+    \"webhookUrl\": \"$LAMBDA_URL\"
+  },
+  \"labels\": {
+    \"severity\": \"critical\",
+    \"alertname\": \"High JVM Threads\",
+    \"cluster\": \"{{ \$labels.cluster }}\",
+    \"cluster_type\": \"{{ \$labels.cluster_type }}\",
+    \"container_name\": \"{{ \$labels.container_name }}\",
+    \"namespace\": \"{{ \$labels.namespace }}\",
+    \"task_pod_id\": \"{{ \$labels.task_pod_id }}\",
+    \"container_ip\": \"{{ \$labels.container_ip }}\"
+  }
+}"
 
-log "🚨 Creating notification policy..."
-POLICY_RESPONSE=$(curl -s -X PUT -H "Content-Type: application/json" \
-  -u "$GRAFANA_USER:$GRAFANA_PASSWORD" \
-  -d "{
-    \"receiver\": \"$CONTACT_POINT_NAME\",
-    \"group_by\": [\"alertname\"],
-    \"routes\": [
-      {
-        \"receiver\": \"$CONTACT_POINT_NAME\",
-        \"group_by\": [\"alertname\", \"pod\"],
-        \"matchers\": [\"severity = critical\"],
-        \"group_wait\": \"30s\",
-        \"group_interval\": \"5m\",
-        \"repeat_interval\": \"4h\"
-      }
-    ],
-    \"group_wait\": \"30s\",
-    \"group_interval\": \"5m\",
-    \"repeat_interval\": \"1h\"
-  }" \
-  "$GRAFANA_URL/api/v1/provisioning/policies")
+# Add folderUID if we have one
+if [[ -n "$FOLDER_UID" ]]; then
+  ALERT_PAYLOAD=$(echo "$ALERT_PAYLOAD" | jq ". + {\"folderUID\": \"$FOLDER_UID\"}")
+fi
 
-log "✅ Alert rule created"
-log "✅ JVM monitoring setup complete"
-log "🌍 Grafana: $GRAFANA_URL"
-log "📊 Dashboard shows jvm_threads_live_threads from both EKS and ECS"
-log "🚨 Alert triggers Lambda thread dump when threads > $THREAD_THRESHOLD, stops when threads < $THREAD_THRESHOLD"
+ALERT_RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" \
+  -u "$GRAFANA_USER:$GRAFANA_PASSWORD" \
+  -d "$ALERT_PAYLOAD" \

infrastructure/scripts/setup/thread-dump-lambda/src/index.py

@@ -79,45 +79,57 @@ def verify_basic_auth(event: Dict[str, Any]) -> bool:
         logger.error(f"Authentication error: {str(e)}")
         return False
 
-def extract_pod_info_from_valuestring(value_string: str) -> Dict[str, str]:
-    """Extract pod information from Grafana alert valueString"""
+def extract_all_metrics_from_valuestring(value_string: str) -> list:
+    """Extract all metric information from Grafana alert valueString"""
     try:
-        logger.info(f"Extracting pod info from valueString: {value_string}")
-
-        # First, extract the labels section
-        labels_match = re.search(r'labels=\{([^}]+)\}', value_string)
-        if not labels_match:
-            logger.warning("Could not find labels section in valueString")
-            return {}
-
-        labels_str = labels_match.group(1)
-        logger.info(f"Extracted labels section: {labels_str}")
-
-        # Now extract individual labels from the labels section
-        labels = {}
-        for label_pair in labels_str.split(', '):
-            if '=' in label_pair:
-                key, value = label_pair.split('=', 1)
-                labels[key.strip()] = value.strip()
-                logger.info(f"Parsed label: {key.strip()}={value.strip()}")
-
-        logger.info(f"All parsed labels: {labels}")
-
-        # Map the labels to our expected keys
-        result = {
-            'cluster': labels.get('cluster'),
-            'cluster_type': labels.get('cluster_type'),
-            'container_name': labels.get('container_name'),
-            'task_pod_id': labels.get('task_pod_id') or labels.get('pod'),  # Use pod if task_pod_id not found
-            'namespace': labels.get('namespace'),
-            'container_ip': labels.get('container_ip') or labels.get('exported_instance')  # Add container_ip extraction
-        }
-
-        logger.info(f"Final extracted values: {result}")
-        return result
+        logger.info(f"Extracting all metrics from valueString: {value_string}")
+
+        # Find all metric entries in the valueString
+        # Pattern: [ var='...' metric='...' labels={...} value=... ]
+        import re
+        metric_pattern = r'\[\s*var=\'[^\']+\'\s*metric=\'[^\']+\'\s*labels=\{([^}]+)\}\s*value=(\d+)\s*\]'
+        matches = re.findall(metric_pattern, value_string)
+
+        if not matches:
+            logger.warning("Could not find any metric entries in valueString")
+            return []
+
+        results = []
+        for i, (labels_str, value) in enumerate(matches):
+            logger.info(f"Processing metric {i+1}: labels={{{labels_str}}} value={value}")
+
+            # Parse individual labels from the labels section
+            labels = {}
+            for label_pair in labels_str.split(', '):
+                if '=' in label_pair:
+                    key, label_value = label_pair.split('=', 1)
+                    labels[key.strip()] = label_value.strip()
+
+            logger.info(f"All parsed labels for metric {i+1}: {labels}")
+
+            # Map the labels to our expected keys
+            task_pod_id = labels.get('task_pod_id')
+            # Use pod if task_pod_id is missing, empty, or "unknown"
+            if not task_pod_id or task_pod_id == 'unknown':
+                task_pod_id = labels.get('pod')
+
+            result = {
+                'cluster': labels.get('cluster'),
+                'cluster_type': labels.get('cluster_type'),
+                'container_name': labels.get('container_name'),
+                'task_pod_id': task_pod_id,
+                'namespace': labels.get('namespace'),
+                'container_ip': labels.get('container_ip') or labels.get('exported_instance'),
+                'thread_count': int(value)
+            }
+
+            logger.info(f"Final extracted values for metric {i+1}: {result}")
+            results.append(result)
+
+        return results
     except Exception as e:
-        logger.error(f"Error extracting pod info from valueString: {str(e)}")
-        return {}
+        logger.error(f"Error extracting metrics from valueString: {str(e)}")
+        return []
 
 def analyze_thread_dump(thread_dump: str) -> str:
     region_name = os.environ.get('AWS_REGION', 'us-east-1')
@@ -298,35 +310,41 @@ def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
                     if any(is_invalid(val) for val in [cluster_type, cluster_name, task_pod_id, container_name]):
                         logger.info("Some labels are missing or invalid, trying to extract from valueString")
 
-                        # Try to extract from valueString
+                        # Try to extract all metrics from valueString
                         if 'valueString' in alert:
-                            extracted_info = extract_pod_info_from_valuestring(alert['valueString'])
-                            logger.info(f"Raw extracted info: {extracted_info}")
-
-                            # Update missing values
-                            if is_invalid(cluster_type) and extracted_info.get('cluster_type'):
-                                cluster_type = extracted_info['cluster_type']
-
-                            if is_invalid(cluster_name) and extracted_info.get('cluster'):
-                                cluster_name = extracted_info['cluster']
-
-                            if is_invalid(task_pod_id) and extracted_info.get('task_pod_id'):
-                                task_pod_id = extracted_info['task_pod_id']
-
-                            if is_invalid(container_name) and extracted_info.get('container_name'):
-                                container_name = extracted_info['container_name']
-
-                            if is_invalid(container_ip) and extracted_info.get('container_ip'):
-                                container_ip = extracted_info['container_ip']
-                                logger.info(f"Using container_ip from valueString: {container_ip}")
-
-                            # IMPORTANT: Always override namespace with extracted value if available
-                            if extracted_info.get('namespace'):
-                                namespace = extracted_info['namespace']
-                                logger.info(f"Overriding namespace with extracted value: {namespace}")
-
-                            logger.info(f"Final extracted values: cluster_type={cluster_type}, cluster={cluster_name}, task_pod_id={task_pod_id}, container_name={container_name}, namespace={namespace}")
-
+                            extracted_metrics = extract_all_metrics_from_valuestring(alert['valueString'])
+                            logger.info(f"Extracted {len(extracted_metrics)} metrics from valueString")
+
+                            # Process each metric separately
+                            for i, metric_info in enumerate(extracted_metrics):
+                                logger.info(f"Processing metric {i+1}: {metric_info}")
+
+                                # Use extracted values for this metric
+                                metric_cluster_type = metric_info.get('cluster_type')
+                                metric_cluster_name = metric_info.get('cluster')
+                                metric_task_pod_id = metric_info.get('task_pod_id')
+                                metric_container_name = metric_info.get('container_name')
+                                metric_namespace = metric_info.get('namespace', 'default')
+                                metric_container_ip = metric_info.get('container_ip')
+
+                                # Validate this metric's inputs
+                                if any(is_invalid(val) for val in [metric_cluster_type, metric_cluster_name, metric_task_pod_id, metric_container_name]):
+                                    logger.warning(f"Skipping metric {i+1} due to missing values: cluster_type={metric_cluster_type}, cluster={metric_cluster_name}, task_pod_id={metric_task_pod_id}, container_name={metric_container_name}")
+                                    continue
+
+                                # Process this metric
+                                try:
+                                    result = process_alert(metric_cluster_type, metric_cluster_name, metric_task_pod_id, metric_container_name, metric_namespace, s3_bucket, metric_container_ip)
+                                    results.append(result)
+                                    logger.info(f"Successfully processed metric {i+1}")
+                                except Exception as e:
+                                    logger.error(f"Error processing metric {i+1}: {str(e)}")
+                                    continue
+
+                            # Skip the original single-metric processing since we handled all metrics
+                            continue
+
+                    # Original single-metric processing (fallback)
                     # Validate inputs after extraction attempt
                     if any(is_invalid(val) for val in [cluster_type, cluster_name, task_pod_id, container_name]):
                         logger.warning(f"Still missing or invalid alert labels after extraction: cluster_type={cluster_type}, cluster={cluster_name}, task_pod_id={task_pod_id}, container_name={container_name}")