K8SPXC-1546: Cache binlog->gtid set pairs

build/Dockerfile

@@ -65,6 +65,7 @@ COPY build/liveness-check.sh /liveness-check.sh
 COPY build/readiness-check.sh /readiness-check.sh
 COPY build/pmm-prerun.sh /pmm-prerun.sh
 COPY build/get-pxc-state /get-pxc-state
+COPY build/wsrep_cmd_notify_handler.sh /wsrep_cmd_notify_handler.sh
 
 COPY build/haproxy-entrypoint.sh /haproxy-entrypoint.sh
 COPY build/haproxy-init-entrypoint.sh /haproxy-init-entrypoint.sh

build/pxc-entrypoint.sh

@@ -41,7 +41,7 @@
 	elif [ "${!fileVar:-}" ]; then
 		val="$(<"${!fileVar}")"
 	elif [ "${3:-}" ] && [ -f "/etc/mysql/mysql-users-secret/$3" ]; then
 		val="$(</etc/mysql/mysql-users-secret/$3)"
 	fi
 	export "$var"="$val"
 	unset "$fileVar"
@@ -60,7 +60,7 @@
 	case "$f" in
 	*.sh)
 		echo "$0: running $f"
 		. "$f"
 		;;
 	*.sql)
 		echo "$0: running $f"
@@ -188,10 +188,12 @@
 	sed -i "/\[mysqld\]/a enforce-gtid-consistency" $CFG
 fi
 
+sed -i "/\[mysqld\]/a wsrep_notify_cmd=/var/lib/mysql/wsrep_cmd_notify_handler.sh" $CFG
+
 # add sst.cpat to exclude pxc-entrypoint, unsafe-bootstrap, pxc-configure-pxc from SST cleanup
 grep -q "^progress=" $CFG && sed -i "s|^progress=.*|progress=1|" $CFG
 grep -q "^\[sst\]" "$CFG" || printf '[sst]\n' >>"$CFG"
-grep -q "^cpat=" "$CFG" || sed '/^\[sst\]/a cpat=.*\\.pem$\\|.*init\\.ok$\\|.*galera\\.cache$\\|.*wsrep_recovery_verbose\\.log$\\|.*readiness-check\\.sh$\\|.*liveness-check\\.sh$\\|.*get-pxc-state$\\|.*sst_in_progress$\\|.*sleep-forever$\\|.*pmm-prerun\\.sh$\\|.*sst-xb-tmpdir$\\|.*\\.sst$\\|.*gvwstate\\.dat$\\|.*grastate\\.dat$\\|.*\\.err$\\|.*\\.log$\\|.*RPM_UPGRADE_MARKER$\\|.*RPM_UPGRADE_HISTORY$\\|.*pxc-entrypoint\\.sh$\\|.*unsafe-bootstrap\\.sh$\\|.*pxc-configure-pxc\\.sh\\|.*peer-list$\\|.*auth_plugin$\\|.*version_info$\\|.*mysql-state-monitor$\\|.*mysql-state-monitor\\.log$\\|.*notify\\.sock$\\|.*mysql\\.state$' "$CFG" 1<>"$CFG"
+grep -q "^cpat=" "$CFG" || sed '/^\[sst\]/a cpat=.*\\.pem$\\|.*init\\.ok$\\|.*galera\\.cache$\\|.*wsrep_recovery_verbose\\.log$\\|.*readiness-check\\.sh$\\|.*liveness-check\\.sh$\\|.*get-pxc-state$\\|.*sst_in_progress$\\|.*sleep-forever$\\|.*pmm-prerun\\.sh$\\|.*sst-xb-tmpdir$\\|.*\\.sst$\\|.*gvwstate\\.dat$\\|.*grastate\\.dat$\\|.*\\.err$\\|.*\\.log$\\|.*RPM_UPGRADE_MARKER$\\|.*RPM_UPGRADE_HISTORY$\\|.*pxc-entrypoint\\.sh$\\|.*unsafe-bootstrap\\.sh$\\|.*pxc-configure-pxc\\.sh\\|.*peer-list$\\|.*auth_plugin$\\|.*version_info$\\|.*mysql-state-monitor$\\|.*mysql-state-monitor\\.log$\\|.*notify\\.sock$\\|.*mysql\\.state$\\|.*wsrep_cmd_notify_handler\\.sh$' "$CFG" 1<>"$CFG"
 
 if [[ $MYSQL_VERSION == '8.0' && $MYSQL_PATCH_VERSION -ge 26 ]] || [[ $MYSQL_VERSION == '8.4' ]]; then
 	grep -q "^skip_replica_start=ON" "$CFG" || sed -i "/\[mysqld\]/a skip_replica_start=ON" $CFG
@@ -251,7 +253,7 @@
 	i=$(curl "$DISCOVERY_SERVICE/v2/keys/pxc-cluster/queue/$CLUSTER_NAME" | jq -r '.node.nodes[].value')
 
 	# this remove my ip from the list
 	i1="${i[@]//$NODE_IP/}"
 
 	# Register the current IP in the discovery service
 	# key set to expire in 30 sec. There is a cronjob that should update them regularly
@@ -261,8 +263,8 @@
 
 	i=$(curl "$DISCOVERY_SERVICE/v2/keys/pxc-cluster/$CLUSTER_NAME/?quorum=true" | jq -r '.node.nodes[]?.key' | awk -F'/' '{print $(NF)}')
 	# this remove my ip from the list
 	i2="${i[@]//$NODE_IP/}"
 	CLUSTER_JOIN=$(join , $i1 $i2)
 
 	sed -r "s|^[#]?wsrep_node_address=.*$|wsrep_node_address=${NODE_IP}|" "${CFG}" 1<>"${CFG}"
 	sed -r "s|^[#]?wsrep_cluster_name=.*$|wsrep_cluster_name=${CLUSTER_NAME}|" "${CFG}" 1<>"${CFG}"
@@ -303,7 +305,7 @@
 
 # if we have CLUSTER_JOIN - then we do not need to perform datadir initialize
 # the data will be copied from another node
 if [ -z "$CLUSTER_JOIN" ] && [ "$1" = 'mysqld' -a -z "$wantHelp" ]; then
 	# still need to check config, container may have started with --user
 	_check_config "$@"
 
@@ -316,7 +318,7 @@
 	if [ ! -d "$DATADIR/mysql" ]; then
 		file_env 'MYSQL_ROOT_PASSWORD' '' 'root'
 		{ set +x; } 2>/dev/null
 		if [ -z "$MYSQL_ROOT_PASSWORD" -a -z "$MYSQL_ALLOW_EMPTY_PASSWORD" -a -z "$MYSQL_RANDOM_ROOT_PASSWORD" ]; then
 			echo >&2 'error: database is uninitialized and password option is not specified '
 			echo >&2 '  You need to specify one of MYSQL_ROOT_PASSWORD, MYSQL_ALLOW_EMPTY_PASSWORD and MYSQL_RANDOM_ROOT_PASSWORD'
 			exit 1
@@ -372,7 +374,7 @@
 		fi
 
 		{ set +x; } 2>/dev/null
 		if [ ! -z "$MYSQL_RANDOM_ROOT_PASSWORD" ]; then
 			MYSQL_ROOT_PASSWORD="$(pwmake 128)"
 			echo "GENERATED ROOT PASSWORD: $MYSQL_ROOT_PASSWORD"
 		fi
@@ -380,7 +382,7 @@
 		rootCreate=
 		# default root to listen for connections from anywhere
 		file_env 'MYSQL_ROOT_HOST' '%'
 		if [ ! -z "$MYSQL_ROOT_HOST" -a "$MYSQL_ROOT_HOST" != 'localhost' ]; then
 			# no, we don't care if read finds a terminating character in this heredoc
 			# https://unix.stackexchange.com/questions/265149/why-is-set-o-errexit-breaking-this-read-heredoc-expression/265151#265151
 			read -r -d '' rootCreate <<-EOSQL || true
@@ -437,7 +439,7 @@
 		EOSQL
 
 		{ set +x; } 2>/dev/null
 		if [ ! -z "$MYSQL_ROOT_PASSWORD" ]; then
 			mysql+=(-p"${MYSQL_ROOT_PASSWORD}")
 		fi
 		set -x
@@ -451,7 +453,7 @@
 		file_env 'MYSQL_USER'
 		file_env 'MYSQL_PASSWORD'
 		{ set +x; } 2>/dev/null
 		if [ "$MYSQL_USER" -a "$MYSQL_PASSWORD" ]; then
 			echo "CREATE USER '$MYSQL_USER'@'%' IDENTIFIED BY '$MYSQL_PASSWORD' ;" | "${mysql[@]}"
 
 			if [ "$MYSQL_DATABASE" ]; then
@@ -469,7 +471,7 @@
 		done
 
 		{ set +x; } 2>/dev/null
 		if [ ! -z "$MYSQL_ONETIME_PASSWORD" ]; then
 			"${mysql[@]}" <<-EOSQL
 				ALTER USER 'root'@'%' PASSWORD EXPIRE;
 			EOSQL
@@ -486,13 +488,13 @@
 	fi
 
 	# exit when MYSQL_INIT_ONLY environment variable is set to avoid starting mysqld
 	if [ ! -z "$MYSQL_INIT_ONLY" ]; then
 		echo 'Initialization complete, now exiting!'
 		exit 0
 	fi
 fi
 
 if [ "$1" = 'mysqld' -a -z "$wantHelp" ]; then
 	# still need to check config, container may have started with --user
 	_check_config "$@"
 
@@ -507,20 +509,20 @@
 	pxc_version=''
 	if [ -f "$DATADIR/version_info" ]; then
 		pxc_version_file="$DATADIR/version_info"
 		pxc_version=$(cat "$pxc_version_file" | awk '{print $3}')
 	elif [ -f "$DATADIR/xtrabackup_info" ]; then
 		pxc_version_file="$DATADIR/xtrabackup_info"
 		pxc_version=$(grep 'server_version' "$pxc_version_file" | awk '{print $3}' | tr -d '\n')
 	fi
 
 	if [[ -f $pxc_version_file && -n $pxc_version && $MYSQL_VERSION == '5.7' ]] && [[ $(cat /tmp/version_info) != $pxc_version ]]; then
 		SOCKET="$(_get_config 'socket' "$@")"
 		"$@" --skip-networking --socket="${SOCKET}" --wsrep-provider='none' &
 		pid="$!"
 
 		mysql=(mysql --protocol=socket -uroot -hlocalhost --socket="${SOCKET}" --password="")
 		{ set +x; } 2>/dev/null
 		if [ ! -z "$MYSQL_ROOT_PASSWORD" ]; then
 			mysql+=(-p"${MYSQL_ROOT_PASSWORD}")
 		fi
 		set -x
@@ -550,16 +552,16 @@
 POD_NAMESPACE=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace)
 
 wsrep_start_position_opt=""
 if [ "$1" = 'mysqld' -a -z "$wantHelp" ]; then
 	DATADIR="$(_get_config 'datadir' "$@")"
 	grastate_loc="${DATADIR}/grastate.dat"
 	wsrep_verbose_logfile="$DATADIR/wsrep_recovery_verbose.log"
 	if [ -f "$wsrep_verbose_logfile" ]; then
 		cat "$wsrep_verbose_logfile" | tee -a "$DATADIR/wsrep_recovery_verbose_history.log"
 		rm -f "$wsrep_verbose_logfile"
 	fi
 
 	if [ -s "$grastate_loc" -a -d "$DATADIR/mysql" ]; then
 		uuid=$(grep 'uuid:' "$grastate_loc" | cut -d: -f2 | tr -d ' ' || :)
 		seqno=$(grep 'seqno:' "$grastate_loc" | cut -d: -f2 | tr -d ' ' || :)
 		safe_to_bootstrap=$(grep 'safe_to_bootstrap:' "$grastate_loc" | cut -d: -f2 | tr -d ' ' || :)
@@ -574,11 +576,11 @@
 		fi
 	fi
 
 	if [ -z "$wsrep_start_position_opt" -a -d "$DATADIR/mysql" ]; then
 		"$@" --wsrep_recover --log-error-verbosity=3 --log_error="$wsrep_verbose_logfile"
 
 		echo >&2 "WSREP: Print recovery logs: "
 		cat "$wsrep_verbose_logfile" | tee -a "$DATADIR/wsrep_recovery_verbose_history.log"
 		if grep ' Recovered position:' "$wsrep_verbose_logfile"; then
 			start_pos="$(
 				grep ' Recovered position:' "$wsrep_verbose_logfile" |
@@ -615,7 +617,7 @@
 				sed -i 's/safe_to_bootstrap: 0/safe_to_bootstrap: 1/g' "$grastate_loc"
 			fi
 			echo "Recovery was finished."
 			exec "$@" $wsrep_start_position_opt
 		}
 		function is_manual_recovery() {
 			set +o xtrace
@@ -668,7 +670,7 @@
 	fi
 fi
 
 test -e /opt/percona/hookscript/hook.sh && source /opt/percona/hookscript/hook.sh
 
 init_opt=""
 if [[ -f /etc/mysql/init-file/init.sql ]]; then
@@ -676,4 +678,4 @@
 	echo "Using init-file: /etc/mysql/init-file/init.sql"
 fi
 
 exec "$@" ${wsrep_start_position_opt} ${init_opt}

build/pxc-init-entrypoint.sh

@@ -12,3 +12,4 @@ install -o "$(id -u)" -g "$(id -g)" -m 0755 -D /peer-list /var/lib/mysql/peer-li
 install -o "$(id -u)" -g "$(id -g)" -m 0755 -D /get-pxc-state /var/lib/mysql/get-pxc-state
 install -o "$(id -u)" -g "$(id -g)" -m 0755 -D /pmm-prerun.sh /var/lib/mysql/pmm-prerun.sh
 install -o "$(id -u)" -g "$(id -g)" -m 0755 -D /mysql-state-monitor /var/lib/mysql/mysql-state-monitor
+install -o "$(id -u)" -g "$(id -g)" -m 0755 -D /wsrep_cmd_notify_handler.sh /var/lib/mysql/wsrep_cmd_notify_handler.sh

build/wsrep_cmd_notify_handler.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+
+set -o errexit
+
+while [ $# -gt 0 ]; do
+	case $1 in
+	--status)
+	STATUS=$2
+	shift
+	;;
+	--uuid)
+	CLUSTER_UUID=$2
+	shift
+	;;
+	--primary)
+	[ "$2" = "yes" ] && PRIMARY="1" || PRIMARY="0"
+	shift
+	;;
+	--index)
+	INDEX=$2
+	shift
+	;;
+	--members)
+	MEMBERS=$2
+	shift
+	;;
+	esac
+
+	shift
+done
+
+CLUSTER_NAME=$(hostname -f | cut -d'-' -f1)
+CLUSTER_FQDN=$(hostname -f | cut -d'.' -f3-)
+
+if [[ "$STATUS" == "joiner" ]]; then
+	PITR_HOST="${CLUSTER_NAME}-pitr.${CLUSTER_FQDN}"
+	if getent hosts "${PITR_HOST}" >/dev/null 2>&1; then
+		curl -d "hostname=$(hostname -f)" "http://${PITR_HOST}:8080/invalidate-cache/"
+	fi
+fi
+
+exit 0

cmd/pitr/collector/cache.go

@@ -0,0 +1,91 @@
+package collector
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log"
+	"time"
+
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/backup/storage"
+	"github.com/pkg/errors"
+)
+
+type BinlogCacheEntry struct {
+	Binlogs map[string]string `json:"binlogs"` // binlog name -> gtid set
+}
+
+type HostBinlogCache struct {
+	// host -> binlogs. we use pointer here to not copy BinlogCacheEntry
+	// in case if it grows big and it'll grow big eventually.
+	Entries       map[string]*BinlogCacheEntry `json:"entries"`
+	Version       int                          `json:"version"`
+	LastUpdatedAt time.Time                    `json:"last_updated_at"`
+}
+
+func (e *BinlogCacheEntry) Get(key string) (string, bool) {
+	value, ok := e.Binlogs[key]
+	return value, ok
+}
+
+func (e *BinlogCacheEntry) Set(key, value string) {
+	e.Binlogs[key] = value
+}
+
+func loadCache(ctx context.Context, s storage.Storage, key string) (*HostBinlogCache, error) {
+	cache := &HostBinlogCache{
+		Entries: make(map[string]*BinlogCacheEntry),
+		Version: 1,
+	}
+
+	objReader, err := s.GetObject(ctx, key)
+	if err != nil {
+		if errors.Is(err, storage.ErrObjectNotFound) {
+			log.Printf("WARNING: cache file %s not found", key)
+			return cache, nil
+		}
+		return nil, errors.Wrap(err, "get cache from storage")
+	}
+	defer objReader.Close()
+
+	if err := json.NewDecoder(objReader).Decode(cache); err != nil {
+		return nil, errors.Wrap(err, "decode cache")
+	}
+
+	return cache, nil
+}
+
+func saveCache(ctx context.Context, s storage.Storage, cache *HostBinlogCache, key string) error {
+	log.Printf("updating binlog cache")
+	cache.LastUpdatedAt = time.Now()
+
+	data, err := json.Marshal(cache)
+	if err != nil {
+		return errors.Wrap(err, "marshal cache")
+	}
+
+	err = s.PutObject(ctx, key, bytes.NewReader(data), int64(len(data)))
+	return errors.Wrap(err, "put cache to s3")
+}
+
+func InvalidateCache(ctx context.Context, c *Collector, hostname string) error {
+	cache, err := loadCache(ctx, c.GetStorage(), c.GetGTIDCacheKey())
+	if err != nil {
+		return errors.Wrap(err, "load cache")
+	}
+
+	_, ok := cache.Entries[hostname]
+	if !ok {
+		return errors.Errorf("failed to find cache for %s", hostname)
+	}
+
+	log.Printf("invalidating cache for %s", hostname)
+	delete(cache.Entries, hostname)
+
+	err = saveCache(ctx, c.GetStorage(), cache, c.GetGTIDCacheKey())
+	if err != nil {
+		return errors.Wrap(err, "save cache")
+	}
+
+	return nil
+}