Reindexing Strategies

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

"""
Zero-Downtime Reindexing with Change Capture
 
This implementation demonstrates the dual-write pattern with
proper change buffering to ensure no updates are lost.
"""
 
from datetime import datetime
from typing import Optional
import asyncio
 
 
class ZeroDowntimeReindexer:
    """
    Orchestrates zero-downtime reindexing with these guarantees:
    - All documents from source are indexed
    - All changes during reindex are captured
    - Production traffic is unaffected
    - Switchover is atomic
    """
    
    def __init__(
        self,
        source_db: 'DatabaseClient',
        search_client: 'ElasticsearchClient', 
        change_source: 'ChangeDataCapture',
        index_alias: str,
    ):
        self.source_db = source_db
        self.es = search_client
        self.cdc = change_source
        self.alias = index_alias
        
    async def reindex(
        self,
        new_index_name: str,
        new_mappings: dict,
        new_settings: dict,
    ) -> ReindexResult:
        """
        Execute full reindex with zero downtime.
        
        Returns details about the reindex operation including
        timing, document counts, and any issues encountered.
        """
        start_time = datetime.utcnow()
        
        # Step 1: Create new index with target mappings/settings
        await self._create_new_index(
            new_index_name, 
            new_mappings, 
            new_settings
        )
        
        # Step 2: Start capturing changes BEFORE bulk load
        # This is critical - we need changes from this moment forward
        change_cursor = await self.cdc.start_capture()
        capture_start_time = datetime.utcnow()
        
        # Step 3: Bulk load all documents from source
        bulk_load_result = await self._bulk_load_from_source(
            new_index_name,
            snapshot_time=capture_start_time
        )
        
        # Step 4: Replay accumulated changes to catch up
        catchup_result = await self._replay_changes(
            new_index_name,
            change_cursor,
            until=datetime.utcnow()
        )
        
        # Step 5: Enter dual-write mode briefly
        # New changes go to both old and new index
        await self._enable_dual_write(new_index_name)
        
        # Step 6: Verify new index consistency
        verification = await self._verify_index(new_index_name)
        if not verification.passed:
            raise ReindexVerificationError(verification.issues)
            
        # Step 7: Atomic alias switch
        old_index = await self._get_current_index()
        await self._atomic_alias_switch(old_index, new_index_name)
        
        # Step 8: Cleanup
        await self._disable_dual_write()
        await self._schedule_old_index_deletion(old_index)
        
        return ReindexResult(
            success=True,
            old_index=old_index,
            new_index=new_index_name,
            documents_loaded=bulk_load_result.count,
            changes_replayed=catchup_result.count,
            duration=datetime.utcnow() - start_time
        )
        
    async def _create_new_index(
        self, 
        name: str, 
        mappings: dict, 
        settings: dict
    ) -> None:
        """Create new index with bulk-optimized settings."""
        # Start with settings optimized for bulk loading
        bulk_settings = {
            **settings,
            "index": {
                **settings.get("index", {}),
                "refresh_interval": "-1",  # Disable refresh during load
                "number_of_replicas": 0,   # No replicas during load
            }
        }
        
        await self.es.indices.create(
            index=name,
            body={
                "mappings": mappings,
                "settings": bulk_settings
            }
        )
        
    async def _bulk_load_from_source(
        self, 
        index_name: str,
        snapshot_time: datetime
    ) -> BulkLoadResult:
        """
        Stream all documents from source database to new index.
        
        Uses cursor-based pagination to handle billions of docs
        without memory issues.
        """
        total_indexed = 0
        batch_size = 5000
        
        async for batch in self.source_db.stream_all_documents(
            batch_size=batch_size,
            as_of=snapshot_time  # Consistent read as of snapshot time
        ):
            transformed = [
                self._transform_document(doc) for doc in batch
            ]
            
            await self.es.bulk_index(
                index=index_name,
                documents=transformed
            )
            
            total_indexed += len(batch)
            
            # Log progress periodically
            if total_indexed % 100000 == 0:
                self._log_progress(total_indexed, "bulk_load")
                
        return BulkLoadResult(count=total_indexed)
        
    async def _replay_changes(
        self,
        index_name: str,
        cursor: 'ChangeCursor',
        until: datetime
    ) -> CatchupResult:
        """
        Apply changes that occurred during bulk load.
        
        This is the "catchup" phase that brings the new index
        up to current state.
        """
        changes_applied = 0
        
        async for change_batch in self.cdc.read_changes(
            cursor=cursor,
            until=until
        ):
            for change in change_batch:
                if change.operation == 'DELETE':
                    await self.es.delete(
                        index=index_name,
                        id=change.document_id,
                        ignore=[404]  # May not exist yet
                    )
                else:
                    doc = self._transform_document(change.document)
                    await self.es.index(
                        index=index_name,
                        id=change.document_id,
                        body=doc
                    )
                changes_applied += 1
                
        return CatchupResult(count=changes_applied)
        
    async def _atomic_alias_switch(
        self, 
        old_index: str, 
        new_index: str
    ) -> None:
        """
        Atomically move alias from old to new index.
        
        This is a single atomic operation - queries see either
        the old index or the new one, never both or neither.
        """
        await self.es.indices.update_aliases(
            body={
                "actions": [
                    {"remove": {"index": old_index, "alias": self.alias}},
                    {"add": {"index": new_index, "alias": self.alias}}
                ]
            }
        )

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

// Basic reindex: copy all documents from source to dest
POST /_reindex
{
  "source": {
    "index": "products_v1"
  },
  "dest": {
    "index": "products_v2"
  }
}
 
// Reindex with transformation using Painless script
POST /_reindex
{
  "source": {
    "index": "products_v1"
  },
  "dest": {
    "index": "products_v2"
  },
  "script": {
    "source": """
      // Add new computed field
      ctx._source.search_boost = 
        ctx._source.sales_count * 0.01 + 
        ctx._source.rating * 0.5;
      
      // Rename field
      ctx._source.product_name = ctx._source.remove('name');
      
      // Remove deprecated field
      ctx._source.remove('legacy_category');
      
      // Conditional transformation
      if (ctx._source.price < 0) {
        ctx._source.price = 0;
      }
    """,
    "lang": "painless"
  }
}
 
// Parallel reindex with slicing (automatic slicing)
POST /_reindex?slices=auto&wait_for_completion=false
{
  "source": {
    "index": "products_v1",
    "size": 5000  // Batch size per slice
  },
  "dest": {
    "index": "products_v2"
  }
}
// Returns task ID for monitoring
 
// Check reindex progress
GET /_tasks/node_id:task_id
 
// Reindex with throttling to limit cluster impact
POST /_reindex?requests_per_second=5000
{
  "source": {
    "index": "products_v1"
  },
  "dest": {
    "index": "products_v2"
  }
}
 
// Reindex from remote cluster
POST /_reindex
{
  "source": {
    "remote": {
      "host": "http://old-cluster:9200",
      "username": "reindex_user",
      "password": "secret"
    },
    "index": "products"
  },
  "dest": {
    "index": "products_v2"
  }
}
 
// Reindex subset with query
POST /_reindex
{
  "source": {
    "index": "products_v1",
    "query": {
      "range": {
        "updated_at": {
          "gte": "2024-01-01"
        }
      }
    }
  },
  "dest": {
    "index": "products_v2"
  }
}

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

/**
 * Comprehensive verification suite for reindexed indexes.
 * 
 * Run this BEFORE switching the production alias.
 */
 
interface VerificationResult {
  passed: boolean;
  checks: CheckResult[];
  summary: string;
}
 
interface CheckResult {
  name: string;
  passed: boolean;
  details: string;
  severity: 'critical' | 'warning' | 'info';
}
 
class ReindexVerification {
  constructor(
    private esClient: ElasticsearchClient,
    private sourceDb: DatabaseClient,
    private oldIndex: string,
    private newIndex: string,
  ) {}
 
  async runFullVerification(): Promise<VerificationResult> {
    const checks: CheckResult[] = [];
 
    // 1. Document count verification
    checks.push(await this.verifyDocumentCount());
 
    // 2. Random sample verification
    checks.push(await this.verifyRandomSample(1000));
 
    // 3. Edge case verification
    checks.push(await this.verifyEdgeCases());
 
    // 4. Query parity verification
    checks.push(await this.verifyQueryParity());
 
    // 5. Performance baseline
    checks.push(await this.verifyPerformance());
 
    // 6. Mapping verification
    checks.push(await this.verifyMappings());
 
    const criticalFailures = checks.filter(
      c => !c.passed && c.severity === 'critical'
    );
 
    return {
      passed: criticalFailures.length === 0,
      checks,
      summary: this.generateSummary(checks),
    };
  }
 
  private async verifyDocumentCount(): Promise<CheckResult> {
    const [sourceCount, newIndexCount, oldIndexCount] = await Promise.all([
      this.sourceDb.count('products'),
      this.esClient.count({ index: this.newIndex }),
      this.esClient.count({ index: this.oldIndex }),
    ]);
 
    const percentDiff = Math.abs(
      (newIndexCount - sourceCount) / sourceCount * 100
    );
 
    const passed = percentDiff < 0.01; // Less than 0.01% difference
 
    return {
      name: 'Document Count',
      passed,
      severity: 'critical',
      details: `Source: ${sourceCount}, New Index: ${newIndexCount}, \
        Old Index: ${oldIndexCount}, Diff: ${percentDiff.toFixed(4)}%`,
    };
  }
 
  private async verifyRandomSample(sampleSize: number): Promise<CheckResult> {
    // Get random document IDs from source
    const randomIds = await this.sourceDb.getRandomIds('products', sampleSize);
 
    let matches = 0;
    let mismatches: Array<{ id: string; field: string; expected: any; actual: any }> = [];
 
    for (const id of randomIds) {
      const [sourceDoc, indexDoc] = await Promise.all([
        this.sourceDb.get('products', id),
        this.esClient.get({ index: this.newIndex, id }).catch(() => null),
      ]);
 
      if (!indexDoc) {
        mismatches.push({ 
          id, 
          field: '_exists', 
          expected: true, 
          actual: false 
        });
        continue;
      }
 
      // Compare key fields
      const fieldsToCheck = ['name', 'price', 'category', 'brand'];
      let docMatches = true;
 
      for (const field of fieldsToCheck) {
        const expected = this.transformField(field, sourceDoc[field]);
        const actual = indexDoc._source[field];
 
        if (JSON.stringify(expected) !== JSON.stringify(actual)) {
          mismatches.push({ id, field, expected, actual });
          docMatches = false;
        }
      }
 
      if (docMatches) matches++;
    }
 
    const matchRate = matches / sampleSize;
    const passed = matchRate >= 0.999; // 99.9% match rate required
 
    return {
      name: 'Random Sample',
      passed,
      severity: 'critical',
      details: `${matches}/${sampleSize} documents matched (\
        ${(matchRate * 100).toFixed(2)}%). \
        Sample mismatches: ${JSON.stringify(mismatches.slice(0, 5))}`,
    };
  }
 
  private async verifyQueryParity(): Promise<CheckResult> {
    // Load sample queries from production logs
    const sampleQueries = await this.loadProductionQueries(100);
 
    let parityCases = 0;
    let divergentCases: Array<{ query: string; oldCount: number; newCount: number }> = [];
 
    for (const query of sampleQueries) {
      const [oldResult, newResult] = await Promise.all([
        this.esClient.search({ index: this.oldIndex, body: query }),
        this.esClient.search({ index: this.newIndex, body: query }),
      ]);
 
      const oldCount = oldResult.hits.total.value;
      const newCount = newResult.hits.total.value;
 
      // Allow 10% divergence (new analyzers may change results)
      const divergence = Math.abs(oldCount - newCount) / Math.max(oldCount, 1);
 
      if (divergence <= 0.1) {
        parityCases++;
      } else {
        divergentCases.push({ 
          query: JSON.stringify(query).slice(0, 100), 
          oldCount, 
          newCount 
        });
      }
    }
 
    const parityRate = parityCases / sampleQueries.length;
    const passed = parityRate >= 0.9; // 90% of queries have parity
 
    return {
      name: 'Query Parity',
      passed,
      severity: 'warning', // Not critical - intentional changes may cause divergence
      details: `${parityCases}/${sampleQueries.length} queries had result parity. \
        Divergent examples: ${JSON.stringify(divergentCases.slice(0, 3))}`,
    };
  }
 
  private async verifyPerformance(): Promise<CheckResult> {
    const testQueries = await this.loadPerformanceTestQueries();
    const results: Array<{ query: string; oldP99: number; newP99: number }> = [];
 
    for (const query of testQueries) {
      const oldLatencies: number[] = [];
      const newLatencies: number[] = [];
 
      // Run each query 10 times to get stable latency
      for (let i = 0; i < 10; i++) {
        const [oldResult, newResult] = await Promise.all([
          this.timeQuery(this.oldIndex, query),
          this.timeQuery(this.newIndex, query),
        ]);
        oldLatencies.push(oldResult);
        newLatencies.push(newResult);
      }
 
      results.push({
        query: JSON.stringify(query).slice(0, 50),
        oldP99: this.percentile(oldLatencies, 99),
        newP99: this.percentile(newLatencies, 99),
      });
    }
 
    // New index should not be more than 20% slower
    const regressions = results.filter(r => r.newP99 > r.oldP99 * 1.2);
    const passed = regressions.length === 0;
 
    return {
      name: 'Performance',
      passed,
      severity: 'warning',
      details: passed 
        ? 'No significant performance regressions detected'
        : `Regressions: ${JSON.stringify(regressions)}`,
    };
  }
}