背景

业务所有的服务器日志都是通过filebeat进行采集，然后写入到一个公共的ES集群中。因为当前使用的集群无法继续扩容了并且版本也较低(5.6.4), 所以需要把集群迁移到一个新的规模更大的集群，并且升级一下ES的版本，升级到6.4.3.

方案制定

迁移的需求是：

1. 日志数据不能停止写入到ES
2. 日志查询不受影响，延迟不能超过1分钟

参考之前写的关于数据迁移的文章Elasticsearch数据迁移与集群容灾，制定出的迁移方案为：

1. 先使用logstash或者snapshot全量同步一次数据到新集群中
2. 使用logstash追平当天的日志索引后，查询入口切换到新的ES集群
3. 日志写入入口切换到新的ES集群

实施步骤

1. 新建6.4.3版本ES集群

如果旧的日志集群处理日志时使用了ingest pipeline， 新的集群也需要同样配置pipeline。

2.使用logstash/snapshot全同步数据

如果数据规模较小，比如几十GB, 则可以使用logstash进行全量同步， logstash配置文件如下：

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "*"
        docinfo => true
        size => 5000
        scroll => "5m"
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
        pipeline => "timezone-pipeline"
    }
}

实施过程中遇到的两个问题：

• 源集群(5.6.4)的.kibana索引也被同步到6.4.3版本的新集群了，造成不兼容，需要在新集群中删除掉.kibana索引
• 源集群中的日志时间戳字段@timestamp是增加了+08:00时区后缀的，经过上述迁移后，同步到新集群中的日志数据中@timestamp没有了时区后缀，这个问题在logstash侧进行了尝试没有解决，所以通过在es侧增加ingest pipeline进行解决：

    "description": "timezone-pipeline",
    "processors": [
      {
        "date": {
          "field": "@timestamp",
          "formats": [
            "ISO8601"
          ],
          "timezone": "Asia/Shanghai",
          "ignore_failure": true
        }
      }
    ],
    "on_failure": [
      {
        "set": {
          "field": "error",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ]

如果数据规模较大，几百GB以上，则可以使用snapshot进行一次全量的迁移，速度较快。

3. 记录新集群中当天索引中数据的最新时间戳

存量的旧的索引不会再写入了，而当天的索引还在持续写入，在步骤2的全量同步数据完成之后(logstash执行完毕后会自动终止进程)， 需要查询出当天索引的数据中已经同步完成的最新的时间戳，在新的集群中执行查询：

GET es-runlog-2019-11-20/_search
{
  "query": {
    "match_all": {}
  },
  "size": 1,
  "sort": [
    {
      "@timestamp": "desc"
    }
  ]
}

记执行上述查询获取到的时间戳为start.

4.增量迁移当天的索引

使用logstash增量迁移当天的索引，logstash配置如下：

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "es-runlog-2019.11.20"
        query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
        size => 5000
        scroll => "5m"
        docinfo => true
        schedule => "* * * * *" #定时任务，每分钟执行一次
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
        pipeline => "timezone-pipeline"
    }
}

上述配置，每分钟执行一次，从源集群中拉取5分钟前到当前分钟的所有数据，同步到新的集群中；因为查询的粒度为分钟，所以每次执行定时任务查询时会有一部分重叠的数据，所以需要在output中配置document_id参数避免重复写入到新集群中。

实施过程中遇到的问题有：

1. 用于运行logstash的机器的规格要比较大，因为logstash比较消耗内存和cpu，机器性能不够，很可能出现数据同步延迟增大
2. 可以通过比较新旧集群当天的索引每分钟doc数据量，判断同步的延迟情况，如果延迟较大，可以通过调整logstash配置或者使用更大的机器运行logstash确保同步过程顺利进行

5. 记录开始迁移的时间

在新的集群中执行以下查询，记录开始进行增量迁移的时间戳：

GET es-runlog-2019-11-20/_search

{
  "query": {
    "range": {
      "@timestamp": {
        "gt": "{start}"
      }
    }
  },
  "size": 1,
  "sort": [
    {
      "@timestamp": "asc"
    }
  ]
}

记获取到的时间戳为end.

6. 追平start和end之间的数据

使用logstash从源集群中获取start和end之间的日志数据，同步到新集群中，配置文件如下：

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "es-runlog-2019.11.20"
        query => '{"query":{"range":{"@timestamp":{"gte":"start","lt":"end"}}}}'
        docinfo => true
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
         document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
         pipeline => "timezone-pipeline"
    }
}

7. 持续观察数据同步过程是否稳定

待步骤6的数据追平过程结束之后，需要持续观察步骤5的增量迁移数据的情况是否稳定，待一段时间，比如几个小时之后，仍然可以稳定的进行同步，此时可以把日志的查询入口切换到新集群中，之后再把数据写入入切换到新集群中，至此，一次在线跨集群迁移数据实施过程完毕。