#!/usr/bin/env python3
"""
Send test metrics to AWS Managed Prometheus.
Usage:
pip install prometheus-client boto3 requests
export PROMETHEUS_WORKSPACE_ID="ws-xxxxx"
export AWS_REGION="us-east-1"
python send_test_metrics.py
"""
import os
import time
import requests
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
WORKSPACE_ID = os.environ.get("PROMETHEUS_WORKSPACE_ID", "ws-becebb13-aa20-4972-8d6f-398b57c71189")
REGION = os.environ.get("AWS_REGION", "us-east-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 query_amp(query: str) -> dict:
"""Query AMP to verify metrics."""
url = f"https://aps-workspaces.{REGION}.amazonaws.com/workspaces/{WORKSPACE_ID}/api/v1/query"
headers = {"Content-Type": "application/x-www-form-urlencoded"}
body = f"query={query}".encode()
signed_headers = sign_request("POST", url, headers, body)
response = requests.post(url, headers=signed_headers, data=body)
return response.json()
def list_labels() -> dict:
"""List all labels in AMP."""
url = f"https://aps-workspaces.{REGION}.amazonaws.com/workspaces/{WORKSPACE_ID}/api/v1/labels"
headers = {}
signed_headers = sign_request("GET", url, headers, b"")
response = requests.get(url, headers=signed_headers)
return response.json()
def list_metrics() -> dict:
"""List all metric names in AMP."""
url = f"https://aps-workspaces.{REGION}.amazonaws.com/workspaces/{WORKSPACE_ID}/api/v1/label/__name__/values"
headers = {}
signed_headers = sign_request("GET", url, headers, b"")
response = requests.get(url, headers=signed_headers)
return response.json()
def main():
print("=" * 60)
print("AWS Managed Prometheus - Test Query")
print("=" * 60)
print()
print(f"Workspace ID: {WORKSPACE_ID}")
print(f"Region: {REGION}")
print()
# Test 1: List labels
print("1. Listing all labels...")
try:
result = list_labels()
if result.get("status") == "success":
labels = result.get("data", [])
print(f" ✓ Found {len(labels)} labels: {labels[:10]}{'...' if len(labels) > 10 else ''}")
else:
print(f" ✗ Error: {result}")
except Exception as e:
print(f" ✗ Exception: {e}")
print()
# Test 2: List metrics
print("2. Listing all metric names...")
try:
result = list_metrics()
if result.get("status") == "success":
metrics = result.get("data", [])
print(f" ✓ Found {len(metrics)} metrics")
if metrics:
print(f" Sample metrics: {metrics[:5]}")
else:
print(f" ✗ Error: {result}")
except Exception as e:
print(f" ✗ Exception: {e}")
print()
# Test 3: Query 'up' metric
print("3. Querying 'up' metric...")
try:
result = query_amp("up")
if result.get("status") == "success":
data = result.get("data", {})
results = data.get("result", [])
print(f" ✓ Found {len(results)} time series")
for r in results[:3]:
metric = r.get("metric", {})
value = r.get("value", [None, None])[1]
print(f" - {metric.get('job', 'unknown')}: {value}")
else:
print(f" ✗ Error: {result}")
except Exception as e:
print(f" ✗ Exception: {e}")
print()
print("=" * 60)
if not labels and not metrics:
print()
print("⚠️ No metrics found in AMP workspace!")
print()
print("To add test metrics, deploy Prometheus with remote_write:")
print()
print("On your bastion, run:")
print("""
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="{url}" \\
--set server.remoteWrite[0].sigv4.region="{region}" \\
--set server.persistentVolume.enabled=false \\
--set alertmanager.enabled=false
""".format(url=REMOTE_WRITE_URL, region=REGION))
else:
print()
print("✓ Metrics are available! You can now test the MCP tools:")
print()
print("In MCP Inspector, try:")
print(" - list_labels")
print(" - list_metrics")
print(" - query_instant with query='up'")
print(" - get_label_values with label_name='job'")
if __name__ == "__main__":
main()
