#!/usr/bin/env python3 import os from datetime import datetime import boto3 import requests from botocore.exceptions import NoCredentialsError, ClientError import logging # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) class MetricsExporter: def __init__(self, base_url, s3_bucket, s3_prefix='metrics_exports', local_dir=None): """ Initialize the MetricsExporter with S3 and export configurations. Args: base_url (str): Base URL for the metrics export API s3_bucket (str): S3 bucket name for storing exported metrics s3_prefix (str): S3 prefix/folder path (default: 'metrics_exports') local_dir (str, optional): Local directory to temporarily store files. If None, uses system temp dir. """ self.base_url = base_url self.s3_bucket = s3_bucket self.s3_prefix = s3_prefix.rstrip('/') self.local_dir = local_dir or os.path.join(os.getcwd(), 'temp_metrics_exports') # Create local directory if it doesn't exist os.makedirs(self.local_dir, exist_ok=True) # Initialize S3 client self.s3_client = boto3.client('s3') def export_metrics(self, start_time, end_time): """ Export metrics for a given time range and upload to S3. Args: start_time (datetime): Start time for metrics export end_time (datetime): End time for metrics export """ current_ts = int(start_time.timestamp()) end_ts = int(end_time.timestamp()) while current_ts < end_ts: next_ts = min(current_ts + 3600, end_ts) # 1 hour window # Format timestamps for filename start_str = datetime.fromtimestamp(current_ts).strftime('%Y%m%d_%H%M') end_str = datetime.fromtimestamp(next_ts).strftime('%Y%m%d_%H%M') # Create local file path filename = f"metrics_{start_str}_{end_str}.prom" local_path = os.path.join(self.local_dir, filename) logger.info(f"Exporting metrics from {datetime.fromtimestamp(current_ts)} to {datetime.fromtimestamp(next_ts)}") try: # Export metrics to local file self._export_to_file(current_ts, next_ts, local_path) # Upload to S3 s3_key = f"{self.s3_prefix}/{filename}" self._upload_to_s3(local_path, s3_key) # Clean up local file after successful upload os.remove(local_path) logger.info(f"Successfully processed and uploaded {filename}") except Exception as e: logger.error(f"Error processing {filename}: {str(e)}") # Keep the file locally if there was an error if os.path.exists(local_path): logger.info(f"Local file kept at: {local_path}") # Move to next hour current_ts = next_ts def _export_to_file(self, start_ts, end_ts, output_path): """Export metrics from API to a local file.""" params = { 'start': f"{start_ts}000", 'end': f"{end_ts}000", 'match[]': '{__name__!=""}' } try: # Log the full URL being called full_url = f"{self.base_url}?" + "&".join([f"{k}={v}" for k, v in params.items()]) logger.info(f"Exporting metrics from URL: {full_url}") response = requests.get(self.base_url, params=params, stream=True) response.raise_for_status() # Write response to file with open(output_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: # filter out keep-alive new chunks f.write(chunk) # Check if file is empty if os.path.getsize(output_path) == 0: os.remove(output_path) raise ValueError("No data returned for the specified time range") except requests.RequestException as e: if os.path.exists(output_path): os.remove(output_path) raise Exception(f"Failed to export metrics: {str(e)}") def _upload_to_s3(self, local_path, s3_key): """Upload a file to S3.""" try: self.s3_client.upload_file( Filename=local_path, Bucket=self.s3_bucket, Key=s3_key ) logger.debug(f"Successfully uploaded {local_path} to s3://{self.s3_bucket}/{s3_key}") except (NoCredentialsError, ClientError) as exc: raise Exception(f"Failed to upload to S3: {str(exc)}") from exc def parse_datetime(dt_str): """Parse datetime string in YYYY-MM-DD HH:MM:SS format.""" try: return datetime.strptime(dt_str, '%Y-%m-%d %H:%M:%S') except ValueError: raise ValueError("Invalid datetime format. Please use 'YYYY-MM-DD HH:MM:SS'") if __name__ == "__main__": import argparse # Set up argument parser parser = argparse.ArgumentParser(description='Export metrics and upload to S3') parser.add_argument('--start', required=True, help='Start time (YYYY-MM-DD HH:MM:SS)') parser.add_argument('--end', required=True, help='End time (YYYY-MM-DD HH:MM:SS)') parser.add_argument( '--base-url', default='http://vmselect-vmcluster.observability.svc.cluster.local:8481/select/39523:22/prometheus/api/v1/export/native', help='Base URL for metrics export API', ) parser.add_argument('--s3-bucket', required=True, help='S3 bucket name') parser.add_argument( '--s3-prefix', default='metrics_exports', help='S3 prefix/folder path (default: metrics_exports)', ) parser.add_argument('--local-dir', help='Local directory for temporary storage (default: ./temp_metrics_exports)') args = parser.parse_args() try: # Parse datetime strings start_time = parse_datetime(args.start) end_time = parse_datetime(args.end) if start_time >= end_time: raise ValueError("Start time must be before end time") # Initialize and run exporter exporter = MetricsExporter( base_url=args.base_url, s3_bucket=args.s3_bucket, s3_prefix=args.s3_prefix, local_dir=args.local_dir ) logger.info(f"Starting metrics export from {start_time} to {end_time}") exporter.export_metrics(start_time, end_time) logger.info("Metrics export and upload completed successfully") except Exception as e: logger.error(f"Error: {str(e)}", exc_info=True) exit(1)