es_backup_handler.py

import os
from datetime import datetime
from uuid import uuid4

import boto3
import requests
from requests_aws4auth import AWS4Auth


# configuration
host = os.environ.get("DOMAIN_HOST")
region = os.environ.get("DOMAIN_REGION", "us-east-1")
snapshot_repo = os.environ.get("DOMAIN_REPO_NAME", "my-snapshot-repo")
snapshot_name_base = os.environ.get("DOMAIN_SNAPSHOT_NAME", "my-snapshot")

# this role is created with Terraform and must be provided
role_arn = os.environ.get("DOMAIN_SNAPSHOT_ROLE_ARN")

# this bucket is assumed to be already created and accessible
s3_bucket = os.environ.get("DOMAIN_REPO_S3BUCKET")

def lambda_handler(event, context):

    print("Elasticsearch backup Lambda")
    print(f"Domain host: {host}")
    print(f"Snapshot name base: {snapshot_name_base}")
    print(f"Snapshot repo: {snapshot_repo}")

    auth = authentication()
    
    register_snapshot_repository(auth)
    take_snapshot(auth)

    return event


def authentication() -> AWS4Auth:

    credentials = boto3.Session().get_credentials()
    awsauth = AWS4Auth(
        credentials.access_key,
        credentials.secret_key,
        region,
        "es",
        session_token=credentials.token,
    )

    return awsauth


def register_snapshot_repository(awsauth: AWS4Auth):
    
    path = f"_snapshot/{snapshot_repo}"
    url = host + path

    payload = {
        "type": "s3",
        "settings": {
            "bucket": s3_bucket, 
            "region": region, 
            "role_arn": role_arn
        }
    }

    headers = {"Content-Type": "application/json"}

    r = requests.put(url, auth=awsauth, json=payload, headers=headers)
    if r.status_code != 200:
        raise Exception(
            f"Cannot register the snapshot repository {snapshot_repo}. Details: {r.text}"
        )
    else:
        print(r.text)


def take_snapshot(awsauth: AWS4Auth):

    name = snapshot_name_base + "-" + datetime.today().strftime('%Y_%m_%d-%H_%M_%S')

    path = f"_snapshot/{snapshot_repo}/{name}"
    url = host + path

    r = requests.put(url, auth=awsauth)
    if r.status_code != 200:
        raise Exception(
            f"Cannot take snapshot {name} in repository {snapshot_repo}. Details: {r.text}"
        )
    else:
        print(r.text)