From 9c2bd062a889e8155e8b7605e7aa8057dfbdee6a Mon Sep 17 00:00:00 2001
From: Matvey Kruglov <kubuzzzz@gmail.com>
Date: Sat, 7 Apr 2018 02:34:47 +0300
Subject: [PATCH] Add repl_oplog_window_s metric to mongodb input (#3964)

---
 plugins/inputs/mongodb/README.md            |  1 +
 plugins/inputs/mongodb/mongodb_data.go      |  1 +
 plugins/inputs/mongodb/mongodb_data_test.go |  1 +
 plugins/inputs/mongodb/mongodb_server.go    | 38 +++++++++++++++++++++
 plugins/inputs/mongodb/mongostat.go         |  8 +++++
 5 files changed, 49 insertions(+)

diff --git a/plugins/inputs/mongodb/README.md b/plugins/inputs/mongodb/README.md
index 5ddf4e2b2..4dce129c9 100644
--- a/plugins/inputs/mongodb/README.md
+++ b/plugins/inputs/mongodb/README.md
@@ -56,6 +56,7 @@ and create a single measurement containing values e.g.
  * ttl_deletes_per_sec
  * ttl_passes_per_sec
  * repl_lag
+ * repl_oplog_window_s
  * jumbo_chunks (only if mongos or mongo config)
 
 If gather_db_stats is set to true, it will also collect per database stats exposed by db.stats()
diff --git a/plugins/inputs/mongodb/mongodb_data.go b/plugins/inputs/mongodb/mongodb_data.go
index 5114e0ad7..2c45f7ea2 100644
--- a/plugins/inputs/mongodb/mongodb_data.go
+++ b/plugins/inputs/mongodb/mongodb_data.go
@@ -60,6 +60,7 @@ var DefaultReplStats = map[string]string{
 	"member_status":         "NodeType",
 	"state":                 "NodeState",
 	"repl_lag":              "ReplLag",
+	"repl_oplog_window_s":   "OplogTimeDiff",
 }
 
 var DefaultClusterStats = map[string]string{
diff --git a/plugins/inputs/mongodb/mongodb_data_test.go b/plugins/inputs/mongodb/mongodb_data_test.go
index 9a12be6a5..e51b4e404 100644
--- a/plugins/inputs/mongodb/mongodb_data_test.go
+++ b/plugins/inputs/mongodb/mongodb_data_test.go
@@ -162,6 +162,7 @@ func TestStateTag(t *testing.T) {
 		"repl_queries_per_sec":  int64(0),
 		"repl_updates_per_sec":  int64(0),
 		"repl_lag":              int64(0),
+		"repl_oplog_window_s":   int64(0),
 		"resident_megabytes":    int64(0),
 		"updates_per_sec":       int64(0),
 		"vsize_megabytes":       int64(0),
diff --git a/plugins/inputs/mongodb/mongodb_server.go b/plugins/inputs/mongodb/mongodb_server.go
index 51eb48a71..d85d41746 100644
--- a/plugins/inputs/mongodb/mongodb_server.go
+++ b/plugins/inputs/mongodb/mongodb_server.go
@@ -22,6 +22,41 @@ func (s *Server) getDefaultTags() map[string]string {
 	return tags
 }
 
+type oplogEntry struct {
+	Timestamp bson.MongoTimestamp `bson:"ts"`
+}
+
+func (s *Server) gatherOplogStats() *OplogStats {
+	stats := &OplogStats{}
+	localdb := s.Session.DB("local")
+
+	op_first := oplogEntry{}
+	op_last := oplogEntry{}
+	query := bson.M{"ts": bson.M{"$exists": true}}
+
+	for _, collection_name := range []string{"oplog.rs", "oplog.$main"} {
+		if err := localdb.C(collection_name).Find(query).Sort("$natural").Limit(1).One(&op_first); err != nil {
+			if err == mgo.ErrNotFound {
+				continue
+			}
+			log.Println("E! Error getting first oplog entry (" + err.Error() + ")")
+			return stats
+		}
+		if err := localdb.C(collection_name).Find(query).Sort("-$natural").Limit(1).One(&op_last); err != nil {
+			if err == mgo.ErrNotFound {
+				continue
+			}
+			log.Println("E! Error getting last oplog entry (" + err.Error() + ")")
+			return stats
+		}
+	}
+
+	op_first_time := time.Unix(int64(op_first.Timestamp>>32), 0)
+	op_last_time := time.Unix(int64(op_last.Timestamp>>32), 0)
+	stats.TimeDiff = int64(op_last_time.Sub(op_first_time).Seconds())
+	return stats
+}
+
 func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error {
 	s.Session.SetMode(mgo.Eventual, true)
 	s.Session.SetSocketTimeout(0)
@@ -66,6 +101,8 @@ func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error
 		log.Println("E! Error getting database shard stats (" + err.Error() + ")")
 	}
 
+	oplogStats := s.gatherOplogStats()
+
 	result_db_stats := &DbStats{}
 	if gatherDbStats == true {
 		names := []string{}
@@ -99,6 +136,7 @@ func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error
 		ClusterStatus: result_cluster,
 		DbStats:       result_db_stats,
 		ShardStats:    resultShards,
+		OplogStats:    oplogStats,
 	}
 
 	defer func() {
diff --git a/plugins/inputs/mongodb/mongostat.go b/plugins/inputs/mongodb/mongostat.go
index daf5a4504..76b2b50d5 100644
--- a/plugins/inputs/mongodb/mongostat.go
+++ b/plugins/inputs/mongodb/mongostat.go
@@ -35,6 +35,7 @@ type MongoStatus struct {
 	ClusterStatus *ClusterStatus
 	DbStats       *DbStats
 	ShardStats    *ShardStats
+	OplogStats    *OplogStats
 }
 
 type ServerStatus struct {
@@ -102,6 +103,11 @@ type ReplSetStatus struct {
 	MyState int64           `bson:"myState"`
 }
 
+// OplogStatus stores information from getReplicationInfo
+type OplogStats struct {
+	TimeDiff int64
+}
+
 // ReplSetMember stores information related to a replica set member
 type ReplSetMember struct {
 	Name       string    `bson:"name"`
@@ -442,6 +448,7 @@ type StatLine struct {
 	// Replicated Opcounter fields
 	InsertR, QueryR, UpdateR, DeleteR, GetMoreR, CommandR int64
 	ReplLag                                               int64
+	OplogTimeDiff                                         int64
 	Flushes                                               int64
 	Mapped, Virtual, Resident, NonMapped                  int64
 	Faults                                                int64
@@ -772,6 +779,7 @@ func NewStatLine(oldMongo, newMongo MongoStatus, key string, all bool, sampleSec
 
 	newClusterStat := *newMongo.ClusterStatus
 	returnVal.JumboChunksCount = newClusterStat.JumboChunksCount
+	returnVal.OplogTimeDiff = newMongo.OplogStats.TimeDiff
 
 	newDbStats := *newMongo.DbStats
 	for _, db := range newDbStats.Dbs {