OpenSearch Repository GCS Plugin Upgrade: Migration and Best Practices#

1
#!/bin/bash
2
OPENSEARCH_URL="https://localhost:9200"
3
AUTH="-u admin:admin"
4

5
echo "=== OpenSearch GCS Plugin Status ==="
6

7
# Check installed plugins
8
echo "\nInstalled plugins:"
9
curl -s $AUTH "$OPENSEARCH_URL/_cat/plugins?v" | grep repository-gcs
10

11
# Check plugin version details
12
echo "\nPlugin details:"
13
curl -s $AUTH "$OPENSEARCH_URL/_nodes/plugins?filter_path=nodes.*.plugins" | \
14
  jq '.nodes[].plugins[] | select(.name == "repository-gcs")'
15

16
# List existing repositories
17
echo "\nGCS repositories:"
18
curl -s $AUTH "$OPENSEARCH_URL/_snapshot?pretty" | \
19
  jq '.[] | select(.type == "gcs") | {name: .type, settings: .settings}'
20

21
# Check active snapshots
22
echo "\nActive snapshots:"
23
for repo in $(curl -s $AUTH "$OPENSEARCH_URL/_snapshot" | jq -r 'keys[]'); do
24
  echo "Repository: $repo"
25
  curl -s $AUTH "$OPENSEARCH_URL/_snapshot/$repo/_current?pretty"
26
done
27

28
# Check snapshot statistics
29
echo "\nSnapshot statistics:"
30
curl -s $AUTH "$OPENSEARCH_URL/_snapshot/_stats?pretty"

1
#!/bin/bash
2
BACKUP_DIR="/backup/opensearch-gcs/$(date +%Y%m%d-%H%M%S)"
3
mkdir -p "$BACKUP_DIR"
4

5
# Export repository settings
6
echo "Backing up repository configurations..."
7
curl -s $AUTH "$OPENSEARCH_URL/_snapshot" > "$BACKUP_DIR/repositories.json"
8

9
# Export snapshot list for each repository
10
for repo in $(curl -s $AUTH "$OPENSEARCH_URL/_snapshot" | jq -r 'keys[]'); do
11
  echo "Backing up snapshots for repository: $repo"
12
  curl -s $AUTH "$OPENSEARCH_URL/_snapshot/$repo/_all" > "$BACKUP_DIR/snapshots-$repo.json"
13
done
14

15
# Backup plugin configuration
16
echo "Backing up plugin configuration..."
17
cp /etc/opensearch/opensearch.yml "$BACKUP_DIR/"
18
cp -r /etc/opensearch/repository-gcs/ "$BACKUP_DIR/" 2>/dev/null || true
19

20
# Document current version
21
echo "Documenting current versions..."
22
cat > "$BACKUP_DIR/version-info.txt" <<EOF
23
OpenSearch Version: $(curl -s $AUTH "$OPENSEARCH_URL" | jq -r '.version.number')
24
Plugin Version: $(curl -s $AUTH "$OPENSEARCH_URL/_nodes/plugins" | jq -r '.nodes[].plugins[] | select(.name == "repository-gcs") | .version')
25
Backup Date: $(date)
26
EOF
27

28
echo "Backup completed: $BACKUP_DIR"

1
#!/bin/bash
2
# Stop OpenSearch
3
sudo systemctl stop opensearch
4

5
# Remove the old plugin
6
sudo -u opensearch /usr/share/opensearch/bin/opensearch-plugin remove repository-gcs
7

8
# Clean up any residual files
9
sudo rm -rf /usr/share/opensearch/plugins/repository-gcs/
10
sudo rm -rf /var/lib/opensearch/repository-gcs/
11

12
# Clear plugin cache
13
sudo rm -rf /tmp/opensearch-*

1
#!/bin/bash
2
# Set the target version
3
PLUGIN_VERSION="2.10.0"
4
OPENSEARCH_VERSION="2.10.0"
5

6
# Install the new plugin
7
sudo -u opensearch /usr/share/opensearch/bin/opensearch-plugin install \
8
  "repository-gcs:${PLUGIN_VERSION}"
9

10
# Verify installation
11
/usr/share/opensearch/bin/opensearch-plugin list | grep repository-gcs

1
#!/bin/bash
2
# Update existing repository with new settings
3
curl -X PUT "$OPENSEARCH_URL/_snapshot/gcs-backup" \
4
  $AUTH \
5
  -H "Content-Type: application/json" \
6
  -d '{
7
    "type": "gcs",
8
    "settings": {
9
      "bucket": "opensearch-snapshots",
10
      "client": "default",
11
      "base_path": "snapshots/prod",
12
      "chunk_size": "1gb",
13
      "compress": true,
14
      "max_restore_bytes_per_sec": "100mb",
15
      "max_snapshot_bytes_per_sec": "40mb",
16
      "readonly": false,
17
      "metadata": {
18
        "cluster_name": "production",
19
        "upgraded_at": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'"
20
      }
21
    }
22
  }'
23

24
# Verify repository
25
curl -X POST "$OPENSEARCH_URL/_snapshot/gcs-backup/_verify" $AUTH

1
#!/bin/bash
2
# 1. Create verification snapshot
3
VERIFY_SNAPSHOT="verify-$(date +%Y%m%d-%H%M%S)"
4
curl -X PUT "$OPENSEARCH_URL/_snapshot/gcs-backup/$VERIFY_SNAPSHOT" \
5
  $AUTH \
6
  -H "Content-Type: application/json" \
7
  -d '{
8
    "indices": ".opensearch",
9
    "include_global_state": false
10
  }'
11

12
# 2. Test restore capability
13
curl -X POST "$OPENSEARCH_URL/_snapshot/gcs-backup/$VERIFY_SNAPSHOT/_restore" \
14
  $AUTH \
15
  -H "Content-Type: application/json" \
16
  -d '{
17
    "indices": ".opensearch",
18
    "rename_pattern": "(.+)",
19
    "rename_replacement": "test_$1"
20
  }'
21

22
# 3. Verify and cleanup
23
curl -X DELETE "$OPENSEARCH_URL/test_*" $AUTH
24
curl -X DELETE "$OPENSEARCH_URL/_snapshot/gcs-backup/$VERIFY_SNAPSHOT" $AUTH

1
#!/bin/bash
2
# Setup new cluster with new plugin version
3
NEW_CLUSTER="https://new-cluster:9200"
4

5
# 1. Configure repository on new cluster
6
curl -X PUT "$NEW_CLUSTER/_snapshot/gcs-migration" \
7
  $AUTH \
8
  -H "Content-Type: application/json" \
9
  -d '{
10
    "type": "gcs",
11
    "settings": {
12
      "bucket": "opensearch-snapshots",
13
      "client": "default",
14
      "base_path": "migration/temp",
15
      "readonly": false
16
    }
17
  }'
18

19
# 2. Create snapshot on old cluster
20
MIGRATION_SNAPSHOT="migration-$(date +%Y%m%d-%H%M%S)"
21
curl -X PUT "$OPENSEARCH_URL/_snapshot/gcs-backup/$MIGRATION_SNAPSHOT?wait_for_completion=false" \
22
  $AUTH \
23
  -H "Content-Type: application/json" \
24
  -d '{
25
    "indices": "*",
26
    "include_global_state": true
27
  }'
28

29
# 3. Monitor snapshot progress
30
while true; do
31
  STATUS=$(curl -s $AUTH "$OPENSEARCH_URL/_snapshot/gcs-backup/$MIGRATION_SNAPSHOT/_status" | \
32
    jq -r '.snapshots[0].state')
33
  if [ "$STATUS" = "SUCCESS" ]; then
34
    break
35
  fi
36
  echo "Snapshot status: $STATUS"
37
  sleep 30
38
done
39

40
# 4. Restore on new cluster
41
curl -X POST "$NEW_CLUSTER/_snapshot/gcs-migration/$MIGRATION_SNAPSHOT/_restore" \
42
  $AUTH \
43
  -H "Content-Type: application/json" \
44
  -d '{
45
    "indices": "*",
46
    "include_global_state": false,
47
    "index_settings": {
48
      "index.number_of_replicas": 0
49
    }
50
  }'

1
#!/usr/bin/env python3
2
import requests
3
import json
4
import time
5
from datetime import datetime, timedelta
6

7
class IncrementalMigration:
8
    def __init__(self, source_url, target_url, auth):
9
        self.source = source_url
10
        self.target = target_url
11
        self.auth = auth
12
        self.bucket = "opensearch-snapshots"
13

14
    def setup_repositories(self):
15
        """Setup GCS repositories on both clusters"""
16
        repo_config = {
17
            "type": "gcs",
18
            "settings": {
19
                "bucket": self.bucket,
20
                "client": "default",
21
                "base_path": "incremental",
22
                "chunk_size": "1gb",
23
                "compress": True
24
            }
25
        }
26

27
        # Setup on source
28
        requests.put(
29
            f"{self.source}/_snapshot/gcs-incremental",
30
            auth=self.auth,
31
            json=repo_config
32
        )
33

34
        # Setup on target
35
        requests.put(
36
            f"{self.target}/_snapshot/gcs-incremental",
37
            auth=self.auth,
38
            json=repo_config
39
        )
40

41
    def get_indices_by_age(self, days_old):
42
        """Get indices older than specified days"""
43
        response = requests.get(
44
            f"{self.source}/_cat/indices?format=json",
45
            auth=self.auth
46
        )
47

48
        indices = []
49
        cutoff_date = datetime.now() - timedelta(days=days_old)
50

51
        for index in response.json():
52
            # Parse index date from name (assuming pattern like logs-2024.01.15)
53
            try:
54
                date_str = index['index'].split('-')[-1]
55
                index_date = datetime.strptime(date_str, '%Y.%m.%d')
56

57
                if index_date < cutoff_date:
58
                    indices.append(index['index'])
59
            except:
60
                continue
61

62
        return indices
63

64
    def migrate_indices_batch(self, indices, batch_name):
65
        """Migrate a batch of indices"""
66
        snapshot_name = f"batch-{batch_name}-{int(time.time())}"
67

68
        # Create snapshot
69
        print(f"Creating snapshot {snapshot_name} for {len(indices)} indices...")
70
        response = requests.put(
71
            f"{self.source}/_snapshot/gcs-incremental/{snapshot_name}",
72
            auth=self.auth,
73
            json={
74
                "indices": ",".join(indices),
75
                "include_global_state": False,
76
                "metadata": {
77
                    "batch": batch_name,
78
                    "index_count": len(indices)
79
                }
80
            }
81
        )
82

83
        # Wait for completion
84
        self.wait_for_snapshot(snapshot_name)
85

86
        # Restore on target
87
        print(f"Restoring snapshot {snapshot_name}...")
88
        response = requests.post(
89
            f"{self.target}/_snapshot/gcs-incremental/{snapshot_name}/_restore",
90
            auth=self.auth,
91
            json={
92
                "indices": ",".join(indices),
93
                "include_global_state": False,
94
                "index_settings": {
95
                    "index.number_of_replicas": 0
96
                }
97
            }
98
        )
99

100
        return snapshot_name
101

102
    def wait_for_snapshot(self, snapshot_name):
103
        """Wait for snapshot to complete"""
104
        while True:
105
            response = requests.get(
106
                f"{self.source}/_snapshot/gcs-incremental/{snapshot_name}",
107
                auth=self.auth
108
            )
109

110
            snapshot = response.json()['snapshots'][0]
111
            state = snapshot['state']
112

113
            if state == 'SUCCESS':
114
                print(f"Snapshot {snapshot_name} completed successfully")
115
                break
116
            elif state == 'FAILED':
117
                raise Exception(f"Snapshot {snapshot_name} failed")
118
            else:
119
                print(f"Snapshot {snapshot_name} state: {state}")
120
                time.sleep(30)
121

122
    def run_incremental_migration(self):
123
        """Run the incremental migration process"""
124
        self.setup_repositories()
125

126
        # Migrate in batches by age
127
        age_ranges = [
128
            (365, "very-old"),    # > 1 year
129
            (180, "old"),         # 6-12 months
130
            (90, "medium"),       # 3-6 months
131
            (30, "recent"),       # 1-3 months
132
            (7, "current"),       # 1 week - 1 month
133
            (0, "latest")         # < 1 week
134
        ]
135

136
        for days, batch_name in age_ranges:
137
            indices = self.get_indices_by_age(days)
138
            if indices:
139
                print(f"\nMigrating {batch_name} indices ({len(indices)} total)...")
140
                self.migrate_indices_batch(indices[:50], batch_name)  # Batch of 50
141

142
                # Verify migration
143
                self.verify_indices(indices[:50])
144

145
    def verify_indices(self, indices):
146
        """Verify indices were migrated successfully"""
147
        for index in indices:
148
            source_count = requests.get(
149
                f"{self.source}/{index}/_count",
150
                auth=self.auth
151
            ).json()['count']
152

153
            target_count = requests.get(
154
                f"{self.target}/{index}/_count",
155
                auth=self.auth
156
            ).json()['count']
157

158
            if source_count != target_count:
159
                print(f"WARNING: Count mismatch for {index}: {source_count} vs {target_count}")
160
            else:
161
                print(f"Verified {index}: {source_count} documents")
162

163
# Run migration
164
if __name__ == "__main__":
165
    migration = IncrementalMigration(
166
        source_url="https://old-cluster:9200",
167
        target_url="https://new-cluster:9200",
168
        auth=('admin', 'admin')
169
    )
170
    migration.run_incremental_migration()

1
#!/bin/bash
2
# Configure snapshot settings
3
curl -X PUT "$OPENSEARCH_URL/_cluster/settings" \
4
  $AUTH \
5
  -H "Content-Type: application/json" \
6
  -d '{
7
    "persistent": {
8
      "repositories.gcs.chunk_size": "1gb",
9
      "repositories.gcs.compress": true,
10
      "repositories.gcs.application_name": "opensearch-prod",
11
      "snapshot.max_restore_bytes_per_sec": "200mb",
12
      "snapshot.max_snapshot_bytes_per_sec": "100mb"
13
    },
14
    "transient": {
15
      "indices.recovery.max_bytes_per_sec": "200mb",
16
      "cluster.routing.allocation.node_concurrent_recoveries": 4
17
    }
18
  }'
19

20
# Create optimized repository
21
curl -X PUT "$OPENSEARCH_URL/_snapshot/gcs-optimized" \
22
  $AUTH \
23
  -H "Content-Type: application/json" \
24
  -d '{
25
    "type": "gcs",
26
    "settings": {
27
      "bucket": "opensearch-snapshots",
28
      "client": "default",
29
      "base_path": "optimized",
30
      "chunk_size": "1gb",
31
      "compress": true,
32
      "max_restore_bytes_per_sec": "200mb",
33
      "max_snapshot_bytes_per_sec": "100mb",
34
      "application_name": "opensearch-optimized"
35
    }
36
  }'

1
#!/usr/bin/env python3
2
import concurrent.futures
3
import requests
4
import json
5
from datetime import datetime
6

7
class ParallelSnapshotManager:
8
    def __init__(self, opensearch_url, auth):
9
        self.url = opensearch_url
10
        self.auth = auth
11
        self.max_workers = 5
12

13
    def create_snapshot_parallel(self, indices_groups, repository):
14
        """Create snapshots in parallel for different index groups"""
15
        timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
16

17
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
18
            futures = []
19

20
            for i, indices in enumerate(indices_groups):
21
                snapshot_name = f"parallel-{timestamp}-group{i}"
22
                future = executor.submit(
23
                    self._create_snapshot,
24
                    repository,
25
                    snapshot_name,
26
                    indices
27
                )
28
                futures.append((snapshot_name, future))
29

30
            # Wait for all snapshots to complete
31
            results = []
32
            for snapshot_name, future in futures:
33
                try:
34
                    result = future.result(timeout=3600)  # 1 hour timeout
35
                    results.append({
36
                        'snapshot': snapshot_name,
37
                        'status': 'success',
38
                        'details': result
39
                    })
40
                except Exception as e:
41
                    results.append({
42
                        'snapshot': snapshot_name,
43
                        'status': 'failed',
44
                        'error': str(e)
45
                    })
46

47
            return results
48

49
    def _create_snapshot(self, repository, snapshot_name, indices):
50
        """Create a single snapshot"""
51
        response = requests.put(
52
            f"{self.url}/_snapshot/{repository}/{snapshot_name}",
53
            auth=self.auth,
54
            json={
55
                "indices": ",".join(indices),
56
                "include_global_state": False,
57
                "metadata": {
58
                    "created_by": "parallel_snapshot_manager",
59
                    "index_count": len(indices)
60
                }
61
            }
62
        )
63

64
        if response.status_code != 200:
65
            raise Exception(f"Failed to create snapshot: {response.text}")
66

67
        # Wait for completion
68
        return self._wait_for_snapshot(repository, snapshot_name)
69

70
    def _wait_for_snapshot(self, repository, snapshot_name):
71
        """Wait for snapshot completion"""
72
        while True:
73
            response = requests.get(
74
                f"{self.url}/_snapshot/{repository}/{snapshot_name}",
75
                auth=self.auth
76
            )
77

78
            if response.status_code != 200:
79
                raise Exception(f"Failed to get snapshot status: {response.text}")
80

81
            snapshot = response.json()['snapshots'][0]
82
            if snapshot['state'] == 'SUCCESS':
83
                return snapshot
84
            elif snapshot['state'] == 'FAILED':
85
                raise Exception(f"Snapshot failed: {snapshot.get('failures', 'Unknown error')}")
86

87
            time.sleep(10)

1
#!/bin/bash
2
while true; do
3
    clear
4
    echo "=== OpenSearch GCS Plugin Monitor ==="
5
    echo "Time: $(date)"
6
    echo ""
7

8
    # Plugin status
9
    echo "Plugin Status:"
10
    curl -s $AUTH "$OPENSEARCH_URL/_nodes/stats/repositories" | \
11
      jq '.nodes[].repositories'
12
    echo ""
13

14
    # Active operations
15
    echo "Active Snapshot Operations:"
16
    curl -s $AUTH "$OPENSEARCH_URL/_snapshot/_status" | \
17
      jq '.snapshots[] | {snapshot: .snapshot, state: .state, progress: .shards_stats.done}'
18
    echo ""
19

20
    # Repository stats
21
    echo "Repository Statistics:"
22
    curl -s $AUTH "$OPENSEARCH_URL/_snapshot/_stats" | \
23
      jq '.stats'
24
    echo ""
25

26
    # GCS metrics
27
    echo "GCS Client Metrics:"
28
    curl -s $AUTH "$OPENSEARCH_URL/_nodes/stats/repositories?include_repository_stats=true" | \
29
      jq '.nodes[].repositories.gcs'
30

31
    sleep 30
32
done

1
#!/usr/bin/env python3
2
import requests
3
import json
4
import hashlib
5
from datetime import datetime
6

7
class GCSPluginValidator:
8
    def __init__(self, opensearch_url, auth):
9
        self.url = opensearch_url
10
        self.auth = auth
11
        self.validation_results = []
12

13
    def run_all_validations(self):
14
        """Run complete validation suite"""
15
        print("Running GCS Plugin Validation Suite...")
16
        print("=" * 50)
17

18
        self.validate_plugin_installation()
19
        self.validate_repository_access()
20
        self.validate_snapshot_operations()
21
        self.validate_restore_operations()
22
        self.validate_performance_metrics()
23
        self.validate_security_settings()
24

25
        self.print_results()
26

27
    def validate_plugin_installation(self):
28
        """Validate plugin is properly installed"""
29
        try:
30
            response = requests.get(
31
                f"{self.url}/_cat/plugins?format=json",
32
                auth=self.auth
33
            )
34

35
            plugins = response.json()
36
            gcs_plugin = next((p for p in plugins if p['component'] == 'repository-gcs'), None)
37

38
            if gcs_plugin:
39
                self.validation_results.append({
40
                    'test': 'Plugin Installation',
41
                    'status': 'PASS',
42
                    'details': f"Version {gcs_plugin['version']} installed"
43
                })
44
            else:
45
                self.validation_results.append({
46
                    'test': 'Plugin Installation',
47
                    'status': 'FAIL',
48
                    'details': 'Plugin not found'
49
                })
50
        except Exception as e:
51
            self.validation_results.append({
52
                'test': 'Plugin Installation',
53
                'status': 'ERROR',
54
                'details': str(e)
55
            })
56

57
    def validate_repository_access(self):
58
        """Validate GCS repository access"""
59
        test_repo = "gcs-validation-test"
60

61
        try:
62
            # Create test repository
63
            response = requests.put(
64
                f"{self.url}/_snapshot/{test_repo}",
65
                auth=self.auth,
66
                json={
67
                    "type": "gcs",
68
                    "settings": {
69
                        "bucket": "opensearch-snapshots",
70
                        "base_path": "validation-test",
71
                        "readonly": False
72
                    }
73
                }
74
            )
75

76
            if response.status_code == 200:
77
                # Verify repository
78
                verify_response = requests.post(
79
                    f"{self.url}/_snapshot/{test_repo}/_verify",
80
                    auth=self.auth
81
                )
82

83
                if verify_response.status_code == 200:
84
                    self.validation_results.append({
85
                        'test': 'Repository Access',
86
                        'status': 'PASS',
87
                        'details': 'Successfully created and verified repository'
88
                    })
89
                else:
90
                    self.validation_results.append({
91
                        'test': 'Repository Access',
92
                        'status': 'FAIL',
93
                        'details': f"Verification failed: {verify_response.text}"
94
                    })
95

96
                # Cleanup
97
                requests.delete(f"{self.url}/_snapshot/{test_repo}", auth=self.auth)
98
            else:
99
                self.validation_results.append({
100
                    'test': 'Repository Access',
101
                    'status': 'FAIL',
102
                    'details': f"Failed to create repository: {response.text}"
103
                })
104

105
        except Exception as e:
106
            self.validation_results.append({
107
                'test': 'Repository Access',
108
                'status': 'ERROR',
109
                'details': str(e)
110
            })
111

112
    def validate_snapshot_operations(self):
113
        """Validate snapshot creation and management"""
114
        # Implementation continues...
115
        pass
116

117
    def print_results(self):
118
        """Print validation results"""
119
        print("\nValidation Results:")
120
        print("=" * 50)
121

122
        for result in self.validation_results:
123
            status_color = {
124
                'PASS': '\033[92m',
125
                'FAIL': '\033[91m',
126
                'ERROR': '\033[93m'
127
            }.get(result['status'], '\033[0m')
128

129
            print(f"{status_color}{result['status']}\033[0m - {result['test']}")
130
            print(f"  Details: {result['details']}")
131
            print()
132

133
# Run validation
134
if __name__ == "__main__":
135
    validator = GCSPluginValidator(
136
        opensearch_url="https://localhost:9200",
137
        auth=('admin', 'admin')
138
    )
139
    validator.run_all_validations()

1
#!/usr/bin/env python3
2
import requests
3
import json
4
from datetime import datetime, timedelta
5

6
class GCSCostOptimizer:
7
    def __init__(self, opensearch_url, auth):
8
        self.url = opensearch_url
9
        self.auth = auth
10

11
    def analyze_snapshot_costs(self):
12
        """Analyze snapshot storage costs"""
13
        # Get all snapshots
14
        response = requests.get(
15
            f"{self.url}/_snapshot/_all",
16
            auth=self.auth
17
        )
18

19
        total_size = 0
20
        old_snapshots = []
21

22
        for repo_name, repo_data in response.json().items():
23
            snapshots_response = requests.get(
24
                f"{self.url}/_snapshot/{repo_name}/_all",
25
                auth=self.auth
26
            )
27

28
            for snapshot in snapshots_response.json()['snapshots']:
29
                # Calculate age
30
                start_time = datetime.fromtimestamp(snapshot['start_time_in_millis'] / 1000)
31
                age = datetime.now() - start_time
32

33
                # Track old snapshots
34
                if age > timedelta(days=90):
35
                    old_snapshots.append({
36
                        'repository': repo_name,
37
                        'snapshot': snapshot['snapshot'],
38
                        'age_days': age.days,
39
                        'size_gb': snapshot.get('total_size', 0) / (1024**3)
40
                    })
41

42
                total_size += snapshot.get('total_size', 0)
43

44
        # Calculate costs (example: $0.02 per GB per month for standard storage)
45
        monthly_cost = (total_size / (1024**3)) * 0.02
46

47
        print(f"\nSnapshot Storage Analysis:")
48
        print(f"Total Size: {total_size / (1024**3):.2f} GB")
49
        print(f"Estimated Monthly Cost: ${monthly_cost:.2f}")
50
        print(f"\nOld Snapshots (>90 days): {len(old_snapshots)}")
51

52
        # Recommendations
53
        if old_snapshots:
54
            potential_savings = sum(s['size_gb'] for s in old_snapshots) * 0.02
55
            print(f"Potential Monthly Savings: ${potential_savings:.2f}")
56
            print("\nRecommended Deletions:")
57
            for snapshot in sorted(old_snapshots, key=lambda x: x['age_days'], reverse=True)[:10]:
58
                print(f"- {snapshot['repository']}/{snapshot['snapshot']} ({snapshot['age_days']} days, {snapshot['size_gb']:.2f} GB)")
59

60
    def implement_lifecycle_policy(self):
61
        """Implement cost-optimized lifecycle policy"""
62
        policy = {
63
            "schedule": "0 0 2 * * ?",  # Daily at 2 AM
64
            "name": "<cost-optimized-{now/d}>",
65
            "repository": "gcs-backup",
66
            "config": {
67
                "indices": ["*"],
68
                "ignore_unavailable": True,
69
                "include_global_state": False,
70
                "partial": False
71
            },
72
            "retention": {
73
                "expire_after": "30d",  # Keep for 30 days
74
                "min_count": 7,         # Always keep at least 7
75
                "max_count": 30         # Never keep more than 30
76
            }
77
        }
78

79
        # Apply policy
80
        response = requests.put(
81
            f"{self.url}/_slm/policy/cost-optimized",
82
            auth=self.auth,
83
            json=policy
84
        )
85

86
        if response.status_code == 200:
87
            print("Cost-optimized lifecycle policy implemented successfully")
88
        else:
89
            print(f"Failed to implement policy: {response.text}")
90

91
# Run cost analysis
92
if __name__ == "__main__":
93
    optimizer = GCSCostOptimizer(
94
        opensearch_url="https://localhost:9200",
95
        auth=('admin', 'admin')
96
    )
97
    optimizer.analyze_snapshot_costs()
98
    optimizer.implement_lifecycle_policy()