data-struct.go

type OpTime struct {
	TS primitive.Timestamp `bson:"ts"`
	T  int64               `bson:"t"`
}

type ReplSetMember struct {
	ID                   int64  `bson:"_id"`
	Name                 string `bson:"name"`
	Health               int64  `bson:"health"`
	State                int64  `bson:"state"`
	StateStr             string `bson:"stateStr"`
	Uptime               int64  `bson:"uptime"`
	Optime               *OpTime
	OptimeDate           time.Time           `bson:"optimeDate"`
	OptimeDurableDate    time.Time           `bson:"optimeDurableDate"`
	LastAppliedWallTime  time.Time           `bson:"lastAppliedWallTime"`
	LastDurableWallTime  time.Time           `bson:"lastDurableWallTime"`
	LastHeartbeatMessage string              `bson:"lastHeartbeatMessage"`
	SyncSourceHost       string              `bson:"syncSourceHost"`
	SyncSourceId         int64               `bson:"syncSourceId"`
	InfoMessage          string              `bson:"infoMessage"`
	ElectionTime         primitive.Timestamp `bson:"electionTime"`
	ElectionDate         time.Time           `bson:"electionDate"`
	ConfigVersion        int64               `bson:"configVersion"`
	ConfigTerm           int64               `bson:"configTerm"`
	Self                 bool                `bson:"self"`
}

type ReplSetGetStatus struct {
	Set                     string    `bson:"set"`
	Date                    time.Time `bson:"date"`
	MyState                 int64     `bson:"myState"`
	Term                    int64     `bson:"term"`
	HeartbeatIntervalMillis int64     `bson:"heartbeatIntervalMillis"`

	Members []ReplSetMember `bson:"members"`
	Ok      int64           `bson:"ok"`
}

GetReplSetStatus.go

func GetReplSetStatus(ctx context.Context, admin *mongo.Database) (ReplSetGetStatus, error) {
	var result bson.M
	command := bson.M{"replSetGetStatus": 1}
	result, err := getCommand(ctx, admin, command)
	if err != nil {
		log.Fatal(err)
	}

	var r ReplSetGetStatus
	bsonBytes, _ := bson.Marshal(result)
	err = bson.Unmarshal(bsonBytes, &r)
	if err != nil {
		log.Fatal(err)
	}
	return r, nil
}

main.go

func main() {
	ctx := context.Background()

	var client *mongo.Client
	client, err := mongo.Connect(ctx, options.Client().ApplyURI(MONGODB_URI))
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	admin := client.Database("admin")

	// handle exiting signals
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sig
		log.Println("signal received, exiting...")
		os.Exit(0)
	}()

	// execute the following code every 5 seconds
	for {
		fmt.Println("***** Get ReplSetStatus *****")
		r, _ := GetReplSetStatus(ctx, admin)
		PrintSecondaryLags(r)
		time.Sleep(5 * time.Second)
	}
}

monitor-mongo-replset.go

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

var MONGODB_URI = "mongodb://ctbx-db-i0.cc:27017,ctbx-db-i1.cc:27017,ctbx-db-i2.cc:27017/?replicaSet=rs0"

type OpTime struct {
	TS primitive.Timestamp `bson:"ts"`
	T  int64               `bson:"t"`
}

type ReplSetMember struct {
	ID                   int64  `bson:"_id"`
	Name                 string `bson:"name"`
	Health               int64  `bson:"health"`
	State                int64  `bson:"state"`
	StateStr             string `bson:"stateStr"`
	Uptime               int64  `bson:"uptime"`
	Optime               *OpTime
	OptimeDate           time.Time           `bson:"optimeDate"`
	OptimeDurableDate    time.Time           `bson:"optimeDurableDate"`
	LastAppliedWallTime  time.Time           `bson:"lastAppliedWallTime"`
	LastDurableWallTime  time.Time           `bson:"lastDurableWallTime"`
	LastHeartbeatMessage string              `bson:"lastHeartbeatMessage"`
	SyncSourceHost       string              `bson:"syncSourceHost"`
	SyncSourceId         int64               `bson:"syncSourceId"`
	InfoMessage          string              `bson:"infoMessage"`
	ElectionTime         primitive.Timestamp `bson:"electionTime"`
	ElectionDate         time.Time           `bson:"electionDate"`
	ConfigVersion        int64               `bson:"configVersion"`
	ConfigTerm           int64               `bson:"configTerm"`
	Self                 bool                `bson:"self"`
}

type ReplSetGetStatus struct {
	Set                     string    `bson:"set"`
	Date                    time.Time `bson:"date"`
	MyState                 int64     `bson:"myState"`
	Term                    int64     `bson:"term"`
	HeartbeatIntervalMillis int64     `bson:"heartbeatIntervalMillis"`

	Members []ReplSetMember `bson:"members"`
	Ok      int64           `bson:"ok"`
}

func getCommand(ctx context.Context, db *mongo.Database, command bson.M) (bson.M, error) {
	var result bson.M
	err := db.RunCommand(ctx, command).Decode(&result)
	if err != nil {
		log.Println(err)
		return nil, err
	}

	return result, nil
}

func GetReplSetStatus(ctx context.Context, admin *mongo.Database) (ReplSetGetStatus, error) {
	var result bson.M
	command := bson.M{"replSetGetStatus": 1}
	result, err := getCommand(ctx, admin, command)
	if err != nil {
		log.Fatal(err)
	}

	var r ReplSetGetStatus
	bsonBytes, _ := bson.Marshal(result)
	err = bson.Unmarshal(bsonBytes, &r)
	if err != nil {
		log.Fatal(err)
	}
	return r, nil
}

func PrintSecondaryLags(r ReplSetGetStatus) {
	// calculate the lag of the primary
	var primary_optime int64 = 0
	for _, member := range r.Members {
		if member.State == 1 {
			primary_optime = int64(member.Optime.TS.T)
		}
	}
	for _, member := range r.Members {
		if member.State == 2 {
			lag := primary_optime - int64(member.Optime.TS.T)
			fmt.Println(member.Name, " lag:", lag)
		}
	}
}

func main() {
	ctx := context.Background()

	var client *mongo.Client
	client, err := mongo.Connect(ctx, options.Client().ApplyURI(MONGODB_URI))
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	admin := client.Database("admin")

	// handle exiting signals
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sig
		log.Println("signal received, exiting...")
		os.Exit(0)
	}()

	// execute the following code every 5 seconds
	for {
		fmt.Println("***** Get ReplSetStatus *****")
		r, _ := GetReplSetStatus(ctx, admin)
		PrintSecondaryLags(r)
		time.Sleep(5 * time.Second)
	}
}

PrintSecondaryLags.go

func PrintSecondaryLags(r ReplSetGetStatus) {
	// calculate the lag of the primary
	var primary_optime int64 = 0
	for _, member := range r.Members {
		if member.State == 1 {
			primary_optime = int64(member.Optime.TS.T)
		}
	}
	for _, member := range r.Members {
		if member.State == 2 {
			lag := primary_optime - int64(member.Optime.TS.T)
			fmt.Println(member.Name, " lag:", lag)
		}
	}
}