#!/usr/bin/env python3
"""
Push test metrics to AWS Managed Prometheus.
This script uses the Prometheus remote write API to push sample metrics
that you can then query with the MCP tools.
Usage:
export PROMETHEUS_WORKSPACE_ID="ws-xxxxx"
export AWS_REGION="us-east-1"
python push_test_metrics.py
"""
import os
import time
import struct
import snappy
import requests
from datetime import datetime
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
# Prometheus remote write protobuf format
# We'll use a simplified approach with the text format instead
WORKSPACE_ID = os.environ.get("PROMETHEUS_WORKSPACE_ID")
REGION = os.environ.get("AWS_REGION", "us-east-1")
if not WORKSPACE_ID:
print("Error: Set PROMETHEUS_WORKSPACE_ID environment variable")
print("Example: export PROMETHEUS_WORKSPACE_ID=ws-becebb13-aa20-4972-8d6f-398b57c71189")
exit(1)
REMOTE_WRITE_URL = f"https://aps-workspaces.{REGION}.amazonaws.com/workspaces/{WORKSPACE_ID}/api/v1/remote_write"
def sign_request(method: str, url: str, headers: dict, body: bytes) -> dict:
"""Sign request with AWS SigV4."""
session = boto3.Session(region_name=REGION)
credentials = session.get_credentials()
request = AWSRequest(method=method, url=url, headers=headers, data=body)
SigV4Auth(credentials, "aps", REGION).add_auth(request)
return dict(request.headers)
def create_write_request(metrics: list) -> bytes:
"""
Create a Prometheus remote write request.
This is a simplified version - for production, use prometheus-client
or a proper protobuf implementation.
"""
# We'll use prometheus_client's exposition format and convert
# For simplicity, let's use the OpenMetrics text format via a different endpoint
pass
def push_metric_via_query(metric_name: str, value: float, labels: dict):
"""
Push a metric using the remote write endpoint.
Note: AMP remote write requires proper protobuf format.
For testing, we'll verify the setup works.
"""
print(f" {metric_name}{labels} = {value}")
def main():
print("=" * 60)
print("AWS Managed Prometheus - Test Metrics Setup")
print("=" * 60)
print()
print(f"Workspace ID: {WORKSPACE_ID}")
print(f"Region: {REGION}")
print(f"Remote Write URL: {REMOTE_WRITE_URL}")
print()
# Test connectivity to AMP
print("Testing connectivity to AMP...")
labels_url = f"https://aps-workspaces.{REGION}.amazonaws.com/workspaces/{WORKSPACE_ID}/api/v1/labels"
headers = {
"Content-Type": "application/json",
}
signed_headers = sign_request("GET", labels_url, headers, b"")
try:
response = requests.get(labels_url, headers=signed_headers)
if response.status_code == 200:
data = response.json()
print(f"✓ Connected to AMP successfully!")
print(f" Existing labels: {data.get('data', [])}")
else:
print(f"✗ AMP returned status {response.status_code}")
print(f" Response: {response.text}")
except Exception as e:
print(f"✗ Connection failed: {e}")
return
print()
print("=" * 60)
print("To add test metrics to AMP, you have several options:")
print("=" * 60)
print()
print("OPTION 1: Deploy a Prometheus server in your EKS cluster")
print("-" * 60)
print("""
On your bastion host, run:
helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
helm repo update
helm install prometheus prometheus-community/prometheus \\
--namespace prometheus-mcp \\
--set server.remoteWrite[0].url="{remote_write_url}" \\
--set server.remoteWrite[0].sigv4.region="{region}" \\
--set server.remoteWrite[0].queue_config.max_samples_per_send=1000
This will scrape Kubernetes metrics and send them to AMP.
""".format(remote_write_url=REMOTE_WRITE_URL, region=REGION))
print()
print("OPTION 2: Use the AWS Distro for OpenTelemetry (ADOT)")
print("-" * 60)
print("""
Deploy ADOT collector to push metrics:
kubectl apply -f https://amazon-eks.s3.amazonaws.com/docs/addons-otel-permissions.yaml
""")
print()
print("OPTION 3: Push metrics using promtool (simplest for testing)")
print("-" * 60)
print("""
On your local machine or bastion, install promtool and use:
# Create a metrics file
cat > /tmp/metrics.txt << 'EOF'
# HELP test_metric A test metric for MCP testing
# TYPE test_metric gauge
test_metric{{job="mcp-test",instance="test-1"}} 42
test_metric{{job="mcp-test",instance="test-2"}} 100
# HELP http_requests_total Total HTTP requests
# TYPE http_requests_total counter
http_requests_total{{method="GET",status="200"}} 1523
http_requests_total{{method="POST",status="200"}} 456
# HELP up Target up status
# TYPE up gauge
up{{job="prometheus",instance="localhost:9090"}} 1
up{{job="node",instance="localhost:9100"}} 1
EOF
# Push using AWS CLI (requires proper protobuf encoding)
# For now, deploy Prometheus with helm (Option 1) - it's the easiest!
""")
print()
print("=" * 60)
print("RECOMMENDED: Use Option 1 (Helm)")
print("=" * 60)
print()
print("Run this on your bastion to deploy Prometheus and start pushing metrics:")
print()
print(f"""
# Install Helm if not present
curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash
# Add repo and install
helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
helm repo update
# Install Prometheus with remote write to AMP
helm install prometheus prometheus-community/prometheus \\
--namespace prometheus-mcp \\
--set server.remoteWrite[0].url="{REMOTE_WRITE_URL}" \\
--set server.remoteWrite[0].sigv4.region="{REGION}" \\
--set serviceAccounts.server.annotations."eks\\.amazonaws\\.com/role-arn"="arn:aws:iam::897945677099:role/prometheus-mcp-pod-role"
""")
print()
print("After a few minutes, metrics will appear in AMP!")
print("Then test with MCP: list_metrics, query_instant('up'), etc.")
if __name__ == "__main__":
main()
