Zero-Downtime Elasticsearch Cloud Migration: Data Sync, Document Comparison, and Cutover

Posted on Edited on Views: Word count in article: 3.6k Reading time ≈ 3 mins.
A zero-downtime Elasticsearch cloud migration: ~40 indexes under 100 GB with no usable update-time fields. Data sync, document comparison, mapping checks, and traffic cutover.

How do you change the engines on a plane mid-flight? This post walks through a complete zero-downtime Elasticsearch migration to the cloud, covering the core problems: data migration, dual-write synchronization, and a smooth cutover.

How do you swap the engines of a plane while it is flying?

Background

  • Business context
    • omitted
  • Technical context
    • The offline cluster has around 40 indexes; the total data volume is small — under 100 GB.
    • Elasticsearch carries authentication traffic for the business, so a downtime cutover was not acceptable.
      • On top of that, the data in Elasticsearch comes from many different services, pushed at unpredictable times, with no complete re-push mechanism — so downtime was off the table.
    • The indexes essentially have no create or update time fields; even where a few had them, they were never used.
      • This ruled out Logstash’s incremental sync.
    • We wanted a straight swap with no changes to the business services.
    • The service is split into a read service and a write service, but the read service can still call write APIs and vice versa.

Architecture

  • Full data sync with Logstash
  • Script-based comparison to surface the differences, then script-based backfill

Zero-downtime Elasticsearch cutover (cloud migration) architecture

Notes:

  • The CLB and proxy layer configurations must have redundancy built in.
  • If a single CLB cannot carry the load, consider one of the following:
    • Option 1: Request multiple CLBs and configure all of their addresses in the application.
    • Option 2: Request a single EIP and put multiple CLBs behind it, so the application only needs to know one address.
    • Option 3: Upgrade the CLB straight to an NLB.
  • CLB documentation
  • NLB documentation
  • The reason for preparing two sets of CLB and proxy layer: the proxy layer is an Nginx cluster, and updating the configuration on each machine one by one and reloading is slow — and during that window it is unclear which Elasticsearch cluster is receiving the writes.

Core comparison logic

  • Fetch all indexes in the offline cluster (skipping system indexes and indexes that do not need to be migrated)
  • Iterate over the index set from step 1
    • Fetch the document counts of the online and offline indexes; if the totals differ, abort the comparison.
    • If the totals match, use search_after to page through both the online and offline indexes and compare the documents.

Note: the sort field set for search_after has a few requirements:

  1. If _id is the business ID, use that field directly.
  2. If _id is an ES-generated ID, sort by the business ID field (that field must be unique within the index; if uniqueness cannot be guaranteed, add other fields to make it unique; the point of uniqueness is that the two indexes being compared should have identical documents at the same position — anything different means there is a problem);
  3. If no set of fields can form a composite primary key, pull the entire index into memory, concatenate all fields according to the mapping to build a composite ID, deduplicate, and then compare one by one. (When index counts differ, a similar approach can also be used to find the cause. The reason for this blunt approach: 1. our indexes are small; 2. this comparison program is a temporary tool, not something that will be used long-term.)

Templates, mappings, index settings — all of these need to be compared too.

Core comparison code

MapFlatUtil.java

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
import java.util.*;

/**
 * @Author jiankunking
 * @Date 2024/9/4 17:13
 * @Description:
 */
public class MapFlatUtil {

    static String PREFIX = ".";

    public static Map<String, Object> flat(Map<String, Object> map) {
        Map<String, Object> configMap = new LinkedHashMap<>();
        map.entrySet().forEach(entry -> {
            if (entry.getValue() instanceof Map) {
                Map<String, Object> subMap = flat(entry.getKey(), (Map<String, Object>) entry.getValue());
                if (!subMap.isEmpty()) {
                    configMap.putAll(subMap);
                }
            } else if (entry.getValue() instanceof List) {
                configMap.put(entry.getKey(), entry.getValue());
            } else {
                configMap.put(entry.getKey(), entry.getValue() == null ? "" : String.valueOf(entry.getValue()));
            }
        });
        return configMap;
    }

    private static Map<String, Object> flat(String parentNode, Map<String, Object> source) {
        Map<String, Object> flatMap = new LinkedHashMap<>();
        Set<Map.Entry<String, Object>> set = source.entrySet();
        set.forEach(entity -> {
            Object value = entity.getValue();
            String key = entity.getKey();
            String newKey = parentNode + PREFIX + key;
            if (value instanceof Map) {
                flatMap.putAll(flat(newKey, (Map<String, Object>) value));
            } else if (value instanceof List) {
                flatMap.put(newKey, value);
            } else {
                flatMap.put(newKey, value == null ? "" : String.valueOf(value));
            }
        });
        return flatMap;
    }
}

MapCompareUtil.java

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
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.jiankunking.branchcompare.es.SortUtil.mapComparator;

/**
 * @Author jiankunking
 * @Date 2024/9/14 9:48
 * @Description:
 */
@Slf4j
public class MapCompareUtil {

    public static boolean isMapEquals(Map<String, Object> offlineMap, Map<String, Object> onlineMap) throws JsonProcessingException {
        offlineMap = MapFlatUtil.flat(offlineMap);
        onlineMap = MapFlatUtil.flat(onlineMap);
        if (offlineMap.size() != onlineMap.size()) {
            return false;
        }

        for (Map.Entry<String, Object> offlineEntry : offlineMap.entrySet()) {
            String offlineEntryKey = offlineEntry.getKey();
            if (!onlineMap.containsKey(offlineEntryKey)) {
                return false;
            }
            Object offlineEntryValue = offlineEntry.getValue();
            Object onlineEntryValue = onlineMap.get(offlineEntryKey);

            Class offlineEntryValueClass = offlineEntryValue.getClass();
            Class onlineEntryValueClass = onlineEntryValue.getClass();

            if (offlineEntryValueClass != onlineEntryValueClass) {
                log.warn("value type not equals,offlineEntryValue:" + offlineEntryValueClass.getName() + ",onlineEntryValue:" + onlineEntryValueClass.getName());
                return false;
            }
            if (offlineEntryValue instanceof Map) {
                Map<String, Object> offlineMapValue = (Map<String, Object>) offlineEntryValue;
                Map<String, Object> onlineMapValue = (Map<String, Object>) onlineEntryValue;
                if (!isMapEquals(offlineMapValue, onlineMapValue)) {
                    return false;
                }
                continue;
            } else if (offlineEntryValue instanceof List) {
                List<Object> offlineList = (List<Object>) offlineEntryValue;
                List<Object> onlineList = (List<Object>) onlineEntryValue;
                if (offlineList.size() != onlineList.size()) {
                    log.warn("list size not equals,offlineList:" + offlineList.size() + ",onlineList:" + onlineList.size());
                    return false;
                }

                // List<Map>
                if (!offlineList.isEmpty() && offlineList.get(0) instanceof Map) {
                    List<Map<String, Object>> offlineEntryValueTmp = (List<Map<String, Object>>) offlineEntryValue;
                    List<Map<String, Object>> onlineEntryValueTmp = (List<Map<String, Object>>) onlineEntryValue;

                    List<SortUtil.Sort> sorts = new ArrayList<>();
                    // 按照map 的key 排序
                    for (Map.Entry<String, Object> entry : offlineEntryValueTmp.get(0).entrySet()) {
                        sorts.add(new SortUtil.Sort(entry.getKey(), SortUtil.Order.ASC));
                    }

                    List<Map<String, Object>> offlineEntryValueSorted = offlineEntryValueTmp.stream()
                                                                                             .sorted(mapComparator(sorts))
                                                                                             .collect(Collectors.toList());
                    List<Map<String, Object>> onlineEntryValueSorted = onlineEntryValueTmp.stream()
                                                                                           .sorted(mapComparator(sorts))
                                                                                           .collect(Collectors.toList());
                    for (int i = 0; i < offlineEntryValueSorted.size(); i++) {
                        Object offlineListItem = offlineEntryValueSorted.get(i);
                        Object onlineListItem = onlineEntryValueSorted.get(i);
                        if (!isMapEquals((Map<String, Object>) offlineListItem, (Map<String, Object>) onlineListItem)) {
                            return false;
                        }
                    }

                } else {
                    // List<简单类型>
                    offlineList.sort(Comparator.comparing(o -> o.toString()));
                    onlineList.sort(Comparator.comparing(o -> o.toString()));
                    for (int i = 0; i < offlineList.size(); i++) {
                        Object offlineListItem = offlineList.get(i);
                        Object onlineListItem = onlineList.get(i);
                        if (!simpleObjectEquals(offlineListItem, onlineListItem)) {
                            log.warn("list item not equals,offlineListItem:" + offlineListItem + ",onlineListItem:" + onlineListItem);
                            return false;
                        }
                    }
                }
                continue;
            }
            if (!simpleObjectEquals(offlineEntryValue, onlineEntryValue)) {
                log.warn("map value not equals,offlineEntryValue:" + offlineEntryValue + ",onlineEntryValue:" + onlineEntryValue);
                return false;
            }
        }
        return true;
    }

    // 只能处理简单对象 不能处理Map List等复杂类型
    private static boolean simpleObjectEquals(Object o1, Object o2) throws JsonProcessingException {
        String offlineJson = new ObjectMapper().writeValueAsString(o1);
        String onlineJson = new ObjectMapper().writeValueAsString(o2);
        if (offlineJson.equals(onlineJson)) {
            return true;
        }
        return false;
    }
}

SortUtil.java

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
import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author jiankunking
 * @Date 2024/9/5 14:00
 * @Description: https://gist.github.com/IOsetting/25ca8d70c12c11390113d343f666cd6e
 */
public class SortUtil {
    public enum Order {ASC, DESC}

    /**
     * @param sorts keys and sort direction
     * @return sorted list
     */
    public static Comparator<Map<String, Object>> mapComparator(List<Sort> sorts) {
        return (o1, o2) -> {
            int ret = 0;
            for (Sort sort : sorts) {
                Object v1 = o1.get(sort.field);
                Object v2 = o2.get(sort.field);
                ret = singleCompare(v1, v2, sort.order == Order.ASC);
                if (ret != 0) {
                    break;
                }
            }
            return ret;
        };
    }

    public static class Sort {
        public String field;
        public Order order;

        public Sort(String field, Order order) {
            this.field = field;
            this.order = order;
        }
    }

    private static int singleCompare(Object ao, Object bo, boolean asc) {
        int ret;
        if (ao == null && bo == null) {
            ret = 0;
        } else if (ao == null) {
            ret = -1;
        } else if (bo == null) {
            ret = 1;
        } else if (ao instanceof BigDecimal) {
            ret = ((BigDecimal) ao).compareTo((BigDecimal) bo);
        } else if (ao instanceof Number) {
            if (((Number) ao).doubleValue() != ((Number) bo).doubleValue()) {
                ret = ((Number) ao).doubleValue() > ((Number) bo).doubleValue() ? 1 : -1;
            } else {
                ret = 0;
            }
        } else if (ao instanceof Date) {
            ret = ((Date) ao).compareTo((Date) bo);
        } else {
            ret = String.valueOf(ao).compareTo(String.valueOf(bo));
        }
        if (!asc) {
            return -ret;
        }
        return ret;
    }

    public static void main(String[] args) {
        List<Map<String, Object>> list = new ArrayList<>();
        List<Sort> sorts = new ArrayList<>();

        List<Map<String, Object>> sorted = list.stream()
                                               .sorted(mapComparator(sorts))
                                               .collect(Collectors.toList());
        for (Map<String, Object> map : sorted) {
            System.out.println(map.get("somekey"));
        }
    }
}

EsQueryUtil.java

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
public static SearchResponse searchAfterByMultiFields(RestHighLevelClient restHighLevelClient, String indexName, List<String> searchAfterSortFields, List<Object> searchAfterValues, int size) throws IOException {
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.size(size);
        builder.trackTotalHits(true);

        builder.query(QueryBuilders.matchAllQuery());

        // USING SEARCH AFTER
        if (searchAfterValues != null && !searchAfterValues.isEmpty()) {
            builder.searchAfter(searchAfterValues.toArray());
        }
        for (String sortField : searchAfterSortFields) {
            builder.sort(sortField, SortOrder.ASC);
        }

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(indexName);
        searchRequest.source(builder);
        // log.info(searchRequest.toString());
        log.info(searchRequest.source().toString());

        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        return response;
    }

    static List<Object> getSearchAfterValues(List<String> searchAfterSortFields, SearchHit hit) {
        List<Object> searchAfterValues = new ArrayList<>(searchAfterSortFields.size());
        Map<String, Object> map = hit.getSourceAsMap();
        for (String field : searchAfterSortFields) {
            if (field.equals("_id")) {
                searchAfterValues.add(hit.getId());
            } else {
                searchAfterValues.add(map.get(field));
            }
        }
        return searchAfterValues;

    }

Postmortem

  • Walk the entire flow end to end with everyone involved, and verify each potential problem together with its response plan.
  • Just because something is temporary does not mean you can relax your guard.
    • The machines we provisioned for the proxy layer came with two disks: a 50 GB system disk and a 500 GB data disk. But when it came time to actually deploy, the cloud vendor’s engineers still put the Nginx access logs on the system disk, which filled it up and degraded the system.
      • We had specifically discussed using that 500 GB disk for access logs, precisely to avoid filling up the machine’s disk.
    • We had also listed in the task plan that the proxy layer should send alerts when problems occurred, but we never verified them one by one — so when the system disk filled up, no alert arrived in time.
    • Anything on the critical path — temporary or not — must be tested and verified, one item at a time.

This article is the English edition of a post originally published in Chinese.