1 概述
logstash是一个对数据进行抽取、转换、输出的工具，能对接多种数据源和目标数据。本文介绍通过它来同步elasticsearch的数据。
2 环境
实验仅仅需要一台logstash机器和两台elasticsearch机器（elasticsearch v7.1.0）。本文用docker来模拟，使用docker-compose来启动elasticsearch容器。
elasticsearch机器A是192.168.243.128:9200。
elasticsearch机器B是192.168.243.128:19200。
2.1 安装elasticsearch机器A
2.1.1 编写docker-compose.yml
version: '2.2'
services:
cerebro:
image: lmenezes/cerebro:0.8.3
container_name: cerebro
ports:
- "9000:9000"
command:
- -Dhosts.0.host=http://elasticsearch:9200
kibana:
image: docker.elastic.co/kibana/kibana:7.1.0
container_name: kibana7
environment:
- I18N_LOCALE=zh-CN
- XPACK_GRAPH_ENABLED=true
- TIMELION_ENABLED=true
- XPACK_MONITORING_COLLECTION_ENABLED="true"
ports:
- "5601:5601"
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:7.1.0
container_name: es7_01
environment:
- cluster.name=xttblog
- node.name=es7_01
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- discovery.seed_hosts=es7_01
- cluster.initial_master_nodes=es7_01,es7_02
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- es7data1:/usr/share/elasticsearch/data
ports:
- 9200:9200
elasticsearch2:
image: docker.elastic.co/elasticsearch/elasticsearch:7.1.0
container_name: es7_02
environment:
- cluster.name=xttblog
- node.name=es7_02
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- discovery.seed_hosts=es7_01
- cluster.initial_master_nodes=es7_01,es7_02
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- es7data2:/usr/share/elasticsearch/data
volumes:
es7data1:
driver: local
es7data2:
driver: local
2.1.2 修改vm.max_map_count
cat >> /etc/sysctl.conf << EOF
vm.max_map_count = 2621440
EOF
sysctl -p
2.1.3 启动
docker-compose up -d
2.1.3 导入样例数据
localhost:5601，用于访问kibana并导入样例数据
localhost:9000，用于访问cerebro
2.2 安装elasticsearch机器B
2.2.1 编写docker-compose.yml
version: '2.2'
services:
cerebro:
image: lmenezes/cerebro:0.8.3
container_name: cerebro-2
ports:
- "19000:9000"
command:
- -Dhosts.0.host=http://elasticsearch:9200
kibana:
image: docker.elastic.co/kibana/kibana:7.1.0
container_name: kibana7-2
environment:
- I18N_LOCALE=zh-CN
- XPACK_GRAPH_ENABLED=true
- TIMELION_ENABLED=true
- XPACK_MONITORING_COLLECTION_ENABLED="true"
ports:
- "15601:5601"
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:7.1.0
container_name: es7_03
environment:
- cluster.name=xttblog
- node.name=es7_03
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- discovery.seed_hosts=es7_03
- cluster.initial_master_nodes=es7_03,es7_04
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- es7data3:/usr/share/elasticsearch/data
ports:
- 19200:9200
elasticsearch2:
image: docker.elastic.co/elasticsearch/elasticsearch:7.1.0
container_name: es7_04
environment:
- cluster.name=xttblog
- node.name=es7_04
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- discovery.seed_hosts=es7_03
- cluster.initial_master_nodes=es7_03,es7_04
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- es7data4:/usr/share/elasticsearch/data
volumes:
es7data3:
driver: local
es7data4:
driver: local
2.2.2 修改vm.max_map_count
cat >> /etc/sysctl.conf << EOF
vm.max_map_count = 2621440
EOF
sysctl -p
2.2.3 启动
docker-compose up -d
2.2 logstash机器上准备python环境
在centos7机器上执行如下命令，可安装python环境：
yum install python2
yum install python-pip
由于python脚本需要导入requests模块哈yaml模块，因此需要执行如下命令在机器上安装python模块：
pip install pyyaml
pip install requests
2.2 logstash机器上准备python脚本
2.2.1 准备migrateConfig.yaml
本文件用于描述源端和目标端是谁，是否开启SSL连接。
一般只需要修改src_ip、dest_ip。
cat > /tmp/migrateConfig.yaml << EOF
es_cluster_new:
# 源集群的名称
clustername: es_cluster_new
# 源Elasticsearch集群的访问地址，加上“http://”。
src_ip: http://x.x.x.x:9200
# 访问源Elasticsearch集群的用户名和密码，如果为非安全集群则设置为""。
src_username: ""
src_password: ""
# 目标Elasticsearch集群的访问地址，加上“http://”。
dest_ip: http://x.x.x.x:9200
# 访问目标Elasticsearch集群的用户名和密码，如果为非安全集群则设置为""。
dest_username: ""
dest_password: ""
# only_mapping 可以不定义，默认值为false,需要搭配“migrateMapping.py”使用，表示是否只处理这个文件中mapping地址的索引。
# 当设置成false时，则迁移源集群中除“.kibana”和“.*”之外的所有索引数据。
# 当设置成true时，则只迁移源集群中和下面mapping的key一致的索引数据；迁移过程中会将索引名称与下面的mapping匹配，如果匹配一致，则使用mapping的value作为目标集群的索引名称；如果匹配不到，则使用源集群原始的索引名称。
only_mapping: false
# 当only_mapping = true时，mapping用于设置要迁移的索引，key为源集群的索引名字，value为目标集群的索引名字。
mapping:
test_index_1: test_index_1
# only_compare_index 可以不定义，默认值为false,需要搭配“checkIndices.py”使用，当设置为false会比较所有的索引和文档数量，当设置为true只比较索引数量。
only_compare_index: false
EOF
2.2.1 准备migrateTemplate.py
用于迁移索引模板，此文件不需要修改，直接复制。
cat > /tmp/migrateTemplate.py << EOF
# -*- coding:UTF-8 -*-
import sys
import yaml
import requests
import json
import os
def printDividingLine():
print("<=============================================================>")
def loadConfig(argv):
if argv is None or len(argv) != 2:
config_yaml = "migrateConfig.yaml"
else:
config_yaml = argv[1]
config_file = open(config_yaml)
# config = yaml.load(config_file, Loader=yaml.FullLoader)
return yaml.load(config_file)
def put_template_to_target(url, template, cluster, template_name, dest_auth=None):
headers = {'Content-Type': 'application/json'}
create_resp = requests.put(url, headers=headers, data=json.dumps(template), auth=dest_auth, verify=False)
if not os.path.exists("templateLogs"):
os.makedirs("templateLogs")
if create_resp.status_code != 200:
print(
"create template " + url + " failed with response: " + str(
create_resp) + ", source template is " + template_name)
print(create_resp.text)
filename = "templateLogs/" + str(cluster) + "#" + template_name
with open(filename + ".json", "w") as f:
json.dump(template, f)
return False
else:
return True
def main(argv):
requests.packages.urllib3.disable_warnings()
print("begin to migration template!")
config = loadConfig(argv)
src_clusters = config.keys()
print("process cluster name:")
for name in src_clusters:
print(name)
print("cluster total number:" + str(src_clusters.__len__()))
for name, value in config.items():
printDividingLine()
source_user = value["src_username"]
source_passwd = value["src_password"]
source_auth = None
if source_user != "":
source_auth = (source_user, source_passwd)
dest_user = value["dest_username"]
dest_passwd = value["dest_password"]
dest_auth = None
if dest_user != "":
dest_auth = (dest_user, dest_passwd)
print("start to process cluster name：" + name)
source_url = value["src_ip"] + "/_template"
response = requests.get(source_url, auth=source_auth, verify=False)
if response.status_code != 200:
print("*** get all template failed. resp statusCode:" + str(
response.status_code) + " response is " + response.text)
continue
all_template = response.json()
migrate_itemplate = []
for template in all_template.keys():
if template.startswith(".") or template == "logstash":
continue
if "index_patterns" in all_template[template]:
for t in all_template[template]["index_patterns"]:
# if "kibana" in template:
if t.startswith("."):
continue
migrate_itemplate.append(template)
for template in migrate_itemplate:
dest_index_url = value["dest_ip"] + "/_template/" + template
result = put_template_to_target(dest_index_url, all_template[template], name, template, dest_auth)
if result is True:
print('[success] delete success, cluster: %-10s, template %-10s ' % (str(name), str(template)))
else:
print('[failure] delete failure, cluster: %-10s, template %-10s ' % (str(name), str(template)))
if __name__ == '__main__':
main(sys.argv)
EOF
2.2.1 准备migrateMapping.py
脚本用于迁移索的表结构，此文件不需要修改，直接复制。
cat > /tmp/migrateMapping.py << EOF
# -*- coding:UTF-8 -*-
import sys
import yaml
import requests
import re
import json
import os
def printDividingLine():
print("<=============================================================>")
def loadConfig(argv):
if argv is None or len(argv) != 2:
config_yaml = "migrateConfig.yaml"
else:
config_yaml = argv[1]
config_file = open(config_yaml)
# config = yaml.load(config_file, Loader=yaml.FullLoader)
return yaml.load(config_file)
def get_cluster_version(url, auth=None):
response = requests.get(url, auth=auth)
if response.status_code != 200:
print("*** get ElasticSearch message failed. resp statusCode:" + str(
response.status_code) + " response is " + response.text)
return False
cluster = response.json()
version = cluster["version"]["number"]
return True
def process_mapping(index_mapping, dest_index):
# remove unnecessary keys
del index_mapping["settings"]["index"]["provided_name"]
del index_mapping["settings"]["index"]["uuid"]
del index_mapping["settings"]["index"]["creation_date"]
del index_mapping["settings"]["index"]["version"]
if "lifecycle" in index_mapping["settings"]["index"]:
del index_mapping["settings"]["index"]["lifecycle"]
# check alias
aliases = index_mapping["aliases"]
for alias in list(aliases.keys()):
if alias == dest_index:
print(
"source index " + dest_index + " alias " + alias + " is the same as dest_index name, will remove this alias.")
del index_mapping["aliases"][alias]
# if index_mapping["settings"]["index"].has_key("lifecycle"):
if "lifecycle" in index_mapping["settings"]["index"]:
lifecycle = index_mapping["settings"]["index"]["lifecycle"]
opendistro = {"opendistro": {"index_state_management":
{"policy_id": lifecycle["name"],
"rollover_alias": lifecycle["rollover_alias"]}}}
index_mapping["settings"].update(opendistro)
# index_mapping["settings"]["opendistro"]["index_state_management"]["rollover_alias"] = lifecycle["rollover_alias"]
del index_mapping["settings"]["index"]["lifecycle"]
# replace synonyms_path
if "analysis" in index_mapping["settings"]["index"]:
analysis = index_mapping["settings"]["index"]["analysis"]
if "filter" in analysis:
filter = analysis["filter"]
if "my_synonym_filter" in filter:
my_synonym_filter = filter["my_synonym_filter"]
if "synonyms_path" in my_synonym_filter:
index_mapping["settings"]["index"]["analysis"]["filter"]["my_synonym_filter"][
"synonyms_path"] = "/rds/datastore/elasticsearch/v7.10.2/package/elasticsearch-7.10.2/plugins/analysis-dynamic-synonym/config/synonyms.txt"
return index_mapping
def getAlias(source, source_auth):
# get all indices
response = requests.get(source + "/_alias", auth=source_auth)
if response.status_code != 200:
print("*** get all index failed. resp statusCode:" + str(
response.status_code) + " response is " + response.text)
exit()
all_index = response.json()
system_index = []
create_index = []
for index in list(all_index.keys()):
if (index.startswith(".")):
system_index.append(index)
else:
create_index.append(index)
return system_index, create_index
def put_mapping_to_target(url, mapping, cluster, source_index, dest_auth=None):
headers = {'Content-Type': 'application/json'}
create_resp = requests.put(url, headers=headers, data=json.dumps(mapping), auth=dest_auth, verify=False)
if not os.path.exists("mappingLogs"):
os.makedirs("mappingLogs")
if create_resp.status_code != 200:
print(
"create index " + url + " failed with response: " + str(create_resp) +
", source index is " + str(source_index))
print(create_resp.text)
filename = "mappingLogs/" + str(cluster) + "#" + str(source_index)
with open(filename + ".json", "w") as f:
json.dump(mapping, f)
return False
else:
return True
def main(argv):
requests.packages.urllib3.disable_warnings()
print("begin to migrate index mapping!")
config = loadConfig(argv)
src_clusters = config.keys()
print("begin to process cluster name :")
for name in src_clusters:
print(name)
print("cluster count:" + str(src_clusters.__len__()))
for name, value in config.items():
printDividingLine()
source = value["src_ip"]
source_user = value["src_username"]
source_passwd = value["src_password"]
source_auth = None
if source_user != "":
source_auth = (source_user, source_passwd)
dest = value["dest_ip"]
dest_user = value["dest_username"]
dest_passwd = value["dest_password"]
dest_auth = None
if dest_user != "":
dest_auth = (dest_user, dest_passwd)
print("start to process cluster: " + name)
# only deal with mapping list
if 'only_mapping' in value and value["only_mapping"]:
for source_index, dest_index in value["mapping"].iteritems():
print("start to process source index" + source_index + ", target index: " + dest_index)
source_url = source + "/" + source_index
response = requests.get(source_url, auth=source_auth)
if response.status_code != 200:
print("*** get ElasticSearch message failed. resp statusCode:" + str(
response.status_code) + " response is " + response.text)
continue
mapping = response.json()
index_mapping = process_mapping(mapping[source_index], dest_index)
dest_url = dest + "/" + dest_index
result = put_mapping_to_target(dest_url, index_mapping, name, source_index, dest_auth)
if result is False:
print("cluster name:" + name + ", " + source_index + ":failure")
continue
print("cluster name:" + name + ", " + source_index + ":success")
else:
# get all indices
system_index, create_index = getAlias(source, source_auth)
success_index = 0
for index in create_index:
source_url = source + "/" + index
index_response = requests.get(source_url, auth=source_auth)
if index_response.status_code != 200:
print("*** get ElasticSearch message failed. resp statusCode:" + str(
index_response.status_code) + " response is " + index_response.text)
continue
mapping = index_response.json()
dest_index = index
if 'mapping' in value:
if index in value["mapping"].keys():
dest_index = value["mapping"][index]
index_mapping = process_mapping(mapping[index], dest_index)
dest_url = dest + "/" + dest_index
result = put_mapping_to_target(dest_url, index_mapping, name, index, dest_auth)
if result is False:
print("[failure]: migrate mapping cluster name: " + name + ", " + index)
continue
print("[success]: migrate mapping cluster name: " + name + ", " + index)
success_index = success_index + 1
print("create index mapping success total: " + str(success_index))
if __name__ == '__main__':
main(sys.argv)
EOF
3 同步数据
3.1 同步元数据
python /tmp/migrateTemplate.py
python /tmp/migrateMapping.py
3.2 同步实际数据
准备logstash.conf文件，放在/tmp/目录中即可。需要在logstash.conf指定源端和目标端，待同步数据的索引有哪些。
实际需要改动字段是input.elasticsearch.hosts 、input.elasticsearch.index，output.elasticsearch.hosts。
vim /tmp/logstash.conf，输入以下内容：
input{
elasticsearch{
# 源Elasticsearch的访问地址，不需要添加协议，添加HTTPS协议会导致报错。
hosts => ["192.168.243.128:9200"]
# 访问源集群的用户名和密码，非安全集群无需配置。
# user => "css_logstash"
# password => "*****"
# 配置待迁移的索引信息，多个索引以逗号隔开，可以使用通配符设置，例如“index*”。
index => "kibana_sample_data_flights,kibana_sample_data_ecommerce"
docinfo => true
slices => 3
size => 3000
# 当源集群是HTTPS访问方式时，则设置ssl => false。
# ssl => false
}
}
# 移除一些logstash增加的字段。
filter {
mutate {
remove_field => ["@metadata", "@version"]
}
}
output{
elasticsearch{
# 目标Elasticsearch集群的访问地址
hosts => ["192.168.243.128:19200"]
# 访问目标集群的用户名和密码，非安全集群无需配置。
# user => "css_logstash"
# password => "*****"
# 配置目标集群的索引，以下配置为索引名称和源端保持一致，保持默认。
index => "%{[@metadata][_index]}"
document_type => "%{[@metadata][_type]}"
document_id => "%{[@metadata][_id]}"
# 当目标集群是HTTPS访问方式时，则需额外配置以下信息。
# 配置集群HTTPS访问证书，CSS集群保持以下不变。
#cacert => "/rds/datastore/logstash/v7.10.0/package/logstash-7.10.0/extend/certs/CloudSearchService.cer"
# 是否开启HTTPS通信，HTTPS访问集群则设置为true。
#ssl => true
# 是否验证服务端Elasticsearch证书，设置为false表示不验证证书。
#ssl_certificate_verification => false
}
}
执行如下命令来启动logstash进程，进程执行完后会自动退出：
docker run -it --rm -v /tmp/logstash.conf:/tmp/logstash.conf docker.elastic.co/logstash/logstash:7.1.0 logstash -f /tmp/logstash.conf
4 小结
logstash可用于同步elasticsearch的数据，不仅可以进行全量同步，其实还能进行增量同步（数据带时间字段，用该字段进行query即可抽取出增量的数据），虽然本文没演示增量同步。
