40% of a 350K Monthly Bill Was Idle: 6 Production Decisions from Tag Attribution to Cost-as-Code in FinOps

# Mandatory tags (5 dimensions)
Owner       = "Zhang San"     # Person responsible; you know who to call when something breaks
Team        = "payment"       # Team, used for cost allocation
Project     = "order-service" # Project / product line
Environment = "production"    # Environment: production/staging/development
CostCenter  = "CC-001"        # Cost center, linked to the finance system

#!/usr/bin/env python3
"""
Tag compliance check script: scans all ECS instances and checks whether
the mandatory tags are present. Used as a resource creation gate in CI/CD
pipelines, or as a scheduled inspection.
"""
import json
import sys
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_tea_openapi.models import Config

REQUIRED_TAGS = ["Owner", "Team", "Project", "Environment", "CostCenter"]

def check_tag_compliance(access_key_id, access_key_secret, region_id="cn-hangzhou"):
    """Scan tag compliance for all ECS instances in the specified region"""
    config = Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com"
    )
    client = EcsClient(config)

    # Get all instances
    instances = client.describe_instances(
        describe_instances_request={}
    ).body.instances.instance

    violations = []
    for inst in instances:
        instance_id = inst.instance_id
        instance_name = inst.instance_name
        existing_tags = {tag.tag_key: tag.tag_value for tag in (inst.tags.tag or [])}

        missing = [t for t in REQUIRED_TAGS if t not in existing_tags]
        if missing:
            violations.append({
                "instance_id": instance_id,
                "instance_name": instance_name,
                "missing_tags": missing,
                "existing_tags": existing_tags
            })

    return violations

if __name__ == "__main__":
    violations = check_tag_compliance(
        access_key_id=sys.argv[1],
        access_key_secret=sys.argv[2]
    )
    if violations:
        print(f"[FAIL] Found {len(violations)} instances with non-compliant tags:")
        for v in violations:
            print(f"  {v['instance_id']} ({v['instance_name']}): missing {v['missing_tags']}")
        sys.exit(1)
    else:
        print("[PASS] All instances have compliant tags")

# Batch fix tag case (Alibaba Cloud CLI example)
# First list all tags and find instances that need fixing
aliyun ecs DescribeTags --RegionId cn-hangzhou | \
  jq '.Tags.Tag[] | select(.TagKey | test("team"))' | \
  jq -r '.InstanceId'

# For each instance, delete the old tag and add the standardized one
for instance_id in $(aliyun ecs DescribeInstances --RegionId cn-hangzhou | \
  jq -r '.Instances.Instance[] | select(.Tags.Tag[]? | .TagKey == "team") | .InstanceId'); do
    # Delete the lowercase tag
    aliyun ecs RemoveTags --ResourceType instance --ResourceId "$instance_id" \
        --Tag.1.Key team --RegionId cn-hangzhou
    # Read the value of the lowercase tag
    tag_value=$(aliyun ecs DescribeInstances --InstanceIds "[\"$instance_id\"]" | \
        jq -r '.Instances.Instance[0].Tags.Tag[] | select(.TagKey == "team") | .TagValue')
    # Add the uppercase tag
    aliyun ecs AddTags --ResourceType instance --ResourceId "$instance_id" \
        --Tag.1.Key Team --Tag.1.Value "$tag_value" --RegionId cn-hangzhou
done

┌─────────────────────────────────────────────────────┐
│                   OpenCost Architecture              │
├─────────────────────────────────────────────────────┤
│                                                     │
│  Prometheus  ──→  OpenCost  ──→  Cost data          │
│  (metrics)        (cost engine)   (API + UI)        │
│                                                     │
│  Inputs:                                            │
│  • node-exporter → node price, specs                │
│  • kube-state-metrics → Pod resource requests       │
│  • cAdvisor → actual resource usage                 │
│  • cloud pricing API → node unit price              │
│                                                     │
│  Outputs:                                           │
│  • namespace cost                                   │
│  • Pod-level cost                                   │
│  • tag-dimension cost                               │
│  • resource efficiency (usage vs requests)          │
│                                                     │
└─────────────────────────────────────────────────────┘

# Install OpenCost (requires an existing Prometheus)
kubectl apply -f https://raw.githubusercontent.com/opencost/opencost/develop/kubernetes/opencost.yaml -n opencost

# Verify the deployment
kubectl get pods -n opencost
# Expected output: opencost 1/1 Running

# Port-forward to access the UI
kubectl port-forward -n opencost service/opencost 9003 9090

# Visit http://localhost:9090 for the cost dashboard

# OpenCost custom cost allocation rules
# Business-line cost attribution via Pod labels
apiVersion: v1
kind: ConfigMap
metadata:
  name: opencost-config
  namespace: opencost
data:
  # Allocate cost by labels
  costLabels: "team,project,env"
  # Node price overrides (required for self-hosted IDC or non-cloud environments)
  # Otherwise OpenCost fetches node prices automatically via the cloud API
  pricing: |
    {
      "defaultCPUHourlyCost": 0.0,
      "defaultRAMHourlyCost": 0.0,
      "customPricesEnabled": true
    }    

# Pod label example: cost attribution labels
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  namespace: production
spec:
  template:
    metadata:
      labels:
        app: order-service
        team: payment          # team
        project: order-system  # project
        env: production        # environment

#!/usr/bin/env python3
"""
ECS right-sizing inspection: generates downsizing recommendations
based on CloudMonitor metrics.
Dependencies: alibabacloud_cloudmonitor20190801
"""
import json
from datetime import datetime, timedelta
from alibabacloud_cloudmonitor20190801.client import Client as CmsClient
from alibabacloud_tea_openapi.models import Config

def get_cpu_utilization(client, instance_id, days=7):
    """Get the CPU utilization P95 of an instance over the past N days"""
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    period = 300  # one data point every 5 minutes

    response = client.describe_metric_list(
        describe_metric_list_request={
            "MetricName": "CPUUtilization",
            "Dimensions": json.dumps({"instanceId": instance_id}),
            "StartTime": start_time.strftime("%Y-%m-%d %H:%M:%S"),
            "EndTime": end_time.strftime("%Y-%m-%d %H:%M:%S"),
            "Period": str(period)
        }
    )

    values = []
    for datapoint in response.body.datapoints:
        dp = json.loads(datapoint) if isinstance(datapoint, str) else datapoint
        if "Average" in dp:
            values.append(dp["Average"])

    if not values:
        return None

    # Compute P95
    values.sort()
    p95_index = int(len(values) * 0.95)
    return values[p95_index]

def generate_rightsizing_report(client, instances):
    """Generate a right-sizing recommendation report"""
    report = []
    for inst in instances:
        instance_id = inst["instance_id"]
        instance_type = inst["instance_type"]

        cpu_p95 = get_cpu_utilization(client, instance_id)
        if cpu_p95 is None:
            continue

        if cpu_p95 < 15:
            suggestion = "Downsize 50%"
            priority = "High"
        elif cpu_p95 < 30:
            suggestion = "Downsize 30%"
            priority = "Medium"
        elif cpu_p95 < 60:
            suggestion = "Keep as-is"
            priority = "Low"
        else:
            suggestion = "Upgrade or scale out"
            priority = "Urgent"

        report.append({
            "instance_id": instance_id,
            "instance_type": instance_type,
            "cpu_p95": round(cpu_p95, 1),
            "suggestion": suggestion,
            "priority": priority
        })

    return report

if __name__ == "__main__":
    config = Config(
        access_key_id="YOUR_AK",
        access_key_secret="YOUR_SK",
        endpoint="metrics.cn-hangzhou.aliyuncs.com"
    )
    cms_client = CmsClient(config)

    # Instance list (actually fetched from the ECS API)
    instances = [
        {"instance_id": "i-bp1xxx", "instance_type": "ecs.c5.2xlarge"},
        {"instance_id": "i-bp1yyy", "instance_type": "ecs.r5.large"},
    ]

    report = generate_rightsizing_report(cms_client, instances)
    print(f"{'InstanceID':<20} {'Spec':<20} {'CPU P95':<10} {'Suggestion':<15} {'Priority'}")
    print("-" * 80)
    for r in report:
        print(f"{r['instance_id']:<20} {r['instance_type']:<20} {r['cpu_p95']:<10} {r['suggestion']:<15} {r['priority']}")

# K8s Spot node pool PodDisruptionBudget
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: spot-service-pdb
  namespace: production
spec:
  minAvailable: 2  # keep at least 2 Pods available
  selector:
    matchLabels:
      node-role: spot
---
# Deployment on Spot nodes
apiVersion: apps/v1
kind: Deployment
metadata:
  name: batch-processor
spec:
  replicas: 6
  template:
    spec:
      nodeSelector:
        node-role: spot
      tolerations:
        - key: "spot-instance"
          operator: "Exists"
      containers:
        - name: processor
          image: registry.example.com/processor:v1.0
          lifecycle:
            preStop:
              exec:
                command:
                  - /bin/sh
                  - -c
                  - |
                    # Graceful exit: deregister from the load balancer, wait for in-flight requests to finish
                    curl -s http://lb-controller/deregister
                    sleep 30                    
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 5

# GitLab CI integration with Infracost cost estimation
# .gitlab-ci.yml
stages:
  - validate
  - cost-check
  - plan
  - apply

infracost-cost-check:
  stage: cost-check
  image: infracost/infracost:latest
  variables:
    INFRACOST_API_KEY: $INFRACOST_API_KEY
  script:
    # Generate the cost estimate report
    - infracost breakdown --path . --format json --out-file infracost.json
    # Generate a diff report (vs. the main branch baseline)
    - infracost diff --path . --compare-to infracost-base.json --format json --out-file infracost-diff.json
    # Extract the monthly cost change
    - |
      MONTHLY_COST=$(jq -r '.projects[0].breakdown.totalMonthlyCost // 0' infracost.json)
      COST_DIFF=$(jq -r '.projects[0].diff.totalMonthlyCost // 0' infracost-diff.json)
      echo "Monthly cost of this change: ${MONTHLY_COST} USD"
      echo "Change vs. baseline: ${COST_DIFF} USD"
      # Budget gate: monthly increase over 5000 USD requires approval
      THRESHOLD=5000
      if (( $(echo "${COST_DIFF} > ${THRESHOLD}" | bc -l) )); then
        echo "⚠️ Cost increase exceeds threshold (${THRESHOLD} USD/month), requires architect approval"
        exit 1
      fi      
  artifacts:
    paths:
      - infracost.json
      - infracost-diff.json
  only:
    - merge_requests

#!/usr/bin/env python3
"""
Daily spend alert: checks whether today's cloud spend exceeds the daily average budget.
Fetches real-time spend via the Alibaba Cloud BSS (billing center) API.
"""
import json
from alibabacloud_bssopenapi20171214.client import Client as BssClient
from alibabacloud_tea_openapi.models import Config

def check_daily_cost(access_key_id, access_key_secret, monthly_budget=280000):
    """
    Check whether today's spend is abnormal.
    monthly_budget: monthly budget (RMB), daily budget = monthly_budget / 30
    """
    config = Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        endpoint="business.aliyuncs.com"
    )
    client = BssClient(config)

    # Query this month's bill
    from alibabacloud_bssopenapi20171214.models import QueryBillRequest
    import datetime
    now = datetime.datetime.now()
    response = client.query_bill(
        query_bill_request=QueryBillRequest(
            billing_cycle=f"{now.year}-{now.month:02d}"
        )
    )

    # Sum up today's spend
    daily_cost = 0
    for item in response.body.data.account_bill_list.items:
        if item.billing_date == now.strftime("%Y-%m-%d"):
            daily_cost += float(item.pretax_amount)

    daily_budget = monthly_budget / 30
    ratio = daily_cost / daily_budget

    if ratio > 1.5:
        level = "CRITICAL"
    elif ratio > 1.2:
        level = "WARNING"
    elif ratio > 1.0:
        level = "NOTICE"
    else:
        level = "OK"

    return {
        "daily_cost": round(daily_cost, 2),
        "daily_budget": round(daily_budget, 2),
        "ratio": round(ratio, 2),
        "level": level
    }

if __name__ == "__main__":
    result = check_daily_cost("YOUR_AK", "YOUR_SK", monthly_budget=280000)
    print(json.dumps(result, indent=2, ensure_ascii=False))
    # Send an alert via DingTalk/Feishu bot when level is CRITICAL

def detect_cost_anomaly(client, instance_ids, days=7):
    """
    Day-over-day anomaly detection: compares today's cost with the daily
    average of the past N days. If today's cost exceeds 200% of the N-day
    average, it is flagged as an anomaly.
    """
    import datetime
    now = datetime.datetime.now()

    costs = []
    for i in range(days + 1):
        date = (now - datetime.timedelta(days=i)).strftime("%Y-%m-%d")
        daily_cost = get_daily_cost(client, date)
        costs.append({"date": date, "cost": daily_cost})

    today_cost = costs[0]["cost"]
    avg_cost = sum(c["cost"] for c in costs[1:]) / days

    if avg_cost > 0 and today_cost > avg_cost * 2:
        return {
            "anomaly": True,
            "today_cost": today_cost,
            "avg_cost": round(avg_cost, 2),
            "increase_ratio": round(today_cost / avg_cost, 2),
            "message": f"Today's cost {today_cost} RMB exceeds the {days}-day average of {round(avg_cost, 2)} RMB by {round(today_cost / avg_cost, 2)}x"
        }
    return {"anomaly": False, "today_cost": today_cost, "avg_cost": round(avg_cost, 2)}

#!/usr/bin/env python3
"""
Idle resource inspection: scans low-CPU ECS instances, SLBs without backends,
and unattached cloud disks. Runs daily; results are sent via DingTalk/Feishu.
"""
import json
import datetime

def scan_idle_ecs(client, threshold=5, days=7):
    """Scan ECS instances whose CPU utilization stays below the threshold"""
    idle_instances = []
    # Get all running instances
    instances = client.describe_instances(
        describe_instances_request={"status": "Running"}
    ).body.instances.instance

    for inst in instances:
        cpu_avg = get_cpu_avg(client, inst.instance_id, days)
        if cpu_avg is not None and cpu_avg < threshold:
            idle_instances.append({
                "instance_id": inst.instance_id,
                "instance_name": inst.instance_name,
                "instance_type": inst.instance_type,
                "cpu_avg": round(cpu_avg, 1),
                "idle_days": days,
                "monthly_cost": estimate_monthly_cost(inst.instance_type),
                "suggestion": "Recommended to release or downsize"
            })
    return idle_instances

def scan_idle_slb(client):
    """Scan SLB instances with no backend servers"""
    idle_slbs = []
    slbs = client.describe_load_balancers(
        describe_load_balancers_request={}
    ).body.load_balancers.load_balancer

    for slb in slbs:
        # Get the backend server list
        backends = client.describe_health_status(
            describe_health_status_request={
                "load_balancer_id": slb.load_balancer_id
            }
        ).body.backend_servers.backend_server

        if not backends:
            idle_slbs.append({
                "slb_id": slb.load_balancer_id,
                "slb_name": slb.load_balancer_name,
                "address": slb.address,
                "monthly_cost": 30,  # SLB base fee is about 30 RMB/month
                "suggestion": "No backend servers, recommended to release"
            })
    return idle_slbs

def scan_unattached_disks(client):
    """Scan unattached cloud disks"""
    idle_disks = []
    disks = client.describe_disks(
        describe_disks_request={"status": "Available"}  # Available = unattached
    ).body.disks.disk

    for disk in disks:
        idle_disks.append({
            "disk_id": disk.disk_id,
            "disk_name": disk.disk_name,
            "size_gb": disk.size,
            "disk_type": disk.category,
            "create_time": disk.creation_time,
            "monthly_cost": estimate_disk_monthly(disk.size, disk.category),
            "suggestion": "Unattached disk, recommended to release or attach"
        })
    return idle_disks

def generate_report(idle_ecs, idle_slbs, idle_disks):
    """Generate the inspection report"""
    total_waste = sum(i["monthly_cost"] for i in idle_ecs) + \
                  sum(s["monthly_cost"] for s in idle_slbs) + \
                  sum(d["monthly_cost"] for d in idle_disks)

    report = f"""
Idle resource inspection report ({datetime.date.today()})

Idle ECS: {len(idle_ecs)} instances, ~{sum(i['monthly_cost'] for i in idle_ecs)} RMB wasted per month
Idle SLB: {len(idle_slbs)} instances, ~{sum(s['monthly_cost'] for s in idle_slbs)} RMB wasted per month
Unattached disks: {len(idle_disks)} disks, ~{sum(d['monthly_cost'] for d in idle_disks)} RMB wasted per month

Total monthly waste: {total_waste} RMB
"""
    if total_waste > 10000:
        report += "\n⚠️ Monthly waste exceeds 10K RMB, recommended to handle immediately!"
    return report