#!/bin/bash

# Backs up every (non-system, open) Elasticsearch index into separate files:
#
#   <index>.meta.json   settings/mappings/aliases, ready to use as the body of
#                       PUT /<index> (uuid, creation_date, etc. are stripped)
#   <index>.ndjson.zst  documents in _bulk format ({"index":{...}} + _source),
#                       .ndjson.gz with ES_COMPRESS=gzip
#
# Cluster level metadata (templates, ingest pipelines) goes into _cluster/.
#
# Restore with restore-elastic (see its header for options).
#
# Optional settings in /etc/backup-daily.conf (defaults shown):
#   ES_URL=http://localhost:9200
#   ES_USER=              ES_PASS=         (basic auth)
#   ES_API_KEY=                            (base64 encoded "id:api_key")
#   ES_CA_CERT=                            (CA file for https)
#   ES_INSECURE=0                          (1 = skip TLS verification)
#   ES_INDEX_EXCLUDE=                      (extended regex of indices to skip)
#   ES_SCROLL_SIZE=1000
#   ES_RETRY=3                             (retries per request, 0 = off)
#   ES_COMPRESS=zstd                       (zstd or gzip)
#   ES_COMPRESS_LEVEL=                     (empty = compressor's default)

#######################################

if ! [ -f /etc/backup-daily.conf ]; then
	echo "/etc/backup-daily.conf not found"
	exit 4
fi

BACKUP_ROOT=
ES_URL=http://localhost:9200
ES_USER=
ES_PASS=
ES_API_KEY=
ES_CA_CERT=
ES_INSECURE=0
ES_INDEX_EXCLUDE=
ES_SCROLL_SIZE=1000
ES_RETRY=3
ES_COMPRESS=zstd
ES_COMPRESS_LEVEL=

. /etc/backup-daily.conf

[ -z $BACKUP_ROOT ] && exit 11
[ -d $BACKUP_ROOT ] || exit 21

BACKUP_DIR=$BACKUP_ROOT/elastic
[ -d $BACKUP_DIR ] || mkdir $BACKUP_DIR

case $ES_COMPRESS in
	gzip) COMPRESS="gzip --rsyncable"; EXT=gz ;;
	zstd) COMPRESS="zstd -q --rsyncable -T0"; EXT=zst ;;
	*) echo "unknown ES_COMPRESS: $ES_COMPRESS"; exit 32 ;;
esac
[ -n "$ES_COMPRESS_LEVEL" ] && COMPRESS="$COMPRESS -$ES_COMPRESS_LEVEL"

for cmd in curl jq; do
	command -v $cmd >/dev/null || { echo "$cmd not found"; exit 31; }
done
command -v $ES_COMPRESS >/dev/null \
	|| { echo "$ES_COMPRESS not found, install it or set ES_COMPRESS"; exit 31; }

#######################################

umask 077

[ -d $BACKUP_DIR/tmp ] && rm -r $BACKUP_DIR/tmp
mkdir -p $BACKUP_DIR/tmp/_cluster

TMP=$BACKUP_DIR/tmp
WORK=$(mktemp -d)
trap 'rm -rf "$WORK"' EXIT

# curl config keeps credentials out of the process list
CURLCFG=$WORK/curl.cfg
curl_quote() { printf '%s' "$1" | sed 's/\\/\\\\/g; s/"/\\"/g'; }
{
	echo 'silent'
	echo 'show-error'
	echo 'fail'
	echo 'header = "Content-Type: application/json"'
	# transient failures only (timeouts, 408/429/5xx, connection refused);
	# bodies read from stdin are buffered by curl, so retries resend them
	echo "retry = $ES_RETRY"
	echo 'retry-connrefused'
	[ -n "$ES_USER" ] && echo "user = \"$(curl_quote "$ES_USER:$ES_PASS")\""
	[ -n "$ES_API_KEY" ] && echo "header = \"Authorization: ApiKey $(curl_quote "$ES_API_KEY")\""
	[ -n "$ES_CA_CERT" ] && echo "cacert = \"$(curl_quote "$ES_CA_CERT")\""
	[ "$ES_INSECURE" = 1 ] && echo 'insecure'
} > $CURLCFG

# es METHOD PATH [curl args...]
es() {
	local method=$1 path=$2
	shift 2
	curl -K "$CURLCFG" -X "$method" "${ES_URL%/}/$path" "$@"
}

FAILED=

# cluster level metadata; missing endpoints (older versions) are ignored
for item in _index_template _component_template _template _ingest/pipeline; do
	es GET "$item" > "$TMP/_cluster/$(echo ${item#_} | tr / _).json" 2>/dev/null \
		|| rm -f "$TMP/_cluster/$(echo ${item#_} | tr / _).json"
done

# open indices, skipping system/hidden ones (starting with a dot)
INDICES="$(es GET '_cat/indices?format=json&h=index,status&expand_wildcards=open' \
	| jq -r '.[] | select(.status == "open") | .index | select(startswith(".") | not)' \
	| sort)"

if [ -z "$INDICES" ]; then
	echo "could not list indices from $ES_URL" >&2
	exit 41
fi

[ -n "$ES_INDEX_EXCLUDE" ] && INDICES="$(echo "$INDICES" | grep -Ev -- "$ES_INDEX_EXCLUDE")"

# only the parts of search responses that are used
SCROLL_FILTER='filter_path=_scroll_id,hits.total,hits.hits._id,hits.hits._routing,hits.hits._source'
SCROLL_ID_RE='^\{"_scroll_id":("[A-Za-z0-9+/=_-]+")'

# page_scroll_id FILE: sets sid to the scroll id in FILE as a JSON string;
# ES writes it first, so usually the page doesn't have to be parsed for it
page_scroll_id() {
	local head
	IFS= read -r -N 65536 head < "$1"
	if [[ $head =~ $SCROLL_ID_RE ]]; then
		sid=${BASH_REMATCH[1]}
	else
		sid=$(jq -c '._scroll_id' "$1")
	fi
	[ -n "$sid" ] && [ "$sid" != null ]
}

backup_index() {
	local idx=$1
	local uidx=$(jq -rn --arg i "$idx" '$i | @uri')
	local PAGE=$WORK/page.json NEXT=$WORK/next.json
	local sid total= count=0 n t fetch= rc tmp

	# settings, mappings, aliases
	es GET "$uidx" > $PAGE || return 1
	jq --arg i "$idx" '.[$i] | {
		settings: { index: (.settings.index | del(
			.uuid, .version, .creation_date, .provided_name,
			.history_uuid, .resize, .routing.allocation.initial_recovery)) },
		mappings: .mappings,
		aliases: .aliases
	}' $PAGE > $TMP/$idx.meta.json || return 1

	# documents, via scroll, compressed on the fly; nothing is locked while
	# the scroll is open, so a slower compressor only delays the backup.
	# The next page is fetched while the current one is converted.
	(
		shopt -s lastpipe
		trap '[ -n "$fetch" ] && kill $fetch 2>/dev/null' EXIT

		es POST "$uidx/_search?scroll=5m&$SCROLL_FILTER" \
			-d "{\"size\":$ES_SCROLL_SIZE,\"sort\":[\"_doc\"]}" > $PAGE || exit 1

		while :; do
			page_scroll_id $PAGE || exit 1
			es POST "_search/scroll?$SCROLL_FILTER" \
				-d "{\"scroll\":\"5m\",\"scroll_id\":$sid}" > $NEXT &
			fetch=$!

			# one pass: "<hits> <total>" line first, then the documents
			# (filter_path drops hits.hits altogether from an empty page)
			jq -rc '(.hits.hits // []) as $h |
				"\($h | length) \(.hits.total.value // .hits.total)",
				($h[] |
					{ index: ({ _id: ._id }
						+ (if ._routing then { routing: ._routing } else {} end)) },
					._source)' $PAGE \
				| { read -r n t; cat; }
			[ "${PIPESTATUS[*]}" = "0 0" ] || exit 1

			wait $fetch
			rc=$?
			fetch=
			[ -n "$total" ] || total=$t
			[ "$n" -gt 0 ] || break
			[ $rc = 0 ] || exit 1
			count=$((count + n))

			tmp=$PAGE; PAGE=$NEXT; NEXT=$tmp
		done

		es DELETE "_search/scroll" -d "{\"scroll_id\":$sid}" >/dev/null 2>&1

		if [ "$count" != "$total" ]; then
			echo "$idx: expected $total documents, got $count" >&2
			exit 1
		fi
	) | $COMPRESS > "$TMP/$idx.ndjson.$EXT"

	local status=("${PIPESTATUS[@]}")
	[ "${status[0]}" = 0 ] && [ "${status[1]}" = 0 ]
}

# save indices into separate files
for idx in $INDICES
do
	if ! backup_index "$idx"; then
		echo "$idx: backup failed" >&2
		rm -f "$TMP/$idx.ndjson.$EXT" "$TMP/$idx.meta.json"
		FAILED="$FAILED $idx"
	fi
done

chown -R 0:0 $TMP
find $TMP -type f -exec chmod 0600 {} +

[ -d $BACKUP_DIR/current ] && rm -r $BACKUP_DIR/current
mv $BACKUP_DIR/tmp $BACKUP_DIR/current

if [ -n "$FAILED" ]; then
	echo "failed indices:$FAILED" >&2
	exit 1
fi

exit 0