1#!/usr/bin/env python3
2"""
3ElastiCache Serverless Cost Estimator
4
5Provides approximate cost estimates for what ElastiCache provisioned
6clusters would cost on ElastiCache Serverless, or for sizing a new
7serverless cache. These are estimates only. Actual serverless costs
8depend on real-time workload characteristics and may differ from the
9values produced by this tool. For accurate billing data, use CloudWatch
10ElastiCacheProcessingUnits and BytesUsedForCache metrics on a running
11serverless cache.
12
13Two estimation modes:
14  Simple:   Provide avg memory + daily command count per cluster.
15            Each command is estimated at MAX(1, avg_payload_kb) ECPUs.

serverless_estimator.py on GitHub

This is a quick approximation for initial cost comparison.

Detailed: Also provide per-command stats (from INFO commandstats).

Classifies commands as simple (1 ECPU) or complex

(estimated as MAX(calls, usec/3) ECPUs based on the

assumption that 1 ECPU corresponds to approximately

3 microseconds of vCPU time, derived from the AWS pricing

blog example). Takes the higher of the CPU and network

components. More accurate than simple mode but still an

approximation.

IMPORTANT: The ECPU formulas used in this tool are heuristic approximations

based on publicly available AWS documentation and pricing examples. AWS does

not publish exact ECPU calculation formulas. The 3-microsecond-per-ECPU

assumption for complex commands is derived from the AWS pricing blog and may

not apply uniformly to all command types. See command_classifier.py for

details on command classification.

Pricing is fetched live from the AWS Bulk Pricing API and reflects current

published rates. See https://aws.amazon.com/elasticache/pricing/

Note: INFO commandstats returns cumulative counts that keep incrementing

from server start. If you include an "uptime_days" column in the

commandstats CSV, the estimator divides calls and usec by that value to

normalize to a daily rate. The --endpoint mode handles this automatically

by taking two snapshots and computing the delta.

Usage:

python3 serverless_estimator.py --input clusters.csv

python3 serverless_estimator.py --input clusters.csv --commandstats stats.csv

python3 serverless_estimator.py --endpoint my-cluster.abc123.use1.cache.amazonaws.com:6379

python3 serverless_estimator.py --endpoint my-cluster.abc123.use1.cache.amazonaws.com --instance-type cache.r7g.large

python3 serverless_estimator.py --input clusters.csv --pricing pricing.csv

python3 serverless_estimator.py --input clusters.csv --output estimate.csv --json

"""

import argparse

import csv

import json

import math

import os

import re

import sys

from typing import Dict, List, Optional

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from command_classifier import estimate_ecpus_from_commandstats

from pricing import PricingLoader

HOURS_PER_MONTH = 730

MIN_STORAGE_GB = {

"valkey": 0.1, # 100 MB

"redis": 1.0, # 1 GB

"memcached": 1.0, # 1 GB

}

# Number of sample points for sinusoidal workload modeling.

# 24 = one sample per hour over a daily cycle.

_SIN_SAMPLES = 24

def _sine_wave_samples(min_val, max_val, n=_SIN_SAMPLES):

"""Generate n hourly samples of a sine wave oscillating between min and max.

Models a daily traffic pattern:

value(t) = min + (max - min) * (1 + sin(2*pi*t/n)) / 2

Returns a list of n float values representing the workload at each hour.

"""

amplitude = (max_val - min_val) / 2.0

midpoint = (max_val + min_val) / 2.0

return [midpoint + amplitude * math.sin(2 * math.pi * i / n) for i in range(n)]

def compute_burst_multiplier(avg_val, min_val=None, max_val=None, peak_to_avg=None):

"""Compute a cost multiplier for bursty (sinusoidal) workloads.

Serverless bills per-hour. A workload that swings between min and peak

follows a sine curve over the day. This function samples the curve at

24 hourly points and returns the ratio of the sine-wave average to the

customer-provided average.

Why this matters: if the customer says "my average is X" but their actual

traffic swings from min to peak, the true average of the sine curve

((min+max)/2) may differ from X. The multiplier corrects for that.

The caller can specify the burst in two ways:

1. min_val + max_val: explicit floor and ceiling of the daily cycle

2. peak_to_avg: ratio of peak to average (e.g., 3.0 means peak is 3x avg)

Returns 1.0 (no adjustment) if no burst parameters are provided or if

the workload is flat.

"""

if peak_to_avg is not None and peak_to_avg > 1.0:

max_val = avg_val * peak_to_avg

min_val = max(0, avg_val * (2 - peak_to_avg))

if min_val is None or max_val is None:

return 1.0

if max_val <= min_val or max_val <= 0:

return 1.0

if avg_val <= 0:

return 1.0

samples = _sine_wave_samples(min_val, max_val)

sine_avg = sum(samples) / len(samples)

return sine_avg / avg_val

def load_clusters(path: str) -> List[Dict]:

"""Load cluster data from CSV.

Required columns:

cluster_name, instance_type, region, engine, node_count,

avg_memory_gb, daily_commands

Optional columns:

primary_nodes -- defaults to max(1, node_count // 2). This is a

rough heuristic; for CMD clusters (1 primary +

N replicas) or CME clusters (1 primary per shard),

provide the actual value via --primary-nodes or

the primary_nodes CSV column for accurate results.

current_monthly_cost -- if known; otherwise computed from list price

avg_payload_kb -- average payload size in KB (default 1.0).

ECPUs scale linearly with payload: 3.2 KB = 3.2 ECPUs per op.

peak_commands -- peak daily commands (for bursty workload modeling)

min_commands -- minimum daily commands (for bursty workload modeling)

peak_to_avg_ratio -- ratio of peak to average commands (alternative to min/max)

peak_memory_gb -- peak memory in GB (for bursty storage modeling)

min_memory_gb -- minimum memory in GB (for bursty storage modeling)

"""

clusters = []

with open(path, newline="") as f:

reader = csv.DictReader(f)

for row in reader:

node_count = int(row.get("node_count", 1))

primary_nodes = int(row.get("primary_nodes", 0))

if primary_nodes == 0:

# Heuristic: assumes half the nodes are primaries. This is only

# accurate for 1-replica setups; actual primary count depends on

# cluster topology (CMD: 1 primary + N replicas, CME: 1 primary

# per shard). Provide the primary_nodes column for accuracy.

primary_nodes = max(1, node_count // 2)

clusters.append({

"cluster_name": row["cluster_name"].strip(),

"instance_type": row.get("instance_type", "").strip(),

"region": row.get("region", "us-east-1").strip(),

"engine": row.get("engine", "valkey").strip().lower(),

"node_count": node_count,

"primary_nodes": primary_nodes,

"avg_memory_gb": float(row.get("avg_memory_gb", 0)),

"daily_commands": float(row.get("daily_commands", 0)),

"current_monthly_cost": float(row.get("current_monthly_cost", 0) or 0),

"avg_payload_kb": float(row.get("avg_payload_kb", 1.0) or 1.0),

"peak_commands": float(row.get("peak_commands", 0) or 0),

"min_commands": float(row.get("min_commands", 0) or 0),

"peak_to_avg_ratio": float(row.get("peak_to_avg_ratio", 0) or 0),

"peak_memory_gb": float(row.get("peak_memory_gb", 0) or 0),

"min_memory_gb": float(row.get("min_memory_gb", 0) or 0),

})

return clusters

def parse_commandstats_file(path: str) -> Dict[str, dict]:

"""Parse a commandstats CSV.

Expected columns: cluster_name, command, calls, usec

Optional columns:

uptime_days - days since server restart. When present, calls and

usec are divided by this value to normalize cumulative

INFO commandstats output to a daily rate. Without it

the raw cumulative values are used as-is.

Returns a dict keyed by cluster_name. Each value is a dict of

{command: {"calls": int, "usec": int}}. A special key

"_normalized" (bool) indicates whether uptime_days normalization

was applied for that cluster.

"""

result = {}

with open(path, newline="") as f:

for row in csv.DictReader(f):

cluster = row["cluster_name"].strip()

cmd = row["command"].strip().lower()

calls = int(row.get("calls", 0))

usec = int(row.get("usec", 0))

uptime_raw = row.get("uptime_days", "").strip()

uptime_days = float(uptime_raw) if uptime_raw else None

if uptime_days and uptime_days > 0:

calls = int(calls / uptime_days)

usec = int(usec / uptime_days)

normalized = True

else:

normalized = False

if cluster not in result:

result[cluster] = {"_normalized": normalized} # type: ignore[dict-item]

result[cluster][cmd] = {"calls": calls, "usec": usec} # type: ignore[assignment]

# If any row for this cluster has uptime_days, mark as normalized

if normalized:

result[cluster]["_normalized"] = True

return result

def parse_commandstats_info(text: str) -> dict:

"""Parse raw INFO commandstats output from valkey-cli (or redis-cli).

Example line: cmdstat_get:calls=1000,usec=1500,usec_per_call=1.50,rejected_calls=0,failed_calls=0

"""

result = {}

for line in text.strip().splitlines():

line = line.strip()

if not line or line.startswith("#"):

continue

match = re.match(r"cmdstat_(\w+):calls=(\d+),usec=(\d+)", line)

if match:

cmd = match.group(1).lower()

result[cmd] = {

"calls": int(match.group(2)),

"usec": int(match.group(3)),

}

return result

def estimate_cluster(

cluster: Dict,

pricing: PricingLoader,

commandstats: Optional[dict] = None,

) -> Dict:

"""Estimate serverless cost for one cluster."""

region = cluster["region"]

engine = cluster["engine"]

node_count = cluster["node_count"]

instance_type = cluster["instance_type"]

# --- Current provisioned cost ---

if cluster["current_monthly_cost"] > 0:

current_cost = cluster["current_monthly_cost"]

cost_source = "provided"

elif instance_type:

try:

hourly = pricing.get_node_hourly_rate(region, instance_type, engine)

current_cost = hourly * HOURS_PER_MONTH * node_count

cost_source = "list_price"

except ValueError:

current_cost = 0

cost_source = "unknown"

else:

current_cost = 0

cost_source = "unknown"

# --- Serverless storage ---

avg_memory_gb = cluster["avg_memory_gb"]

min_gb = MIN_STORAGE_GB.get(engine, 1.0)

billing_memory_gb = max(avg_memory_gb, min_gb)

used_minimum = avg_memory_gb < min_gb

# Bursty storage: if peak/min memory provided, adjust

storage_burst = compute_burst_multiplier(

avg_memory_gb,

min_val=cluster.get("min_memory_gb") or None,

max_val=cluster.get("peak_memory_gb") or None,

)

adjusted_memory_gb = billing_memory_gb * storage_burst

monthly_gb_hours = adjusted_memory_gb * HOURS_PER_MONTH

storage_rate = pricing.get_serverless_storage_rate(region, engine)

storage_cost = monthly_gb_hours * storage_rate

# --- Serverless ECPUs ---

ecpu_rate = pricing.get_serverless_ecpu_rate(region, engine)

# Bursty commands: if peak/min commands or peak_to_avg_ratio provided, adjust

ecpu_burst = compute_burst_multiplier(

cluster["daily_commands"],

min_val=cluster.get("min_commands") or None,

max_val=cluster.get("peak_commands") or None,

peak_to_avg=cluster.get("peak_to_avg_ratio") or None,

)

if commandstats and engine == "memcached":

print("WARNING: commandstats mode uses Redis/Valkey INFO format, which is "

"incompatible with Memcached. ECPU estimates for Memcached cluster '{}' "

"will be inaccurate. Falling back to simple estimation.".format(

cluster["cluster_name"]),

file=sys.stderr)

commandstats = None

if commandstats:

# _normalized is a metadata flag, not a command; exclude before estimation

cs_normalized = commandstats.get("_normalized", False)

cs_data = {k: v for k, v in commandstats.items() if k != "_normalized"}

ecpu_result = estimate_ecpus_from_commandstats(cs_data)

daily_cpu_ecpus = ecpu_result["total_ecpus"]

# Network component: each command costs MAX(1, avg_payload_kb) ECPUs.

# This approximates the "1 ECPU per KB transferred" pricing rule.

# See: https://aws.amazon.com/elasticache/pricing/

avg_payload_kb = cluster.get("avg_payload_kb", 1.0) or 1.0

ecpu_per_request = max(1.0, avg_payload_kb)

daily_net_ecpus = cluster["daily_commands"] * ecpu_per_request

# Serverless charges the higher of vCPU time or data transferred.

# See: https://aws.amazon.com/blogs/database/unlock-on-demand-cost-optimized-performance-with-amazon-elasticache-serverless/

daily_ecpus = max(daily_cpu_ecpus, daily_net_ecpus)

monthly_ecpus = daily_ecpus * 30 * ecpu_burst

ecpu_mode = "detailed"

fixed_ecpus = ecpu_result["fixed_ecpus"] * 30

nonfixed_ecpus = ecpu_result["nonfixed_ecpus"] * 30

internal_calls = ecpu_result["internal_calls"] * 30

else:

avg_payload_kb = cluster.get("avg_payload_kb", 1.0) or 1.0

ecpu_per_request = max(1.0, avg_payload_kb)

monthly_ecpus = cluster["daily_commands"] * 30 * ecpu_per_request * ecpu_burst

ecpu_mode = "simple"

fixed_ecpus = None

nonfixed_ecpus = None

internal_calls = None

ecpu_cost = monthly_ecpus * ecpu_rate

total_serverless = storage_cost + ecpu_cost

# --- Comparison ---

diff = total_serverless - current_cost if current_cost > 0 else None

diff_pct = (diff / current_cost * 100) if diff is not None and current_cost > 0 else None

# --- Notes ---

notes = []

if used_minimum:

notes.append("Min storage applied ({} GB)".format(min_gb))

if cost_source == "list_price":

notes.append("Current cost: on-demand list price")

if cost_source == "unknown":

notes.append("Current cost unknown - provide current_monthly_cost or instance_type")

if ecpu_mode == "simple":

if avg_payload_kb > 1:

notes.append("ECPU: {} ECPUs per request ({} KB avg payload)".format(

round(ecpu_per_request, 1), round(avg_payload_kb, 1)))

else:

notes.append("ECPU: 1 cmd = 1 ECPU (provide avg_payload_kb for better accuracy)")

if commandstats and not cs_normalized:

notes.append("Commandstats not normalized by uptime; provide uptime_days column for accuracy")

if ecpu_burst > 1.0:

notes.append("Burst adjustment applied: {:.2f}x (peak/avg commands)".format(ecpu_burst))

if storage_burst > 1.0:

notes.append("Burst adjustment applied: {:.2f}x (peak/avg memory)".format(storage_burst))

# Serverless compatibility warnings

if engine in ("redis", "valkey"):

notes.append("Serverless requires cluster-mode-enabled clients and TLS; Global Data Store and Data Tiering are not supported")

return {

"cluster_name": cluster["cluster_name"],

"region": region,

"engine": engine,

"instance_type": instance_type,

"node_count": node_count,

"primary_nodes": cluster["primary_nodes"],

"current_monthly_cost": round(current_cost, 2),

"cost_source": cost_source,

"avg_memory_gb": round(avg_memory_gb, 4),

"billing_memory_gb": round(billing_memory_gb, 4),

"monthly_gb_hours": round(monthly_gb_hours, 2),

"storage_rate_per_gb_hour": storage_rate,

"storage_cost": round(storage_cost, 2),

"monthly_ecpus": round(monthly_ecpus),

"ecpu_rate_per_million": round(ecpu_rate * 1_000_000, 4),

"ecpu_cost": round(ecpu_cost, 2),

"ecpu_mode": ecpu_mode,

"fixed_ecpus": round(fixed_ecpus) if fixed_ecpus is not None else None,

"nonfixed_ecpus": round(nonfixed_ecpus) if nonfixed_ecpus is not None else None,

"internal_calls_excluded": round(internal_calls) if internal_calls is not None else None,

"serverless_total": round(total_serverless, 2),

"diff": round(diff, 2) if diff is not None else None,

"diff_pct": round(diff_pct, 1) if diff_pct is not None else None,

"notes": "; ".join(notes),

}

def write_csv(results: List[Dict], path: str):

"""Write results to CSV."""

if not results:

return

fields = [

"cluster_name", "region", "engine", "instance_type",

"node_count", "primary_nodes",

"current_monthly_cost", "cost_source",

"avg_memory_gb", "billing_memory_gb",

"monthly_gb_hours", "storage_cost",

"monthly_ecpus", "ecpu_cost", "ecpu_mode",

"serverless_total", "diff", "diff_pct", "notes",

]

with open(path, "w", newline="") as f:

writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")

writer.writeheader()

writer.writerows(results)

print("Output written to: {}".format(path))

def print_summary(results: List[Dict]):

"""Print summary to console."""

total_current = sum(r["current_monthly_cost"] for r in results)

total_sl = sum(r["serverless_total"] for r in results)

total_stor = sum(r["storage_cost"] for r in results)

total_ecpu = sum(r["ecpu_cost"] for r in results)

cheaper = sum(1 for r in results if r["diff"] is not None and r["diff"] < 0)

more_exp = sum(1 for r in results if r["diff"] is not None and r["diff"] > 0)

print()

print("=" * 65)

print(" ElastiCache Serverless Cost Estimate")

print("=" * 65)

print(" Clusters analyzed: {}".format(len(results)))

print(" Total nodes: {}".format(sum(r["node_count"] for r in results)))

print()

print(" Current provisioned cost: ${:>12,.2f} /month".format(total_current))

print(" Estimated serverless: ${:>12,.2f} /month".format(total_sl))

print(" Data storage: ${:>12,.2f}".format(total_stor))

print(" ECPUs: ${:>12,.2f}".format(total_ecpu))

print()

if total_current > 0:

savings = total_current - total_sl

pct = savings / total_current * 100

print(" Estimated savings: ${:>12,.2f} ({:.1f}%)".format(savings, pct))

print(" Clusters cheaper on SL: {}".format(cheaper))

print(" Clusters more expensive: {}".format(more_exp))

print("=" * 65)

print()

print(" {:<30s} {:>10s} {:>10s} {:>8s}".format(

"Cluster", "Current", "Serverless", "Change"))

print(" {} {} {} {}".format("-" * 30, "-" * 10, "-" * 10, "-" * 8))

for r in sorted(results, key=lambda x: x.get("diff") or 0):

name = r["cluster_name"][:30]

cur = "${:,.0f}".format(r["current_monthly_cost"])

sl = "${:,.0f}".format(r["serverless_total"])

chg = "{:+.0f}%".format(r["diff_pct"]) if r["diff_pct"] is not None else "N/A"

print(" {:<30s} {:>10s} {:>10s} {:>8s}".format(name, cur, sl, chg))

print()

print(" ECPU estimates assume 1 KB avg payload unless avg_payload_kb is provided.")

print(" ECPUs scale linearly with payload size (e.g. 3 KB = 3x ECPUs).")

print()

print(" IMPORTANT: These are approximate estimates. Actual serverless costs depend")

print(" on real-time workload characteristics and may differ. For accurate billing,")

print(" use CloudWatch metrics on a running serverless cache.")

print(" See https://aws.amazon.com/elasticache/pricing/")

print()

def collect_from_endpoint(endpoint, cluster_name=None, instance_type="",

region="us-east-1", engine="valkey", node_count=2,

avg_payload_kb=1.0, use_tls=True, sample_seconds=60):

"""Connect to a live Valkey/Redis endpoint and collect metrics.

Takes two INFO snapshots separated by sample_seconds and computes

deltas. This gives the actual traffic rate during the sample window,

which is more accurate than dividing cumulative totals by uptime.

Returns (cluster_dict, commandstats_dict) ready for estimate_cluster().

"""

try:

import valkey as client_lib

except ImportError:

print("Error: the 'valkey' package is required for --endpoint mode.")

print("Install with: pip install valkey")

sys.exit(1)

import time

# Parse host:port

if ":" in endpoint and not endpoint.startswith("["):

host, port_str = endpoint.rsplit(":", 1)

port = int(port_str)

else:

host = endpoint

port = 6379

if not cluster_name:

cluster_name = host.split(".")[0]

print("Connecting to {}:{}{}...".format(host, port, " (TLS)" if use_tls else ""))

r = client_lib.Redis(host=host, port=port, ssl=use_tls,

ssl_cert_reqs=None, decode_responses=True,

socket_connect_timeout=10)

print(" PING: {}".format(r.ping()))

# Collect memory and replication (point-in-time, no delta needed)

mem_info = r.info("memory")

repl_info = r.info("replication")

dataset_bytes = mem_info.get("used_memory_dataset", mem_info.get("used_memory", 0))

dataset_gb = dataset_bytes / (1024 ** 3)

role = repl_info.get("role", "unknown")

print(" Role: {}".format(role))

print(" Memory: {:.4f} GB ({:,.0f} bytes)".format(dataset_gb, dataset_bytes))

# Snapshot 1

print(" Taking snapshot 1...")

cs1 = r.info("commandstats")

stats1 = r.info("stats")

t1 = time.time()

# Wait

print(" Sampling for {} seconds...".format(sample_seconds))

time.sleep(sample_seconds)

# Snapshot 2

print(" Taking snapshot 2...")

cs2 = r.info("commandstats")

stats2 = r.info("stats")

t2 = time.time()

elapsed = t2 - t1

elapsed_days = elapsed / 86400

# Compute commandstats deltas

commandstats = {}

total_delta_calls = 0

for key in cs2:

cmd = key.replace("cmdstat_", "")

calls_delta = cs2[key].get("calls", 0) - cs1.get(key, {}).get("calls", 0)

usec_delta = cs2[key].get("usec", 0) - cs1.get(key, {}).get("usec", 0)

if calls_delta > 0:

# Scale to daily rate

daily_calls = int(calls_delta / elapsed_days)

daily_usec = int(usec_delta / elapsed_days)

commandstats[cmd] = {"calls": daily_calls, "usec": daily_usec}

total_delta_calls += calls_delta

commandstats["_normalized"] = True # type: ignore[assignment]

daily_commands = total_delta_calls / elapsed_days if elapsed_days > 0 else 0

# Also get total commands delta for simple mode

total_cmds_delta = stats2.get("total_commands_processed", 0) - stats1.get("total_commands_processed", 0)

daily_commands_total = total_cmds_delta / elapsed_days if elapsed_days > 0 else 0

print(" Sample window: {:.0f} seconds".format(elapsed))

print(" Commands in window: {:,}".format(total_cmds_delta))

print(" Daily rate: {:,.0f} commands/day".format(daily_commands_total))

top_cmds = sorted(commandstats.items(),

key=lambda x: x[1].get("calls", 0) if isinstance(x[1], dict) else 0,

reverse=True)[:5]

print(" Top commands: {}".format(

", ".join("{} ({:,}/day)".format(k, v["calls"])

for k, v in top_cmds if isinstance(v, dict))))

cluster = {

"cluster_name": cluster_name,

"instance_type": instance_type,

"region": region,

"engine": engine.lower(),

"node_count": node_count,

"primary_nodes": max(1, node_count // 2),

"avg_memory_gb": dataset_gb,

"daily_commands": daily_commands_total,

"current_monthly_cost": 0,

"avg_payload_kb": avg_payload_kb,

"peak_commands": 0,

"min_commands": 0,

"peak_to_avg_ratio": 0,

"peak_memory_gb": 0,

"min_memory_gb": 0,

}

return cluster, commandstats

def main():

parser = argparse.ArgumentParser(

description="Estimate ElastiCache Serverless costs from provisioned cluster metrics"

)

parser.add_argument("--input", "-i",

help="CSV with cluster data (see README for format)")

parser.add_argument("--endpoint", "-e",

help="Connect directly to a Valkey/Redis endpoint (host:port or host)")

parser.add_argument("--cluster-name", default=None,

help="Cluster name (used with --endpoint, default: derived from host)")

parser.add_argument("--instance-type", default="",

help="Instance type (used with --endpoint, e.g., cache.r7g.large)")

parser.add_argument("--region", default="us-east-1",

help="AWS region (used with --endpoint, default: us-east-1)")

parser.add_argument("--engine", default="valkey",

help="Engine (used with --endpoint, default: valkey)")

parser.add_argument("--node-count", type=int, default=2,

help="Node count (used with --endpoint, default: 2)")

parser.add_argument("--avg-payload-kb", type=float, default=1.0,

help="Average payload size in KB (default: 1.0)")

parser.add_argument("--no-tls", action="store_true",

help="Disable TLS when connecting via --endpoint")

parser.add_argument("--sample-seconds", type=int, default=60,

help="Seconds to sample when using --endpoint (default: 60). "

"Takes two snapshots this far apart and computes deltas.")

parser.add_argument("--commandstats", "-c",

help="CSV with per-command stats for detailed ECPU estimation")

parser.add_argument("--pricing", "-p",

help="Pricing CSV (optional - fetches live from AWS if omitted)")

parser.add_argument("--output", "-o", default="serverless_estimate.csv",

help="Output CSV path (default: serverless_estimate.csv)")

parser.add_argument("--json", action="store_true",

help="Also write JSON output")

args = parser.parse_args()

if not args.input and not args.endpoint:

parser.error("Either --input (CSV) or --endpoint (host:port) is required")

pricing = PricingLoader(args.pricing)

if args.endpoint:

# Direct endpoint mode: connect, collect, normalize, estimate

cluster, commandstats = collect_from_endpoint(

args.endpoint,

cluster_name=args.cluster_name,

instance_type=args.instance_type,

region=args.region,

engine=args.engine,

node_count=args.node_count,

avg_payload_kb=args.avg_payload_kb,

use_tls=not args.no_tls,

sample_seconds=args.sample_seconds,

)

results = [estimate_cluster(cluster, pricing, commandstats)]

else:

print("Loading clusters from: {}".format(args.input))

clusters = load_clusters(args.input)

print(" {} clusters loaded".format(len(clusters)))

all_commandstats = None

if args.commandstats:

print("Loading commandstats from: {}".format(args.commandstats))

all_commandstats = parse_commandstats_file(args.commandstats)

print(" Stats for {} clusters".format(len(all_commandstats)))

results = []

for cluster in clusters:

cs = all_commandstats.get(cluster["cluster_name"]) if all_commandstats else None

results.append(estimate_cluster(cluster, pricing, cs))

write_csv(results, args.output)

print_summary(results)

if args.json:

base, _ = os.path.splitext(args.output)

json_path = base + ".json"

with open(json_path, "w") as f:

json.dump(results, f, indent=2, default=str)

print("JSON written to: {}".format(json_path))

if __name__ == "__main__":

main()