1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class StorageCosts:
    """Storage cost parameters."""
    cost_per_gb_month: float          # $/GB/month
    retention_months: float            # How long data is kept
    
    @property
    def cost_per_gb_lifetime(self) -> float:
        return self.cost_per_gb_month * self.retention_months
 
@dataclass
class ComputeCosts:
    """Compute cost parameters."""
    cost_per_core_hour: float         # $/core-hour
    cost_per_gb_ram_month: float      # $/GB RAM/month
    
@dataclass
class OptimizationProfile:
    """Profile for a compression/dedup configuration."""
    name: str
    compression_ratio: float           # e.g., 3.0 for 3:1
    cpu_hours_per_tb: float           # Core-hours to compress 1 TB
    ram_gb_per_tb_unique: float       # RAM needed per TB unique data (0 for compression only)
    decompression_cpu_per_tb: float   # Core-hours to decompress 1 TB
    typical_dedup_ratio: float = 1.0  # Additional dedup savings
 
class OptimizationROICalculator:
    """
    Calculate ROI of different storage optimization strategies.
    """
    
    # Common optimization profiles
    PROFILES = {
        'none': OptimizationProfile(
            name='No optimization',
            compression_ratio=1.0,
            cpu_hours_per_tb=0,
            ram_gb_per_tb_unique=0,
            decompression_cpu_per_tb=0,
        ),
        'lz4_fast': OptimizationProfile(
            name='LZ4 Fast',
            compression_ratio=2.1,
            cpu_hours_per_tb=0.4,
            ram_gb_per_tb_unique=0,
            decompression_cpu_per_tb=0.07,
        ),
        'zstd_fast': OptimizationProfile(
            name='Zstd Level 1',
            compression_ratio=2.8,
            cpu_hours_per_tb=0.6,
            ram_gb_per_tb_unique=0,
            decompression_cpu_per_tb=0.2,
        ),
        'zstd_default': OptimizationProfile(
            name='Zstd Level 3',
            compression_ratio=3.0,
            cpu_hours_per_tb=1.1,
            ram_gb_per_tb_unique=0,
            decompression_cpu_per_tb=0.25,
        ),
        'zstd_high': OptimizationProfile(
            name='Zstd Level 15',
            compression_ratio=3.4,
            cpu_hours_per_tb=25,
            ram_gb_per_tb_unique=0,
            decompression_cpu_per_tb=0.25,
        ),
        'dedup_lz4': OptimizationProfile(
            name='Dedup + LZ4',
            compression_ratio=2.1,
            cpu_hours_per_tb=5.5,  # Hash + index + LZ4
            ram_gb_per_tb_unique=7,
            decompression_cpu_per_tb=0.1,
            typical_dedup_ratio=3.0,  # Common for backup workloads
        ),
        'dedup_zstd': OptimizationProfile(
            name='Dedup + Zstd',
            compression_ratio=3.0,
            cpu_hours_per_tb=6.5,
            ram_gb_per_tb_unique=7,
            decompression_cpu_per_tb=0.3,
            typical_dedup_ratio=3.0,
        ),
    }
    
    def __init__(self, storage_costs: StorageCosts, compute_costs: ComputeCosts):
        self.storage = storage_costs
        self.compute = compute_costs
    
    def calculate_roi(
        self,
        profile: OptimizationProfile,
        data_size_tb: float,
        reads_per_tb_per_month: float = 1.0,  # Read amplification
    ) -> dict:
        """
        Calculate ROI for an optimization profile.
        
        Returns cost breakdown and net savings.
        """
        data_size_gb = data_size_tb * 1024
        
        # Combined reduction from dedup + compression
        total_ratio = profile.compression_ratio * profile.typical_dedup_ratio
        
        # Storage calculations
        physical_gb = data_size_gb / total_ratio
        storage_saved_gb = data_size_gb - physical_gb
        storage_savings = storage_saved_gb * self.storage.cost_per_gb_lifetime
        
        # Compute costs for compression
        compress_cpu_hours = data_size_tb * profile.cpu_hours_per_tb
        compress_cost = compress_cpu_hours * self.compute.cost_per_core_hour
        
        # Compute costs for decompression (ongoing reads)
        total_reads_tb = data_size_tb * reads_per_tb_per_month * self.storage.retention_months
        decompress_cpu_hours = total_reads_tb * profile.decompression_cpu_per_tb
        decompress_cost = decompress_cpu_hours * self.compute.cost_per_core_hour
        
        # Memory costs (for dedup index, over retention period)
        unique_tb = data_size_tb / profile.typical_dedup_ratio
        ram_gb_needed = unique_tb * profile.ram_gb_per_tb_unique
        ram_cost = ram_gb_needed * self.compute.cost_per_gb_ram_month * self.storage.retention_months
        
        # Total compute cost
        total_compute_cost = compress_cost + decompress_cost + ram_cost
        
        # Net ROI
        net_savings = storage_savings - total_compute_cost
        roi_percent = (net_savings / max(total_compute_cost, 0.01)) * 100
        
        return {
            'profile': profile.name,
            'effective_ratio': f'{total_ratio:.1f}:1',
            'storage_saved_gb': storage_saved_gb,
            'storage_savings_usd': storage_savings,
            'compress_cost_usd': compress_cost,
            'decompress_cost_usd': decompress_cost,
            'ram_cost_usd': ram_cost,
            'total_compute_cost_usd': total_compute_cost,
            'net_savings_usd': net_savings,
            'roi_percent': roi_percent,
            'cost_effective': net_savings > 0,
        }
    
    def compare_profiles(
        self,
        data_size_tb: float,
        reads_per_tb_per_month: float = 1.0,
    ) -> list[dict]:
        """Compare all profiles for a given workload."""
        results = []
        for profile in self.PROFILES.values():
            result = self.calculate_roi(profile, data_size_tb, reads_per_tb_per_month)
            results.append(result)
        
        # Sort by net savings
        results.sort(key=lambda x: x['net_savings_usd'], reverse=True)
        return results
 
 
# Example usage with real-world costs
if __name__ == '__main__':
    # AWS-like pricing (illustrative)
    storage = StorageCosts(
        cost_per_gb_month=0.023,  # S3 Standard
        retention_months=12,
    )
    compute = ComputeCosts(
        cost_per_core_hour=0.04,   # Approximation
        cost_per_gb_ram_month=0.005,
    )
    
    calculator = OptimizationROICalculator(storage, compute)
    
    # Analyze 100 TB backup workload
    results = calculator.compare_profiles(
        data_size_tb=100,
        reads_per_tb_per_month=0.5,  # Read 50% of data per month
    )
    
    for r in results:
        print(f"{r['profile']:20} | Ratio: {r['effective_ratio']:6} | "
              f"Net: ${r['net_savings_usd']:, .0f
                        }")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

from dataclasses import dataclass
 
@dataclass
class StorageMedia:
    """Characteristics of a storage medium."""
    name: str
    read_latency_ms: float          # Time to first byte
    read_bandwidth_mb_s: float      # Sequential read speed
    iops: int                       # Random read IOPS
 
@dataclass
class CompressionConfig:
    """Compression configuration."""
    algorithm: str
    decompress_bandwidth_mb_s: float
    compression_ratio: float
 
# Common storage media profiles
MEDIA = {
    'nvme': StorageMedia('NVMe SSD', 0.02, 3000, 500000),
    'sata_ssd': StorageMedia('SATA SSD', 0.1, 500, 50000),
    'hdd': StorageMedia('HDD', 10, 150, 150),
    'cloud_ssd': StorageMedia('Cloud SSD (gp3)', 0.5, 400, 16000),
    'cloud_hdd': StorageMedia('Cloud HDD (st1)', 20, 250, 500),
}
 
# Common compression profiles (decompression matters for reads)
COMPRESSION = {
    'none': CompressionConfig('None', float('inf'), 1.0),
    'lz4': CompressionConfig('LZ4', 4000, 2.1),
    'zstd_fast': CompressionConfig('Zstd-1', 1400, 2.8),
    'zstd_default': CompressionConfig('Zstd-3', 1300, 3.0),
    'zstd_high': CompressionConfig('Zstd-15', 1100, 3.4),
}
 
def analyze_read_latency(
    logical_size_kb: float,
    block_size_kb: float,
    media: StorageMedia,
    compression: CompressionConfig,
) -> dict:
    """
    Analyze read latency for a request.
    
    Accounts for:
    1. Read amplification (reading full block for partial access)
    2. Decompression time
    3. Media latency
    """
    # Read amplification: how many blocks must we read?
    blocks_needed = max(1, (logical_size_kb + block_size_kb - 1) // block_size_kb)
    logical_read_kb = blocks_needed * block_size_kb
    
    # Compressed size to read from disk
    physical_read_kb = logical_read_kb / compression.compression_ratio
    
    # Time components
    disk_latency_ms = media.read_latency_ms
    disk_transfer_ms = (physical_read_kb / 1024) / media.read_bandwidth_mb_s * 1000
    decompress_ms = (logical_read_kb / 1024) / compression.decompress_bandwidth_mb_s * 1000
    
    total_ms = disk_latency_ms + disk_transfer_ms + decompress_ms
    
    return {
        'request_size_kb': logical_size_kb,
        'block_size_kb': block_size_kb,
        'blocks_read': blocks_needed,
        'read_amplification': logical_read_kb / logical_size_kb,
        'physical_read_kb': physical_read_kb,
        'disk_latency_ms': disk_latency_ms,
        'disk_transfer_ms': disk_transfer_ms,
        'decompress_ms': decompress_ms,
        'total_latency_ms': total_ms,
        'decompress_fraction': decompress_ms / total_ms,
    }
 
def find_optimal_config(
    logical_size_kb: float,
    latency_budget_ms: float,
    media: StorageMedia,
) -> list[dict]:
    """
    Find compression configurations that meet latency budget.
    """
    results = []
    
    for block_size in [4, 8, 16, 32, 64, 128]:
        for comp_name, compression in COMPRESSION.items():
            analysis = analyze_read_latency(
                logical_size_kb, block_size, media, compression
            )
            
            meets_budget = analysis['total_latency_ms'] <= latency_budget_ms
            
            results.append({
                'block_size_kb': block_size,
                'compression': comp_name,
                'ratio': compression.compression_ratio,
                'latency_ms': analysis['total_latency_ms'],
                'meets_budget': meets_budget,
                'decompress_fraction': analysis['decompress_fraction'],
            })
    
    # Sort by compression ratio (best first) among those meeting budget
    meeting = [r for r in results if r['meets_budget']]
    meeting.sort(key=lambda x: x['ratio'], reverse=True)
    
    return meeting[:5]  # Top 5 options

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

from enum import Enum
from dataclasses import dataclass
import time
import zstandard as zstd
import lz4.frame
 
class DataTemperature(Enum):
    HOT = 'hot'
    WARM = 'warm'
    COLD = 'cold'
    FROZEN = 'frozen'
 
@dataclass
class CompressionStrategy:
    """Strategy for a data temperature tier."""
    algorithm: str
    level: int
    skip_threshold: float  # Skip if ratio below this
 
class AdaptiveCompressor:
    """
    Compression system that adapts to data characteristics and access patterns.
    """
    
    STRATEGIES = {
        DataTemperature.HOT: CompressionStrategy('lz4', 0, 1.05),
        DataTemperature.WARM: CompressionStrategy('zstd', 3, 1.1),
        DataTemperature.COLD: CompressionStrategy('zstd', 9, 1.15),
        DataTemperature.FROZEN: CompressionStrategy('zstd', 19, 1.2),
    }
    
    # File extensions that shouldn't be compressed
    SKIP_EXTENSIONS = {
        '.jpg', '.jpeg', '.png', '.gif', '.webp',  # Images
        '.mp4', '.avi', '.mkv', '.webm',           # Video
        '.mp3', '.aac', '.flac', '.ogg',           # Audio
        '.zip', '.gz', '.bz2', '.xz', '.7z',       # Already compressed
        '.pdf',                                      # Often has compressed streams
    }
    
    def __init__(self):
        self.zstd_compressors = {
            level: zstd.ZstdCompressor(level=level)
            for level in [3, 9, 15, 19]
        }
        self.zstd_decompressor = zstd.ZstdDecompressor()
    
    def _should_skip_by_extension(self, filename: str) -> bool:
        """Check if file extension indicates incompressible data."""
        ext = '.' + filename.rsplit('.', 1)[-1].lower() if '.' in filename else ''
        return ext in self.SKIP_EXTENSIONS
    
    def _check_compressibility(self, data: bytes, threshold: float) -> tuple[bool, float]:
        """
        Quick check if data is worth compressing.
        
        Returns (should_compress, sample_ratio)
        """
        sample = data[:4096]
        if len(sample) < 100:
            return True, 0  # Too small to check
        
        try:
            compressed = lz4.frame.compress(sample)
            ratio = len(sample) / len(compressed)
            return ratio >= threshold, ratio
        except:
            return True, 0  # If check fails, try compressing
    
    def compress(
        self,
        data: bytes,
        temperature: DataTemperature,
        filename: str = '',
    ) -> tuple[bytes, dict]:
        """
        Compress data adaptively based on temperature and content.
        
        Returns (compressed_data, metadata)
        """
        strategy = self.STRATEGIES[temperature]
        
        # Skip by extension
        if filename and self._should_skip_by_extension(filename):
            return data, {
                'compressed': False,
                'reason': 'extension_skip',
                'algorithm': 'none',
            }
        
        # Compressibility check
        worth_it, sample_ratio = self._check_compressibility(data, strategy.skip_threshold)
        if not worth_it:
            return data, {
                'compressed': False,
                'reason': 'low_ratio',
                'sample_ratio': sample_ratio,
                'algorithm': 'none',
            }
        
        # Compress with appropriate algorithm
        start = time.time()
        
        if strategy.algorithm == 'lz4':
            compressed = lz4.frame.compress(data)
        else:
            compressor = self.zstd_compressors.get(strategy.level)
            compressed = compressor.compress(data)
        
        elapsed = time.time() - start
        ratio = len(data) / len(compressed)
        
        # Final check: did compression actually help?
        if ratio < strategy.skip_threshold:
            return data, {
                'compressed': False,
                'reason': 'actual_ratio_low',
                'actual_ratio': ratio,
                'algorithm': 'none',
            }
        
        return compressed, {
            'compressed': True,
            'algorithm': strategy.algorithm,
            'level': strategy.level,
            'original_size': len(data),
            'compressed_size': len(compressed),
            'ratio': ratio,
            'compress_time_ms': elapsed * 1000,
        }
    
    def decompress(self, data: bytes, metadata: dict) -> bytes:
        """Decompress based on stored metadata."""
        if not metadata.get('compressed', False):
            return data
        
        algorithm = metadata['algorithm']
        
        if algorithm == 'lz4':
            return lz4.frame.decompress(data)
        elif algorithm == 'zstd':
            return self.zstd_decompressor.decompress(data)
        else:
            raise ValueError(f"Unknown algorithm: {algorithm}")