[PML-54] allow to pause/resume change replication (#31)

Author: defbin

.github/workflows/go.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - uses: actions/setup-go@v4
+      - uses: actions/setup-go@v5
       - run: go test -v ./...
 
   gofmt:
@@ -21,8 +21,10 @@ jobs:
       - uses: actions/setup-go@v5
       - run: go install golang.org/x/tools/cmd/goimports@latest
       - run: go install mvdan.cc/gofumpt@latest
+      - run: go install github.com/segmentio/golines@latest
       - run: goimports -w -local "github.com/percona" .
       - run: gofumpt -w -extra .
+      - run: golines -w --shorten-comments --chain-split-dots -m 100 .
       - uses: reviewdog/action-suggester@v1
         with:
           tool_name: gofmt
@@ -33,8 +35,6 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
-        with:
-          go-version: "1.23"
       - name: golangci-lint
         uses: reviewdog/action-golangci-lint@v2
         with:

.golangci.yml

@@ -12,6 +12,7 @@ linters:
     - dupl
     - funlen
     - tenv
+    - wsl
 
 linters-settings:
   govet:

.vscode/settings.json

@@ -28,8 +28,7 @@
   "go.useLanguageServer": true,
   "gopls": {
     "analyses": {
-      "composites": false,
-      "fieldalignment": false
+      "composites": false
     },
     "formatting.gofumpt": true,
     "formatting.local": "github.com/percona",

README.md

@@ -8,7 +8,7 @@ Percona MongoLink is a tool for replicating data from a source MongoDB cluster t
 - **Real-Time Replication**: Tail the oplog to keep your target cluster up to date.
 - **Namespace Filtering**: Specify which databases and collections to include or exclude.
 - **Automatic Index Management**: Ensure necessary indexes are created on the target.
-- **HTTP API**: Start, finalize, and check replication status via REST endpoints.
+- **HTTP API**: Start, finalize, pause, resume, and check replication status via REST endpoints.
 
 ## Setup
 
@@ -87,6 +87,38 @@ bin/mongolink finalize
 curl -X POST http://localhost:2242/finalize
 ```
 
+### Pausing the Replication
+
+To pause the replication process, you can either use the command-line interface or send a POST request to the `/pause` endpoint:
+
+#### Using Command-Line Interface
+
+```sh
+bin/mongolink pause
+```
+
+#### Using HTTP API
+
+```sh
+curl -X POST http://localhost:2242/pause
+```
+
+### Resuming the Replication
+
+To resume the replication process, you can either use the command-line interface or send a POST request to the `/resume` endpoint:
+
+#### Using Command-Line Interface
+
+```sh
+bin/mongolink resume
+```
+
+#### Using HTTP API
+
+```sh
+curl -X POST http://localhost:2242/resume
+```
+
 ### Checking the Status
 
 To check the current status of the replication process, you can either use the command-line interface or send a GET request to the `/status` endpoint:
@@ -131,12 +163,10 @@ When using the `--log-json` option, the logs will be output in JSON format with
 - `s`: Scope of the log entry.
 - `ns`: Namespace (database.collection format).
 - `elapsed_secs`: The duration in seconds for the specific operation to complete.
-- `total_elapsed_secs`: The cumulative time in seconds for the entire process, including all operations.
 
 Example:
 
 ```json
-
 { "level": "info",
   "s": "clone",
   "ns": "db_1.coll_1",
@@ -149,7 +179,6 @@ Example:
   "elapsed_secs": 0,
   "time": "2025-02-23 11:26:03.857",
   "message": "Change replication stopped at 1740335163.1740335163 source cluster time" }
-
 ```
 
 ## HTTP API
@@ -202,29 +231,61 @@ Example:
 }
 ```
 
+### POST /pause
+
+Pauses the replication process.
+
+#### Response
+
+- `ok`: Boolean indicating if the operation was successful.
+- `error` (optional): Error message if the operation failed.
+
+Example:
+
+```json
+{
+  "ok": true
+}
+```
+
+### POST /resume
+
+Resumes the replication process.
+
+#### Response
+
+- `ok`: Boolean indicating if the operation was successful.
+- `error` (optional): Error message if the operation failed.
+
+Example:
+
+```json
+{
+  "ok": true
+}
+```
+
 ### GET /status
 
-The /status endpoint provides the current state of the MongoLink replication
-process, including its progress, lag, and event processing details.
+The /status endpoint provides the current state of the MongoLink replication process, including its progress, lag, and event processing details.
 
 #### Response
 
 - `ok`: indicates if the operation was successful.
 - `state`: the current state of the replication.
-- `info` (optional): provides additional information about the current state.
+- `info`: provides additional information about the current state.
 - `error` (optional): the error message if the operation failed.
 
-- `lagTime` (optional): the current lag time in logical seconds between source and target clusters.
-- `eventsProcessed` (optional): the number of events processed.
-- `lastReplicatedOpTime` (optional): the last replicated operation time.
+- `lagTime`: the current lag time in logical seconds between source and target clusters.
+- `eventsProcessed`: the number of events processed.
+- `lastReplicatedOpTime`: the last replicated operation time.
 
-- `initialSync.pauseOnInitialSync` (optional): indicates if the replication is paused on initial sync.
 - `initialSync.completed`: indicates if the initial sync is completed.
 - `initialSync.lagTime`: the lag time in logical seconds until the initial sync completed.
 
 - `initialSync.cloneCompleted`: indicates if the cloning process is completed.
-- `initialSync.estimatedCloneSize` (optional): the estimated total size of the clone.
-- `initialSync.clonedSize` (optional): the size of the data that has been cloned.
+- `initialSync.estimatedCloneSize`: the estimated total size of the clone.
+- `initialSync.clonedSize`: the size of the data that has been cloned.
 
 Example:
 

config/const.go

@@ -37,7 +37,8 @@ const (
 	ReplInitialSyncCheckInterval = time.Second
 )
 
-// CloneMaxWriteSizePerCollection is the maximum write size per collection during the cloning process.
+// CloneMaxWriteSizePerCollection is the maximum write size per collection during the cloning
+// process.
 const CloneMaxWriteSizePerCollection = 100 * MB
 
 // MaxBSONSize is hardcoded maximum BSON document size. 16 mebibytes.

errors/errors.go

@@ -5,56 +5,100 @@ import (
 	"fmt"
 )
 
+// ErrUnsupported indicates an unsupported operation.
+var ErrUnsupported = errors.ErrUnsupported
+
+// wrappedError wraps an error with an additional message.
 type wrappedError struct {
 	cause error
 	msg   string
 }
 
-func (w *wrappedError) Error() string { return w.msg + ": " + w.cause.Error() }
-func (w *wrappedError) Unwrap() error { return w.cause }
+func (w *wrappedError) Error() string {
+	return w.msg + ": " + w.cause.Error()
+}
+
+func (w *wrappedError) Unwrap() error {
+	return w.cause
+}
 
-// New calls [errors.New].
+// New creates a new error with the given text.
+//
+// Calls [errors.New].
+//
+//go:inline
 func New(text string) error {
 	return errors.New(text) //nolint:err113
 }
 
-// Errorf calls [fmt.Errorf].
+// Errorf formats an error according to a format specifier.
+//
+// Calls [fmt.Errorf].
+//
+//go:inline
 func Errorf(format string, vals ...any) error {
 	return fmt.Errorf(format, vals...) //nolint:err113
 }
 
+// Wrap adds a message to an existing error.
 func Wrap(cause error, text string) error {
 	if cause == nil {
 		return nil
 	}
 
+	if text == "" {
+		return cause
+	}
+
 	return &wrappedError{cause: cause, msg: text}
 }
 
+// Wrapf adds a formatted message to an existing error.
 func Wrapf(cause error, format string, vals ...any) error {
 	if cause == nil {
 		return nil
 	}
 
-	return &wrappedError{cause: cause, msg: fmt.Sprintf(format, vals...)}
+	msg := fmt.Sprintf(format, vals...)
+	if msg == "" {
+		return cause
+	}
+
+	return &wrappedError{cause: cause, msg: msg}
 }
 
-// Unwrap calls [errors.Unwrap].
+// Unwrap returns the cause of the error.
+//
+// Calls [errors.Unwrap].
+//
+//go:inline
 func Unwrap(err error) error {
 	return errors.Unwrap(err)
 }
 
-// Join calls [errors.Join].
+// Join combines multiple errors into one.
+//
+// Calls [errors.Join].
+//
+//go:inline
 func Join(errs ...error) error {
 	return errors.Join(errs...)
 }
 
-// Is calls [errors.Is].
+// Is checks if an error matches a target error.
+//
+// Calls [errors.Is].
+//
+//go:inline
 func Is(err, target error) bool {
 	return errors.Is(err, target)
 }
 
-// As calls [errors.As].
+// As checks if an error can be cast to a target type.
+//
+// Calls [errors.As].
+//
+//go:inline
 func As(err error, target any) bool {
 	return errors.As(err, target)
 }

go.mod

@@ -5,12 +5,12 @@ go 1.24
 require (
 	github.com/rs/zerolog v1.33.0
 	github.com/spf13/cobra v1.9.1
-	go.mongodb.org/mongo-driver/v2 v2.0.1
-	golang.org/x/sync v0.11.0
+	go.mongodb.org/mongo-driver/v2 v2.1.0
+	golang.org/x/sync v0.12.0
 )
 
 require (
-	github.com/golang/snappy v0.0.4 // indirect
+	github.com/golang/snappy v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
@@ -20,7 +20,7 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
-	golang.org/x/crypto v0.33.0 // indirect
-	golang.org/x/sys v0.30.0 // indirect
-	golang.org/x/text v0.22.0 // indirect
+	golang.org/x/crypto v0.36.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
 )