Backup Best Practices

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

# Enterprise Backup Retention Policy Configuration
# This YAML defines retention rules for automated enforcement
 
version: "1.0"
policy_name: "Enterprise Database Retention Policy"
effective_date: "2024-01-01"
approved_by: "CTO, Legal, Compliance"
 
# Default retention (applies unless overridden)
default_retention:
  daily_backups: 14
  weekly_backups: 8
  monthly_backups: 12
  annual_backups: 7
  transaction_logs: 72  # hours
 
# Data classification-specific retention
classifications:
  tier1_critical:
    description: "Mission-critical production databases"
    databases:
      - "production_orders"
      - "production_customers"
      - "production_financial"
    retention:
      daily_backups: 30
      weekly_backups: 12
      monthly_backups: 24
      annual_backups: 10
      transaction_logs: 168  # 7 days
    storage_tiers:
      - age_days: 0
        storage_class: "hot_ssd"
      - age_days: 14
        storage_class: "standard_hdd"
      - age_days: 90
        storage_class: "cloud_archive"
      - age_days: 365
        storage_class: "deep_archive"
 
  tier2_operational:
    description: "Business operational databases"
    databases:
      - "operations_inventory"
      - "operations_shipping"
      - "analytics_warehouse"
    retention:
      daily_backups: 14
      weekly_backups: 8
      monthly_backups: 12
      annual_backups: 7
      transaction_logs: 72
 
  tier3_development:
    description: "Development and test databases"
    databases:
      - "dev_*"
      - "test_*"
      - "staging_*"
    retention:
      daily_backups: 7
      weekly_backups: 2
      monthly_backups: 0  # No monthly retention
      annual_backups: 0
      transaction_logs: 24
 
# Regulatory overrides (supersede classification defaults)
regulatory_requirements:
  sox_financial:
    applies_to: ["production_financial", "production_orders"]
    minimum_retention_years: 7
    audit_trail_required: true
    
  pci_dss:
    applies_to: ["production_payment*"]
    minimum_retention_years: 1
    encryption_required: true
    access_logging_required: true
 
# Legal hold configuration
legal_hold:
  enabled: true
  notification_email: "legal-team@company.com"
  hold_database: "backup_management.legal_holds"
 
# Expiration and deletion rules
deletion:
  grace_period_days: 7  # Warning before deletion
  require_approval_for:
    - "tier1_critical"
    - files_larger_than_gb: 100
  verification_required: true  # Confirm deletion successful

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
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238

#!/usr/bin/env python3
"""
Enterprise Backup Retention Automation System
Implements automated lifecycle management based on retention policies
"""
 
import os
import yaml
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
import boto3  # For S3 tier migration example
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
class StorageClass(Enum):
    HOT_SSD = "hot_ssd"
    STANDARD_HDD = "standard_hdd"
    CLOUD_STANDARD = "STANDARD"
    CLOUD_IA = "STANDARD_IA"
    CLOUD_GLACIER = "GLACIER"
    DEEP_ARCHIVE = "DEEP_ARCHIVE"
 
@dataclass
class Backup:
    backup_id: str
    database_name: str
    backup_type: str  # full, incremental, transaction_log
    created_at: datetime
    size_bytes: int
    storage_class: StorageClass
    expiration_date: Optional[datetime]
    legal_hold: bool = False
    parent_backup_id: Optional[str] = None
 
@dataclass
class RetentionAction:
    backup_id: str
    action: str  # migrate, expire, hold
    target_storage_class: Optional[StorageClass]
    reason: str
    scheduled_date: datetime
 
class RetentionPolicyEngine:
    """
    Evaluates backups against retention policies and generates actions
    """
    
    def __init__(self, policy_path: str):
        with open(policy_path, 'r') as f:
            self.policy = yaml.safe_load(f)
        self.s3_client = boto3.client('s3')
        
    def get_classification(self, database_name: str) -> dict:
        """Determine which classification applies to a database"""
        for class_name, class_config in self.policy['classifications'].items():
            for pattern in class_config['databases']:
                if self._matches_pattern(database_name, pattern):
                    return class_config['retention']
        return self.policy['default_retention']
    
    def _matches_pattern(self, name: str, pattern: str) -> bool:
        """Simple wildcard pattern matching"""
        if pattern.endswith('*'):
            return name.startswith(pattern[:-1])
        return name == pattern
    
    def calculate_expiration(self, backup: Backup) -> datetime:
        """Calculate when a backup should expire based on policy"""
        retention = self.get_classification(backup.database_name)
        
        if backup.backup_type == 'transaction_log':
            hours = retention.get('transaction_logs', 72)
            return backup.created_at + timedelta(hours=hours)
        
        # Determine if this is a daily, weekly, monthly, or annual backup
        # This is simplified - real implementation would check backup schedule
        age_days = (datetime.now() - backup.created_at).days
        
        if age_days < 30:
            return backup.created_at + timedelta(days=retention['daily_backups'])
        elif age_days < 90:
            return backup.created_at + timedelta(weeks=retention['weekly_backups'])
        elif age_days < 365:
            return backup.created_at + timedelta(days=30 * retention['monthly_backups'])
        else:
            return backup.created_at + timedelta(days=365 * retention['annual_backups'])
    
    def evaluate_tier_migration(self, backup: Backup) -> Optional[RetentionAction]:
        """Determine if backup should be migrated to different storage tier"""
        age_days = (datetime.now() - backup.created_at).days
        
        # Find appropriate tier based on age
        classification = self._get_classification_config(backup.database_name)
        storage_tiers = classification.get('storage_tiers', [])
        
        target_tier = None
        for tier in sorted(storage_tiers, key=lambda x: x['age_days'], reverse=True):
            if age_days >= tier['age_days']:
                target_tier = StorageClass(tier['storage_class'])
                break
        
        if target_tier and target_tier != backup.storage_class:
            return RetentionAction(
                backup_id=backup.backup_id,
                action='migrate',
                target_storage_class=target_tier,
                reason=f"Age {age_days} days exceeds threshold for current tier",
                scheduled_date=datetime.now()
            )
        return None
    
    def _get_classification_config(self, database_name: str) -> dict:
        """Get full classification config for a database"""
        for class_name, class_config in self.policy['classifications'].items():
            for pattern in class_config['databases']:
                if self._matches_pattern(database_name, pattern):
                    return class_config
        return {'retention': self.policy['default_retention'], 'storage_tiers': []}
    
    def check_legal_holds(self, backup: Backup, holds: List[dict]) -> bool:
        """Check if backup is under any legal hold"""
        for hold in holds:
            if hold['status'] == 'active':
                if backup.database_name in hold.get('scope_databases', []):
                    return True
                if backup.created_at >= hold['hold_date']:
                    # Backup created after hold initiated
                    if backup.database_name in hold.get('scope_databases', []):
                        return True
        return False
    
    def generate_actions(self, backups: List[Backup], 
                         legal_holds: List[dict]) -> List[RetentionAction]:
        """Generate all retention actions for a list of backups"""
        actions = []
        
        for backup in backups:
            # Check legal holds first - they override everything
            if self.check_legal_holds(backup, legal_holds):
                logger.info(f"Backup {backup.backup_id} under legal hold, skipping")
                continue
            
            # Check for expiration
            expiration = self.calculate_expiration(backup)
            if expiration < datetime.now():
                # Check for dependent backups before expiring
                if not self._has_dependents(backup, backups):
                    actions.append(RetentionAction(
                        backup_id=backup.backup_id,
                        action='expire',
                        target_storage_class=None,
                        reason=f"Exceeded retention: expired {expiration}",
                        scheduled_date=datetime.now()
                    ))
                else:
                    logger.warning(
                        f"Backup {backup.backup_id} expired but has dependents"
                    )
            
            # Check for tier migration
            migration = self.evaluate_tier_migration(backup)
            if migration:
                actions.append(migration)
        
        return actions
    
    def _has_dependents(self, backup: Backup, all_backups: List[Backup]) -> bool:
        """Check if any backups depend on this one (for incremental chains)"""
        if backup.backup_type != 'full':
            return False
        return any(
            b.parent_backup_id == backup.backup_id 
            for b in all_backups 
            if b.backup_id != backup.backup_id
        )
 
class RetentionExecutor:
    """
    Executes retention actions and logs results
    """
    
    def __init__(self, dry_run: bool = True):
        self.dry_run = dry_run
        self.s3_client = boto3.client('s3')
        
    def execute(self, actions: List[RetentionAction]) -> dict:
        """Execute a list of retention actions"""
        results = {'succeeded': 0, 'failed': 0, 'skipped': 0}
        
        for action in actions:
            try:
                if action.action == 'expire':
                    self._execute_expiration(action)
                elif action.action == 'migrate':
                    self._execute_migration(action)
                results['succeeded'] += 1
            except Exception as e:
                logger.error(f"Action failed: {action.backup_id}: {e}")
                results['failed'] += 1
        
        return results
    
    def _execute_expiration(self, action: RetentionAction):
        """Delete expired backup"""
        logger.info(f"{'[DRY RUN] ' if self.dry_run else ''}Expiring backup: {action.backup_id}")
        if not self.dry_run:
            # Actual deletion logic here
            pass
    
    def _execute_migration(self, action: RetentionAction):
        """Migrate backup to new storage tier"""
        logger.info(
            f"{'[DRY RUN] ' if self.dry_run else ''}"
            f"Migrating {action.backup_id} to {action.target_storage_class}"
        )
        if not self.dry_run:
            # S3 storage class change example
            # self.s3_client.copy_object(...)
            pass
 
# Main execution
if __name__ == "__main__":
    engine = RetentionPolicyEngine("retention_policy.yaml")
    executor = RetentionExecutor(dry_run=True)
    
    # In practice, backups would come from backup catalog database
    backups = []  # Load from catalog
    legal_holds = []  # Load from legal hold table
    
    actions = engine.generate_actions(backups, legal_holds)
    logger.info(f"Generated {len(actions)} retention actions")
    
    results = executor.execute(actions)
    logger.info(f"Execution results: {results}")

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

-- Retention Policy Compliance Dashboard Queries
 
-- 1. Overall Retention Compliance Status
SELECT 
    database_classification,
    COUNT(*) AS total_databases,
    SUM(CASE WHEN compliant = true THEN 1 ELSE 0 END) AS compliant_count,
    ROUND(100.0 * SUM(CASE WHEN compliant = true THEN 1 ELSE 0 END) / COUNT(*), 1) AS compliance_pct
FROM (
    SELECT 
        d.database_name,
        d.classification AS database_classification,
        CASE 
            WHEN MAX(b.created_at) > NOW() - INTERVAL '2 days' THEN true
            ELSE false 
        END AS compliant
    FROM databases d
    LEFT JOIN backup_catalog b ON d.database_name = b.database_name
    GROUP BY d.database_name, d.classification
) compliance
GROUP BY database_classification;
 
-- 2. Storage Consumption by Retention Tier
SELECT 
    CASE 
        WHEN age_days <= 14 THEN 'Short-term (0-14 days)'
        WHEN age_days <= 90 THEN 'Medium-term (15-90 days)'
        WHEN age_days <= 365 THEN 'Long-term (91-365 days)'
        ELSE 'Archive (365+ days)'
    END AS retention_tier,
    COUNT(*) AS backup_count,
    pg_size_pretty(SUM(size_bytes)) AS total_size,
    pg_size_pretty(AVG(size_bytes)) AS avg_backup_size,
    storage_class
FROM (
    SELECT 
        *,
        EXTRACT(DAY FROM NOW() - created_at) AS age_days
    FROM backup_catalog
) aged_backups
GROUP BY 
    CASE 
        WHEN age_days <= 14 THEN 'Short-term (0-14 days)'
        WHEN age_days <= 90 THEN 'Medium-term (15-90 days)'
        WHEN age_days <= 365 THEN 'Long-term (91-365 days)'
        ELSE 'Archive (365+ days)'
    END,
    storage_class
ORDER BY 
    CASE retention_tier
        WHEN 'Short-term (0-14 days)' THEN 1
        WHEN 'Medium-term (15-90 days)' THEN 2
        WHEN 'Long-term (91-365 days)' THEN 3
        ELSE 4
    END;
 
-- 3. Expiration Forecast
SELECT 
    DATE(expiration_date) AS expiration_day,
    COUNT(*) AS backups_expiring,
    pg_size_pretty(SUM(size_bytes)) AS storage_reclaimed,
    STRING_AGG(DISTINCT database_name, ', ') AS affected_databases
FROM backup_catalog
WHERE expiration_date BETWEEN NOW() AND NOW() + INTERVAL '30 days'
    AND NOT legal_hold
GROUP BY DATE(expiration_date)
ORDER BY expiration_day;
 
-- 4. Legal Hold Impact Report
SELECT 
    lh.matter_name,
    lh.hold_date,
    COUNT(bhi.backup_id) AS backups_held,
    pg_size_pretty(SUM(b.size_bytes)) AS storage_consumed,
    MIN(b.created_at) AS oldest_backup_held,
    MAX(b.expiration_date) AS original_expiration_latest
FROM backup_legal_holds lh
JOIN backup_hold_items bhi ON lh.hold_id = bhi.hold_id
JOIN backup_catalog b ON bhi.backup_id = b.backup_id
WHERE lh.status = 'active'
GROUP BY lh.hold_id, lh.matter_name, lh.hold_date
ORDER BY storage_consumed DESC;
 
-- 5. Deletion Audit Trail
SELECT 
    deleted_at,
    backup_id,
    database_name,
    backup_type,
    original_created_at,
    original_size_bytes,
    deletion_reason,
    deleted_by,
    verification_status
FROM backup_deletion_audit
WHERE deleted_at > NOW() - INTERVAL '30 days'
ORDER BY deleted_at DESC;