diff --git a/ChangeLog.md b/ChangeLog.md index d8589989..f04dee22 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,8 +1,11 @@ # v2.8.1 NEW FEATURES +- add `general.callback_url` (env `CALLBACK_URL`) and `general.callback_timeout` (env `CALLBACK_TIMEOUT`, default `5s`) — HTTP POST completion notification for API, one-shot CLI commands, and each `watch` iteration; API `?callback=` overrides the global URL when non-empty; the payload keeps the existing `status`/`error`/`operation_id` fields and adds `command` and `duration`; `status` is now also `cancel` when the operation was killed via `/backup/kill`; callbacks are sent asynchronously, failures are logged and never change the backup result; one-shot CLI commands and `watch` iterations are now registered in `/backup/status` like API operations, fix [#1481](https://github.com/Altinity/clickhouse-backup/issues/1481) +- add `general.status_history_size` (env `STATUS_HISTORY_SIZE`, default `1000`) — upper bound on how many finished operations are kept in the in-memory status list exposed by `/backup/status` and `system.backup_actions`; needed because `watch` now records one operation per iteration, so the history would otherwise grow for as long as the process lives; operations still running are never dropped, whatever their age - `delete local|remote ` and `POST /backup/delete/{where}/{name}` now refuse to delete a backup which other backups require via `required_backup` and report the dependent backup names, instead of silently breaking the incremental backups chain (the breakage surfaced only later, when a descendant was downloaded or restored, and for object disks the descendant `required` parts blobs were deleted together with the parent); pass `--force` (`force=1` for the API) to get the old behavior, or set `general.rebase_during_delete: true` (env `REBASE_DURING_DELETE`, default `false`) to rebase every dependent increment first (same as the `rebase` command) so the chain stays restorable — rebase copies the deleted backup parts into its dependents, so deletion time grows with the copied data size and a rebase failure aborts the delete. `backups_to_keep_local`/`backups_to_keep_remote` retention is not affected, fix [#1493](https://github.com/Altinity/clickhouse-backup/issues/1493) BUG FIXES +- `GET /backup/kill` and `kill` in `POST /backup/actions` now actually cancel commands which were started through `POST /backup/actions` — such commands re-enter the CLI app in process with `--command-id` passed *before* the command name, so the value lands in the application flag set, but every command action read it with `c.Int("command-id")`, which in urfave/cli v1 only looks at the command's own flag set (each command re-declares the application flags via `Flags: append(cliapp.Flags, ...)`, so it saw the `-1` default). The command therefore ran with a fresh background context instead of the one owned by its status row and ignored cancellation; broken since [0a26ee69](https://github.com/Altinity/clickhouse-backup/commit/0a26ee69) (v2.1.0), which moved `command-id` from the commands to the application but kept the `c.Int` lookups - drop a stale `upload.state2` when `upload --resume` runs for a backup which doesn't exist on remote storage anymore — the resumable state survives a successful upload and is removed only together with the local backup, so `create_remote --resume` + `delete remote` + `upload --resume` skipped every data file and uploaded a backup containing `metadata.json` only, which looked valid in `list remote`; an interrupted upload still leaves the backup folder on remote, so a real resume is not affected, fix [#1492](https://github.com/Altinity/clickhouse-backup/issues/1492) # v2.8.0 diff --git a/ReadMe.md b/ReadMe.md index fc9e23c4..ac3e0a99 100644 --- a/ReadMe.md +++ b/ReadMe.md @@ -172,6 +172,19 @@ general: retries_jitter: 30 # RETRIES_JITTER, percent of RETRIES_PAUSE for jitter to avoid same time retries from parallel operations delete_batch_size: 1000 # DELETE_BATCH_SIZE, default batch size for bulk DeleteObjects() requests in remote storages that support batch delete (e.g. S3); upper bound for one API call + # callback_url - CALLBACK_URL, optional HTTP endpoint notified with POST application/json when a backup command completes + # (API, one-shot CLI commands, and each watch-loop iteration). API `?callback=` overrides this when non-empty. + # Payload always includes status (success|error|cancel), error (empty string on success), operation_id, command and duration. + # Read-only commands (list, tables, status, ...) never send a callback. + # Callbacks are sent asynchronously; failures are logged and never change the backup command exit code / result. + callback_url: "" + callback_timeout: 5s # CALLBACK_TIMEOUT, max wait for the completion callback HTTP POST + + # status_history_size - STATUS_HISTORY_SIZE, how many finished operations are kept in the in-memory + # status list exposed by `/backup/status` and `system.backup_actions`. `watch` records one operation + # per iteration, so the history needs an upper bound. Operations still running are never dropped. + status_history_size: 1000 + watch_interval: 1h # WATCH_INTERVAL, use only for `watch` command, backup will create every 1h full_interval: 24h # FULL_INTERVAL, use only for `watch` command, full backup will create every 24h watch_backup_name_template: "shard{shard}-{type}-{time:20060102150405}" # WATCH_BACKUP_NAME_TEMPLATE, used only for `watch` command, macros values will apply from `system.macros` for time:XXX, look format in https://go.dev/src/time/format.go @@ -542,7 +555,7 @@ Create new backup: `curl -s localhost:7171/backup/create -X POST | jq .` - Optional boolean query argument `configs-only` or `configs_only` works the same as the `--configs-only` CLI argument (backup only configs). - Optional boolean query argument `skip-check-parts-columns` or `skip_check_parts_columns` works the same as the `--skip-check-parts-columns` CLI argument (allow backup inconsistent column types for data parts). - Optional boolean query argument `resume` works the same as the `--resume` CLI argument (resume upload for object disk data). -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Additional example: `curl -s 'localhost:7171/backup/create?table=default.billing&name=billing_test' -X POST` @@ -566,7 +579,7 @@ Create new backup and upload to remote storage: `curl -s localhost:7171/backup/c - Optional string query argument `skip-projections` or `skip_projections` works the same as the `--skip-projections` CLI argument. - Optional boolean query argument `delete-source` or `delete_source` works the same as `--delete-source` CLI argument. - Optional boolean query argument `resume` works the same as the `--resume` CLI argument (resume upload for object disk data). -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -618,7 +631,7 @@ Upload backup to remote storage: `curl -s localhost:7171/backup/upload/"}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -642,7 +655,7 @@ Download backup from remote storage: `curl -s localhost:7171/backup/download/"}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -650,7 +663,7 @@ Note: this operation is asynchronous, so the API will return once the operation Copy required parts from the `required_backup` chain into remote backup and remove the `required_backup` dependency, so the incremental backup becomes a full one: `curl -s localhost:7171/backup/rebase/ -X POST | jq .` -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -660,7 +673,7 @@ Move data parts inside local backup between disks to match the current `system.p - Optional string query argument `table` works the same as the `--tables value` CLI argument. - Optional boolean query argument `dry-run` works the same as the `--dry-run` CLI argument (only log which parts would move between disks, change nothing). -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -684,7 +697,7 @@ Create schema and restore data from backup: `curl -s localhost:7171/backup/resto - Optional boolean query argument `resume` works the same as the `--resume` CLI argument (resume download for object disk data). - Optional boolean query argument `skip_empty_tables` or `skip-empty-tables` works the same as the `--skip-empty-tables` CLI argument (skip restoring tables that have no data). - Optional boolean query argument `rebind_replica_path_if_exists` or `rebind-replica-path-if-exists` works the same as the `--rebind-replica-path-if-exists` CLI argument (overrides `clickhouse.rebind_replica_path_if_exists` for this request, rebind a restored ReplicatedMergeTree to `default_replica_path` when the original ZK path still has leftover state but our replica entry is absent). WARNING: never set during a concurrent HA multi-replica restore. -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. @@ -711,7 +724,7 @@ Download and restore data from remote backup: `curl -s localhost:7171/backup/res - Optional boolean query argument `hardlink_exists_files` or `hardlink-exists-files` works the same as the `--hardlink-exists-files` CLI argument (Create hardlinks for existing files instead of downloading). - Optional boolean query argument `skip_empty_tables` or `skip-empty-tables` works the same as the `--skip-empty-tables` CLI argument (skip restoring tables that have no data). - Optional boolean query argument `rebind_replica_path_if_exists` or `rebind-replica-path-if-exists` works the same as the `--rebind-replica-path-if-exists` CLI argument (overrides `clickhouse.rebind_replica_path_if_exists` for this request, rebind a restored ReplicatedMergeTree to `default_replica_path` when the original ZK path still has leftover state but our replica entry is absent). WARNING: never set during a concurrent HA multi-replica restore. -- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens", "operation_id" : ""}`. +- Optional string query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success|cancel","error":"not empty when error happens", "operation_id" : "", "command":"", "duration":""}`. When omitted or empty, falls back to `general.callback_url` if configured. Note: this operation is asynchronous, so the API will return once the operation has started. The response includes an `operation_id` field that can be used to track the operation status via `/backup/status?operationid=`. diff --git a/cmd/clickhouse-backup/cli_callback.go b/cmd/clickhouse-backup/cli_callback.go new file mode 100644 index 00000000..ac538e04 --- /dev/null +++ b/cmd/clickhouse-backup/cli_callback.go @@ -0,0 +1,69 @@ +package main + +import ( + "strings" + + "github.com/Altinity/clickhouse-backup/v2/pkg/config" + "github.com/Altinity/clickhouse-backup/v2/pkg/status" + + "github.com/google/uuid" + "github.com/urfave/cli" +) + +// registerCLIStatus wraps every command Action so a one-shot CLI run registers +// itself in status.Current, exactly like an API request does. Completion +// callbacks are then emitted from the single place which owns them, +// status.AsyncStatus.Stop, instead of a CLI specific dispatcher. +// +// Which commands are worth notifying about is decided by status.CallbackEligible, +// this package deliberately holds no list of command names. +func registerCLIStatus(commands []cli.Command) { + for i := range commands { + cmd := &commands[i] + registerCLIStatus(cmd.Subcommands) + if cmd.Action == nil || !status.CallbackEligible(cmd.Name) { + continue + } + action := cmd.Action + name := cmd.Name + cmd.Action = func(c *cli.Context) error { + return runWithCLIStatus(c, name, action) + } + } +} + +func runWithCLIStatus(c *cli.Context, name string, action interface{}) error { + // The API server re-enters this same cli.App in process, see + // APIServer.httpBackupActionsHandler. Such runs are already tracked and + // notified by the handler which started them — or deliberately untracked, when + // the command is listed in api.backup_actions_skip_commands, in which case + // --command-id is status.NotFromAPI and only the server mode marker tells the + // two apart. + if status.APIServerMode() || commandIdFromCli(c) != status.NotFromAPI { + return cli.HandleAction(action, c) + } + cfg := config.GetConfigFromCli(c) + commandId, _ := status.Current.StartWithCallback(cliFullCommand(c, name), uuid.NewString(), cliCallback(cfg)) + err := cli.HandleAction(action, c) + status.Current.Stop(commandId, err) + return err +} + +// cliFullCommand renders the command the way API handlers do, name first so +// status.CallbackEligible and /backup/status filters see the same shape. +func cliFullCommand(c *cli.Context, name string) string { + if args := c.Args(); args.Present() { + return name + " " + strings.Join(args, " ") + } + return name +} + +func cliCallback(cfg *config.Config) *status.CallbackConfig { + if cfg == nil || cfg.General.CallbackURL == "" { + return nil + } + return &status.CallbackConfig{ + URLs: []string{cfg.General.CallbackURL}, + Timeout: cfg.General.CallbackTimeoutDuration, + } +} diff --git a/cmd/clickhouse-backup/main.go b/cmd/clickhouse-backup/main.go index 8cc71115..e6ea9d41 100644 --- a/cmd/clickhouse-backup/main.go +++ b/cmd/clickhouse-backup/main.go @@ -29,6 +29,28 @@ var ( buildArch = "unknown" ) +// commandIdFromCli reads --command-id from wherever urfave/cli v1 parsed it. +// +// `command-id` is declared once in cliapp.Flags, but every command re-declares it +// through `Flags: append(cliapp.Flags, ...)`, so it exists in two flag sets. In +// urfave/cli v1 (unlike v2) c.Int does NOT walk the lineage: it reads only the +// command's own copy, which holds the default unless the flag follows the command +// name. The API server passes `--command-id N` *before* the command name +// (APIServer.actionsAsyncCommandsHandler and friends), so the value lands in the +// app flag set and is only reachable via c.GlobalInt. +// +// Reading just c.Int made every command started through POST /backup/actions see +// status.NotFromAPI, so it ran with a fresh background context instead of the one +// owned by its status row and /backup/kill could not cancel it. Broken since +// 0a26ee69 (v2.1.0), which moved the flag from the commands to the app but kept +// the c.Int lookups. config.GetConfigPath handles the same duplication the same way. +func commandIdFromCli(c *cli.Context) int { + if id := c.GlobalInt("command-id"); id != status.NotFromAPI { + return id + } + return c.Int("command-id") +} + func main() { log.Logger = log_helper.SetupLogger(os.Stderr) //log.Logger = zerolog.New(os.Stdout).With().Timestamp().Caller().Logger() @@ -142,7 +164,7 @@ func main() { Description: "Create new backup", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.CreateBackup(c.Args().First(), c.String("diff-from-remote"), c.String("t"), c.StringSlice("partitions"), c.Bool("s"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("skip-check-parts-columns"), c.StringSlice("skip-projections"), c.Bool("resume"), version, c.Int("command-id")) + return b.CreateBackup(c.Args().First(), c.String("diff-from-remote"), c.String("t"), c.StringSlice("partitions"), c.Bool("s"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("skip-check-parts-columns"), c.StringSlice("skip-projections"), c.Bool("resume"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -225,7 +247,7 @@ func main() { Description: "Create and upload", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.CreateToRemote(c.Args().First(), c.Bool("delete-source"), c.String("diff-from"), c.String("diff-from-remote"), c.String("tables"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("skip-check-parts-columns"), c.Bool("resume"), version, c.Int("command-id")) + return b.CreateToRemote(c.Args().First(), c.Bool("delete-source"), c.String("diff-from"), c.String("diff-from-remote"), c.String("tables"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("skip-check-parts-columns"), c.Bool("resume"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -317,7 +339,7 @@ func main() { UsageText: "clickhouse-backup upload [-t, --tables=.] [--partitions=] [-s, --schema] [--diff-from=] [--diff-from-remote=] [--resumable] ", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.Upload(c.Args().First(), c.Bool("delete-source"), c.String("diff-from"), c.String("diff-from-remote"), c.String("t"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac-only"), c.Bool("configs-only"), c.Bool("named-collections-only"), c.Bool("resume"), version, c.Int("command-id")) + return b.Upload(c.Args().First(), c.Bool("delete-source"), c.String("diff-from"), c.String("diff-from-remote"), c.String("t"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac-only"), c.Bool("configs-only"), c.Bool("named-collections-only"), c.Bool("resume"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -406,7 +428,7 @@ func main() { UsageText: "clickhouse-backup download [-t, --tables=.
] [--partitions=] [-s, --schema] [--resumable] ", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.Download(c.Args().First(), c.String("t"), c.StringSlice("partitions"), c.Bool("schema"), c.Bool("rbac-only"), c.Bool("configs-only"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("hardlink-exists-files"), version, c.Int("command-id")) + return b.Download(c.Args().First(), c.String("t"), c.StringSlice("partitions"), c.Bool("schema"), c.Bool("rbac-only"), c.Bool("configs-only"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("hardlink-exists-files"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -467,7 +489,7 @@ func main() { log.Err(fmt.Errorf("backup name must be defined")).Send() cli.ShowCommandHelpAndExit(c, c.Command.Name, 1) } - return b.Rebase(c.Args().First(), c.Int("command-id")) + return b.Rebase(c.Args().First(), commandIdFromCli(c)) }, Flags: cliapp.Flags, }, @@ -481,7 +503,7 @@ func main() { log.Err(fmt.Errorf("backup name must be defined")).Send() cli.ShowCommandHelpAndExit(c, c.Command.Name, 1) } - return b.Rebalance(c.Args().First(), c.String("tables"), c.Bool("dry-run"), c.Int("command-id")) + return b.Rebalance(c.Args().First(), c.String("tables"), c.Bool("dry-run"), commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -502,7 +524,7 @@ func main() { UsageText: "clickhouse-backup restore [-t, --tables=.
] [-m, --restore-database-mapping=:[,<...>]] [--tm, --restore-table-mapping=:[,<...>]] [--partitions=] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--named-collections] [--resume] [--skip-empty-tables] ", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.Restore(c.Args().First(), c.String("tables"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("data"), c.Bool("drop"), c.Bool("ignore-dependencies"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("restore-schema-as-attach"), c.Bool("replicated-copy-to-detached"), c.Bool("skip-empty-tables"), version, c.Int("command-id")) + return b.Restore(c.Args().First(), c.String("tables"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("data"), c.Bool("drop"), c.Bool("ignore-dependencies"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("restore-schema-as-attach"), c.Bool("replicated-copy-to-detached"), c.Bool("skip-empty-tables"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -619,7 +641,7 @@ func main() { UsageText: "clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=.
] [-m, --restore-database-mapping=:[,<...>]] [--tm, --restore-table-mapping=:[,<...>]] [--partitions=] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--named-collections] [--resumable] [--skip-empty-tables] ", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.RestoreFromRemote(c.Args().First(), c.String("tables"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("d"), c.Bool("rm"), c.Bool("i"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("restore-schema-as-attach"), c.Bool("replicated-copy-to-detached"), c.Bool("skip-empty-tables"), c.Bool("hardlink-exists-files"), version, c.Int("command-id")) + return b.RestoreFromRemote(c.Args().First(), c.String("tables"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("d"), c.Bool("rm"), c.Bool("i"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("named-collections"), c.Bool("named-collections-only"), c.Bool("resume"), c.Bool("restore-schema-as-attach"), c.Bool("replicated-copy-to-detached"), c.Bool("skip-empty-tables"), c.Bool("hardlink-exists-files"), version, commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -744,7 +766,7 @@ func main() { log.Err(fmt.Errorf("backup name must be defined")).Send() cli.ShowCommandHelpAndExit(c, c.Command.Name, 1) } - return b.Delete(c.Args().Get(0), c.Args().Get(1), c.Bool("force"), c.Int("command-id")) + return b.Delete(c.Args().Get(0), c.Args().Get(1), c.Bool("force"), commandIdFromCli(c)) }, Flags: append(cliapp.Flags, cli.BoolFlag{ @@ -835,7 +857,7 @@ func main() { Description: "Execute create_remote + delete local, create full backup every `--full-interval`, create and upload incremental backup every `--watch-interval` use previous backup as base with `--diff-from-remote` option, use `backups_to_keep_remote` config option for properly deletion remote backups, will delete old backups which not have references from other backups. Use `--schedule` instead of intervals to run backups on cron expressions", Action: func(c *cli.Context) error { b := backup.NewBackuper(config.GetConfigFromCli(c)) - return b.Watch(c.String("watch-interval"), c.String("full-interval"), c.String("watch-backup-name-template"), c.StringSlice("schedule"), c.String("tables"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac"), c.Bool("configs"), c.Bool("named-collections"), c.Bool("skip-check-parts-columns"), c.Bool("delete-source"), version, c.Int("command-id"), nil, c) + return b.Watch(c.String("watch-interval"), c.String("full-interval"), c.String("watch-backup-name-template"), c.StringSlice("schedule"), c.String("tables"), c.StringSlice("partitions"), c.StringSlice("skip-projections"), c.Bool("schema"), c.Bool("rbac"), c.Bool("configs"), c.Bool("named-collections"), c.Bool("skip-check-parts-columns"), c.Bool("delete-source"), version, commandIdFromCli(c), nil, c) }, Flags: append(cliapp.Flags, cli.StringFlag{ @@ -989,6 +1011,7 @@ func main() { } return cli.ShowAppHelp(c) } + registerCLIStatus(cliapp.Commands) if err := cliapp.Run(os.Args); err != nil { log.Fatal().Stack().Err(err).Send() } diff --git a/cmd/clickhouse-backup/main_test.go b/cmd/clickhouse-backup/main_test.go new file mode 100644 index 00000000..1deec662 --- /dev/null +++ b/cmd/clickhouse-backup/main_test.go @@ -0,0 +1,226 @@ +package main + +import ( + "encoding/json" + "errors" + "flag" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "testing" + "time" + + "github.com/Altinity/clickhouse-backup/v2/pkg/status" + "github.com/stretchr/testify/require" + "github.com/urfave/cli" +) + +func TestCLIStatus_CallbackDispatchedOnCommandSuccess(t *testing.T) { + r := require.New(t) + payloads, srv := callbackReceiver(t) + defer srv.Close() + + err := runWithCLIStatus(newTestCLIContext(t, srv.URL, "create"), "create", func(c *cli.Context) error { + return nil + }) + r.NoError(err) + + got := awaitCallback(t, payloads) + r.Equal(status.SuccessStatus, got.Status) + r.Equal("", got.Error) + r.Equal("create", got.Command) + r.NotEmpty(got.Duration) + r.NotEmpty(got.OperationId) +} + +func TestCLIStatus_CallbackDispatchedOnCommandFailure(t *testing.T) { + r := require.New(t) + payloads, srv := callbackReceiver(t) + defer srv.Close() + + actionErr := errors.New("invalid table pattern") + err := runWithCLIStatus(newTestCLIContext(t, srv.URL, "create"), "create", func(c *cli.Context) error { + return actionErr + }) + r.ErrorIs(err, actionErr) + + got := awaitCallback(t, payloads) + r.Equal(status.ErrorStatus, got.Status) + r.Equal(actionErr.Error(), got.Error) + r.Equal("create", got.Command) +} + +// A broken or slow callback receiver must not change what the command returns. +func TestCLIStatus_CallbackFailureDoesNotAffectExitCode(t *testing.T) { + r := require.New(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + err := runWithCLIStatus(newTestCLIContext(t, srv.URL, "create"), "create", func(c *cli.Context) error { + return nil + }) + r.NoError(err) +} + +// Runs re-entered by the API server carry --command-id and are already tracked +// and notified by the handler which started them, so they must not notify twice. +func TestCLIStatus_SkippedWhenSpawnedFromAPI(t *testing.T) { + r := require.New(t) + payloads, srv := callbackReceiver(t) + defer srv.Close() + + ctx := newTestCLIContextWithCommandId(t, srv.URL, "create", 7) + err := runWithCLIStatus(ctx, "create", func(c *cli.Context) error { return nil }) + r.NoError(err) + + select { + case p := <-payloads: + r.Failf("unexpected callback", "API spawned run must not notify, got %+v", p) + case <-time.After(300 * time.Millisecond): + } +} + +// The command line reported to the callback keeps the command name first, so +// status.CallbackEligible and /backup/status filters see the same shape as API runs. +func TestCLIStatus_FullCommandIncludesArguments(t *testing.T) { + r := require.New(t) + payloads, srv := callbackReceiver(t) + defer srv.Close() + + ctx := newTestCLIContext(t, srv.URL, "create_remote") + err := runWithCLIStatus(ctx, "create_remote", func(c *cli.Context) error { return nil }) + r.NoError(err) + + got := awaitCallback(t, payloads) + r.Equal("create_remote backup-name", got.Command) +} + +// registerCLIStatus must wrap eligible commands wherever they are declared and +// leave everything else untouched, without any command name list of its own. +func TestRegisterCLIStatus_WrapsEligibleCommandsRecursively(t *testing.T) { + r := require.New(t) + noop := func(c *cli.Context) error { return nil } + commands := []cli.Command{ + {Name: "create", Action: noop}, + {Name: "list", Action: noop}, + {Name: "server", Action: noop, Subcommands: []cli.Command{{Name: "restore", Action: noop}}}, + } + original := commands[1].Action + + registerCLIStatus(commands) + + r.NotNil(commands[0].Action) + r.False(sameAction(commands[0].Action, noop), "eligible command `create` must be wrapped") + r.True(sameAction(commands[1].Action, original), "read-only command `list` must stay untouched") + r.True(sameAction(commands[2].Action, noop), "supervisor command `server` must stay untouched") + r.False(sameAction(commands[2].Subcommands[0].Action, noop), "nested eligible command `restore` must be wrapped") +} + +func sameAction(a, b interface{}) bool { + return reflect.ValueOf(a).Pointer() == reflect.ValueOf(b).Pointer() +} + +func callbackReceiver(t *testing.T) (chan status.CallbackPayload, *httptest.Server) { + t.Helper() + payloads := make(chan status.CallbackPayload, 4) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + body, err := io.ReadAll(req.Body) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + var p status.CallbackPayload + if err := json.Unmarshal(body, &p); err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + payloads <- p + w.WriteHeader(http.StatusOK) + })) + return payloads, srv +} + +func awaitCallback(t *testing.T, payloads chan status.CallbackPayload) status.CallbackPayload { + t.Helper() + select { + case p := <-payloads: + return p + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for callback") + return status.CallbackPayload{} + } +} + +func newTestCLIContext(t *testing.T, callbackURL, commandName string) *cli.Context { + t.Helper() + return newTestCLIContextWithCommandId(t, callbackURL, commandName, status.NotFromAPI) +} + +func newTestCLIContextWithCommandId(t *testing.T, callbackURL, commandName string, commandId int) *cli.Context { + t.Helper() + configPath := filepath.Join(t.TempDir(), "config.yml") + content := "general:\n callback_url: \"" + callbackURL + "\"\n callback_timeout: \"2s\"\n" + if err := os.WriteFile(configPath, []byte(content), 0644); err != nil { + t.Fatalf("write config: %v", err) + } + app := cli.NewApp() + app.Commands = []cli.Command{{Name: commandName}} + + // Mirror the real flag layout: main.go declares command-id at app level and + // every command re-declares it via `Flags: append(cliapp.Flags, ...)`. The API + // server passes --command-id *before* the command name, so it lands in the app + // flag set while the command keeps its own default. A helper that puts it only + // on the command flag set cannot catch a lookup reading the wrong one. + appSet := flag.NewFlagSet("clickhouse-backup", flag.ContinueOnError) + appSet.String("config", configPath, "") + appSet.Int("command-id", commandId, "") + parent := cli.NewContext(app, appSet, nil) + + cmdSet := flag.NewFlagSet(commandName, flag.ContinueOnError) + cmdSet.String("config", configPath, "") + cmdSet.Int("command-id", status.NotFromAPI, "") + if commandName == "create_remote" { + if err := cmdSet.Parse([]string{"backup-name"}); err != nil { + t.Fatalf("parse args: %v", err) + } + } + ctx := cli.NewContext(app, cmdSet, parent) + ctx.Command = app.Commands[0] + return ctx +} + +// commandIdFromCli must find --command-id where the API server actually puts it: +// before the command name, i.e. in the app flag set, not the command's own copy. +func TestCommandIdFromCli(t *testing.T) { + r := require.New(t) + r.Equal(7, commandIdFromCli(newTestCLIContextWithCommandId(t, "", "create", 7)), + "--command-id passed by the API server before the command name must be visible") + r.Equal(status.NotFromAPI, commandIdFromCli(newTestCLIContext(t, "", "create")), + "a plain CLI run must report NotFromAPI") +} + +// In an API server process no CLI re-entry may register a status row, even when +// the handler deliberately passes NotFromAPI because the command is listed in +// api.backup_actions_skip_commands. +func TestCLIStatus_SkippedInAPIServerMode(t *testing.T) { + r := require.New(t) + payloads, srv := callbackReceiver(t) + defer srv.Close() + + status.SetAPIServerMode() + defer status.ResetAPIServerModeForTest() + + err := runWithCLIStatus(newTestCLIContext(t, srv.URL, "create"), "create", func(c *cli.Context) error { return nil }) + r.NoError(err) + + select { + case p := <-payloads: + r.Failf("unexpected callback", "API server mode must not register CLI rows, got %+v", p) + case <-time.After(300 * time.Millisecond): + } +} diff --git a/pkg/backup/watch.go b/pkg/backup/watch.go index 7690688b..870f3936 100644 --- a/pkg/backup/watch.go +++ b/pkg/backup/watch.go @@ -12,6 +12,7 @@ import ( "github.com/Altinity/clickhouse-backup/v2/pkg/config" "github.com/Altinity/clickhouse-backup/v2/pkg/server/metrics" "github.com/Altinity/clickhouse-backup/v2/pkg/status" + "github.com/google/uuid" "github.com/pkg/errors" "github.com/rs/zerolog/log" "github.com/urfave/cli" @@ -19,6 +20,19 @@ import ( var watchBackupTemplateTimeRE = regexp.MustCompile(`{time:([^}]+)}`) +// watchIterationCallback builds the completion callback attached to every watch +// iteration, so each cycle notifies general.callback_url separately with its own +// operation_id. Returns nil when no callback URL is configured. +func (b *Backuper) watchIterationCallback() *status.CallbackConfig { + if b.cfg == nil || b.cfg.General.CallbackURL == "" { + return nil + } + return &status.CallbackConfig{ + URLs: []string{b.cfg.General.CallbackURL}, + Timeout: b.cfg.General.CallbackTimeoutDuration, + } +} + func (b *Backuper) NewBackupWatchName(ctx context.Context, backupType string) (string, error) { return b.newBackupWatchNameFromTemplate(ctx, b.cfg.General.WatchBackupNameTemplate, backupType) } @@ -92,6 +106,9 @@ func (b *Backuper) Watch(watchInterval, fullInterval, watchBackupNameTemplate st } ctx, cancel = context.WithCancel(ctx) defer cancel() + // every iteration registers a status row, so the history bound matters here even + // for a standalone CLI `watch` which never goes through the API config reload + status.SetMaxFinishedRows(b.cfg.General.StatusHistorySize) // standalone CLI graceful shutdown, server mode cancels the command context via status.Current.CancelAll on SIGTERM if commandId == status.NotFromAPI { var stopSignals context.CancelFunc @@ -155,6 +172,7 @@ func (b *Backuper) Watch(watchInterval, fullInterval, watchBackupNameTemplate st if backupType == "increment" { diffFromRemote = prevBackupName } + iterationCommandId, _ := status.Current.StartWithCallback("create_remote "+backupName, uuid.NewString(), b.watchIterationCallback()) if metrics != nil { createRemoteErr, createRemoteErrCount = metrics.ExecuteWithMetrics("create_remote", createRemoteErrCount, func() error { return b.CreateToRemote(backupName, deleteSource, "", diffFromRemote, tablePattern, partitions, skipProjections, schemaOnly, backupRBAC, false, backupConfigs, false, backupNamedCollections, false, skipCheckPartsColumns, false, version, commandId) @@ -213,6 +231,11 @@ func (b *Backuper) Watch(watchInterval, fullInterval, watchBackupNameTemplate st } } + if createRemoteErr != nil { + status.Current.Stop(iterationCommandId, createRemoteErr) + } else { + status.Current.Stop(iterationCommandId, deleteLocalErr) + } if (createRemoteErrCount > b.cfg.General.BackupsToKeepRemote && b.cfg.General.BackupsToKeepRemote >= 0) || (deleteLocalErrCount > b.cfg.General.BackupsToKeepLocal && b.cfg.General.BackupsToKeepLocal >= 0) { return errors.Errorf("too many errors create_remote: %d, delete local: %d, during watch full_interval: %s, abort watching", createRemoteErrCount, deleteLocalErrCount, b.cfg.General.FullInterval) diff --git a/pkg/backup/watch_schedule.go b/pkg/backup/watch_schedule.go index 14670ffa..3e00a057 100644 --- a/pkg/backup/watch_schedule.go +++ b/pkg/backup/watch_schedule.go @@ -9,8 +9,10 @@ import ( "github.com/Altinity/clickhouse-backup/v2/pkg/config" "github.com/Altinity/clickhouse-backup/v2/pkg/server/metrics" + "github.com/Altinity/clickhouse-backup/v2/pkg/status" "github.com/Altinity/clickhouse-backup/v2/pkg/storage" + "github.com/google/uuid" "github.com/pkg/errors" cron "github.com/robfig/cron/v3" "github.com/rs/zerolog/log" @@ -250,10 +252,12 @@ func (b *Backuper) executeScheduledBackup(ctx context.Context, st *watchSchedule if rebaseRequired { diffFromRemote = st.prevBackupName } + iterationCommandId, _ := status.Current.StartWithCallback("create_remote "+backupName, uuid.NewString(), b.watchIterationCallback()) createRemote := func() error { return b.CreateToRemote(backupName, deleteSource, "", diffFromRemote, tablePattern, partitions, skipProjections, schemaOnly, backupRBAC, false, backupConfigs, false, backupNamedCollections, false, skipCheckPartsColumns, false, version, commandId) } var createRemoteErr error + var deleteLocalErr error if metrics != nil { createRemoteErr, *createRemoteErrCount = metrics.ExecuteWithMetrics("create_remote", *createRemoteErrCount, createRemote) if createRemoteErr == nil && rebaseRequired { @@ -275,7 +279,6 @@ func (b *Backuper) executeScheduledBackup(ctx context.Context, st *watchSchedule removeLocal := func() error { return b.RemoveBackupLocal(ctx, backupName, nil, true) } - var deleteLocalErr error if metrics != nil { deleteLocalErr, *deleteLocalErrCount = metrics.ExecuteWithMetrics("delete", *deleteLocalErrCount, removeLocal) } else { @@ -285,6 +288,11 @@ func (b *Backuper) executeScheduledBackup(ctx context.Context, st *watchSchedule log.Error().Str("schedule", st.schedule.Name).Msgf("delete local `%s` return error: %v", backupName, deleteLocalErr) } } + if createRemoteErr != nil { + status.Current.Stop(iterationCommandId, createRemoteErr) + } else { + status.Current.Stop(iterationCommandId, deleteLocalErr) + } if createRemoteErr != nil { return } diff --git a/pkg/config/config.go b/pkg/config/config.go index 0b053741..f500ba48 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -93,9 +93,9 @@ type GeneralConfig struct { // rebase every dependent increment first (same as the `rebase` command), so the chain stays restorable and the backup becomes deletable; // rebase copies the deleted backup parts into its dependents, so deletion time grows with the copied data size, // rebase failure is fatal and the backup is not deleted, see https://github.com/Altinity/clickhouse-backup/issues/1493 - RebaseDuringDelete bool `yaml:"rebase_during_delete" envconfig:"REBASE_DURING_DELETE"` - UploadMaxBytesPerSecond uint64 `yaml:"upload_max_bytes_per_second" envconfig:"UPLOAD_MAX_BYTES_PER_SECOND"` - DownloadMaxBytesPerSecond uint64 `yaml:"download_max_bytes_per_second" envconfig:"DOWNLOAD_MAX_BYTES_PER_SECOND"` + RebaseDuringDelete bool `yaml:"rebase_during_delete" envconfig:"REBASE_DURING_DELETE"` + UploadMaxBytesPerSecond uint64 `yaml:"upload_max_bytes_per_second" envconfig:"UPLOAD_MAX_BYTES_PER_SECOND"` + DownloadMaxBytesPerSecond uint64 `yaml:"download_max_bytes_per_second" envconfig:"DOWNLOAD_MAX_BYTES_PER_SECOND"` // MaxBrokenPartRatio - maximum allowed fraction (0..1) of broken data parts that still produces a // successful but partial backup during backup creation (`create`, and the create stage of // `create_remote`). 0 (default) preserves legacy behavior where any broken part aborts the whole @@ -127,6 +127,11 @@ type GeneralConfig struct { WatchInterval string `yaml:"watch_interval" envconfig:"WATCH_INTERVAL"` FullInterval string `yaml:"full_interval" envconfig:"FULL_INTERVAL"` WatchBackupNameTemplate string `yaml:"watch_backup_name_template" envconfig:"WATCH_BACKUP_NAME_TEMPLATE"` + // CallbackURL - optional HTTP endpoint notified when a backup command completes (API, CLI, or watch iteration). + // API query param `?callback=` overrides this when non-empty. + CallbackURL string `yaml:"callback_url" envconfig:"CALLBACK_URL"` + // CallbackTimeout - max wait for the completion callback HTTP POST (duration string, default "5s"). + CallbackTimeout string `yaml:"callback_timeout" envconfig:"CALLBACK_TIMEOUT"` // WatchSchedules - named cron driven watch chains, alternative to watch_interval/full_interval, in env use ';' as separator between schedules, see https://github.com/Altinity/clickhouse-backup/issues/1354 WatchSchedules WatchSchedules `yaml:"watch_schedules" envconfig:"WATCH_SCHEDULES"` ShardedOperationMode string `yaml:"sharded_operation_mode" envconfig:"SHARDED_OPERATION_MODE"` @@ -137,9 +142,15 @@ type GeneralConfig struct { ConfigBackupAlways bool `yaml:"config_backup_always" envconfig:"CONFIG_BACKUP_ALWAYS"` NamedCollectionsBackupAlways bool `yaml:"named_collections_backup_always" envconfig:"NAMED_COLLECTIONS_BACKUP_ALWAYS"` DeleteBatchSize int `yaml:"delete_batch_size" envconfig:"DELETE_BATCH_SIZE"` - RetriesDuration time.Duration - WatchDuration time.Duration - FullDuration time.Duration + // StatusHistorySize bounds how many finished operations are kept in the in-memory + // async status (`/backup/status`, `system.backup_actions`). Long living `watch` + // processes record one operation per iteration, so the history needs an upper bound. + // Operations still running are never dropped, whatever their age. + StatusHistorySize int `yaml:"status_history_size" envconfig:"STATUS_HISTORY_SIZE"` + RetriesDuration time.Duration + WatchDuration time.Duration + FullDuration time.Duration + CallbackTimeoutDuration time.Duration } // GCSConfig - GCS settings section @@ -746,6 +757,21 @@ func ValidateConfig(cfg *Config) error { } else { return errors.New("empty retries pause") } + if cfg.General.CallbackTimeout != "" { + if duration, err := time.ParseDuration(cfg.General.CallbackTimeout); err != nil { + return errors.Wrap(err, "invalid callback timeout") + } else if duration <= 0 { + return errors.Errorf("invalid callback timeout `%s`, it must be > 0", cfg.General.CallbackTimeout) + } else { + cfg.General.CallbackTimeoutDuration = duration + } + } else { + cfg.General.CallbackTimeout = "5s" + cfg.General.CallbackTimeoutDuration = 5 * time.Second + } + if cfg.General.StatusHistorySize <= 0 { + return errors.Errorf("invalid status_history_size `%d`, it must be > 0", cfg.General.StatusHistorySize) + } if cfg.General.WatchInterval != "" { if duration, err := time.ParseDuration(cfg.General.WatchInterval); err != nil { return errors.Wrap(err, "invalid watch interval") @@ -844,6 +870,9 @@ func DefaultConfig() *Config { RetriesOnFailure: 3, RetriesPause: "5s", RetriesDuration: 5 * time.Second, + CallbackTimeout: "5s", + CallbackTimeoutDuration: 5 * time.Second, + StatusHistorySize: 1000, WatchInterval: "1h", WatchDuration: 1 * time.Hour, FullInterval: "24h", diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 7e95bb7c..7371d5fa 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "testing" + "time" "github.com/rs/zerolog" "github.com/rs/zerolog/log" @@ -687,3 +688,122 @@ func TestDefaultCompleteResumableAfterRestartCommands(t *testing.T) { } } } + +func TestConfig_ParseCallbackURL_FromYAML(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + content := "general:\n callback_url: \"http://example.com/webhook\"\n" + if err := os.WriteFile(configPath, []byte(content), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.General.CallbackURL != "http://example.com/webhook" { + t.Fatalf("expected CallbackURL %q, got %q", "http://example.com/webhook", cfg.General.CallbackURL) + } +} + +func TestConfig_ParseCallbackURL_FromENV(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + content := "general:\n callback_url: \"http://from-yaml.example/webhook\"\n" + if err := os.WriteFile(configPath, []byte(content), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + t.Setenv("CALLBACK_URL", "http://from-env.example/webhook") + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.General.CallbackURL != "http://from-env.example/webhook" { + t.Fatalf("expected CALLBACK_URL env override %q, got %q", "http://from-env.example/webhook", cfg.General.CallbackURL) + } +} + +func TestConfig_ParseCallbackTimeout_Default(t *testing.T) { + cfg := DefaultConfig() + if err := ValidateConfig(cfg); err != nil { + t.Fatalf("ValidateConfig: %v", err) + } + if cfg.General.CallbackTimeout != "5s" { + t.Fatalf("expected default CallbackTimeout %q, got %q", "5s", cfg.General.CallbackTimeout) + } + if cfg.General.CallbackTimeoutDuration != 5*time.Second { + t.Fatalf("expected default CallbackTimeoutDuration %v, got %v", 5*time.Second, cfg.General.CallbackTimeoutDuration) + } +} + +func TestConfig_ParseCallbackTimeout_FromYAML(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + content := "general:\n callback_timeout: \"10s\"\n" + if err := os.WriteFile(configPath, []byte(content), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.General.CallbackTimeout != "10s" { + t.Fatalf("expected CallbackTimeout %q, got %q", "10s", cfg.General.CallbackTimeout) + } + if cfg.General.CallbackTimeoutDuration != 10*time.Second { + t.Fatalf("expected CallbackTimeoutDuration %v, got %v", 10*time.Second, cfg.General.CallbackTimeoutDuration) + } +} + +func TestConfig_ParseCallbackTimeout_RejectsNonPositive(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + content := "general:\n callback_timeout: \"0s\"\n" + if err := os.WriteFile(configPath, []byte(content), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + _, err := LoadConfig(configPath) + if err == nil { + t.Fatal("expected LoadConfig to reject callback_timeout: 0s") + } + if !strings.Contains(err.Error(), "callback timeout") { + t.Fatalf("expected callback timeout validation error, got: %v", err) + } +} + +func TestConfig_StatusHistorySize_Default(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(configPath, []byte("general:\n remote_storage: none\n"), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.General.StatusHistorySize != 1000 { + t.Fatalf("expected default StatusHistorySize 1000, got %d", cfg.General.StatusHistorySize) + } +} + +func TestConfig_StatusHistorySize_FromYAML(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(configPath, []byte("general:\n status_history_size: 25\n"), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.General.StatusHistorySize != 25 { + t.Fatalf("expected StatusHistorySize 25, got %d", cfg.General.StatusHistorySize) + } +} + +func TestConfig_StatusHistorySize_RejectsNonPositive(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(configPath, []byte("general:\n status_history_size: 0\n"), 0644); err != nil { + t.Fatalf("can't write config file: %v", err) + } + _, err := LoadConfig(configPath) + if err == nil { + t.Fatal("expected LoadConfig to reject status_history_size: 0") + } + if !strings.Contains(err.Error(), "status_history_size") { + t.Fatalf("expected status_history_size validation error, got: %v", err) + } +} diff --git a/pkg/server/callback.go b/pkg/server/callback.go index 10f31a7a..9c6a05d0 100644 --- a/pkg/server/callback.go +++ b/pkg/server/callback.go @@ -1,70 +1,38 @@ package server import ( - "bytes" - "context" - "encoding/json" - "fmt" - "net/http" "net/url" + "strings" + "time" + "github.com/Altinity/clickhouse-backup/v2/pkg/status" "github.com/pkg/errors" ) -// callbackFn is a function which will post a callback when invoked -type callbackFn func(ctx context.Context, v interface{}) []error - -// parseCallback parses a callback URL from URL query values and returns a closure which can send -// a payload back to the specified URL when invoked. -func parseCallback(query url.Values) (callbackFn, error) { - encodedURLs, exist := query["callback"] - if !exist { - noOpCallback := func(_ context.Context, _ interface{}) []error { - return nil +// parseCallback resolves the completion callback for a request. The `callback` +// query parameter may be repeated, and takes precedence over general.callback_url, +// which is used only when no non-empty query parameter is present. +// A nil result means "do not notify". +func parseCallback(query url.Values, fallbackURL string, timeout time.Duration) (*status.CallbackConfig, error) { + var urls []string + for _, v := range query["callback"] { + if strings.TrimSpace(v) == "" { + continue } - return noOpCallback, nil - } - - decodedURLs := make([]string, len(encodedURLs)) - for i, v := range encodedURLs { - d, err := url.QueryUnescape(v) + decoded, err := url.QueryUnescape(v) if err != nil { return nil, errors.Wrapf(err, "could not decode url %q", v) } - decodedURLs[i] = d - } - - client := &http.Client{} - return func(ctx context.Context, v interface{}) []error { - payload, err := json.Marshal(v) - if err != nil { - return []error{errors.Wrapf(err, "error encoding %v", v)} + if strings.TrimSpace(decoded) == "" { + continue } - - var errs []error - for _, callBackURL := range decodedURLs { - reader := bytes.NewReader(payload) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, callBackURL, reader) - if err != nil { - errs = append(errs, errors.Wrapf(err, "error creating request to %q", callBackURL)) - continue - } - req.Header.Set("Content-Type", "application/json") - resp, err := client.Do(req) - if err != nil { - errs = append( - errs, - errors.Wrapf(err, "error while posting callback to %q", callBackURL), - ) - continue - } - if resp.StatusCode != http.StatusOK { - errs = append( - errs, - fmt.Errorf("error while posting callback to %q: status code %d", callBackURL, resp.StatusCode), - ) - } - } - return errs - }, nil + urls = append(urls, decoded) + } + if len(urls) == 0 && strings.TrimSpace(fallbackURL) != "" { + urls = []string{fallbackURL} + } + if len(urls) == 0 { + return nil, nil + } + return &status.CallbackConfig{URLs: urls, Timeout: timeout}, nil } diff --git a/pkg/server/callback_test.go b/pkg/server/callback_test.go index 2e8fdcd1..19402558 100644 --- a/pkg/server/callback_test.go +++ b/pkg/server/callback_test.go @@ -1,200 +1,64 @@ package server import ( - "context" - "encoding/json" - "net/http" - "net/http/httptest" "net/url" - "reflect" "testing" + "time" - "github.com/gorilla/mux" + "github.com/stretchr/testify/require" ) -func TestParseCallback(t *testing.T) { - ctx := context.Background() - - // Test server setup - type payload struct { - Key string `json:"key"` - Val string `json:"val"` - } - - goodEndpoint1 := "/good1" - goodEndpoint2 := "/good2" - badEndpoint := "/bad" - goodChan1 := make(chan *payload, 5) - goodChan2 := make(chan *payload, 5) - - passToChanHandler := func(ch chan *payload) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - defer func() { - if err := r.Body.Close(); err != nil { - t.Fatalf("can't close r.Body: %v", err) - } - }() - - var data payload - if err := json.NewDecoder(r.Body).Decode(&data); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - } - ch <- &data - if _, err := w.Write(nil); err != nil { - t.Fatalf("unexpected error while writing response from test server: %v", err) - } - } - } - returnErrHandler := http.HandlerFunc( - func(w http.ResponseWriter, _ *http.Request) { - http.Error(w, "bad endpoint error", http.StatusInternalServerError) - }) - - router := mux.NewRouter() - router.Handle(goodEndpoint1, passToChanHandler(goodChan1)) - router.Handle(goodEndpoint2, passToChanHandler(goodChan2)) - router.Handle(badEndpoint, returnErrHandler) - - srv := httptest.NewServer(router) - defer srv.Close() - - t.Run("Test empty callback", - func(t *testing.T) { - values := url.Values{} - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for empty values: %v", err) - } - if err := cb(ctx, nil); err != nil { - t.Fatalf("unexpected error when calling callback for empty values: %v", err) - } - }, - ) - - t.Run("Test invalid callback URL", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - "valid", - "invalid%", - }, - } - _, err := parseCallback(values) - if err == nil { - t.Fatalf("expected error when passing invalid callback URL") - } - }, - ) - - t.Run("Test normal callbacks", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - url.QueryEscape(srv.URL + goodEndpoint1), - url.QueryEscape(srv.URL + goodEndpoint2), - }, - } - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for good endpoints: %v", err) - } - pl := payload{Key: "a", Val: "b"} - if err := cb(ctx, pl); err != nil { - t.Fatalf("unexpected error when calling callbacks for good endpoints: %v", err) - } - - if val1 := <-goodChan1; !reflect.DeepEqual(val1, &pl) { - t.Fatalf("expected %v, got %v", pl, val1) - } - if val2 := <-goodChan2; !reflect.DeepEqual(val2, &pl) { - t.Fatalf("expected %v, got %v", pl, val2) - } - }, - ) - - t.Run("Test bad callback - unresponsive host", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - url.QueryEscape(srv.URL + goodEndpoint1), - url.QueryEscape("invalid.url.local"), - }, - } - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for bad host: %v", err) - } - pl := payload{Key: "c", Val: "d"} - if err := cb(ctx, pl); err == nil { - t.Fatalf("expected error when calling bad host callback") - } +func TestParseCallback_NoCallbackConfigured(t *testing.T) { + r := require.New(t) + cb, err := parseCallback(url.Values{}, "", time.Second) + r.NoError(err) + r.Nil(cb) +} - if val1 := <-goodChan1; !reflect.DeepEqual(val1, &pl) { - t.Fatalf("expected %v, got %v", pl, val1) - } - }, - ) +func TestParseCallback_DecodesAndKeepsEveryQueryURL(t *testing.T) { + r := require.New(t) + cb, err := parseCallback(url.Values{"callback": []string{ + "http://localhost:1/good1", + url.QueryEscape("http://localhost:1/good2?a=b"), + }}, "", time.Second) + r.NoError(err) + r.NotNil(cb) + r.Equal([]string{"http://localhost:1/good1", "http://localhost:1/good2?a=b"}, cb.URLs) + r.Equal(time.Second, cb.Timeout) +} - t.Run("Test bad callback - invalid endpoint", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - url.QueryEscape(srv.URL + goodEndpoint1), - url.QueryEscape(srv.URL + badEndpoint), - }, - } - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for bad endpoint: %v", err) - } - pl := payload{Key: "e", Val: "f"} - if err := cb(ctx, pl); err == nil { - t.Fatalf("expected error when calling bad endpoint callback") - } +func TestParseCallback_RejectsUndecodableURL(t *testing.T) { + r := require.New(t) + cb, err := parseCallback(url.Values{"callback": []string{"%zz"}}, "", time.Second) + r.Error(err) + r.Nil(cb) +} - if val1 := <-goodChan1; !reflect.DeepEqual(val1, &pl) { - t.Fatalf("expected %v, got %v", pl, val1) - } - }, - ) +// general.callback_url is a fallback, an explicit ?callback= always wins. +func TestParseCallback_QueryParamOverridesGlobalCallback(t *testing.T) { + r := require.New(t) + cb, err := parseCallback(url.Values{"callback": []string{"http://localhost:1/from-query"}}, "http://localhost:1/global", time.Second) + r.NoError(err) + r.NotNil(cb) + r.Equal([]string{"http://localhost:1/from-query"}, cb.URLs) +} - t.Run("Test nil context error", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - url.QueryEscape(srv.URL + goodEndpoint1), - }, - } - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for good endpoint: %v", err) - } - pl := payload{} - if err := cb(nil, pl); err == nil { - t.Fatalf("expected error when passing nil context to callback function") - } - }, - ) +func TestParseCallback_GlobalCallbackUsedWhenQueryParamMissing(t *testing.T) { + r := require.New(t) + cb, err := parseCallback(url.Values{}, "http://localhost:1/global", time.Second) + r.NoError(err) + r.NotNil(cb) + r.Equal([]string{"http://localhost:1/global"}, cb.URLs) +} - t.Run("Test bad payload", - func(t *testing.T) { - values := url.Values{ - "callback": []string{ - url.QueryEscape(srv.URL + goodEndpoint1), - }, - } - cb, err := parseCallback(values) - if err != nil { - t.Fatalf("unexpected error when getting callback for good endpoint: %v", err) - } - type recursive struct { - Ref *recursive `json:"Ref"` - } - badPl := recursive{} - badPl.Ref = &badPl - if err := cb(ctx, badPl); err == nil { - t.Fatalf("expected error when passing unmarshalable payload") - } - }, - ) +// `?callback=` with an empty or blank value is treated as absent, not as a +// request to disable the globally configured callback. +func TestParseCallback_EmptyQueryParamFallsBackToGlobal(t *testing.T) { + r := require.New(t) + for _, raw := range []string{"", " ", "%20"} { + cb, err := parseCallback(url.Values{"callback": []string{raw}}, "http://localhost:1/global", time.Second) + r.NoError(err, "raw=%q", raw) + r.NotNil(cb, "raw=%q", raw) + r.Equal([]string{"http://localhost:1/global"}, cb.URLs, "raw=%q", raw) + } } diff --git a/pkg/server/server.go b/pkg/server/server.go index cd3dca77..b314bb09 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -79,6 +79,9 @@ func Run(cliCtx *cli.Context, cliApp *cli.App, configPath string, clickhouseBack cfg *config.Config err error ) + // from here on every re-entry into cliApp comes from an API handler which owns + // its own status row, see status.SetAPIServerMode + status.SetAPIServerMode() log.Debug().Msg("Wait for ClickHouse") for { cfg, err = config.LoadConfig(configPath) @@ -1114,14 +1117,14 @@ func (api *APIServer) httpCreateHandler(w http.ResponseWriter, r *http.Request) fullCommand = fmt.Sprintf("%s %s", fullCommand, backupName) } - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "create", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("create", 0, func() error { b := backup.NewBackuper(cfg) @@ -1130,7 +1133,6 @@ func (api *APIServer) httpCreateHandler(w http.ResponseWriter, r *http.Request) if err != nil { log.Error().Msgf("API /backup/create error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), true); metricsErr != nil { @@ -1138,7 +1140,6 @@ func (api *APIServer) httpCreateHandler(w http.ResponseWriter, r *http.Request) } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusCreated, struct { Status string `json:"status"` @@ -1251,14 +1252,14 @@ func (api *APIServer) httpCreateRemoteHandler(w http.ResponseWriter, r *http.Req fullCommand = fmt.Sprintf("%s %s", fullCommand, backupName) } - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "create_remote", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("create_remote", 0, func() error { b := backup.NewBackuper(cfg) @@ -1267,7 +1268,6 @@ func (api *APIServer) httpCreateRemoteHandler(w http.ResponseWriter, r *http.Req if err != nil { log.Error().Msgf("API /backup/create_remote error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), false); metricsErr != nil { @@ -1275,7 +1275,6 @@ func (api *APIServer) httpCreateRemoteHandler(w http.ResponseWriter, r *http.Req } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusCreated, struct { Status string `json:"status"` @@ -1564,14 +1563,14 @@ func (api *APIServer) httpUploadHandler(w http.ResponseWriter, r *http.Request) fullCommand = fmt.Sprint(fullCommand, " ", name) - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "upload", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("upload", 0, func() error { b := backup.NewBackuper(cfg) @@ -1580,14 +1579,12 @@ func (api *APIServer) httpUploadHandler(w http.ResponseWriter, r *http.Request) if err != nil { log.Error().Msgf("Upload error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), false); metricsErr != nil { log.Error().Stack().Err(metricsErr).Msgf("UpdateBackupMetrics return error") } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -1623,14 +1620,14 @@ func (api *APIServer) httpRebaseHandler(w http.ResponseWriter, r *http.Request) fullCommand := fmt.Sprint("rebase ", name) operationId, _ := uuid.NewUUID() - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "rebase", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("rebase", 0, func() error { b := backup.NewBackuper(cfg) @@ -1639,14 +1636,12 @@ func (api *APIServer) httpRebaseHandler(w http.ResponseWriter, r *http.Request) if err != nil { log.Error().Msgf("Rebase error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), false); metricsErr != nil { log.Error().Stack().Err(metricsErr).Msgf("UpdateBackupMetrics return error") } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -1688,14 +1683,14 @@ func (api *APIServer) httpRebalanceHandler(w http.ResponseWriter, r *http.Reques } operationId, _ := uuid.NewUUID() - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "rebalance", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("rebalance", 0, func() error { b := backup.NewBackuper(cfg) @@ -1704,14 +1699,12 @@ func (api *APIServer) httpRebalanceHandler(w http.ResponseWriter, r *http.Reques if err != nil { log.Error().Msgf("Rebalance error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), false); metricsErr != nil { log.Error().Stack().Err(metricsErr).Msgf("UpdateBackupMetrics return error") } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -1932,14 +1925,14 @@ func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request) name := utils.CleanBackupNameRE.ReplaceAllString(vars["name"], "") fullCommand += fmt.Sprintf(" %s", name) - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "restore", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("restore", 0, func() error { b := backup.NewBackuper(cfg) @@ -1951,10 +1944,8 @@ func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request) status.Current.Stop(commandId, err) if err != nil { log.Error().Msgf("API /backup/restore error: %v", err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -2178,14 +2169,14 @@ func (api *APIServer) httpRestoreRemoteHandler(w http.ResponseWriter, r *http.Re name := utils.CleanBackupNameRE.ReplaceAllString(vars["name"], "") fullCommand += fmt.Sprintf(" %s", name) - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "restore_remote", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("restore_remote", 0, func() error { b := backup.NewBackuper(cfg) @@ -2197,10 +2188,8 @@ func (api *APIServer) httpRestoreRemoteHandler(w http.ResponseWriter, r *http.Re status.Current.Stop(commandId, err) if err != nil { log.Error().Msgf("API /backup/restore_remote error: %v", err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -2281,14 +2270,14 @@ func (api *APIServer) httpDownloadHandler(w http.ResponseWriter, r *http.Request fullCommand += fmt.Sprintf(" %s", name) - callback, err := parseCallback(query) + callback, err := parseCallback(query, cfg.General.CallbackURL, cfg.General.CallbackTimeoutDuration) if err != nil { log.Error().Err(err).Send() api.writeError(w, http.StatusBadRequest, "download", err) return } - commandId, _ := status.Current.StartWithOperationId(fullCommand, operationId.String()) + commandId, _ := status.Current.StartWithCallback(fullCommand, operationId.String(), callback) go func() { err, _ := api.metrics.ExecuteWithMetrics("download", 0, func() error { b := backup.NewBackuper(cfg) @@ -2297,14 +2286,12 @@ func (api *APIServer) httpDownloadHandler(w http.ResponseWriter, r *http.Request if err != nil { log.Error().Msgf("API /backup/download error: %v", err) status.Current.Stop(commandId, err) - api.errorCallback(context.Background(), err, operationId.String(), callback) return } if metricsErr := api.UpdateBackupMetrics(context.Background(), true); metricsErr != nil { log.Error().Stack().Err(metricsErr).Msgf("UpdateBackupMetrics return error") } status.Current.Stop(commandId, nil) - api.successCallback(context.Background(), operationId.String(), callback) }() api.sendJSONEachRow(w, http.StatusOK, struct { Status string `json:"status"` @@ -2584,6 +2571,7 @@ func (api *APIServer) ReloadConfig(w http.ResponseWriter, command string) (*conf api.metrics.NumberBackupsRemoteExpected.Set(float64(cfg.General.BackupsToKeepRemote)) api.metrics.NumberBackupsLocalExpected.Set(float64(cfg.General.BackupsToKeepLocal)) status.SetCancelWaitTimeout(cfg.API.CancelOperationTimeoutDuration) + status.SetMaxFinishedRows(cfg.General.StatusHistorySize) return cfg, nil } diff --git a/pkg/server/utils.go b/pkg/server/utils.go index 8604ab7b..9cf1d589 100644 --- a/pkg/server/utils.go +++ b/pkg/server/utils.go @@ -1,7 +1,6 @@ package server import ( - "context" "encoding/json" "fmt" "net/http" @@ -66,34 +65,3 @@ func (api *APIServer) sendJSONEachRow(w http.ResponseWriter, statusCode int, v i log.Warn().Msgf("%#v doesn't support Flusher interface", w) } } - -// CallbackResponse is the response that is returned to callers -type CallbackResponse struct { - Status string `json:"status"` - Error string `json:"error"` - OperationId string `json:"operation_id"` -} - -// errorCallback executes callbacks with a payload notifying callers that the operation has failed -func (api *APIServer) errorCallback(ctx context.Context, err error, operationId string, callback callbackFn) { - payload := &CallbackResponse{ - Status: "error", - Error: err.Error(), - OperationId: operationId, - } - for _, e := range callback(ctx, payload) { - log.Error().Err(e).Send() - } -} - -// successCallback executes callbacks with a payload notifying callers that the operation succeeded -func (api *APIServer) successCallback(ctx context.Context, operationId string, callback callbackFn) { - payload := &CallbackResponse{ - Status: "success", - Error: "", - OperationId: operationId, - } - for _, e := range callback(ctx, payload) { - log.Error().Err(e).Send() - } -} diff --git a/pkg/status/callback.go b/pkg/status/callback.go new file mode 100644 index 00000000..5c376a09 --- /dev/null +++ b/pkg/status/callback.go @@ -0,0 +1,104 @@ +package status + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/pkg/errors" + "github.com/rs/zerolog/log" +) + +// CallbackPayload is the JSON body posted to callback URLs on command completion. +// Status, Error, and OperationId match the legacy API callback payload for +// backward compatibility (Error has no omitempty so success still sends ""). +// Command and Duration are optional extras. +type CallbackPayload struct { + Status string `json:"status"` + Error string `json:"error"` + OperationId string `json:"operation_id"` + Command string `json:"command,omitempty"` + Duration string `json:"duration,omitempty"` +} + +// CallbackConfig describes where to notify when a command finishes. +// It is attached to a status row by the caller which starts the command +// (API handler, CLI wrapper or watch iteration). +type CallbackConfig struct { + URLs []string + Timeout time.Duration +} + +// DefaultCallbackTimeout is used when CallbackConfig.Timeout is not positive. +const DefaultCallbackTimeout = 5 * time.Second + +// callbackCommands are the command names which produce a completion callback. +// Read-only commands (list, tables, status, ...) and long-running supervisors +// (server, watch) are deliberately absent, watch notifies per iteration instead. +var callbackCommands = map[string]struct{}{ + "create": {}, + "create_remote": {}, + "upload": {}, + "download": {}, + "restore": {}, + "restore_remote": {}, + "delete": {}, + "rebase": {}, + "rebalance": {}, + "clean": {}, + "clean_remote_broken": {}, + "clean_local_broken": {}, + "clean_broken_retention": {}, +} + +// CallbackEligible reports whether a full command line ("create_remote --tables=x name") +// belongs to a command which produces a completion callback. Only the first token matters. +func CallbackEligible(fullCommand string) bool { + name, _, _ := strings.Cut(strings.TrimSpace(fullCommand), " ") + _, ok := callbackCommands[name] + return ok +} + +// notify posts the completion payload to every configured URL. It is invoked +// from Stop in a separate goroutine, so the status lock is never held here and +// a slow or broken receiver can not stall the command which just finished. +func notify(cb *CallbackConfig, payload CallbackPayload) { + timeout := cb.Timeout + if timeout <= 0 { + timeout = DefaultCallbackTimeout + } + for _, callbackURL := range cb.URLs { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + err := SendCallback(ctx, callbackURL, payload) + cancel() + if err != nil { + log.Error().Err(err).Str("callback_url", callbackURL).Str("command", payload.Command).Msg("callback failed") + } + } +} + +// SendCallback POSTs payload as JSON to callbackURL. The caller owns timeouts via ctx. +func SendCallback(ctx context.Context, callbackURL string, payload CallbackPayload) error { + body, err := json.Marshal(payload) + if err != nil { + return errors.Wrap(err, "error encoding callback payload") + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, callbackURL, bytes.NewReader(body)) + if err != nil { + return errors.Wrapf(err, "error creating callback request to %q", callbackURL) + } + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + return errors.Wrapf(err, "error while posting callback to %q", callbackURL) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("error while posting callback to %q: status code %d", callbackURL, resp.StatusCode) + } + return nil +} diff --git a/pkg/status/callback_test.go b/pkg/status/callback_test.go new file mode 100644 index 00000000..0f2408ac --- /dev/null +++ b/pkg/status/callback_test.go @@ -0,0 +1,253 @@ +package status + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestSendCallback_Success(t *testing.T) { + r := require.New(t) + var ( + mu sync.Mutex + gotBody []byte + gotCT string + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + mu.Lock() + defer mu.Unlock() + gotCT = req.Header.Get("Content-Type") + body, err := io.ReadAll(req.Body) + if err != nil { + t.Errorf("read body: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + gotBody = body + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + payload := CallbackPayload{ + Status: SuccessStatus, + Error: "", + OperationId: "op-123", + Command: "create my_backup", + Duration: "1.5s", + } + err := SendCallback(context.Background(), srv.URL, payload) + r.NoError(err) + + mu.Lock() + defer mu.Unlock() + r.Equal("application/json", gotCT) + + var got map[string]interface{} + r.NoError(json.Unmarshal(gotBody, &got)) + r.Equal("success", got["status"]) + r.Equal("", got["error"]) + r.Equal("op-123", got["operation_id"]) + r.Equal("create my_backup", got["command"]) + r.Equal("1.5s", got["duration"]) +} + +func TestSendCallback_HTTPError_DoesNotPanic(t *testing.T) { + r := require.New(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + err := SendCallback(context.Background(), srv.URL, CallbackPayload{ + Status: ErrorStatus, + Error: "boom", + OperationId: "op-err", + }) + r.Error(err) +} + +func TestSendCallback_Timeout(t *testing.T) { + r := require.New(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(200 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + err := SendCallback(ctx, srv.URL, CallbackPayload{ + Status: SuccessStatus, + OperationId: "op-timeout", + }) + r.Error(err) +} + +func TestCallbackEligible(t *testing.T) { + r := require.New(t) + for _, command := range []string{"create", "create_remote my_backup", "restore_remote --tables=db.t my_backup", "clean_remote_broken"} { + r.True(CallbackEligible(command), "command %q must notify", command) + } + for _, command := range []string{"", "list", "list remote", "tables", "status", "watch", "server", "create_lightweight"} { + r.False(CallbackEligible(command), "command %q must not notify", command) + } +} + +func TestStop_SendsCallbackOnSuccess(t *testing.T) { + r := require.New(t) + payloads, srv := newCallbackReceiver(t) + defer srv.Close() + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("create_remote my_backup", "op-1", &CallbackConfig{URLs: []string{srv.URL}, Timeout: 2 * time.Second}) + s.Stop(commandId, nil) + + got := awaitPayload(t, payloads) + r.Equal(SuccessStatus, got.Status) + r.Equal("", got.Error) + r.Equal("op-1", got.OperationId) + r.Equal("create_remote my_backup", got.Command) + r.NotEmpty(got.Duration) +} + +func TestStop_SendsCallbackOnError(t *testing.T) { + r := require.New(t) + payloads, srv := newCallbackReceiver(t) + defer srv.Close() + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("create_remote my_backup", "op-2", &CallbackConfig{URLs: []string{srv.URL}, Timeout: 2 * time.Second}) + s.Stop(commandId, errors.New("disk is full")) + + got := awaitPayload(t, payloads) + r.Equal(ErrorStatus, got.Status) + r.Equal("disk is full", got.Error) +} + +// Every configured URL is notified, the API accepts a repeated ?callback= param. +func TestStop_SendsCallbackToEveryURL(t *testing.T) { + r := require.New(t) + payloads1, srv1 := newCallbackReceiver(t) + defer srv1.Close() + payloads2, srv2 := newCallbackReceiver(t) + defer srv2.Close() + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("upload my_backup", "op-3", &CallbackConfig{URLs: []string{srv1.URL, srv2.URL}, Timeout: 2 * time.Second}) + s.Stop(commandId, nil) + + r.Equal("op-3", awaitPayload(t, payloads1).OperationId) + r.Equal("op-3", awaitPayload(t, payloads2).OperationId) +} + +func TestStop_NoCallbackForReadOnlyCommand(t *testing.T) { + r := require.New(t) + payloads, srv := newCallbackReceiver(t) + defer srv.Close() + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("list remote", "op-4", &CallbackConfig{URLs: []string{srv.URL}, Timeout: 2 * time.Second}) + s.Stop(commandId, nil) + + r.Nil(s.byId[commandId].callback) + expectNoPayload(t, payloads) +} + +// A broken receiver is logged and swallowed, the command result is unaffected +// and the caller is never blocked by callback IO. +func TestStop_BrokenReceiverDoesNotBlockCommand(t *testing.T) { + r := require.New(t) + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + <-release + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + defer close(release) + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("create my_backup", "op-5", &CallbackConfig{URLs: []string{srv.URL}, Timeout: 5 * time.Second}) + + done := make(chan struct{}) + go func() { + s.Stop(commandId, nil) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + r.Fail("Stop blocked on a slow callback receiver") + } + r.Equal(SuccessStatus, s.GetStatusByOperationId("op-5")[0].Status) +} + +// A killed command still owes its caller exactly one notification, sent when the +// command goroutine finally returns and calls Stop. +func TestStop_NotifiesOnceAfterCancel(t *testing.T) { + r := require.New(t) + payloads, srv := newCallbackReceiver(t) + defer srv.Close() + + s := &AsyncStatus{} + commandId, _ := s.StartWithCallback("restore my_backup", "op-6", &CallbackConfig{URLs: []string{srv.URL}, Timeout: 2 * time.Second}) + go s.Stop(commandId, nil) + _, err := s.Cancel("restore my_backup", errors.New("canceled by user")) + r.NoError(err) + + got := awaitPayload(t, payloads) + r.Equal(CancelStatus, got.Status) + r.Equal("canceled by user", got.Error) + + // second Stop must stay silent + s.Stop(commandId, nil) + expectNoPayload(t, payloads) +} + +func newCallbackReceiver(t *testing.T) (chan CallbackPayload, *httptest.Server) { + t.Helper() + payloads := make(chan CallbackPayload, 8) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + body, err := io.ReadAll(req.Body) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + var p CallbackPayload + if err := json.Unmarshal(body, &p); err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + payloads <- p + w.WriteHeader(http.StatusOK) + })) + return payloads, srv +} + +func awaitPayload(t *testing.T, payloads chan CallbackPayload) CallbackPayload { + t.Helper() + select { + case p := <-payloads: + return p + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for callback") + return CallbackPayload{} + } +} + +func expectNoPayload(t *testing.T, payloads chan CallbackPayload) { + t.Helper() + select { + case p := <-payloads: + t.Fatalf("unexpected callback %+v", p) + case <-time.After(300 * time.Millisecond): + } +} diff --git a/pkg/status/status.go b/pkg/status/status.go index 13482349..873c4120 100644 --- a/pkg/status/status.go +++ b/pkg/status/status.go @@ -5,6 +5,7 @@ import ( stderrors "errors" "strings" "sync" + "sync/atomic" "time" "github.com/Altinity/clickhouse-backup/v2/pkg/common" @@ -23,11 +24,52 @@ var Current = &AsyncStatus{} const NotFromAPI = int(-1) +// apiServerMode is set once when the API server starts. The server re-enters the +// same cli.App in process for POST /backup/actions, and every such re-entry comes +// from a handler which already owns its status row — or deliberately has none, +// when the command is listed in api.backup_actions_skip_commands. Either way the +// CLI wrapper must not register anything of its own in this process. +var apiServerMode atomic.Bool + +// SetAPIServerMode marks this process as an API server. Never reset. +func SetAPIServerMode() { + apiServerMode.Store(true) +} + +// APIServerMode reports whether this process runs the API server. +func APIServerMode() bool { + return apiServerMode.Load() +} + type AsyncStatus struct { - commands []ActionRow + // commands is the ordered history. Rows are addressed by a monotonic id + // rather than by position, because trimLocked drops finished rows from + // anywhere in the history while ids handed out earlier must stay valid. + commands []*ActionRow + byId map[int]*ActionRow + nextId int sync.RWMutex } +// DefaultMaxFinishedRows is used until SetMaxFinishedRows is called from a loaded +// config, so a status list created before any config is read is still bounded. +const DefaultMaxFinishedRows = 1000 + +// maxFinishedRows bounds how many finished rows are kept in memory. Long running +// `watch` processes register one row per iteration, so without a bound the history +// grows for as long as the process lives, see +// https://github.com/Altinity/clickhouse-backup/issues/1481 +var maxFinishedRows = DefaultMaxFinishedRows + +// SetMaxFinishedRows applies general.status_history_size. Safe to call from the +// API server on config reload, and from `watch` before its first iteration. +// Non-positive values are ignored, ValidateConfig already rejects them. +func SetMaxFinishedRows(n int) { + if n > 0 { + maxFinishedRows = n + } +} + type ActionRowStatus struct { Command string `json:"command"` Status string `json:"status"` @@ -39,12 +81,19 @@ type ActionRowStatus struct { type ActionRow struct { ActionRowStatus + // id is the value handed to callers as commandId, stable for the row's life. + id int Ctx context.Context Cancel context.CancelFunc // Done is closed by Stop when the command goroutine has fully returned. // Cancel/CancelAll wait on this so callers know the operation really // finished (e.g. defers like pidlock.RemovePidFile have already run). Done chan struct{} + // startedAt is the monotonic counterpart of ActionRowStatus.Start, used to + // report an exact duration in the completion callback. + startedAt time.Time + // callback is nil when the command must not produce a completion callback. + callback *CallbackConfig } // CancelWaitTimeout bounds how long Cancel/CancelAll wait for the command @@ -67,23 +116,97 @@ func (status *AsyncStatus) Start(command string) (int, context.Context) { } func (status *AsyncStatus) StartWithOperationId(command string, operationId string) (int, context.Context) { + return status.StartWithCallback(command, operationId, nil) +} + +// StartWithCallback registers a command and attaches the completion callback +// configuration to it. Passing a nil callback, an empty URL list or a command +// which is not CallbackEligible means no callback is sent when it finishes. +func (status *AsyncStatus) StartWithCallback(command string, operationId string, callback *CallbackConfig) (int, context.Context) { status.Lock() defer status.Unlock() + now := time.Now() ctx, cancel := context.WithCancel(context.Background()) - status.commands = append(status.commands, ActionRow{ + if callback != nil && (len(callback.URLs) == 0 || !CallbackEligible(command)) { + callback = nil + } + if status.byId == nil { + status.byId = map[int]*ActionRow{} + } + row := &ActionRow{ ActionRowStatus: ActionRowStatus{ Command: command, - Start: time.Now().Format(common.TimeFormat), + Start: now.Format(common.TimeFormat), Status: InProgressStatus, OperationId: operationId, }, - Ctx: ctx, - Cancel: cancel, - Done: make(chan struct{}), - }) - lastCommandId := len(status.commands) - 1 - log.Debug().Msgf("api.status.Start -> status.commands[%d] == %+v", lastCommandId, status.commands[lastCommandId]) - return lastCommandId, ctx + id: status.nextId, + Ctx: ctx, + Cancel: cancel, + Done: make(chan struct{}), + startedAt: now, + callback: callback, + } + status.nextId++ + status.commands = append(status.commands, row) + status.byId[row.id] = row + log.Debug().Msgf("api.status.Start -> status.commands[%d] == %+v", row.id, *row) + status.trimLocked() + return row.id, ctx +} + +// trimLocked drops the oldest finished rows once more than maxFinishedRows of +// them accumulate. Rows still in progress are always kept regardless of age, so +// a long living `watch` or `server` row does not block trimming of the finished +// iterations recorded after it. Lock MUST be held. +func (status *AsyncStatus) trimLocked() { + finished := 0 + for _, row := range status.commands { + if trimmableLocked(row) { + finished++ + } + } + drop := finished - maxFinishedRows + if drop <= 0 { + return + } + kept := make([]*ActionRow, 0, len(status.commands)-drop) + for _, row := range status.commands { + if drop > 0 && trimmableLocked(row) { + delete(status.byId, row.id) + drop-- + continue + } + kept = append(kept, row) + } + status.commands = kept +} + +// trimmableLocked reports whether a row can be forgotten. A terminal status is +// not enough: Cancel/CancelAll mark a row canceled while its command goroutine +// is still running, and that goroutine still has to reach Stop to close Done, +// release the Cancel waiter and send the callback it owes. Dropping the row +// before that would strand /backup/kill for CancelWaitTimeout and lose the +// notification. Lock MUST be held. +func trimmableLocked(row *ActionRow) bool { + if row.Status == InProgressStatus { + return false + } + if row.Done == nil { + return true + } + select { + case <-row.Done: + return true + default: + return false + } +} + +// rowLocked resolves a commandId to a row, or nil when it was already trimmed +// or never existed. Lock MUST be held. +func (status *AsyncStatus) rowLocked(commandId int) *ActionRow { + return status.byId[commandId] } func (status *AsyncStatus) CheckCommandInProgress(command string) bool { @@ -119,40 +242,81 @@ func (status *AsyncStatus) GetContextWithCancel(commandId int) (context.Context, ctx, cancel := context.WithCancel(context.Background()) return ctx, cancel, nil } - if commandId >= len(status.commands) { + row := status.rowLocked(commandId) + if row == nil { return nil, nil, errors.Errorf("commandId=%d not exists in current running commands", commandId) } - if status.commands[commandId].Ctx == nil { - return nil, nil, errors.Errorf("commands[%d]=%s have nil context ", commandId, status.commands[commandId].Command) + if row.Ctx == nil { + return nil, nil, errors.Errorf("commands[%d]=%s have nil context ", commandId, row.Command) } // for create_remote and restore_remote API call - if stderrors.Is(status.commands[commandId].Ctx.Err(), context.Canceled) && strings.Contains(status.commands[commandId].Command, "_remote") { - status.commands[commandId].Ctx, status.commands[commandId].Cancel = context.WithCancel(context.Background()) + if stderrors.Is(row.Ctx.Err(), context.Canceled) && strings.Contains(row.Command, "_remote") { + row.Ctx, row.Cancel = context.WithCancel(context.Background()) } - return status.commands[commandId].Ctx, status.commands[commandId].Cancel, nil + return row.Ctx, row.Cancel, nil } func (status *AsyncStatus) Stop(commandId int, err error) { status.Lock() - defer status.Unlock() + row := status.rowLocked(commandId) + if row == nil { + status.Unlock() + log.Warn().Msgf("api.status.stop -> commandId=%d not found", commandId) + return + } // Always signal "goroutine finished" to any Cancel waiter, even if the // row was already moved to cancel/error/success state by a concurrent // Cancel() call. - closeDoneLocked(&status.commands[commandId]) - if status.commands[commandId].Status != InProgressStatus { + closeDoneLocked(row) + if row.Status != InProgressStatus { + // Already terminal, typically moved to CancelStatus by Cancel/CancelAll. + // The callback is still owed to the caller, Cancel() itself does not send + // it because the command goroutine has not returned yet at that point. + callback, payload := status.finishLocked(row) + status.Unlock() + if callback != nil { + go notify(callback, payload) + } return } - status.commands[commandId].Cancel() + row.Cancel() s := SuccessStatus if err != nil { s = ErrorStatus - status.commands[commandId].Error = err.Error() + row.Error = err.Error() + } + row.Status = s + row.Finish = time.Now().Format(common.TimeFormat) + row.Ctx = nil + row.Cancel = nil + log.Debug().Msgf("api.status.stop -> status.commands[%d] == %+v", commandId, *row) + + callback, payload := status.finishLocked(row) + status.Unlock() + + // Fired outside the lock and asynchronously, a slow or broken callback + // receiver must never stall or fail the command which just finished. + if callback != nil { + go notify(callback, payload) + } +} + +// finishLocked builds the completion callback for a row which reached a terminal +// state and clears it, so a command notifies at most once. It returns a nil +// config when the row has no callback or was already notified. Lock MUST be held. +func (status *AsyncStatus) finishLocked(row *ActionRow) (*CallbackConfig, CallbackPayload) { + callback := row.callback + if callback == nil { + return nil, CallbackPayload{} + } + row.callback = nil + return callback, CallbackPayload{ + Status: row.Status, + Error: row.Error, + OperationId: row.OperationId, + Command: row.Command, + Duration: time.Since(row.startedAt).String(), } - status.commands[commandId].Status = s - status.commands[commandId].Finish = time.Now().Format(common.TimeFormat) - status.commands[commandId].Ctx = nil - status.commands[commandId].Cancel = nil - log.Debug().Msgf("api.status.stop -> status.commands[%d] == %+v", commandId, status.commands[commandId]) } // closeDoneLocked closes row.Done idempotently. Must be called with the @@ -325,3 +489,8 @@ func (status *AsyncStatus) GetStatusByOperationId(operationId string) []ActionRo } return make([]ActionRowStatus, 0) } + +// ResetAPIServerModeForTest clears the API server marker. Tests only. +func ResetAPIServerModeForTest() { + apiServerMode.Store(false) +} diff --git a/pkg/status/status_test.go b/pkg/status/status_test.go index 08977c93..f654f6f8 100644 --- a/pkg/status/status_test.go +++ b/pkg/status/status_test.go @@ -130,3 +130,83 @@ func TestSetCancelWaitTimeout(t *testing.T) { SetCancelWaitTimeout(-1 * time.Second) r.Equal(42*time.Second, CancelWaitTimeout) } + +// Long running `watch` registers one row per iteration, so finished rows must be +// dropped while already handed out commandIds keep resolving to their own row. +func TestTrimFinishedRows_KeepsCommandIdStable(t *testing.T) { + r := require.New(t) + defer func(old int) { maxFinishedRows = old }(maxFinishedRows) + maxFinishedRows = 5 + + s := &AsyncStatus{} + // a `watch`/`server` row stays in progress for the whole process lifetime and + // must not keep the finished iterations recorded after it alive + stuckId, _ := s.Start("watch") + firstIterationId, _ := s.Start("create_remote iteration-0") + s.Stop(firstIterationId, nil) + var lastId int + for i := 1; i < 50; i++ { + lastId, _ = s.Start("create_remote iteration") + s.Stop(lastId, nil) + } + + s.RLock() + kept := len(s.commands) + s.RUnlock() + // trimming runs on Start, so the row finished after the last Start may linger, + // plus the one still in progress + r.LessOrEqual(kept, maxFinishedRows+2, "finished rows must be trimmed") + + // the in progress row is never dropped, and its id still resolves + _, _, err := s.GetContextWithCancel(stuckId) + r.NoError(err) + + // the newest row still resolves to itself, not to a shifted neighbour + _, _, err = s.GetContextWithCancel(lastId) + r.Error(err, "finished rows have no context") + r.Equal(SuccessStatus, s.GetStatus(false, "iteration", 1)[0].Status) + + // a trimmed id resolves to nothing instead of hitting the wrong row + _, _, err = s.GetContextWithCancel(firstIterationId) + r.Error(err) + r.Empty(s.GetStatus(false, "iteration-0", 1), "trimmed row must be gone from history") +} + +// A row canceled via /backup/kill is terminal but its command goroutine may still +// be running, and that goroutine must still reach Stop to close Done and release +// the Cancel waiter. Trimming it early would strand Cancel for CancelWaitTimeout. +func TestTrimKeepsCanceledRowUntilGoroutineReturns(t *testing.T) { + r := require.New(t) + defer func(old int) { maxFinishedRows = old }(maxFinishedRows) + maxFinishedRows = 1 + // Cancel waits for the goroutine, which in this test only returns later + defer func(old time.Duration) { CancelWaitTimeout = old }(CancelWaitTimeout) + CancelWaitTimeout = 100 * time.Millisecond + + s := &AsyncStatus{} + canceledId, _ := s.Start("restore my_backup") + _, err := s.Cancel("restore my_backup", stderrors.New("canceled by user")) + r.NoError(err) + + // push far past the limit while the canceled goroutine has not returned yet + for i := 0; i < 20; i++ { + id, _ := s.Start("create iteration") + s.Stop(id, nil) + } + + s.RLock() + row := s.byId[canceledId] + s.RUnlock() + r.NotNil(row, "canceled row must survive trimming until its goroutine returns") + + // once the goroutine reaches Stop the row becomes trimmable + s.Stop(canceledId, nil) + for i := 0; i < 20; i++ { + id, _ := s.Start("create iteration") + s.Stop(id, nil) + } + s.RLock() + _, stillThere := s.byId[canceledId] + s.RUnlock() + r.False(stillThere, "canceled row must be trimmed after its goroutine returned") +} diff --git a/test/integration/kill_test.go b/test/integration/kill_test.go index 2aebc464..a8e91671 100644 --- a/test/integration/kill_test.go +++ b/test/integration/kill_test.go @@ -11,17 +11,100 @@ import ( "testing" "time" + "github.com/Altinity/clickhouse-backup/v2/pkg/status" "github.com/rs/zerolog/log" "github.com/stretchr/testify/require" ) -// TestKill reproduces https://github.com/Altinity/clickhouse-backup/issues/1365. -// An `upload` action is started via the REST API and killed while in-progress; the -// .pid file must be removed by the kill handler so that subsequent operations -// (delete in particular) do not falsely report "another command is already running". +// assertActionCanceled fails unless the killed command's row in /backup/actions +// ended as "cancel". +// +// NOTE: this proves the kill handler reached the row, NOT that the command obeyed +// the cancellation. status.Cancel sets CancelStatus itself and status.Stop leaves +// an already terminal row alone, so the row reads "cancel" even when the worker +// ignored its context and ran to completion (verified by re-running these tests +// against the pre-fix c.Int("command-id") lookup: they still passed). The checks +// that actually have teeth are assertBackupNotComplete, assertBackupAbsent and the +// row-count check in TestKillRestore — a command which was not canceled finishes +// its work and leaves a complete artifact behind. +func assertActionCanceled(r *require.Assertions, env *TestEnvironment, cmdPrefix, nameNeedle string, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for { + out, err := env.DockerExecOut("clickhouse-backup", "bash", "-ce", + execCurlWithFailBody("'http://127.0.0.1:7171/backup/actions'")) + r.NoError(err) + row := "" + for _, line := range strings.Split(out, "\n") { + if strings.Contains(line, `"command":"`+cmdPrefix) && strings.Contains(line, nameNeedle) { + row = line + } + } + switch { + case strings.Contains(row, `"status":"`+status.CancelStatus+`"`): + return + case strings.Contains(row, `"status":"`+status.SuccessStatus+`"`): + r.FailNow(fmt.Sprintf( + "%s ... %s ended as %q instead of %q: the command ran to completion, so /backup/kill did not cancel it\nrow: %s", + cmdPrefix, nameNeedle, status.SuccessStatus, status.CancelStatus, row)) + case time.Now().After(deadline): + r.FailNow(fmt.Sprintf( + "timeout waiting for %s ... %s to reach status %q\nlast row: %s\nall rows:\n%s", + cmdPrefix, nameNeedle, status.CancelStatus, row, out)) + } + time.Sleep(300 * time.Millisecond) + } +} + +// assertBackupNotComplete proves the killed command never finished writing its +// backup: no row in system.backup_list for it is a complete one. +// +// `desc` carries the data format for a usable backup and the GetLocalBackups +// "broken ..." reason for an unusable one, so "no non-broken row" covers both +// legitimate post-kill states — the directory was never created, or it exists +// without the metadata.json that create/download/upload write last. A command +// which ignored cancellation runs to completion and leaves a complete backup. +// +// Unlike a `delete` based check this does not depend on how far the command got +// before the kill landed: deleting works for a broken backup too, so its result +// cannot tell a canceled command from a finished one. +func assertBackupNotComplete(r *require.Assertions, env *TestEnvironment, location, backupName string) { + var complete uint64 + r.NoError(env.ch.SelectSingleRowNoCtx(&complete, + "SELECT count() FROM system.backup_list WHERE location=? AND name=? AND desc NOT LIKE '%broken%' SETTINGS empty_result_for_aggregation_by_empty_set=0", + location, backupName)) + r.Equal(uint64(0), complete, + "%s backup %q is complete after kill, so the killed command ran to completion instead of being canceled", + location, backupName) +} + +// assertBackupAbsent is the stricter variant used for `create` only: a canceled +// create removes the partial backup it was building, so nothing must be left at +// all. Do not use it for download/upload, which have no such cleanup — there the +// leftovers are legitimate and only assertBackupNotComplete applies. +// +// It doubles as the stale pid lock regression check (issue #1365): a leftover pid +// file makes delete answer "another clickhouse-backup ..." instead. +func assertBackupAbsent(r *require.Assertions, env *TestEnvironment, where, backupName string) { + out, _ := postActionAllowError(env, fmt.Sprintf("delete %s %s", where, backupName)) + r.NotContains(out, "another clickhouse-backup", "delete must not see a stale pid lock: %s", out) + r.Contains(out, fmt.Sprintf("is not found on %s storage", where), + "the killed command ran to completion and left a usable %s backup %q, so /backup/kill did not cancel it: %s", + where, backupName, out) +} + +// TestKillUpload reproduces https://github.com/Altinity/clickhouse-backup/issues/1365. +// An `upload` is started via the REST API and killed while in-progress; the .pid +// file must be removed by the kill handler so that subsequent operations (delete in +// particular) do not falsely report "another command is already running". // Also verifies that /backup/kill blocks until the upload goroutine actually // finished (sync wait), observable via the upload_finish metric. -func TestKill(t *testing.T) { +// +// It is not just "the upload variant" of the other TestKill* tests: it is the only +// one exercising the `GET /backup/kill` endpoint and the only one killing a command +// started through a dedicated REST endpoint (POST /backup/upload/{name}, which keeps +// the commandId in process). The others start and kill through POST /backup/actions, +// which re-enters the CLI app with --command-id. Keep both paths covered. +func TestKillUpload(t *testing.T) { env, r := NewTestEnvironment(t) env.connectWithWait(t, r, 0*time.Second, 1*time.Second, 1*time.Minute) r.NoError(env.DockerCP("configs/config-s3.yml", "clickhouse-backup:/etc/clickhouse-backup/config.yml")) @@ -50,7 +133,7 @@ func TestKill(t *testing.T) { }() defer func() { if out, err := env.DockerExecOut("clickhouse-backup", "clickhouse-backup", "delete", "remote", backupName); err != nil && !strings.Contains(out, fmt.Sprintf("'%s' is not found on remote storage", backupName)) { - t.Errorf("TestKill teardown error=%+v: delete remote %s: %s", err, backupName, out) + t.Errorf("TestKillUpload teardown error=%+v: delete remote %s: %s", err, backupName, out) } // The killed mid-flight upload leaves an incomplete remote backup that // `delete remote` reports as "not found" and never removes. Purge the @@ -59,7 +142,7 @@ func TestKill(t *testing.T) { // asserts the remote backup path is empty). _ = env.DockerExec("minio", "rm", "-rf", env.minioBackupFSPath(r, "config-s3.yml", backupName)) if err := env.dropDatabase(dbName, true); err != nil { - t.Errorf("TestKill teardown: drop database %s, error=%+v", dbName, err) + t.Errorf("TestKillUpload teardown: drop database %s, error=%+v", dbName, err) } }() time.Sleep(3 * time.Second) @@ -119,6 +202,9 @@ func TestKill(t *testing.T) { "sync wait did not block until the upload goroutine returned", finishBefore, finishAfter) + // The upload must have ended canceled, not completed — see assertActionCanceled. + assertActionCanceled(r, env, "upload", backupName, 10*time.Second) + // 5. Pid file must be gone immediately after kill — this is the regression check. checkOut, _ := env.DockerExecOut("clickhouse-backup", "bash", "-ce", "if [ -f "+pidPath+" ]; then echo EXISTS; cat "+pidPath+"; else echo GONE; fi") @@ -133,6 +219,12 @@ func TestKill(t *testing.T) { "delete must not see a stale pid lock: %s", deleteOut) r.NotContains(deleteOut, "\"status\":\"error\"", "delete must succeed: %s", deleteOut) + // 7. The killed upload must not have finished: Upload writes the remote + // metadata.json after every data file, so an interrupted one leaves either no + // remote backup at all or one the listing reports as broken. An upload which + // ignored cancellation runs to completion and leaves a usable remote backup. + assertBackupNotComplete(r, env, "remote", backupName) + // Remote backup, database, and env-pool return are handled by the defers // registered above so they run on both success and mid-test failure. } @@ -185,8 +277,17 @@ func TestKillDownload(t *testing.T) { // 2. start download and kill it mid-flight (start happens inside observeInProgressAndKill). observeInProgressAndKill(r, env, "download "+backupName, backupName, "download", 15*time.Second) - // 3. a follow-up delete must not trip on a stale pid lock. - delOut = postAction(r, env, "delete local "+backupName) + // 3. The killed download must not have finished: Download writes the local + // metadata.json as its very last step, so an interrupted one leaves either no + // backup directory at all or a directory GetLocalBackups reports as broken. + // A download which ignored cancellation runs to completion and shows up as a + // regular, complete local backup instead. + assertBackupNotComplete(r, env, "local", backupName) + + // 4. a follow-up delete must not trip on a stale pid lock. It has to tolerate + // both post-kill states: the backup directory is removed when it exists, and + // reported as missing when the kill landed before it was created. + delOut, _ = postActionAllowError(env, "delete local "+backupName) r.NotContains(delOut, "another clickhouse-backup", "delete must not see a stale pid lock: %s", delOut) } @@ -222,9 +323,9 @@ func TestKillCreate(t *testing.T) { // finish before the kill is issued. observeInProgressAndKill(r, env, fmt.Sprintf("create --tables=%s.* %s", dbName, backupName), backupName, "create", 15*time.Second) - // a follow-up delete must not trip on a stale pid lock. - delOut := postAction(r, env, "delete local "+backupName) - r.NotContains(delOut, "another clickhouse-backup", "delete must not see a stale pid lock: %s", delOut) + // The canceled create must have removed the partial backup it was building, + // and the follow-up delete must not trip on a stale pid lock. + assertBackupAbsent(r, env, "local", backupName) } // TestKillRestore kills an in-progress restore and verifies the restore @@ -256,6 +357,9 @@ func TestKillRestore(t *testing.T) { time.Sleep(3 * time.Second) // create a local backup, drop the table so restore has to recreate+attach. + var fullRows uint64 + r.NoError(env.ch.SelectSingleRowNoCtx(&fullRows, fmt.Sprintf("SELECT count() FROM %s.t1 SETTINGS empty_result_for_aggregation_by_empty_set=0", dbName))) + r.Greater(fullRows, uint64(0), "the table to restore must not be empty") runActionWait(r, env, fmt.Sprintf("create --tables=%s.* %s", dbName, backupName), "create", backupName, 60*time.Second) // SYNC keyword not supported before 21.x dropSQL := fmt.Sprintf("DROP TABLE %s.t1", dbName) @@ -267,6 +371,16 @@ func TestKillRestore(t *testing.T) { // start happens inside observeInProgressAndKill so a fast restore cannot // finish before the kill is issued. observeInProgressAndKill(r, env, "restore "+backupName, backupName, "restore", 15*time.Second) + + // The canceled restore must not have put the whole table back. Either the + // table was never re-attached (query fails) or it holds fewer rows than the + // backup contains. A restore which ignored cancellation finishes and the row + // count matches the original exactly. + var restoredRows uint64 + if err := env.ch.SelectSingleRowNoCtx(&restoredRows, fmt.Sprintf("SELECT count() FROM %s.t1 SETTINGS empty_result_for_aggregation_by_empty_set=0", dbName)); err == nil { + r.Less(restoredRows, fullRows, + "the killed restore completed (%d of %d rows restored), so /backup/kill did not cancel it", restoredRows, fullRows) + } } // readUploadFinishMetric scrapes /metrics and parses the value of @@ -327,6 +441,14 @@ func postAction(r *require.Assertions, env *TestEnvironment, command string) str return out } +// postActionAllowError is postAction for calls whose HTTP status is part of what +// the test inspects, so a 4xx/5xx body is returned instead of failing the test. +func postActionAllowError(env *TestEnvironment, command string) (string, error) { + body := fmt.Sprintf(`{"command":%q}`, command) + return env.DockerExecOut("clickhouse-backup", "bash", "-ce", + execCurlWithFailBody("-XPOST 'http://127.0.0.1:7171/backup/actions' -d '"+body+"'")) +} + // runActionWait starts an async action and blocks until it reports success. func runActionWait(r *require.Assertions, env *TestEnvironment, command, cmdPrefix, nameNeedle string, timeout time.Duration) { out := postAction(r, env, command) @@ -451,6 +573,9 @@ func observeInProgressAndKill(r *require.Assertions, env *TestEnvironment, comma "the %s goroutine did not return", metricCommand, finishBefore, finishAfter, metricCommand) r.Contains(out, "PID=GONE", "pid file %s must be removed by kill:\n%s", pidPath, out) + + cmdPrefix, _, _ := strings.Cut(command, " ") + assertActionCanceled(r, env, cmdPrefix, backupName, 10*time.Second) } // scriptField returns the value after the first line of observe+kill script diff --git a/test/integration/serverAPI_test.go b/test/integration/serverAPI_test.go index 665f222b..44593e8e 100644 --- a/test/integration/serverAPI_test.go +++ b/test/integration/serverAPI_test.go @@ -672,21 +672,40 @@ func testAPIBackupActionsSkipCommands(t *testing.T, r *require.Assertions, env * var createRows uint64 runClickHouseClientInsertSystemBackupActions(r, env, []string{"create skip_commands_test"}, true) r.NoError(env.ch.SelectSingleRowNoCtx(&createRows, "SELECT count() FROM system.backup_actions WHERE command='create skip_commands_test' AND status=?", status.SuccessStatus)) + actionsDump := "" if createRows != 1 { - // The command was recorded but ended non-success (CI-only flake). Surface the - // actual status/error and the server log so the root cause is diagnosable. - createActions := make([]struct { - Status string `ch:"status"` - Error string `ch:"error"` - }, 0) - r.NoError(env.ch.StructSelect(&createActions, "SELECT status, error FROM system.backup_actions WHERE command='create skip_commands_test'")) + // Either the command ended non-success, or it was recorded more than once + // (e.g. registered both by the API handler and by the CLI it re-enters). + // Dump every row so the assertion message says which of the two happened. + actionsDump = dumpBackupActions(r, env) logOut, _ := env.DockerExecOut("clickhouse-backup", "bash", "-ce", "tail -n 200 /tmp/clickhouse-backup-server.log") - log.Error().Msgf("create skip_commands_test did not succeed, actions=%+v\nclickhouse-backup server log tail:\n%s", createActions, logOut) + log.Error().Msgf("unexpected `create skip_commands_test` rows\n%s\nclickhouse-backup server log tail:\n%s", actionsDump, logOut) } - r.Equal(uint64(1), createRows, "non-skipped commands must still be recorded in system.backup_actions") + r.Equal(uint64(1), createRows, "`create skip_commands_test` must be recorded in system.backup_actions exactly once with status=%s, got %d\n%s", status.SuccessStatus, createRows, actionsDump) runClickHouseClientInsertSystemBackupActions(r, env, []string{"delete local skip_commands_test"}, false) } +// dumpBackupActions renders the whole in-memory async status list, so an assertion +// about system.backup_actions can report what is actually recorded there instead of +// just a row count. +func dumpBackupActions(r *require.Assertions, env *TestEnvironment) string { + rows := make([]struct { + Command string `ch:"command"` + Status string `ch:"status"` + Start string `ch:"start"` + Finish string `ch:"finish"` + Error string `ch:"error"` + OperationId string `ch:"operation_id"` + }, 0) + r.NoError(env.ch.StructSelect(&rows, "SELECT command, status, start, finish, error, operation_id FROM system.backup_actions ORDER BY start")) + dump := fmt.Sprintf("system.backup_actions (%d rows):", len(rows)) + for _, row := range rows { + dump += fmt.Sprintf("\n command=%q status=%q start=%q finish=%q operation_id=%q error=%q", + row.Command, row.Status, row.Start, row.Finish, row.OperationId, row.Error) + } + return dump +} + func testAPIMetrics(r *require.Assertions, env *TestEnvironment) { log.Debug().Msg("Check /metrics clickhouse_backup_last_backup_size_remote") var lastRemoteSize uint64 diff --git a/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot b/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot index 95684da3..c5b1c579 100644 --- a/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot +++ b/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot @@ -1,4 +1,4 @@ -default_config = r"""'[\'general:\', \' remote_storage: none\', \' backups_to_keep_local: 0\', \' backups_to_keep_remote: 0\', \' log_level: info\', \' disable_environment_override: false\', \' allow_empty_backups: false\', \' rebase_before_remove_old_remote: false\', \' rebase_during_delete: false\', \' pipe_buffer_size: 131072\', \' download_copy_buffer_size: 0\', \' compression_use_multi_thread: true\', \' compression_threads: 0\', \' compression_buffer_size: 0\', \' allow_object_disk_streaming: false\', \' use_resumable_state: true\', \' restore_schema_on_cluster: ""\', \' upload_by_part: true\', \' download_by_part: true\', \' restore_database_mapping: {}\', \' restore_table_mapping: {}\', \' retries_on_failure: 3\', \' retries_pause: 5s\', \' retries_jitter: 0\', \' watch_interval: 1h\', \' full_interval: 24h\', \' watch_backup_name_template: shard{shard}-{type}-{time:20060102150405}\', \' watch_schedules: []\', \' sharded_operation_mode: ""\', \' cpu_nice_priority: 15\', \' io_nice_priority: idle\', \' rbac_backup_always: true\', \' rbac_conflict_resolution: recreate\', \' config_backup_always: false\', \' named_collections_backup_always: false\', \' delete_batch_size: 1000\', \' retriesduration: 5s\', \' watchduration: 1h0m0s\', \' fullduration: 24h0m0s\', \'clickhouse:\', \' username: default\', \' password: ""\', \' host: localhost\', \' port: 9000\', \' disk_mapping: {}\', \' skip_tables:\', \' - system.*\', \' - INFORMATION_SCHEMA.*\', \' - information_schema.*\', \' - _temporary_and_external_tables.*\', \' skip_table_engines: []\', \' skip_disks: []\', \' skip_disk_types: []\', \' timeout: 30m\', \' freeze_by_part: false\', \' freeze_by_part_where: ""\', \' use_embedded_backup_restore: false\', \' use_embedded_backup_restore_cluster: ""\', \' embedded_backup_disk: ""\', \' backup_mutations: true\', \' restore_as_attach: false\', \' restore_distributed_cluster: ""\', \' check_parts_columns: true\', \' parts_columns_batch_size: 25\', \' secure: false\', \' skip_verify: false\', \' sync_replicated_tables: false\', \' log_sql_queries: true\', \' config_dir: /etc/clickhouse-server/\', \' restart_command: exec:systemctl restart clickhouse-server\', \' ignore_not_exists_error_during_freeze: true\', \' check_replicas_before_attach: true\', \' default_replica_path: /clickhouse/tables/{cluster}/{shard}/{database}/{table}\', " default_replica_name: \'{replica}\'", \' rebind_replica_path_if_exists: false\', \' tls_key: ""\', \' tls_cert: ""\', \' tls_ca: ""\', \' debug: false\', \' force_rebalance: false\', \'s3:\', \' access_key: ""\', \' secret_key: ""\', \' bucket: ""\', \' endpoint: ""\', \' region: us-east-1\', \' acl: private\', \' assume_role_arn: ""\', \' force_path_style: false\', \' path: ""\', \' object_disk_path: ""\', \' disable_ssl: false\', \' compression_level: 1\', \' compression_format: tar\', \' sse: ""\', \' sse_kms_key_id: ""\', \' sse_customer_algorithm: ""\', \' sse_customer_key: ""\', \' sse_customer_key_md5: ""\', \' sse_kms_encryption_context: ""\', \' disable_cert_verification: false\', \' use_custom_storage_class: false\', \' storage_class: STANDARD\', \' custom_storage_class_map: {}\', \' allow_multipart_download: false\', \' object_labels: {}\', \' request_payer: ""\', \' check_sum_algorithm: ""\', \' request_content_md5: false\', \' retry_mode: standard\', \' chunk_size: 5242880\', \' debug: false\', \' http_write_buffer_size: 0\', \' http_read_buffer_size: 0\', \' http_idle_conn_timeout: ""\', \'gcs:\', \' credentials_file: ""\', \' credentials_json: ""\', \' credentials_json_encoded: ""\', \' sa_email: ""\', \' embedded_access_key: ""\', \' embedded_secret_key: ""\', \' skip_credentials: false\', \' bucket: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_level: 1\', \' compression_format: tar\', \' debug: false\', \' force_http: false\', \' disable_http2: false\', \' endpoint: ""\', \' storage_class: STANDARD\', \' object_labels: {}\', \' custom_storage_class_map: {}\', \' chunk_size: 16777216\', \' encryption_key: ""\', \' upload_buffer_size: 131072\', \' allow_multipart_upload: false\', \' multipart_upload_min_size: 1073741824\', \' allow_multipart_download: false\', \'cos:\', \' url: ""\', \' timeout: 2m\', \' secret_id: ""\', \' secret_key: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' allow_multipart_download: false\', \' debug: false\', \'api:\', \' listen: localhost:7171\', \' enable_metrics: true\', \' enable_pprof: false\', \' username: ""\', \' password: ""\', \' secure: false\', \' certificate_file: ""\', \' private_key_file: ""\', \' ca_cert_file: ""\', \' ca_key_file: ""\', \' create_integration_tables: false\', \' integration_tables_host: ""\', \' allow_parallel: false\', \' complete_resumable_after_restart: true\', \' complete_resumable_after_restart_commands:\', \' - upload\', \' - download\', \' watch_is_main_process: false\', \' backup_actions_skip_commands: []\', \' cancel_operation_timeout: 1800s\', \'ftp:\', \' address: ""\', \' timeout: 2m\', \' username: ""\', \' password: ""\', \' tls: false\', \' skip_tls_verify: false\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' debug: false\', \'sftp:\', \' address: ""\', \' port: 22\', \' username: ""\', \' password: ""\', \' key: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' debug: false\', \'azblob:\', \' endpoint_schema: https\', \' endpoint_suffix: core.windows.net\', \' account_name: ""\', \' account_key: ""\', \' sas: ""\', \' use_managed_identity: false\', \' container: ""\', \' assume_container_exists: false\', \' path: ""\', \' object_disk_path: ""\', \' compression_level: 1\', \' compression_format: tar\', \' sse_key: ""\', \' buffer_count: 3\', \' timeout: 4h\', \' debug: false\', \'custom:\', \' upload_command: ""\', \' download_command: ""\', \' list_command: ""\', \' delete_command: ""\', \' command_timeout: 4h\', \' commandtimeoutduration: 4h0m0s\']'""" +default_config = r"""'[\'general:\', \' remote_storage: none\', \' backups_to_keep_local: 0\', \' backups_to_keep_remote: 0\', \' log_level: info\', \' disable_environment_override: false\', \' allow_empty_backups: false\', \' rebase_before_remove_old_remote: false\', \' rebase_during_delete: false\', \' pipe_buffer_size: 131072\', \' download_copy_buffer_size: 0\', \' compression_use_multi_thread: true\', \' compression_threads: 0\', \' compression_buffer_size: 0\', \' allow_object_disk_streaming: false\', \' use_resumable_state: true\', \' restore_schema_on_cluster: ""\', \' upload_by_part: true\', \' download_by_part: true\', \' restore_database_mapping: {}\', \' restore_table_mapping: {}\', \' retries_on_failure: 3\', \' retries_pause: 5s\', \' retries_jitter: 0\', \' watch_interval: 1h\', \' full_interval: 24h\', \' watch_backup_name_template: shard{shard}-{type}-{time:20060102150405}\', \' callback_url: ""\', \' callback_timeout: 5s\', \' watch_schedules: []\', \' sharded_operation_mode: ""\', \' cpu_nice_priority: 15\', \' io_nice_priority: idle\', \' rbac_backup_always: true\', \' rbac_conflict_resolution: recreate\', \' config_backup_always: false\', \' named_collections_backup_always: false\', \' delete_batch_size: 1000\', \' status_history_size: 1000\', \' retriesduration: 5s\', \' watchduration: 1h0m0s\', \' fullduration: 24h0m0s\', \' callbacktimeoutduration: 5s\', \'clickhouse:\', \' username: default\', \' password: ""\', \' host: localhost\', \' port: 9000\', \' disk_mapping: {}\', \' skip_tables:\', \' - system.*\', \' - INFORMATION_SCHEMA.*\', \' - information_schema.*\', \' - _temporary_and_external_tables.*\', \' skip_table_engines: []\', \' skip_disks: []\', \' skip_disk_types: []\', \' timeout: 30m\', \' freeze_by_part: false\', \' freeze_by_part_where: ""\', \' use_embedded_backup_restore: false\', \' use_embedded_backup_restore_cluster: ""\', \' embedded_backup_disk: ""\', \' backup_mutations: true\', \' restore_as_attach: false\', \' restore_distributed_cluster: ""\', \' check_parts_columns: true\', \' parts_columns_batch_size: 25\', \' secure: false\', \' skip_verify: false\', \' sync_replicated_tables: false\', \' log_sql_queries: true\', \' config_dir: /etc/clickhouse-server/\', \' restart_command: exec:systemctl restart clickhouse-server\', \' ignore_not_exists_error_during_freeze: true\', \' check_replicas_before_attach: true\', \' default_replica_path: /clickhouse/tables/{cluster}/{shard}/{database}/{table}\', " default_replica_name: \'{replica}\'", \' rebind_replica_path_if_exists: false\', \' tls_key: ""\', \' tls_cert: ""\', \' tls_ca: ""\', \' debug: false\', \' force_rebalance: false\', \'s3:\', \' access_key: ""\', \' secret_key: ""\', \' bucket: ""\', \' endpoint: ""\', \' region: us-east-1\', \' acl: private\', \' assume_role_arn: ""\', \' force_path_style: false\', \' path: ""\', \' object_disk_path: ""\', \' disable_ssl: false\', \' compression_level: 1\', \' compression_format: tar\', \' sse: ""\', \' sse_kms_key_id: ""\', \' sse_customer_algorithm: ""\', \' sse_customer_key: ""\', \' sse_customer_key_md5: ""\', \' sse_kms_encryption_context: ""\', \' disable_cert_verification: false\', \' use_custom_storage_class: false\', \' storage_class: STANDARD\', \' custom_storage_class_map: {}\', \' allow_multipart_download: false\', \' object_labels: {}\', \' request_payer: ""\', \' check_sum_algorithm: ""\', \' request_content_md5: false\', \' retry_mode: standard\', \' chunk_size: 5242880\', \' debug: false\', \' http_write_buffer_size: 0\', \' http_read_buffer_size: 0\', \' http_idle_conn_timeout: ""\', \'gcs:\', \' credentials_file: ""\', \' credentials_json: ""\', \' credentials_json_encoded: ""\', \' sa_email: ""\', \' embedded_access_key: ""\', \' embedded_secret_key: ""\', \' skip_credentials: false\', \' bucket: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_level: 1\', \' compression_format: tar\', \' debug: false\', \' force_http: false\', \' disable_http2: false\', \' endpoint: ""\', \' storage_class: STANDARD\', \' object_labels: {}\', \' custom_storage_class_map: {}\', \' chunk_size: 16777216\', \' encryption_key: ""\', \' upload_buffer_size: 131072\', \' allow_multipart_upload: false\', \' multipart_upload_min_size: 1073741824\', \' allow_multipart_download: false\', \'cos:\', \' url: ""\', \' timeout: 2m\', \' secret_id: ""\', \' secret_key: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' allow_multipart_download: false\', \' debug: false\', \'api:\', \' listen: localhost:7171\', \' enable_metrics: true\', \' enable_pprof: false\', \' username: ""\', \' password: ""\', \' secure: false\', \' certificate_file: ""\', \' private_key_file: ""\', \' ca_cert_file: ""\', \' ca_key_file: ""\', \' create_integration_tables: false\', \' integration_tables_host: ""\', \' allow_parallel: false\', \' complete_resumable_after_restart: true\', \' complete_resumable_after_restart_commands:\', \' - upload\', \' - download\', \' watch_is_main_process: false\', \' backup_actions_skip_commands: []\', \' cancel_operation_timeout: 1800s\', \'ftp:\', \' address: ""\', \' timeout: 2m\', \' username: ""\', \' password: ""\', \' tls: false\', \' skip_tls_verify: false\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' debug: false\', \'sftp:\', \' address: ""\', \' port: 22\', \' username: ""\', \' password: ""\', \' key: ""\', \' path: ""\', \' object_disk_path: ""\', \' compression_format: tar\', \' compression_level: 1\', \' debug: false\', \'azblob:\', \' endpoint_schema: https\', \' endpoint_suffix: core.windows.net\', \' account_name: ""\', \' account_key: ""\', \' sas: ""\', \' use_managed_identity: false\', \' container: ""\', \' assume_container_exists: false\', \' path: ""\', \' object_disk_path: ""\', \' compression_level: 1\', \' compression_format: tar\', \' sse_key: ""\', \' buffer_count: 3\', \' timeout: 4h\', \' debug: false\', \'custom:\', \' upload_command: ""\', \' download_command: ""\', \' list_command: ""\', \' delete_command: ""\', \' command_timeout: 4h\', \' commandtimeoutduration: 4h0m0s\']'""" help_flag = r"""'NAME:\n clickhouse-backup - Tool for easy backup of ClickHouse with cloud supportUSAGE:\n clickhouse-backup [-t, --tables=.
] DESCRIPTION:\n Run as \'root\' or \'clickhouse\' userCOMMANDS:\n tables List of tables, exclude skip_tables\n create Create new backup\n create_remote Create and upload new backup\n upload Upload backup to remote storage\n list List of backups\n download Download backup from remote storage\n rebase Copy required parts from `required_backup` chain into remote backup and remove `required_backup` dependency, so backup becomes full\n rebalance Move data parts inside local backup between disks to match current system.parts layout and storage policy, skip parts on object disks\n restore Create schema and restore data from backup\n restore_remote Download and restore\n delete Delete specific backup\n default-config Print default config\n print-config Print current config merged with environment variables\n clean Remove data in \'shadow\' folder from all \'path\' folders available from \'system.disks\'\n clean_remote_broken Remove all broken remote backups\n clean_local_broken Remove all broken local backups\n clean_broken_retention Remove orphan entries under remote `path` and `object_disks_path` that are not in the live backup list\n watch Run infinite loop which create full + incremental backup sequence to allow efficient backup sequences\n acvp Run ACVP wrapper protocol over stdin/stdout\n server Run API server\n help, h Shows a list of commands or help for one commandGLOBAL OPTIONS:\n --config value, -c value Config \'FILE\' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]\n --environment-override value, --env value override any environment variable via CLI parameter\n --fips-info Display FIPS build/runtime info and exit (no Go toolchain required).\n --help, -h show help\n --version, -v print the version'"""