Elasticsearch不停机切换(上云)方案

丝  论坛元老 | 2024-9-23 09:35:30 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1737|帖子 1737|积分 5211

怎样给飞行中的飞机换引擎?
   配景



  • 业务配景

    • 略

  • 技术配景

    • 线下集群40个索引左右，总数据量不大,不到100G
    • 因为ES承担的业务鉴权业务，以是不能接受停机割接

      • 尚有就是ES中数据来自各个业务方，推送的时机不定，也没有完备的重推机制，以是不能停机割接

    • 索引中根本都没有创建或者更新时间字段，纵然部分有，也没有用起来

      • 也就无法使用logstash的增量同步功能。

    • 盼望不进行业务改造，直接替换。
    • 虽然服务分为了读写服务，但通过读服务还是可以调用写入的API，通过写服务也可以调用读的API。

架构方案



  • 全量数据同步logstash
  • 脚步比对出来的差别数据，脚步补数

注意：


  • CLB及署理层的配置一定有冗余
  • 如果个CLB支撑不了，可以思量

    • 方式一：直接申请多个CLB,并将这多个CLB的地点配置到应用中
    • 方式二：先申请一个EIP，在EIP的后面配置多个CLB，这样应用只配置一个EIP的地点就可以了
    • 方式三：CLB直接升配到NLB

  • CLB文档
  • NLB文档
  • 准备两套CLB及署理层的缘故原由是：署理层是个Nginx集群，手动一台一台更新配置然后reload很慢，这时候数据写入的主ES是不确定的。
比对核心逻辑



  • 获取线下集群所有索引(跳过系统以是及不需要迁徙的索引)
  • 遍历第一步获取到的索引聚集

    • 获取线上、线下索引的文档总数，如果总数不一样,终止比对；
    • 如果总数一样，则通过search after(需要)分页分别从线上、线下获取数据比对。

注意：search_after的排序字段聚集有几个要求

  • 如果_id就是业务ID，则直接使用该字段；
  • 如果_id是ES自动生成的ID,则需要使用业务ID字段来排序(需要保证该业务ID索引内部不重复；如果不能保证，则需要添加其他字段来保证唯一；保证唯一的目的就是比对的两个索引在相同位置的文档就应该是一样的，不一样就是有问题)；
  • 如果无法找到能构建复合主键的字段，则需要将索引数据完备的拉到内存中，然后根据mapping将所有字段拼接构建组合ID，然后去重，再依次比对。(索引条数不一样的，也可以通过类似的方式来查找非常的缘故原由;采取这种简朴粗暴方式的缘故原由是：1、我们这种范例索引的数据量不大 2、这个比对程序其实就是个临时的工具，不会长期使用)
   模板、mapping、index setting这些都需要比对。
  比对核心代码

MapFlatUtil.java
  1. import java.util.*;
  3. /**
  4. * @Author jiankunking
  5. * @Date 2024/9/4 17:13
  6. * @Description:
  7. */
  8. public class MapFlatUtil {
  10.     static String PREFIX = ".";
  12.     public static Map<String, Object> flat(Map<String, Object> map) {
  13.         Map<String, Object> configMap = new LinkedHashMap<>();
  14.         map.entrySet().forEach(entry -> {
  15.             if (entry.getValue() instanceof Map) {
  16.                 Map<String, Object> subMap = flat(entry.getKey(), (Map<String, Object>) entry.getValue());
  17.                 if (!subMap.isEmpty()) {
  18.                     configMap.putAll(subMap);
  19.                 }
  20.             } else if (entry.getValue() instanceof List) {
  21.                 configMap.put(entry.getKey(), entry.getValue());
  22.             } else {
  23.                 configMap.put(entry.getKey(), entry.getValue() == null ? "" : String.valueOf(entry.getValue()));
  24.             }
  25.         });
  26.         return configMap;
  27.     }
  29.     private static Map<String, Object> flat(String parentNode, Map<String, Object> source) {
  30.         Map<String, Object> flatMap = new LinkedHashMap<>();
  31.         Set<Map.Entry<String, Object>> set = source.entrySet();
  32.         set.forEach(entity -> {
  33.             Object value = entity.getValue();
  34.             String key = entity.getKey();
  35.             String newKey = parentNode + PREFIX + key;
  36.             if (value instanceof Map) {
  37.                 flatMap.putAll(flat(newKey, (Map<String, Object>) value));
  38.             } else if (value instanceof List) {
  39.                 flatMap.put(newKey, value);
  40.             } else {
  41.                 flatMap.put(newKey, value == null ? "" : String.valueOf(value));
  42.             }
  43.         });
  44.         return flatMap;
  45.     }
  46. }
复制代码
MapCompareUtil.java
  1. import com.fasterxml.jackson.core.JsonProcessingException;
  2. import com.fasterxml.jackson.databind.ObjectMapper;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.util.ArrayList;
  6. import java.util.Comparator;
  7. import java.util.List;
  8. import java.util.Map;
  9. import java.util.stream.Collectors;
  11. import static com.jiankunking.branchcompare.es.SortUtil.mapComparator;
  13. /**
  14. * @Author jiankunking
  15. * @Date 2024/9/14 9:48
  16. * @Description:
  17. */
  18. @Slf4j
  19. public class MapCompareUtil {
  21.     public static boolean isMapEquals(Map<String, Object> offlineMap, Map<String, Object> onlineMap) throws JsonProcessingException {
  22.         offlineMap = MapFlatUtil.flat(offlineMap);
  23.         onlineMap = MapFlatUtil.flat(onlineMap);
  24.         if (offlineMap.size() != onlineMap.size()) {
  25.             return false;
  26.         }
  28.         for (Map.Entry<String, Object> offlineEntry : offlineMap.entrySet()) {
  29.             String offlineEntryKey = offlineEntry.getKey();
  30.             if (!onlineMap.containsKey(offlineEntryKey)) {
  31.                 return false;
  32.             }
  33.             Object offlineEntryValue = offlineEntry.getValue();
  34.             Object onlineEntryValue = onlineMap.get(offlineEntryKey);
  36.             Class offlineEntryValueClass = offlineEntryValue.getClass();
  37.             Class onlineEntryValueClass = onlineEntryValue.getClass();
  39.             if (offlineEntryValueClass != onlineEntryValueClass) {
  40.                 log.warn("value type not equals,offlineEntryValue:" + offlineEntryValueClass.getName() + ",onlineEntryValue:" + onlineEntryValueClass.getName());
  41.                 return false;
  42.             }
  43.             if (offlineEntryValue instanceof Map) {
  44.                 Map<String, Object> offlineMapValue = (Map<String, Object>) offlineEntryValue;
  45.                 Map<String, Object> onlineMapValue = (Map<String, Object>) onlineEntryValue;
  46.                 if (!isMapEquals(offlineMapValue, onlineMapValue)) {
  47.                     return false;
  48.                 }
  49.                 continue;
  50.             } else if (offlineEntryValue instanceof List) {
  51.                 List<Object> offlineList = (List<Object>) offlineEntryValue;
  52.                 List<Object> onlineList = (List<Object>) onlineEntryValue;
  53.                 if (offlineList.size() != onlineList.size()) {
  54.                     log.warn("list size not equals,offlineList:" + offlineList.size() + ",onlineList:" + onlineList.size());
  55.                     return false;
  56.                 }
  58.                 // List<Map>
  59.                 if (!offlineList.isEmpty() && offlineList.get(0) instanceof Map) {
  60.                     List<Map<String, Object>> offlineEntryValueTmp = (List<Map<String, Object>>) offlineEntryValue;
  61.                     List<Map<String, Object>> onlineEntryValueTmp = (List<Map<String, Object>>) onlineEntryValue;
  63.                     List<SortUtil.Sort> sorts = new ArrayList<>();
  64.                     // 按照map 的key 排序
  65.                     for (Map.Entry<String, Object> entry : offlineEntryValueTmp.get(0).entrySet()) {
  66.                         sorts.add(new SortUtil.Sort(entry.getKey(), SortUtil.Order.ASC));
  67.                     }
  69.                     List<Map<String, Object>> offlineEntryValueSorted = offlineEntryValueTmp.stream()
  70.                                                                                             .sorted(mapComparator(sorts))
  71.                                                                                             .collect(Collectors.toList());
  72.                     List<Map<String, Object>> onlineEntryValueSorted = onlineEntryValueTmp.stream()
  73.                                                                                           .sorted(mapComparator(sorts))
  74.                                                                                           .collect(Collectors.toList());
  75.                     for (int i = 0; i < offlineEntryValueSorted.size(); i++) {
  76.                         Object offlineListItem = offlineEntryValueSorted.get(i);
  77.                         Object onlineListItem = onlineEntryValueSorted.get(i);
  78.                         if (!isMapEquals((Map<String, Object>) offlineListItem, (Map<String, Object>) onlineListItem)) {
  79.                             return false;
  80.                         }
  81.                     }
  83.                 } else {
  84.                     // List<简单类型>
  85.                     offlineList.sort(Comparator.comparing(o -> o.toString()));
  86.                     onlineList.sort(Comparator.comparing(o -> o.toString()));
  87.                     for (int i = 0; i < offlineList.size(); i++) {
  88.                         Object offlineListItem = offlineList.get(i);
  89.                         Object onlineListItem = onlineList.get(i);
  90.                         if (!simpleObjectEquals(offlineListItem, onlineListItem)) {
  91.                             log.warn("list item not equals,offlineListItem:" + offlineListItem + ",onlineListItem:" + onlineListItem);
  92.                             return false;
  93.                         }
  94.                     }
  95.                 }
  96.                 continue;
  97.             }
  98.             if (!simpleObjectEquals(offlineEntryValue, onlineEntryValue)) {
  99.                 log.warn("map value not equals,offlineEntryValue:" + offlineEntryValue + ",onlineEntryValue:" + onlineEntryValue);
  100.                 return false;
  101.             }
  102.         }
  103.         return true;
  104.     }
  106.     // 只能处理简单对象 不能处理Map List等复杂类型
  107.     private static boolean simpleObjectEquals(Object o1, Object o2) throws JsonProcessingException {
  108.         String offlineJson = new ObjectMapper().writeValueAsString(o1);
  109.         String onlineJson = new ObjectMapper().writeValueAsString(o2);
  110.         if (offlineJson.equals(onlineJson)) {
  111.             return true;
  112.         }
  113.         return false;
  114.     }
  115. }
复制代码
SortUtil.java
  1. import java.math.BigDecimal;
  2. import java.util.*;
  3. import java.util.stream.Collectors;
  5. /**
  6. * @Author jiankunking
  7. * @Date 2024/9/5 14:00
  8. * @Description: https://gist.github.com/IOsetting/25ca8d70c12c11390113d343f666cd6e
  9. */
  10. public class SortUtil {
  11.     public enum Order {ASC, DESC}
  13.     /**
  14.      * @param sorts keys and sort direction
  15.      * @return sorted list
  16.      */
  17.     public static Comparator<Map<String, Object>> mapComparator(List<Sort> sorts) {
  18.         return (o1, o2) -> {
  19.             int ret = 0;
  20.             for (Sort sort : sorts) {
  21.                 Object v1 = o1.get(sort.field);
  22.                 Object v2 = o2.get(sort.field);
  23.                 ret = singleCompare(v1, v2, sort.order == Order.ASC);
  24.                 if (ret != 0) {
  25.                     break;
  26.                 }
  27.             }
  28.             return ret;
  29.         };
  30.     }
  32.     public static class Sort {
  33.         public String field;
  34.         public Order order;
  36.         public Sort(String field, Order order) {
  37.             this.field = field;
  38.             this.order = order;
  39.         }
  40.     }
  42.     private static int singleCompare(Object ao, Object bo, boolean asc) {
  43.         int ret;
  44.         if (ao == null && bo == null) {
  45.             ret = 0;
  46.         } else if (ao == null) {
  47.             ret = -1;
  48.         } else if (bo == null) {
  49.             ret = 1;
  50.         } else if (ao instanceof BigDecimal) {
  51.             ret = ((BigDecimal) ao).compareTo((BigDecimal) bo);
  52.         } else if (ao instanceof Number) {
  53.             if (((Number) ao).doubleValue() != ((Number) bo).doubleValue()) {
  54.                 ret = ((Number) ao).doubleValue() > ((Number) bo).doubleValue() ? 1 : -1;
  55.             } else {
  56.                 ret = 0;
  57.             }
  58.         } else if (ao instanceof Date) {
  59.             ret = ((Date) ao).compareTo((Date) bo);
  60.         } else {
  61.             ret = String.valueOf(ao).compareTo(String.valueOf(bo));
  62.         }
  63.         if (!asc) {
  64.             return -ret;
  65.         }
  66.         return ret;
  67.     }
  69.     public static void main(String[] args) {
  70.         List<Map<String, Object>> list = new ArrayList<>();
  71.         List<Sort> sorts = new ArrayList<>();
  73.         List<Map<String, Object>> sorted = list.stream()
  74.                                                .sorted(mapComparator(sorts))
  75.                                                .collect(Collectors.toList());
  76.         for (Map<String, Object> map : sorted) {
  77.             System.out.println(map.get("somekey"));
  78.         }
  79.     }
  80. }
复制代码
EsQueryUtil.java
  1. public static SearchResponse searchAfterByMultiFields(RestHighLevelClient restHighLevelClient, String indexName, List<String> searchAfterSortFields, List<Object> searchAfterValues, int size) throws IOException {
  2.         SearchSourceBuilder builder = new SearchSourceBuilder();
  3.         builder.size(size);
  4.         builder.trackTotalHits(true);
  6.         builder.query(QueryBuilders.matchAllQuery());
  8.         // USING SEARCH AFTER
  9.         if (searchAfterValues != null && !searchAfterValues.isEmpty()) {
  10.             builder.searchAfter(searchAfterValues.toArray());
  11.         }
  12.         for (String sortField : searchAfterSortFields) {
  13.             builder.sort(sortField, SortOrder.ASC);
  14.         }
  16.         SearchRequest searchRequest = new SearchRequest();
  17.         searchRequest.indices(indexName);
  18.         searchRequest.source(builder);
  19.         // log.info(searchRequest.toString());
  20.         log.info(searchRequest.source().toString());
  22.         SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
  24.         return response;
  25.     }
  27.     static List<Object> getSearchAfterValues(List<String> searchAfterSortFields, SearchHit hit) {
  28.         List<Object> searchAfterValues = new ArrayList<>(searchAfterSortFields.size());
  29.         Map<String, Object> map = hit.getSourceAsMap();
  30.         for (String field : searchAfterSortFields) {
  31.             if (field.equals("_id")) {
  32.                 searchAfterValues.add(hit.getId());
  33.             } else {
  34.                 searchAfterValues.add(map.get(field));
  35.             }
  36.         }
  37.         return searchAfterValues;
  39.     }
复制代码
反思



  • 要拉通全流程及相关人员，查对每个可能出现的问题及应对方案
  • 有些东西不能因为是临时的就放松鉴戒性

    • 好比本次署理层申请的机器是有两块的盘：1、一个50G的系统盘 2、一个500G的数据盘；但最终落地的时候云厂商同砚还是把nginx的访问日记落到了系统盘，导致系统盘满了，系统受到的影响。

      • 这个500G的盘其时还讨论过，要用来存储访问日记，防止机器磁盘写满。

    • 任务列表也梳理了署理层遇到问题要发送告警，但没有一一核实，导致系统盘满的时候，没有第一时间收到告警。
    • 只要是在核心链路上的，不管是不是临时的，必须一一测试、验证。


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

丝

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表