def _get_recommended_configs()

in cost-optimization/hpa-config-recommender/src/hpaconfigrecommender/plan_workload_simulation.py [0:0]

• 54 lines of code
• 6 McCabe index (conditional complexity)

def _get_recommended_configs(
    config: Config,
    plan: WorkloadPlan,
    workload_df: pd.DataFrame
    ) -> Tuple[Optional[WorkloadPlan], Optional[str]]:
    """
    Calculates the recommended HPA configurations based on the workload
    analysis and usage slopes.

    Args:
        workload_df (pd.DataFrame): The DataFrame containing the workload
            data.
        plan (WorkloadPlan): The HPA plan with initial
            recommendations.
        config (Config): Run configurations.

    Returns:
        Tuple[WorkloadPlan,str]: The updated HPA plan
        with recommended target CPU and limits, and a reason if the plan
        is skipped or invalid.
    """
    reason = {}
    # Filter only the points above what is recommended as baseline requests
    plan_request_baseline = plan.recommended_cpu_request
    filtered_df = workload_df[
        workload_df["avg_container_cpu_usage"] >= plan_request_baseline
    ]
    if filtered_df.empty:
        reason = (
            f"Skip HPA Plan {plan.method}. "
            f"No usage above CPU baseline requests:{plan_request_baseline:.2f}."
        )
        logger.info(reason)
        return None, reason

    # Check if slopes are too big
    max_usage_slope_up_ratio = round(
        filtered_df["max_usage_slope_up_ratio"].max(), 2
    )
    if max_usage_slope_up_ratio > config.HPA_SCALE_LIMIT:
        reason = (
            f"Skip HPA Plan {plan.method}. Slope ratio "
            f"{max_usage_slope_up_ratio} exceeds HPA scale limit "
            f"{config.HPA_SCALE_LIMIT}."
        )
        logger.info(reason)
        return None, reason

    plan.max_usage_slope_up_ratio = max_usage_slope_up_ratio
    plan.recommended_hpa_target_cpu = round(
        (
            (1 - config.HPA_TARGET_BUFFER) /
            filtered_df["max_usage_slope_up_ratio"]
        ).min(), 2
    )
    min_hpa_target_cpu = config.MIN_HPA_TARGET_CPU
    max_hpa_target_cpu = config.MAX_HPA_TARGET_CPU
    if plan.recommended_hpa_target_cpu < min_hpa_target_cpu or \
            plan.recommended_hpa_target_cpu > max_hpa_target_cpu:
        reason = (
            f"Skip HPA Plan {plan.method}. Recommended Target CPU "
            f"{plan.recommended_hpa_target_cpu} not between "
            f"{min_hpa_target_cpu} and {max_hpa_target_cpu}."
        )
        logger.info(reason)
        return None, reason

    plan.recommended_cpu_limit_or_unbounded = np.ceil(
        plan.recommended_cpu_request + (
            filtered_df[
                "max_cpu_usage_in_workload_e2e_startup_latency"
            ].max() / plan.recommended_max_replicas
        )
    )
    return plan, None