PCSM-167: Status response improvement (#143)

Author: inelpandzic

README.md

@@ -297,16 +297,19 @@ The /status endpoint provides the current state of the PCSM replication process,
 - `info`: provides additional information about the current state.
 - `error` (optional): the error message if the operation failed.
 
-- `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.
+- `lagTimeSeconds`: the current lag time in logical seconds between source and target clusters.
+- `eventsApplied`: the number of events applied to the target cluster.
+- `lastReplicatedOpTime`: the last replicated operation time
+- `lastReplicatedOpTime.ts`: op time timestamp
+- `lastReplicatedOpTime.isoDate`: op time ts in human-readable form
+
 
 - `initialSync.completed`: indicates if the initial sync is completed.
-- `initialSync.lagTime`: the lag time in logical seconds until the initial sync completed.
+- `initialSync.lagTimeSeconds`: the lag time in logical seconds until the initial sync completed.
 
 - `initialSync.cloneCompleted`: indicates if the cloning process is completed.
-- `initialSync.estimatedCloneSize`: the estimated total size of the clone.
-- `initialSync.clonedSize`: the size of the data that has been cloned.
+- `initialSync.estimatedCloneSizeBytes`: the estimated total size of the clone.
+- `initialSync.clonedSizeBytes`: the size of the data that has been cloned.
 
 Example:
 
@@ -316,17 +319,20 @@ Example:
     "state": "running",
     "info": "Initial Sync",
 
-    "lagTime": 22,
-    "eventsProcessed": 5000,
-    "lastReplicatedOpTime": "1740335200.5",
+    "lagTimeSeconds": 22,
+    "eventsApplied": 5000,
+    "lastReplicatedOpTime": {
+        "ts": "1762241863.1",
+        "isoDate": "2025-11-04T07:37:43Z"
+    },
 
     "initialSync": {
         "completed": false,
-        "lagTime": 5,
+        "lagTimeSeconds": 5,
 
         "cloneCompleted": false,
-        "estimatedCloneSize": 5000000000,
-        "clonedSize": 2500000000
+        "estimatedCloneSizeBytes": 5000000000,
+        "clonedSizeBytes": 2500000000
     }
 }
 ```

main.go

@@ -689,22 +689,31 @@ func (s *server) handleStatus(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	res.EventsProcessed = status.Repl.EventsProcessed
-	res.LagTime = status.TotalLagTime
+	res.EventsRead = status.Repl.EventsRead
+	res.EventsApplied = status.Repl.EventsApplied
+	res.LagTimeSeconds = status.TotalLagTimeSeconds
 
 	if !status.Repl.LastReplicatedOpTime.IsZero() {
-		res.LastReplicatedOpTime = fmt.Sprintf("%d.%d",
+		ts := fmt.Sprintf("%d.%d",
 			status.Repl.LastReplicatedOpTime.T,
 			status.Repl.LastReplicatedOpTime.I)
+
+		isoDate := time.Unix(int64(status.Repl.LastReplicatedOpTime.T),
+			int64(status.Repl.LastReplicatedOpTime.I)).UTC()
+
+		res.LastReplicatedOpTime = &lastReplicatedOpTime{
+			TS:      ts,
+			ISODate: isoDate.Format(time.RFC3339),
+		}
 	}
 
 	res.InitialSync = &statusInitialSyncResponse{
-		Completed: status.InitialSyncCompleted,
-		LagTime:   status.InitialSyncLagTime,
+		Completed:      status.InitialSyncCompleted,
+		LagTimeSeconds: status.InitialSyncLagTimeSeconds,
 
-		CloneCompleted:     status.Clone.IsFinished(),
-		EstimatedCloneSize: status.Clone.EstimatedTotalSize,
-		ClonedSize:         status.Clone.CopiedSize,
+		CloneCompleted:          status.Clone.IsFinished(),
+		EstimatedCloneSizeBytes: status.Clone.EstimatedTotalSizeBytes,
+		ClonedSizeBytes:         status.Clone.CopiedSizeBytes,
 	}
 
 	switch {
@@ -992,26 +1001,33 @@ type statusResponse struct {
 	// Info provides additional information about the current state.
 	Info string `json:"info,omitempty"`
 
-	// LagTime is the current lag time in logical seconds.
-	LagTime int64 `json:"lagTime"`
-	// EventsProcessed is the number of events processed.
-	EventsProcessed int64 `json:"eventsProcessed"`
+	// LagTimeSeconds is the current lag time in logical seconds.
+	LagTimeSeconds int64 `json:"lagTimeSeconds"`
+	// EventsRead is the number of events read from the source. Not counting tick events.
+	EventsRead int64 `json:"eventsRead"`
+	// EventsApplied is the number of events applied.
+	EventsApplied int64 `json:"eventsApplied"`
 	// LastReplicatedOpTime is the last replicated operation time.
-	LastReplicatedOpTime string `json:"lastReplicatedOpTime,omitempty"`
+	LastReplicatedOpTime *lastReplicatedOpTime `json:"lastReplicatedOpTime,omitempty"`
 
 	// InitialSync contains the initial sync status details.
 	InitialSync *statusInitialSyncResponse `json:"initialSync,omitempty"`
 }
 
+type lastReplicatedOpTime struct {
+	TS      string `json:"ts"`
+	ISODate string `json:"isoDate"`
+}
+
 // statusInitialSyncResponse represents the initial sync status in the /status response.
 type statusInitialSyncResponse struct {
-	// LagTime is the lag time in logical seconds until the initial sync completed.
-	LagTime int64 `json:"lagTime,omitempty"`
+	// LagTimeSeconds is the lag time in logical seconds until the initial sync completed.
+	LagTimeSeconds int64 `json:"lagTimeSeconds,omitempty"`
 
-	// EstimatedCloneSize is the estimated total size of the clone.
-	EstimatedCloneSize uint64 `json:"estimatedCloneSize,omitempty"`
-	// ClonedSize is the size of the data that has been cloned.
-	ClonedSize uint64 `json:"clonedSize"`
+	// EstimatedCloneSizeBytes is the estimated total size of the clone in bytes.
+	EstimatedCloneSizeBytes uint64 `json:"estimatedCloneSizeBytes,omitempty"`
+	// ClonedSizeBytes is the size of the data that has been cloned.
+	ClonedSizeBytes uint64 `json:"clonedSizeBytes"`
 
 	// Completed indicates if the initial sync is completed.
 	Completed bool `json:"completed"`

metrics/metrics.go

@@ -12,9 +12,16 @@ const metricNamespace = "percona_clustersync_mongodb"
 // Counters.
 var (
 	//nolint:gochecknoglobals
-	eventsProcessedTotal = prometheus.NewCounter(prometheus.CounterOpts{
-		Name:      "events_processed_total",
-		Help:      "Total number of events processed.",
+	eventsReadTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Name:      "events_read_total",
+		Help:      "Total number of events read from the source.",
+		Namespace: metricNamespace,
+	})
+
+	//nolint:gochecknoglobals
+	eventsAppliedTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Name:      "events_applied_total",
+		Help:      "Total number of events applied.",
 		Namespace: metricNamespace,
 	})
 
@@ -101,7 +108,7 @@ func Init(reg prometheus.Registerer) {
 		copyReadBatchDurationSeconds,
 		copyInsertBatchDurationSeconds,
 
-		eventsProcessedTotal,
+		eventsAppliedTotal,
 		lagTimeSeconds,
 		intialSyncLagTimeSeconds,
 	)
@@ -144,9 +151,14 @@ func SetCopyInsertBatchDurationSeconds(dur time.Duration) {
 	copyInsertBatchDurationSeconds.Set(float64(dur.Seconds()))
 }
 
-// AddEventsProcessed increments the total number of events processed counter.
-func AddEventsProcessed(v int) {
-	eventsProcessedTotal.Add(float64(v))
+// IncEventsRead increments the total number of events read counter.
+func IncEventsRead() {
+	eventsReadTotal.Inc()
+}
+
+// AddEventsApplied increments the total number of events applied counter.
+func AddEventsApplied(v int) {
+	eventsAppliedTotal.Add(float64(v))
 }
 
 // SetLagTimeSeconds sets the lag time in seconds gauge.

pcsm/catalog.go

@@ -134,7 +134,6 @@ func (c *Catalog) UnlockWrite() {
 func (c *Catalog) Checkpoint() *catalogCheckpoint { //nolint:revive
 	// do not call [sync.RWMutex.RLock] to avoid deadlock through recursive read-locking
 	// that may happen during clone or change replication
-
 	if len(c.Databases) == 0 {
 		return nil
 	}

pcsm/clone.go

@@ -47,8 +47,8 @@ type Clone struct {
 
 // CloneStatus represents the status of the cloning process.
 type CloneStatus struct {
-	EstimatedTotalSize uint64 // Estimated total bytes to be copied
-	CopiedSize         uint64 // Bytes copied so far
+	EstimatedTotalSizeBytes uint64 // Estimated total bytes to be copied
+	CopiedSizeBytes         uint64 // Bytes copied so far
 
 	StartTS  bson.Timestamp
 	FinishTS bson.Timestamp
@@ -148,13 +148,13 @@ func (c *Clone) Status() CloneStatus {
 	defer c.lock.Unlock()
 
 	return CloneStatus{
-		EstimatedTotalSize: c.totalSize,
-		CopiedSize:         c.copiedSize.Load(),
-		StartTS:            c.startTS,
-		FinishTS:           c.finishTS,
-		StartTime:          c.startTime,
-		FinishTime:         c.finishTime,
-		Err:                c.err,
+		EstimatedTotalSizeBytes: c.totalSize,
+		CopiedSizeBytes:         c.copiedSize.Load(),
+		StartTS:                 c.startTS,
+		FinishTS:                c.finishTS,
+		StartTime:               c.startTime,
+		FinishTime:              c.finishTime,
+		Err:                     c.err,
 	}
 }
 

pcsm/pcsm.go

@@ -57,10 +57,10 @@ type Status struct {
 	// Error is the error message if the operation failed.
 	Error error
 
-	// TotalLagTime is the current lag time in logical seconds between source and target clusters.
-	TotalLagTime int64
-	// InitialSyncLagTime is the lag time during the initial sync.
-	InitialSyncLagTime int64
+	// TotalLagTimeSeconds is the current lag time in logical seconds between source and target clusters.
+	TotalLagTimeSeconds int64
+	// InitialSyncLagTimeSeconds is the lag time during the initial sync.
+	InitialSyncLagTimeSeconds int64
 	// InitialSyncCompleted indicates if the initial sync is completed.
 	InitialSyncCompleted bool
 
@@ -261,15 +261,15 @@ func (ml *PCSM) Status(ctx context.Context) *Status {
 		switch {
 		case !s.Repl.LastReplicatedOpTime.IsZero():
 			totalLag := int64(sourceTime.T) - int64(s.Repl.LastReplicatedOpTime.T)
-			s.TotalLagTime = totalLag
+			s.TotalLagTimeSeconds = totalLag
 		case !s.Clone.StartTS.IsZero():
 			totalLag := int64(sourceTime.T) - int64(s.Clone.StartTS.T)
-			s.TotalLagTime = totalLag
+			s.TotalLagTimeSeconds = totalLag
 		}
 	}
 
 	if !s.InitialSyncCompleted {
-		s.InitialSyncLagTime = s.TotalLagTime
+		s.InitialSyncLagTimeSeconds = s.TotalLagTimeSeconds
 	}
 
 	return s

pcsm/repl.go

@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.mongodb.org/mongo-driver/v2/bson"
@@ -40,7 +41,8 @@ type Repl struct {
 	lock sync.Mutex
 	err  error
 
-	eventsProcessed int64
+	eventsRead    atomic.Int64
+	eventsApplied int64
 
 	startTime time.Time
 	pauseTime time.Time
@@ -61,7 +63,8 @@ type ReplStatus struct {
 	PauseTime time.Time
 
 	LastReplicatedOpTime bson.Timestamp // Last applied operation time
-	EventsProcessed      int64          // Number of events processed
+	EventsRead           int64          // Number of events read from the source
+	EventsApplied        int64          // Number of events applied
 
 	Err error
 }
@@ -95,7 +98,8 @@ func NewRepl(source, target *mongo.Client, catalog *Catalog, nsFilter sel.NSFilt
 type replCheckpoint struct {
 	StartTime            time.Time      `bson:"startTime,omitempty"`
 	PauseTime            time.Time      `bson:"pauseTime,omitempty"`
-	EventsProcessed      int64          `bson:"events,omitempty"`
+	EventsRead           int64          `bson:"eventsRead,omitempty"`
+	EventsApplied        int64          `bson:"events,omitempty"`
 	LastReplicatedOpTime bson.Timestamp `bson:"lastOpTS,omitempty"`
 	Error                string         `bson:"error,omitempty"`
 	UseClientBulkWrite   bool           `bson:"clientBulk,omitempty"`
@@ -112,7 +116,8 @@ func (r *Repl) Checkpoint() *replCheckpoint { //nolint:revive
 	cp := &replCheckpoint{
 		StartTime:            r.startTime,
 		PauseTime:            r.pauseTime,
-		EventsProcessed:      r.eventsProcessed,
+		EventsRead:           r.eventsRead.Load(),
+		EventsApplied:        r.eventsApplied,
 		LastReplicatedOpTime: r.lastReplicatedOpTime,
 	}
 
@@ -145,7 +150,7 @@ func (r *Repl) Recover(cp *replCheckpoint) error {
 
 	r.startTime = cp.StartTime
 	r.pauseTime = pauseTime
-	r.eventsProcessed = cp.EventsProcessed
+	r.eventsApplied = cp.EventsApplied
 	r.lastReplicatedOpTime = cp.LastReplicatedOpTime
 
 	if cp.UseClientBulkWrite {
@@ -168,7 +173,8 @@ func (r *Repl) Status() ReplStatus {
 
 	return ReplStatus{
 		LastReplicatedOpTime: r.lastReplicatedOpTime,
-		EventsProcessed:      r.eventsProcessed,
+		EventsRead:           r.eventsRead.Load(),
+		EventsApplied:        r.eventsApplied,
 
 		StartTime: r.startTime,
 		PauseTime: r.pauseTime,
@@ -481,14 +487,17 @@ func (r *Repl) run(opts *options.ChangeStreamOptionsBuilder) {
 			continue
 		}
 
+		r.eventsRead.Add(1)
+		metrics.IncEventsRead()
+
 		if change.Namespace.Database == config.PCSMDatabase {
 			if r.bulkWrite.Empty() {
 				r.lock.Lock()
 				r.lastReplicatedOpTime = change.ClusterTime
-				r.eventsProcessed++
+				r.eventsApplied++
 				r.lock.Unlock()
 
-				metrics.AddEventsProcessed(1)
+				metrics.AddEventsApplied(1)
 			}
 
 			continue
@@ -498,10 +507,10 @@ func (r *Repl) run(opts *options.ChangeStreamOptionsBuilder) {
 			if r.bulkWrite.Empty() {
 				r.lock.Lock()
 				r.lastReplicatedOpTime = change.ClusterTime
-				r.eventsProcessed++
+				r.eventsApplied++
 				r.lock.Unlock()
 
-				metrics.AddEventsProcessed(1)
+				metrics.AddEventsApplied(1)
 			}
 
 			continue
@@ -552,10 +561,10 @@ func (r *Repl) run(opts *options.ChangeStreamOptionsBuilder) {
 
 			r.lock.Lock()
 			r.lastReplicatedOpTime = change.ClusterTime
-			r.eventsProcessed++
+			r.eventsApplied++
 			r.lock.Unlock()
 
-			metrics.AddEventsProcessed(1)
+			metrics.AddEventsApplied(1)
 
 			switch change.OperationType { //nolint:exhaustive
 			case Create, Rename, Drop, DropDatabase:
@@ -603,10 +612,10 @@ func (r *Repl) doBulkOps(ctx context.Context) bool {
 
 	r.lock.Lock()
 	r.lastReplicatedOpTime = r.bulkTS
-	r.eventsProcessed += int64(size)
+	r.eventsApplied += int64(size)
 	r.lock.Unlock()
 
-	metrics.AddEventsProcessed(size)
+	metrics.AddEventsApplied(size)
 
 	log.New("bulk:write").
 		With(log.Int64("size", int64(size)), log.Elapsed(time.Since(r.lastBulkDoneAt))).

tests/pcsm.py

@@ -233,7 +233,8 @@ def wait_for_clone_completed(self):
     def last_applied_op(self):
         """Get the last applied operation time."""
         status = self.pcsm.status()
-        if last_replicated_op_time := status.get("lastReplicatedOpTime"):
+        last_replicated_op_time = status.get("lastReplicatedOpTime", {}).get("ts")
+        if last_replicated_op_time:
             t_s, i_s = last_replicated_op_time.split(".")
             return bson.Timestamp(int(t_s), int(i_s))