K8SPS-715: Replication based point-in-time recovery - #1334
Draft
egegunes wants to merge 9 commits into
Draft
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Comment on lines
+15
to
+18
| --admin-address=127.0.0.1 | ||
| --user=mysql | ||
| --gtid-mode=ON | ||
| --enforce-gtid-consistency=ON |
Contributor
There was a problem hiding this comment.
[shfmt] reported by reviewdog 🐶
Suggested change
| --admin-address=127.0.0.1 | |
| --user=mysql | |
| --gtid-mode=ON | |
| --enforce-gtid-consistency=ON | |
| --admin-address=127.0.0.1 | |
| --user=mysql | |
| --gtid-mode=ON | |
| --enforce-gtid-consistency=ON |
Comment on lines
+21
to
+26
| if [[ "${PITR_METHOD}" == "replication" ]]; then | ||
| MYSQLD_ARGS+=( | ||
| --skip-replica-start | ||
| --read-only=ON | ||
| --super-read-only=ON | ||
| ) |
Contributor
There was a problem hiding this comment.
[shfmt] reported by reviewdog 🐶
Suggested change
| if [[ "${PITR_METHOD}" == "replication" ]]; then | |
| MYSQLD_ARGS+=( | |
| --skip-replica-start | |
| --read-only=ON | |
| --super-read-only=ON | |
| ) | |
| if [[ ${PITR_METHOD} == "replication" ]]; then | |
| MYSQLD_ARGS+=( | |
| --skip-replica-start | |
| --read-only=ON | |
| --super-read-only=ON | |
| ) |
| log "waiting for mysqld to be ready" | ||
| until mysqladmin -u operator -p"$(</etc/mysql/mysql-users-secret/operator)" ping --silent 2>/dev/null; do | ||
| sleep 1; | ||
| sleep 1 |
Contributor
There was a problem hiding this comment.
[shfmt] reported by reviewdog 🐶
Suggested change
| sleep 1 | |
| sleep 1 |
Comment on lines
+50
to
+61
| binlog-replay) | ||
| /opt/percona/pitr replay | ||
| ;; | ||
| replication) | ||
| /opt/percona/pitr init | ||
| /opt/percona/pitr setup | ||
| /opt/percona/pitr apply | ||
| ;; | ||
| *) | ||
| log "unknown PITR_METHOD: ${PITR_METHOD}" | ||
| exit 1 | ||
| ;; |
Contributor
There was a problem hiding this comment.
[shfmt] reported by reviewdog 🐶
Suggested change
| binlog-replay) | |
| /opt/percona/pitr replay | |
| ;; | |
| replication) | |
| /opt/percona/pitr init | |
| /opt/percona/pitr setup | |
| /opt/percona/pitr apply | |
| ;; | |
| *) | |
| log "unknown PITR_METHOD: ${PITR_METHOD}" | |
| exit 1 | |
| ;; | |
| binlog-replay) | |
| /opt/percona/pitr replay | |
| ;; | |
| replication) | |
| /opt/percona/pitr init | |
| /opt/percona/pitr setup | |
| /opt/percona/pitr apply | |
| ;; | |
| *) | |
| log "unknown PITR_METHOD: ${PITR_METHOD}" | |
| exit 1 | |
| ;; |
Contributor
There was a problem hiding this comment.
[shfmt] reported by reviewdog 🐶
percona-server-mysql-operator/build/run-pitr-restore.sh
Lines 6 to 7 in b82e55e
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a new PITR selection mechanism so restores can run either the existing binlog replay flow or a new replication-based flow, wiring the choice through the Restore CRD, PITR Job env, and the PITR helper binary.
Changes:
- Add
spec.pitr.method(binlog-replay|replication) toPerconaServerMySQLRestoreCRD and API types (defaulting tobinlog-replay). - Extend PITR restore Job/env generation to pass
PITR_METHODand update the restore runner script to execute the correct PITR flow. - Introduce replication-method stages (
init/setup/apply) incmd/pitrand add DB helpers for channel-based replication operations.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
api/v1/perconaservermysqlrestore_types.go |
Adds PITR method enum and defaulting to the Restore API. |
config/crd/bases/ps.percona.com_perconaservermysqlrestores.yaml |
CRD schema updated to include pitr.method with enum/default. |
deploy/bundle.yaml |
Generated bundle CRD updated for pitr.method. |
deploy/crd.yaml |
Generated CRD updated for pitr.method. |
deploy/cw-bundle.yaml |
Generated CW bundle updated for pitr.method. |
deploy/backup/restore.yaml |
Example restore manifest updated (commented) to mention method. |
pkg/pitr/pitr.go |
Adds PITR_METHOD env var to the PITR restore Job. |
pkg/pitr/pitr_test.go |
Extends PITR Job tests to assert PITR_METHOD behavior. |
build/run-pitr-restore.sh |
Selects PITR flow based on PITR_METHOD and runs appropriate subcommands. |
cmd/pitr/main.go |
Adds replay/init/setup/apply subcommands and replication PITR implementation. |
cmd/pitr/main_test.go |
Adds tests for new replication stages and refactors existing test for replay. |
cmd/internal/db/db.go |
Adds channel-aware replication helpers used by replication PITR. |
cmd/bootstrap/async/async_replication.go |
Updates callers to new StopReplication/ResetReplication signatures. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+270
to
+280
| indexPath := filepath.Join(mysqlDir, hostname+"-relay-bin.index") | ||
| indexData, err := os.ReadFile(indexPath) | ||
| require.NoError(t, err, "relay log index file must exist") | ||
|
|
||
| indexContent := string(indexData) | ||
| assert.Contains(t, indexContent, hostname+"-relay-bin.000001") | ||
| assert.Contains(t, indexContent, hostname+"-relay-bin.000002") | ||
|
|
||
| wantMagic := []byte{0xfe, 0x62, 0x69, 0x6e} | ||
| for i := 1; i <= len(tc.entries); i++ { | ||
| relayLog := filepath.Join(mysqlDir, fmt.Sprintf("%s-relay-bin.%06d", hostname, i)) |
Comment on lines
+373
to
+381
| if tc.checkRelayLogs { | ||
| hostname, err := os.Hostname() | ||
| require.NoError(t, err) | ||
| for i, wantContent := range []string{"binlogdata1", "binlogdata2"} { | ||
| relayLog := filepath.Join(mysqlDir, fmt.Sprintf("%s-relay-bin.%06d", hostname, i+1)) | ||
| data, err := os.ReadFile(relayLog) | ||
| require.NoErrorf(t, err, "relay log %d must exist", i+1) | ||
| assert.Equalf(t, wantContent, string(data), "relay log %d content mismatch", i+1) | ||
| } |
Comment on lines
+436
to
+443
| if pitrType == "date" { | ||
| lastRelayLog := relayLogName(hostname, pitrChannelName, len(entries)) | ||
| lastRelayLogPath := filepath.Join(mysqlDir, lastRelayLog) | ||
| pitrGTID, err = getGTIDByDatetime(lastRelayLogPath, pitrDate) | ||
| if err != nil { | ||
| return fmt.Errorf("get latest GTID for date %s: %w", pitrDate, err) | ||
| } | ||
| log.Printf("latest GTID for date %s: %s", pitrDate, pitrGTID) |
Comment on lines
+591
to
+600
| cmd := exec.Command("bash", "-c", | ||
| fmt.Sprintf("mysqlbinlog --stop-datetime='%s' %s | grep GTID_NEXT | grep -v AUTOMATIC | tail -n 1", | ||
| stopDatetime, relayLogPath)) | ||
|
|
||
| output, err := cmd.Output() | ||
| if err != nil { | ||
| return "", fmt.Errorf("failed to execute mysqlbinlog pipeline: %w", err) | ||
| } | ||
|
|
||
| line := strings.TrimSpace(string(output)) |
Comment on lines
+419
to
+442
| func (d *DB) ChangeReplicationSourceRelay(ctx context.Context, relayLogFile string, relayLogPos int, channel string) error { | ||
| _, err := d.db.ExecContext(ctx, fmt.Sprintf( | ||
| "CHANGE REPLICATION SOURCE TO RELAY_LOG_FILE='%s', RELAY_LOG_POS=%d, SOURCE_HOST='dummy'%s", | ||
| relayLogFile, relayLogPos, forChannelClause(channel))) | ||
| return errors.Wrap(err, "change replication source to relay log") | ||
| } | ||
|
|
||
| // ChangeReplicationFilterIgnoreDB sets REPLICATE_IGNORE_DB on the given | ||
| // channel so the SQL thread skips events whose default database is in dbs. | ||
| // Database names are emitted unquoted as required by CHANGE REPLICATION FILTER. | ||
| func (d *DB) ChangeReplicationFilterIgnoreDB(ctx context.Context, dbs []string, channel string) error { | ||
| if len(dbs) == 0 { | ||
| return errors.New("no databases provided") | ||
| } | ||
| _, err := d.db.ExecContext(ctx, fmt.Sprintf( | ||
| "CHANGE REPLICATION FILTER REPLICATE_IGNORE_DB = (%s)%s", | ||
| strings.Join(dbs, ", "), forChannelClause(channel))) | ||
| return errors.Wrap(err, "change replication filter") | ||
| } | ||
|
|
||
| func (d *DB) StartReplicaUntilGTID(ctx context.Context, gtid string, channel string) error { | ||
| _, err := d.db.ExecContext(ctx, fmt.Sprintf( | ||
| "START REPLICA SQL_THREAD UNTIL SQL_AFTER_GTIDS='%s'%s", gtid, forChannelClause(channel))) | ||
| return errors.Wrap(err, "start replica until GTID") |
| # date: "2024-11-18T11:10:48Z" | ||
| # force: false | ||
| # gtid: a3e5ff70-83e2-11ef-8e57-7a62caf7e1e3:1-36 | ||
| # method: "" |
Comment on lines
+132
to
+141
| path := filepath.Join(dir, fmt.Sprintf("myhost-relay-bin.%06d", i)) | ||
| data, err := os.ReadFile(path) | ||
| require.NoErrorf(t, err, "placeholder %d must exist", i) | ||
| assert.Equalf(t, wantMagic, data, "placeholder %d must contain only the binlog magic", i) | ||
| } | ||
|
|
||
| indexData, err := os.ReadFile(filepath.Join(dir, "myhost-relay-bin.index")) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, | ||
| "./myhost-relay-bin.000001\n./myhost-relay-bin.000002\n./myhost-relay-bin.000003\n", |
Collaborator
commit: 1a8dec7 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
CHANGE DESCRIPTION
Problem:
Short explanation of the problem.
Cause:
Short explanation of the root cause of the issue if applicable.
Solution:
Short explanation of the solution we are providing with this PR.
CHECKLIST
Jira
Needs Doc) and QA (Needs QA)?Tests
Config/Logging/Testability