# 创建项目目录
mkdir -p ~/storage-tiering-agent/{scripts,config,logs}
cd ~/storage-tiering-agent

buckets:
  - name: user-uploads-bucket
    region: us-east-1
    owner: web-team
    retention_days: 30  # 最短保留天数
  
  - name: backup-archive-bucket
    region: us-east-1
    owner: ops-team
    retention_days: 90
  
  - name: logs-archival-bucket
    region: us-east-1
    owner: devops-team
    retention_days: 365

#!/usr/bin/env python3
"""Collect last access timestamps from S3 access logs."""

import csv
import json
from datetime import datetime, timezone
from pathlib import Path
import boto3

S3_LOG_BUCKET = "your-access-logs-bucket"
OUTPUT_PATH = Path("logs/access_summary.json")

def summarize_access_logs():
    """Summarize object last-access time from S3 server access logs."""
    s3 = boto3.client('s3')
    
    # List log objects for the past 7 days (adjust as needed)
    prefix = f"logs/7days/"
    log_objects = s3.list_objects_v2(Bucket=S3_LOG_BUCKET, Prefix=prefix)
    
    object_last_access = {}  # {key: last_access_timestamp}
    
    if 'Contents' not in log_objects:
        print("No access logs found.")
        return object_last_access
    
    for obj in log_objects['Contents']:
        key = obj['Key']
        response = s3.get_object(Bucket=S3_LOG_BUCKET, Key=key)
        content = response['Body'].read().decode('utf-8')
        
        # Parse each line of the access log (standard S3 format)
        reader = csv.reader(csv.StringIO(content))
        for row in reader:
            if len(row) < 8 or row[0].startswith('#'):
                continue
            # Column 6 is time, column 7 is requester, column 8 is bucket/object
            # Column 9 is request-type (GET, PUT, etc.)
            timestamp_str = f"{row[5]} {row[6]}"
            try:
                ts = datetime.strptime(timestamp_str, "%d/%b/%Y:%H:%M:%S %z").timestamp()
                obj_key = row[8].split('"')[1]  # Extract bucket/object
                if obj_key not in object_last_access or ts > object_last_access[obj_key]:
                    object_last_access[obj_key] = ts
            except (IndexError, ValueError):
                continue
    
    return object_last_access

if __name__ == "__main__":
    summary = summarize_access_logs()
    with open(OUTPUT_PATH, 'w', encoding='utf-8') as f:
        json.dump({"last_access": summary, "collected_at": datetime.now(timezone.utc).isoformat()}, 
                  f, indent=2, ensure_ascii=False)
    print(f"Collected {len(summary)} object access records.")

#!/usr/bin/env python3
"""Decision engine for storage tiering based on access patterns."""

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

ACCESS_SUMMARY = Path("logs/access_summary.json")
BUCKET_CONFIG = Path("config/buckets.yaml")  # Simplified - use yaml loader
OUTPUT_DIR = Path("decisions")

def load_config():
    """Load bucket configurations (simplified JSON representation)."""
    return {
        "user-uploads-bucket": {"region": "us-east-1", "owner": "web-team", "retention_days": 30},
        "backup-archive-bucket": {"region": "us-east-1", "owner": "ops-team", "retention_days": 90},
        "logs-archival-bucket": {"region": "us-east-1", "owner": "devops-team", "retention_days": 365},
    }

def classify_object(obj_key, last_access_ts, bucket_name, config):
    """Classify an object into a tier based on access time and policy."""
    now = datetime.now(timezone.utc).timestamp()
    days_inactive = (now - last_access_ts) / 86400
    
    # Minimum retention check
    min_retain = config.get("retention_days", 30)
    
    if days_inactive <= 30:
        tier = "HOT"
        action = "none"
    elif days_inactive <= 90:
        tier = "WARM"
        action = "move_to_ia"
    elif days_inactive <= 365:
        tier = "COLD"
        action = "move_to_glacier"
    else:
        tier = "ARCHIVE"
        action = "move_to_deep_archive"
    
    # Enforce minimum retention
    if days_inactive < min_retain:
        tier = "HOT"  # Keep hot until minimum reached
        action = "none"
    
    return {
        "object": obj_key,
        "bucket": bucket_name,
        "tier": tier,
        "action": action,
        "days_inactive": round(days_inactive, 1),
        "last_access": datetime.fromtimestamp(last_access_ts, tz=timezone.utc).isoformat(),
    }

def main():
    OUTPUT_DIR.mkdir(exist_ok=True)
    
    # Load access summary
    with open(ACCESS_SUMMARY, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    last_access = data["last_access"]
    config = load_config()
    
    all_decisions = []
    
    # Group decisions by bucket (simplified — in practice you'd parse bucket from object key)
    for obj_key, ts in last_access.items():
        # Determine bucket from key prefix (example pattern)
        bucket_name = obj_key.split('/')[0] if '/' in obj_key else "default-bucket"
        bucket_cfg = config.get(bucket_name, {"retention_days": 30})
        
        decision = classify_object(obj_key, ts, bucket_name, bucket_cfg)
        all_decisions.append(decision)
    
    # Write decisions for execution
    output = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "total_objects": len(all_decisions),
        "by_tier": {d["tier"]: sum(1 for d in all_decisions if d["tier"] == d["tier"]) 
                    for d in set(d["tier"] for d in all_decisions)}
    }
    
    with open(OUTPUT_DIR / "decisions.json", 'w', encoding='utf-8') as f:
        json.dump(output, f, indent=2, ensure_ascii=False)
    
    print(f"Generated {len(all_decisions)} decisions. Tier distribution:")
    for tier, count in output["by_tier"].items():
        print(f"  {tier}: {count}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Execute storage tier migrations based on decision output."""

import json
import boto3
from datetime import datetime, timezone
from pathlib import Path

DECISION_FILE = Path("decisions/decisions.json")
S3_CLIENTS = {
    "us-east-1": boto3.client('s3'),
    "us-west-2": boto3.client('s3', region_name='us-west-2'),
    # Add other regions as needed
}

def apply_lifecycle_policy(bucket_name, region, rules):
    """Apply S3 lifecycle configuration to a bucket."""
    s3 = S3_CLIENTS.get(region) or boto3.client('s3', region_name=region)
    
    config = {
        'LifecycleConfiguration': {
            'Rules': rules
        }
    }
    
    s3.put_bucket_lifecycle_configuration(
        Bucket=bucket_name,
        LifecycleConfiguration=config
    )
    print(f"Applied lifecycle policy to {bucket_name} ({region})")

def execute_migration(decisions):
    """Process individual migration decisions."""
    bucket_actions = {}  # {bucket: [(region, actions)]}
    
    for d in decisions:
        bucket = d["bucket"]
        if bucket not in bucket_actions:
            bucket_actions[bucket] = {"region": "us-east-1", "actions": []}
        
        bucket_actions[bucket]["actions"].append(d)
    
    # Generate lifecycle rules per bucket (simplified example)
    for bucket_info in bucket_actions.values():
        bucket_name = bucket_info["bucket"]  # Need to extract this properly
        region = bucket_info["region"]
        rules = bucket_info["actions"]
        
        # In production, you would group rules by bucket and create
        # proper lifecycle policies with filters and transitions
        pass

def main():
    with open(DECISION_FILE, 'r', encoding='utf-8') as f:
        decisions_data = json.load(f)
    
    # Simple report
    print(f"\nMigration Plan Generated at {decisions_data['generated_at']}")
    print(f"Total objects to process: {decisions_data['total_objects']}")
    print("\nBy Tier:")
    for tier, count in decisions_data["by_tier"].items():
        print(f"  {tier}: {count} objects")
    
    print("\n✅ Next steps:")
    print("   1. Review decisions in /decisions/decisions.json")
    print("   2. Apply S3 lifecycle policies via AWS Console or CLI")
    print("   3. Or execute direct copy/delete operations for cold/archival tiers")

if __name__ == "__main__":
    main()

{
  "Rules": [
    {
      "ID": "StandardToIA",
      "Status": "Enabled",
      "Filter": {"Prefix": ""},
      "Transitions": [
        {
          "Days": 30,
          "StorageClass": "STANDARD_IA"
        },
        {
          "Days": 90,
          "StorageClass": "GLACIER"
        },
        {
          "Days": 365,
          "StorageClass": "GLACIER_DEEP_ARCHIVE"
        }
      ],
      "NoncurrentVersionTransitions": [
        {
          "NoncurrentDays": 30,
          "StorageClass": "STANDARD_IA"
        },
        {
          "NoncurrentDays": 90,
          "StorageClass": "GLACIER"
        }
      ]
    }
  ]
}

aws sapi put-bucket-lifecycle-configuration \
  --bucket your-bucket-name \
  --file lifecycle-config.json

# 设置对象默认存储类别为 NEARLINE（冷线）
gsutil defstorageclass set NEARLINE gs://your-bucket

# 使用 Lifecycle Rules 在 90 天后转为 COLDLINE
gsutil lifecycle set lifecycle-config.json gs://your-bucket

{
  "lifecycle": {
    "rule": [
      {
        "action": {"type": "SetStorageClass", "storageClass": "NEARLINE"},
        "condition": {"age": 30}
      },
      {
        "action": {"type": "SetStorageClass", "storageClass": "COLDLINE"},
        "condition": {"age": 90}
      },
      {
        "action": {"type": "Delete"},
        "condition": {"age": 365}
      }
    ]
  }
}

# 生成存储用量报告（AWS）
aws s3api list-object-v2 --bucket your-bucket > before.json

# 或者使用 Cost Explorer
aws ce get_cost_and_usage --time-period Start=2026-07-01,End=2026-07-31 \
  --metrics "SUM(TotalCost)" --group-by Type=DIMENSIONS,Values=StorageType

aws cloudwatch put-metric-alarm \
  --alarm-name "S3-Cost-Alert" \
  --alarm-description "Alert when S3 spending exceeds threshold" \
  --metric-name "StorageCost" \
  --namespace "AWS/S3" \
  --statistic "Average" \
  --period 86400 \
  --unit "Dollar" \
  --threshold 100 \
  --comparison-operator "GreaterThanThreshold" \
  --dimensions "Name=BucketName,Value=your-bucket-name" \
  --alarm-actions "arn:aws:sns:us-east-1:123456789012:alerts"

# 查看对象存储类别分布
aws s3api list-objects-v2 --bucket your-bucket | grep -E '"ETag":"|StorageClass"'