[Collect RightSizing info from Cost Explorer and Upload to S3 - Lambda Fix]

[samratmk]

Customer Domain: GFS-SI
Prioriry: High

Situation:

• Customer wanted to leverage the existing CID dashboard for Rightsizing information and share it back across to the different business units (BU) to optimize their AWS infrastructure spend whilst identifying opportunities to save costs.

• The existing dashboard, post deployment had issues and the Lambda underneath feeding the data was breaking.

Task:

• Unblock the customer and ensure that the dashboard functions as expected.

• Ensure the patch fix gets back-ported into our knowledge base.

• The problem lies in the AWS Cost Explorer API's limitations. When you call get_rightsizing_recommendation() with a PageSize of 5000, you're exceeding the maximum allowed page size for this particular API.

• AWS Cost Explorer's GetRightsizingRecommendation API has a maximum page size limit of 1000 items per request, not 5000. When you attempted to request 5000 items at once, the API rejected your request, causing the error.

Action:

• This updated code fixes the issue with the Lambda, retrospectively the CID dashboard.

Result:

• TAM shared an updated and patched codebase for the Lambda that the customer used and has now fixed their issue in Production.

[samratmk]

• Added enhanced logging
• Updated pagesize to 100.

""" Collect RightSizing info from Cost Explorer and Upload to S3
"""**
import os
import json
import logging
import traceback
from botocore.config import Config
from datetime import date
import boto3
 
# Set up more detailed logging
logger = logging.getLogger(__name__)
logger.setLevel(getattr(logging, os.environ.get('LOG_LEVEL', 'INFO').upper(), logging.INFO))
# Add a stream handler if not running in Lambda environment for local testing
if not os.environ.get('AWS_EXECUTION_ENV'):
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
 
BUCKET = os.environ.get('BUCKET_NAME')
ROLE = os.environ.get('ROLENAME')
PREFIX = os.environ.get('PREFIX')
BENEFITS_CONSIDERED = True
TARGETS = {
    "RightsizingRecommendationsSameFamily": 'SAME_INSTANCE_FAMILY',
    "RightsizingRecommendationsCrossFamily": 'CROSS_INSTANCE_FAMILY',
}
 
def store_data_to_s3(data, payer_id):
    if not data:
        logger.info("No data to store in S3")
        return
   
    try:
        key = date.today().strftime(f"{PREFIX}/{PREFIX}-data/payer_id={payer_id}/year=%Y/month=%m/day=%d/%Y-%m-%d.json")
        logger.info(f"Attempting to upload data to s3://{BUCKET}/{key}")
       
        s3_client = boto3.client('s3')
        s3_client.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=json.dumps(data, default=str)
        )
        logger.info(f'File upload successful to s3://{BUCKET}/{key}')
    except Exception as e:
        logger.error(f"Error uploading to S3: {str(e)}")
        logger.error(traceback.format_exc())
        raise
 
def process_one_management_acc(management_account_id):
    logger.info(f"Processing account: {management_account_id}")
   
    try:
        logger.info('Assuming role for account')
        region = boto3.session.Session().region_name
        logger.debug(f"Current region: {region}")
       
        partition = boto3.session.Session().get_partition_for_region(region_name=region)
        role_arn = f"arn:{partition}:iam::{management_account_id}:role/{ROLE}"
        logger.info(f"Assuming role: {role_arn}")
       
        sts_client = boto3.client('sts')
        try:
            cred = sts_client.assume_role(
                RoleArn=role_arn,
                RoleSessionName="data_collection"
            )['Credentials']
            logger.info(f"Successfully assumed role for account {management_account_id}")
        except Exception as e:
            logger.error(f"Failed to assume role: {str(e)}")
            logger.error(traceback.format_exc())
            raise
       
        logger.info('Initializing Cost Explorer client in us-east-1')
        cost_explorer = boto3.client(
            "ce",
            "us-east-1",  # Must be "us-east-1"
            config=Config(retries={"max_attempts": 30, "mode": "adaptive"}),
            aws_access_key_id=cred['AccessKeyId'],
            aws_secret_access_key=cred['SecretAccessKey'],
            aws_session_token=cred['SessionToken'],
        )
       
        logger.info('Pulling rightsizing recommendations')
        result = {
            "RecommendationDate": date.today().strftime('%Y-%m-%d'),
            "RightsizingRecommendationsSameFamily": [],
            "RightsizingRecommendationsCrossFamily": [],
        }
       
        for key, target in TARGETS.items():
            logger.info(f"Getting recommendations for target: {target}")
            token = None
            page_count = 0
           
            while True:
                params = {
                    'Service': 'AmazonEC2',
                    'Configuration': {
                        'RecommendationTarget': target,
                        'BenefitsConsidered': BENEFITS_CONSIDERED,
                    },
                    'PageSize': 1000,
                }
               
                if token:
                    params["NextPageToken"] = token
                   
                logger.info(f"Making Cost Explorer API call for {target} (page {page_count + 1})")
                try:
                    response = cost_explorer.get_rightsizing_recommendation(**params)
                    recommendations_count = len(response.get("RightsizingRecommendations", []))
                    logger.info(f"Received {recommendations_count} recommendations for {target}")
                   
                    result[key] += response.get("RightsizingRecommendations", [])
                    token = response.get("NextPageToken")
                    page_count += 1
                   
                    if not token:
                        logger.info(f"No more pages for {target}")
                        break
                    else:
                        logger.info(f"More pages exist for {target}, continuing with token")
                except Exception as e:
                    logger.error(f"Error getting rightsizing recommendations: {str(e)}")
                    logger.error(f"Parameters used: {json.dumps(params, default=str)}")
                    logger.error(traceback.format_exc())
                    raise
       
        logger.info(f"Total recommendations collected - Same Family: {len(result['RightsizingRecommendationsSameFamily'])}, Cross Family: {len(result['RightsizingRecommendationsCrossFamily'])}")
        store_data_to_s3(result, management_account_id)
       
    except Exception as e:
        logger.error(f"Error processing account {management_account_id}: {str(e)}")
        logger.error(traceback.format_exc())
        raise
 
def lambda_handler(event, context):
    try:
        logger.info(f"Lambda invoked with event: {json.dumps(event)}")
       
        # Validate environment variables
        if not BUCKET or not ROLE or not PREFIX:
            missing_vars = []
            if not BUCKET: missing_vars.append("BUCKET_NAME")
            if not ROLE: missing_vars.append("ROLENAME")
            if not PREFIX: missing_vars.append("PREFIX")
            error_msg = f"Missing required environment variables: {', '.join(missing_vars)}"
            logger.error(error_msg)
            raise ValueError(error_msg)
       
        if 'account' not in event:
            error_msg = "Please do not trigger this Lambda manually. Find the corresponding state machine in Step Functions and Trigger from there."
            logger.error(error_msg)
            raise ValueError(error_msg)
       
        account = json.loads(event["account"])
        logger.info(f"Processing account: {account}")
       
        process_one_management_acc(account["account_id"])
        return {"status": "success", "message": f"Successfully processed account {account['account_id']}"}
   
    except Exception as e:
        logger.error(f"Lambda execution failed: {str(e)}")
        logger.error(traceback.format_exc())
        return {
            "status": "error",
            "message": str(e),
            "traceback": traceback.format_exc()
        }

[iakov-aws]

Hello

Please note that we no longer recommend RightSizing from Cost Explorer as Cost Optimization Hub Data has more info and we recommend Compute Optimizer Dashboard or CORA that leverages CostOptimHub Data

For this Rightsizing module you can submit a fix but we will deprecate this one in future versions.