K8SPXC-1748: add checks before creating functions in collector

build/pxc-entrypoint.sh

@@ -472,6 +472,12 @@ if [ -z "$CLUSTER_JOIN" ] && [ "$1" = 'mysqld' ] && [ -z "$wantHelp" ]; then
 		fi
 		set -x
 
+		if [ "$MYSQL_VERSION" == '8.0' ]; then
+			echo "CREATE FUNCTION IF NOT EXISTS get_last_record_timestamp_by_binlog RETURNS INTEGER SONAME 'binlog_utils_udf.so'" | "${mysql[@]}"
+			echo "CREATE FUNCTION IF NOT EXISTS get_gtid_set_by_binlog RETURNS STRING SONAME 'binlog_utils_udf.so'" | "${mysql[@]}"
+			echo "CREATE FUNCTION IF NOT EXISTS get_first_record_timestamp_by_binlog RETURNS INTEGER SONAME 'binlog_utils_udf.so'" | "${mysql[@]}"
+		fi
+
 		echo
 		ls /docker-entrypoint-initdb.d/ >/dev/null
 		for f in /docker-entrypoint-initdb.d/*; do

cmd/pitr/collector/collector.go

@@ -208,7 +208,7 @@ func (c *Collector) Init(ctx context.Context) error {
 	switch {
 	case strings.HasPrefix(version, "8.0"):
 		log.Println("creating collector functions")
-		if err := db.CreateCollectorFunctions(ctx); err != nil {
+		if err := c.CreateCollectorFunctions(ctx); err != nil {
 			return errors.Wrap(err, "init 8.0: create collector functions")
 		}
 	case strings.HasPrefix(version, "8.4"):
@@ -221,6 +221,37 @@ func (c *Collector) Init(ctx context.Context) error {
 	return nil
 }
 
+func (c *Collector) CreateCollectorFunctions(ctx context.Context) error {
+	nodes, err := pxc.GetNodesByServiceName(ctx, c.pxcServiceName)
+	if err != nil {
+		return errors.Wrap(err, "get nodes by service name")
+	}
+
+	create := func(node string) error {
+		nodeArr := strings.Split(node, ":")
+		host := nodeArr[0]
+		db, err := pxc.NewPXC(host, c.pxcUser, c.pxcPass)
+		if err != nil {
+			return errors.Errorf("creating connection for host %s: %v", host, err)
+		}
+		defer db.Close()
+		if err := db.CreateCollectorFunctions(ctx); err != nil {
+			return errors.Wrap(err, "create collector functions")
+		}
+		return nil
+	}
+
+	for _, node := range nodes {
+		if strings.Contains(node, "wsrep_ready:ON:wsrep_connected:ON:wsrep_local_state_comment:Synced:wsrep_cluster_status:Primary") {
+			if err := create(node); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 func (c *Collector) Run(ctx context.Context) error {
 	err := c.newDB(ctx)
 	if err != nil {

cmd/pitr/pxc/pxc.go

@@ -34,6 +34,7 @@ func NewPXC(addr string, user, pass string) (*PXC, error) {
 	config.Params = map[string]string{
 		"interpolateParams": "true",
 		"tls":               "preferred",
+		"multiStatements":   "true",
 	}
 	config.DBName = "mysql"
 
@@ -205,7 +206,7 @@ func (p *PXC) SubtractGTIDSet(ctx context.Context, set, subSet string) (string,
 	return result, nil
 }
 
-func getNodesByServiceName(ctx context.Context, pxcServiceName string) ([]string, error) {
+func GetNodesByServiceName(ctx context.Context, pxcServiceName string) ([]string, error) {
 	cmd := exec.CommandContext(ctx, "/opt/percona/peer-list", "-on-start=/opt/percona/get-pxc-state.sh", "-service="+pxcServiceName)
 	out, err := cmd.CombinedOutput()
 	if err != nil {
@@ -215,7 +216,7 @@ func getNodesByServiceName(ctx context.Context, pxcServiceName string) ([]string
 }
 
 func GetPXCFirstHost(ctx context.Context, pxcServiceName string) (string, error) {
-	nodes, err := getNodesByServiceName(ctx, pxcServiceName)
+	nodes, err := GetNodesByServiceName(ctx, pxcServiceName)
 	if err != nil {
 		return "", errors.Wrap(err, "get nodes by service name")
 	}
@@ -236,7 +237,7 @@ func GetPXCFirstHost(ctx context.Context, pxcServiceName string) (string, error)
 }
 
 func GetPXCOldestBinlogHost(ctx context.Context, pxcServiceName, user, pass string) (string, error) {
-	nodes, err := getNodesByServiceName(ctx, pxcServiceName)
+	nodes, err := GetNodesByServiceName(ctx, pxcServiceName)
 	if err != nil {
 		return "", errors.Wrap(err, "get nodes by service name")
 	}
@@ -351,19 +352,27 @@ func (p *PXC) UninstallBinlogUDFComponent(ctx context.Context) error {
 }
 
 func (p *PXC) CreateCollectorFunctions(ctx context.Context) error {
-	_, err := p.db.ExecContext(ctx, "CREATE FUNCTION IF NOT EXISTS get_last_record_timestamp_by_binlog RETURNS INTEGER SONAME 'binlog_utils_udf.so'")
-	if err != nil {
-		return errors.Wrap(err, "create function get_first_record_timestamp_by_binlog")
+	m := map[string]string{
+		"get_last_record_timestamp_by_binlog":  "INTEGER",
+		"get_gtid_set_by_binlog":               "STRING",
+		"get_first_record_timestamp_by_binlog": "INTEGER",
 	}
 
-	_, err = p.db.ExecContext(ctx, "CREATE FUNCTION IF NOT EXISTS get_gtid_set_by_binlog RETURNS STRING SONAME 'binlog_utils_udf.so'")
-	if err != nil {
-		return errors.Wrap(err, "create function get_gtid_set_by_binlog")
-	}
+	for functionName, returnType := range m {
+		var x int
+		err := p.db.QueryRowContext(ctx, `SELECT 1 FROM mysql.func WHERE name = ? LIMIT 1`, functionName).Scan(&x)
+		if err != nil && !errors.Is(err, sql.ErrNoRows) {
+			return errors.Wrapf(err, "check if function %s exists", functionName)
+		}
+		if err == nil {
+			log.Printf("function %s already exists", functionName)
+			continue
+		}
 
-	_, err = p.db.ExecContext(ctx, "CREATE FUNCTION IF NOT EXISTS get_first_record_timestamp_by_binlog RETURNS INTEGER SONAME 'binlog_utils_udf.so'")
-	if err != nil {
-		return errors.Wrap(err, "create function get_first_record_timestamp_by_binlog")
+		createQ := fmt.Sprintf("SET SESSION wsrep_on = OFF; CREATE FUNCTION IF NOT EXISTS %s RETURNS %s SONAME 'binlog_utils_udf.so'; SET SESSION wsrep_on = ON", functionName, returnType)
+		if _, err := p.db.ExecContext(ctx, createQ); err != nil {
+			return errors.Wrapf(err, "create function %s", functionName)
+		}
 	}
 
 	return nil

e2e-tests/pitr/run

@@ -79,6 +79,18 @@ main() {
 	cluster="pitr"
 	spinup_pxc "$cluster" "$test_dir/conf/$cluster.yml"
 
+	pitr_pod=$(
+		kubectl_bin get pods \
+			--selector="app.kubernetes.io/component=pitr" \
+			-o 'jsonpath={.items[].metadata.name}'
+	)
+	wait_pod "$pitr_pod"
+	function_exists_count=$(kubectl_bin logs -l "app.kubernetes.io/component=pitr" --tail=-1 2>/dev/null | grep -c 'already exists')
+	if [ "$function_exists_count" -eq 0 ]; then
+		echo "There are no 'function ... already exists' logs"
+		exit 1
+	fi
+
 	run_backup "$cluster" "on-pitr-minio"
 
 	# Temporarily skipping this check