Migrates all indices (except .kibana) from an old AWS-hosted ES domain to a new one. Set up an S3 bucket and IAM role beforehand, as described here: https://docs.aws.amazon.com/elasticsearch-service/latest/developerguide/es-managedomains-snapshots.html#es-managedomains-snapshot-prerequisites

aws_migrate_elasticsearch.py

import requests

from time import sleep, time

from boto3 import Session
from requests_aws4auth import AWS4Auth

OLD_ES_ENDPOINT = ''
NEW_ES_ENDPOINT = ''
INDEX = ''      # e.g. 'index1', or 'index1,index2,index3', or '*'

S3_BUCKET = ''
S3_REGION = ''
ROLE_ARN = ''


OLD_ES_REGION = OLD_ES_ENDPOINT.split('.')[1]
NEW_ES_REGION = NEW_ES_ENDPOINT.split('.')[1]
SNAPSHOT_REPO = 'snapshots' + str(int(time()))

creds = Session().get_credentials()
headers = {'Content-Type': 'application/json'}


def register_snapshot_repo(endpoint, region):

    payload = {
        'type': 's3',
        'settings': {
            'region': S3_REGION,
            'bucket': S3_BUCKET,
            'role_arn': ROLE_ARN
        }
    }

    auth = AWS4Auth(creds.access_key,
                    creds.secret_key,
                    region,
                    'es',
                    session_token=creds.token)

    url = endpoint + '/_snapshot/' + SNAPSHOT_REPO
    requests.put(url, auth=auth, json=payload, headers=headers)


def create_snapshot(name):

    auth = AWS4Auth(creds.access_key,
                    creds.secret_key,
                    OLD_ES_REGION,
                    'es',
                    session_token=creds.token)

    url = OLD_ES_ENDPOINT + '/_snapshot/' + SNAPSHOT_REPO + '/' + name
    requests.put(url, auth=auth, headers=headers)


def wait_for_snapshot(name):

    PERIOD = 10

    auth = AWS4Auth(creds.access_key,
                    creds.secret_key,
                    OLD_ES_REGION,
                    'es',
                    session_token=creds.token)

    url = OLD_ES_ENDPOINT + '/_snapshot/' + SNAPSHOT_REPO + '/_all?pretty'
    res = requests.get(url, auth=auth, headers=headers)

    for snapshot in res.json()['snapshots']:
        if snapshot['snapshot'] == name:
            if snapshot['state'] == 'SUCCESS':
                return
            else:
                print(snapshot['state'], 'sleeping', PERIOD)
                sleep(PERIOD)
                wait_for_snapshot(name)


def restore_snapshot(name):

    payload = {'indices': INDEX + ',-.kibana*'}

    auth = AWS4Auth(creds.access_key,
                    creds.secret_key,
                    NEW_ES_REGION,
                    'es',
                    session_token=creds.token)

    url = NEW_ES_ENDPOINT + '/_snapshot/' + SNAPSHOT_REPO + '/' + name + '/_restore'
    res = requests.post(url, auth=auth, json=payload, headers=headers)
    print(res.status_code, res.text)


if __name__ == '__main__':

    register_snapshot_repo(OLD_ES_ENDPOINT, OLD_ES_REGION)
    register_snapshot_repo(NEW_ES_ENDPOINT, NEW_ES_REGION)

    OLD_ES_HOSTNAME = OLD_ES_ENDPOINT.split('://')[1].split('.')[0]

    name = OLD_ES_HOSTNAME + '_' + str(int(time()))
    create_snapshot(name)
    wait_for_snapshot(name)

    restore_snapshot(name)