Retrieves RDS major engine versions and their support lifecycle dates.

rds-support-timeline.py

#!/usr/bin/env python3
"""
AWS RDS Database Engine Support Timeline Script
Retrieves RDS major engine versions and their support lifecycle dates.
"""

import os
import json
import boto3
import urllib.error
import urllib.request
from datetime import datetime
from typing import List, Dict


def get_rds_engine_support_info() -> List[Dict]:
    """Fetch RDS major engine versions and support lifecycle information."""
    rds = boto3.client('rds', region_name='us-east-1')

    try:
        # Discover all engines using built-in paginator
        print("- Discovering available RDS engines...")
        paginator = rds.get_paginator('describe_db_engine_versions')
        all_versions = []
        for page in paginator.paginate(IncludeAll=True, PaginationConfig={'PageSize': 100}):
            all_versions.extend(page.get('DBEngineVersions', []))

        engines = sorted(set(v['Engine'] for v in all_versions if v.get('Engine')))
        print(f"- Found {len(engines)} engines: {engines}")

        # Get lifecycle support for each engine
        results = []
        for engine in engines:
            try:
                paginator = rds.get_paginator('describe_db_major_engine_versions')
                for page in paginator.paginate(Engine=engine):
                    for v in page.get('DBMajorEngineVersions', []):
                        if not (major_ver := v.get('MajorEngineVersion')):
                            continue

                        standard_end = extended_end = None
                        for lifecycle in v.get('SupportedEngineLifecycles', []):
                            name = lifecycle.get('LifecycleSupportName', '').lower()
                            date = lifecycle.get('LifecycleSupportEndDate')
                            if 'standard' in name:
                                standard_end = date
                            elif 'extended' in name:
                                extended_end = date

                        results.append({'Engine': engine, 'Version': major_ver,
                                      'StandardSupportEnd': standard_end, 'ExtendedSupportEnd': extended_end})
            except rds.exceptions.ClientError:
                pass  # Engine doesn't support this API

        return results
    except Exception as e:
        print(f"- Error: {e}")
        return []

def format_date(date_obj) -> str:
    """Format datetime to YYYY-MM-DD or 'N/A'."""
    if not date_obj:
        return "N/A"
    return date_obj.strftime('%Y-%m-%d') if isinstance(date_obj, datetime) else str(date_obj)

def print_table(data: List[Dict]) -> None:
    """Print data as a formatted table."""
    if not data:
        print("- No data available.")
        return

    # Filter out entries with no support dates
    filtered_data = [x for x in data if x.get('StandardSupportEnd') or x.get('ExtendedSupportEnd')]
    
    if not filtered_data:
        print("- No data available.")
        return

    sorted_data = sorted(filtered_data, key=lambda x: (x['Engine'], x['Version']))
    max_engine = max(len(x['Engine']) for x in sorted_data)
    max_version = max(len(x['Version']) for x in sorted_data)

    print("\n" + "="*100)
    print(f"{'Engine':<{max_engine}} | {'Version':<{max_version}} | {'Standard Support End':<20} | {'Extended Support End':<20}")
    print("="*100)

    for item in sorted_data:
        print(f"{item['Engine']:<{max_engine}} | {item['Version']:<{max_version}} | "
              f"{format_date(item.get('StandardSupportEnd')):<20} | {format_date(item.get('ExtendedSupportEnd')):<20}")

    print("="*100 + "\n")

def create_slack_message(data: List[Dict]) -> Dict:
    """Create Slack message blocks from data."""
    now = datetime.now()
    
    blocks = [
        {"type": "header", "text": {"type": "plain_text", 
         "text": "🗓️ AWS RDS Support Timeline", "emoji": True}},
        {"type": "section", "text": {"type": "mrkdwn", 
         "text": f"*Report Date:* {now.strftime('%B %d, %Y')}"}}
    ]

    # Collect all expiring support for next 3 months (current + 2)
    all_expiring_std = []
    all_expiring_ext = []
    
    for months_ahead in range(3):
        expiring = get_expiring_support(data, months_ahead)
        all_expiring_std.extend(expiring['standard'])
        all_expiring_ext.extend(expiring['extended'])
    
    # Show expiring support section
    if all_expiring_std or all_expiring_ext:
        blocks.append({"type": "divider"})
        blocks.append({
            "type": "section",
            "text": {"type": "mrkdwn", "text": "*⚠️ Support Ending in the Upcoming Months*"}
        })
        
        if all_expiring_std:
            std_items = sorted(all_expiring_std, key=lambda x: (x.get('StandardSupportEnd'), x['Engine'], x['Version']))
            std_text = "*Standard Support:*\n```\n"
            std_text += f"{'Engine':<20} {'Version':<15} {'End Date':<12}\n"
            std_text += "-" * 47 + "\n"
            std_text += "\n".join(
                f"{item['Engine']:<20} {item['Version']:<15} {format_date(item['StandardSupportEnd'])}"
                for item in std_items
            ) + "\n```"
            blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": std_text}})
        
        if all_expiring_ext:
            ext_items = sorted(all_expiring_ext, key=lambda x: (x.get('ExtendedSupportEnd'), x['Engine'], x['Version']))
            ext_text = "*Extended Support:*\n```\n"
            ext_text += f"{'Engine':<20} {'Version':<15} {'End Date':<12}\n"
            ext_text += "-" * 47 + "\n"
            ext_text += "\n".join(
                f"{item['Engine']:<20} {item['Version']:<15} {format_date(item['ExtendedSupportEnd'])}"
                for item in ext_items
            ) + "\n```"
            blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": ext_text}})

    # Add full table (single table, not grouped)
    blocks.append({"type": "divider"})
    blocks.append({
        "type": "section",
        "text": {"type": "mrkdwn", "text": "*📊 Complete Support Timeline*"}
    })
    
    # Filter and sort data
    filtered_data = [x for x in data if x.get('StandardSupportEnd') or x.get('ExtendedSupportEnd')]
    sorted_data = sorted(filtered_data, key=lambda x: (x['Engine'], x['Version']))
    
    # Create single unified table
    table_text = "```\n"
    table_text += f"{'Engine':<20} {'Version':<15} {'Standard End':<15} {'Extended End':<15}\n"
    table_text += "-" * 65 + "\n"
    
    for item in sorted_data:
        std_date = format_date(item.get('StandardSupportEnd'))
        ext_date = format_date(item.get('ExtendedSupportEnd'))
        table_text += f"{item['Engine']:<20} {item['Version']:<15} {std_date:<15} {ext_date:<15}\n"
    
    table_text += "```"
    blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": table_text}})

    return {"blocks": blocks}

def send_to_slack(webhook_url: str, data: List[Dict]) -> bool:
    """Send formatted message to Slack webhook using urllib (no external dependencies)."""
    try:
        message = create_slack_message(data)
        json_data = json.dumps(message).encode('utf-8')
        
        req = urllib.request.Request(
            webhook_url,
            data=json_data,
            headers={'Content-Type': 'application/json'},
            method='POST'
        )
        
        with urllib.request.urlopen(req) as response:
            if response.status == 200:
                print("✓ Message sent to Slack successfully!")
                return True
            else:
                print(f"✗ Failed to send to Slack. Status code: {response.status}")
                return False
                
    except urllib.error.HTTPError as e:
        print(f"✗ HTTP Error sending to Slack: {e.code} - {e.reason}")
        return False
    except urllib.error.URLError as e:
        print(f"✗ URL Error sending to Slack: {e.reason}")
        return False
    except Exception as e:
        print(f"✗ Error sending to Slack: {e}")
        return False

def get_expiring_support(data: List[Dict], months_ahead: int = 0) -> Dict[str, List[Dict]]:
    """Find engines with support ending in a specific month (0=current, 1=next, 2=two months ahead)."""
    now = datetime.now()
    target_month = now.month + months_ahead
    target_year = now.year
    
    # Handle year rollover
    while target_month > 12:
        target_month -= 12
        target_year += 1

    expiring = {'standard': [], 'extended': []}

    for item in data:
        # Check standard support
        if std_date := item.get('StandardSupportEnd'):
            if isinstance(std_date, datetime) and std_date.month == target_month and std_date.year == target_year:
                expiring['standard'].append(item)

        # Check extended support
        if ext_date := item.get('ExtendedSupportEnd'):
            if isinstance(ext_date, datetime) and ext_date.month == target_month and ext_date.year == target_year:
                expiring['extended'].append(item)

    return expiring


def get_expiring_this_month(data: List[Dict]) -> Dict[str, List[Dict]]:
    """Find engines with support ending this month."""
    return get_expiring_support(data, 0)

def print_expiring_this_month(data: List[Dict]) -> None:
    """Print engines with support ending this month."""
    expiring = get_expiring_this_month(data)

    if not expiring['standard'] and not expiring['extended']:
        print("- No engines have support ending this month.")
        return

    print("\n" + "!"*100)
    print(f"SUPPORT ENDING IN {datetime.now().strftime('%B %Y').upper()}")
    print("!"*100)

    if expiring['standard']:
        print("\n🔴 STANDARD SUPPORT ENDING:")
        for item in sorted(expiring['standard'], key=lambda x: (x['Engine'], x['Version'])):
            print(f"  • {item['Engine']} {item['Version']} - {format_date(item['StandardSupportEnd'])}")

    if expiring['extended']:
        print("\n🔴 EXTENDED SUPPORT ENDING:")
        for item in sorted(expiring['extended'], key=lambda x: (x['Engine'], x['Version'])):
            print(f"  • {item['Engine']} {item['Version']} - {format_date(item['ExtendedSupportEnd'])}")

    print("!"*100 + "\n")

def lambda_handler(event, context):
    """AWS Lambda handler."""
    main()
    return {
        'statusCode': 200,
        'body': json.dumps('Execution completed.')
    }

def main():
    """Main function."""
    print("- Fetching AWS RDS engine support information...")
    data = get_rds_engine_support_info()

    if not data:
        print("- No data retrieved. Exiting.")
        return

    # Print table and expiring support
    print_table(data)
    print_expiring_this_month(data)

    # Send to Slack if webhook URL is provided
    if webhook_url := os.environ.get('SLACK_WEBHOOK_URL'):
        print("- Sending to Slack...")
        send_to_slack(webhook_url, data)


if __name__ == "__main__":
    main()