Gather per collections stats in mongodb input plugin (#6137)

etc/telegraf.conf

@@ -3166,6 +3166,11 @@
 #
 #   ## When true, collect per database stats
 #   # gather_perdb_stats = false
+#   ## When true, collect per collection stats
+#   # gather_col_stats = false
+#   ## List of db where collections stats are collected
+#   ## If empty, all db are concerned
+#   # col_stats_dbs = ["local"]
 #
 #   ## Optional TLS Config
 #   # tls_ca = "/etc/telegraf/ca.pem"

plugins/inputs/mongodb/README.md

@@ -13,6 +13,11 @@
 
   ## When true, collect per database stats
   # gather_perdb_stats = false
+  ## When true, collect per collection stats
+  # gather_col_stats = false
+  ## List of db where collections stats are collected
+  ## If empty, all db are concerned
+  # col_stats_dbs = ["local"]
 
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
@@ -147,6 +152,20 @@ Error in input [mongodb]: not authorized on admin to execute command { serverSta
     - storage_size (integer)
     - type (string)
 
+- mongodb_col_stats
+  - tags:
+    - hostname
+    - collection
+    - db_name
+  - fields:
+    - size (integer)
+    - avg_obj_size (integer)
+    - storage_size (integer)
+    - total_index_size (integer)
+    - ok (integer)
+    - count (integer)
+    - type (tring)
+
 - mongodb_shard_stats
   - tags:
     - hostname
@@ -161,5 +180,6 @@ Error in input [mongodb]: not authorized on admin to execute command { serverSta
 mongodb,hostname=127.0.0.1:27017 active_reads=0i,active_writes=0i,commands=1335i,commands_per_sec=7i,connections_available=814i,connections_current=5i,connections_total_created=0i,cursor_no_timeout=0i,cursor_no_timeout_count=0i,cursor_pinned=0i,cursor_pinned_count=1i,cursor_timed_out=0i,cursor_timed_out_count=0i,cursor_total=0i,cursor_total_count=1i,deletes=0i,deletes_per_sec=0i,document_deleted=0i,document_inserted=0i,document_returned=13i,document_updated=0i,flushes=5i,flushes_per_sec=0i,getmores=269i,getmores_per_sec=0i,inserts=0i,inserts_per_sec=0i,jumbo_chunks=0i,member_status="PRI",net_in_bytes=986i,net_in_bytes_count=358006i,net_out_bytes=23906i,net_out_bytes_count=661507i,open_connections=5i,percent_cache_dirty=0,percent_cache_used=0,queries=18i,queries_per_sec=3i,queued_reads=0i,queued_writes=0i,repl_commands=0i,repl_commands_per_sec=0i,repl_deletes=0i,repl_deletes_per_sec=0i,repl_getmores=0i,repl_getmores_per_sec=0i,repl_inserts=0i,repl_inserts_per_sec=0i,repl_lag=0i,repl_oplog_window_sec=24355215i,repl_queries=0i,repl_queries_per_sec=0i,repl_updates=0i,repl_updates_per_sec=0i,resident_megabytes=62i,state="PRIMARY",total_available=0i,total_created=0i,total_in_use=0i,total_refreshing=0i,ttl_deletes=0i,ttl_deletes_per_sec=0i,ttl_passes=23i,ttl_passes_per_sec=0i,updates=0i,updates_per_sec=0i,vsize_megabytes=713i,wtcache_app_threads_page_read_count=13i,wtcache_app_threads_page_read_time=74i,wtcache_app_threads_page_write_count=0i,wtcache_bytes_read_into=55271i,wtcache_bytes_written_from=125402i,wtcache_current_bytes=117050i,wtcache_max_bytes_configured=1073741824i,wtcache_pages_evicted_by_app_thread=0i,wtcache_pages_queued_for_eviction=0i,wtcache_server_evicting_pages=0i,wtcache_tracked_dirty_bytes=0i,wtcache_worker_thread_evictingpages=0i 1547159491000000000
 mongodb_db_stats,db_name=admin,hostname=127.0.0.1:27017 avg_obj_size=241,collections=2i,data_size=723i,index_size=49152i,indexes=3i,num_extents=0i,objects=3i,ok=1i,storage_size=53248i,type="db_stat" 1547159491000000000
 mongodb_db_stats,db_name=local,hostname=127.0.0.1:27017 avg_obj_size=813.9705882352941,collections=6i,data_size=55350i,index_size=102400i,indexes=5i,num_extents=0i,objects=68i,ok=1i,storage_size=204800i,type="db_stat" 1547159491000000000
+mongodb_col_stats,collection=foo,db_name=local,hostname=127.0.0.1:27017 size=375005928i,avg_obj_size=5494,type="col_stat",storage_size=249307136i,total_index_size=2138112i,ok=1i,count=68251i 1547159491000000000
 mongodb_shard_stats,hostname=127.0.0.1:27017,in_use=3i,available=3i,created=4i,refreshing=0i 1522799074000000000
 ```

plugins/inputs/mongodb/mongodb.go

@@ -22,6 +22,8 @@ type MongoDB struct {
 	Ssl              Ssl
 	mongos           map[string]*Server
 	GatherPerdbStats bool
+	GatherColStats   bool
+	ColStatsDbs      []string
 	tlsint.ClientConfig
 }
 
@@ -40,6 +42,11 @@ var sampleConfig = `
 
   ## When true, collect per database stats
   # gather_perdb_stats = false
+  ## When true, collect per collection stats
+  # gather_col_stats = false
+  ## List of db where collections stats are collected
+  ## If empty, all db are concerned
+  # col_stats_dbs = ["local"]
 
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
@@ -164,7 +171,7 @@ func (m *MongoDB) gatherServer(server *Server, acc telegraf.Accumulator) error {
 		}
 		server.Session = sess
 	}
-	return server.gatherData(acc, m.GatherPerdbStats)
+	return server.gatherData(acc, m.GatherPerdbStats, m.GatherColStats, m.ColStatsDbs)
 }
 
 func init() {

plugins/inputs/mongodb/mongodb_data.go

@@ -13,6 +13,7 @@ type MongodbData struct {
 	Fields        map[string]interface{}
 	Tags          map[string]string
 	DbData        []DbData
+	ColData       []ColData
 	ShardHostData []DbData
 }
 
@@ -21,6 +22,12 @@ type DbData struct {
 	Fields map[string]interface{}
 }
 
+type ColData struct {
+	Name   string
+	DbName string
+	Fields map[string]interface{}
+}
+
 func NewMongodbData(statLine *StatLine, tags map[string]string) *MongodbData {
 	return &MongodbData{
 		StatLine: statLine,
@@ -159,6 +166,15 @@ var DbDataStats = map[string]string{
 	"ok":           "Ok",
 }
 
+var ColDataStats = map[string]string{
+	"count":            "Count",
+	"size":             "Size",
+	"avg_obj_size":     "AvgObjSize",
+	"storage_size":     "StorageSize",
+	"total_index_size": "TotalIndexSize",
+	"ok":               "Ok",
+}
+
 func (d *MongodbData) AddDbStats() {
 	for _, dbstat := range d.StatLine.DbStatsLines {
 		dbStatLine := reflect.ValueOf(&dbstat).Elem()
@@ -175,6 +191,23 @@ func (d *MongodbData) AddDbStats() {
 	}
 }
 
+func (d *MongodbData) AddColStats() {
+	for _, colstat := range d.StatLine.ColStatsLines {
+		colStatLine := reflect.ValueOf(&colstat).Elem()
+		newColData := &ColData{
+			Name:   colstat.Name,
+			DbName: colstat.DbName,
+			Fields: make(map[string]interface{}),
+		}
+		newColData.Fields["type"] = "col_stat"
+		for key, value := range ColDataStats {
+			val := colStatLine.FieldByName(value).Interface()
+			newColData.Fields[key] = val
+		}
+		d.ColData = append(d.ColData, *newColData)
+	}
+}
+
 func (d *MongodbData) AddShardHostStats() {
 	for host, hostStat := range d.StatLine.ShardHostStatsLines {
 		hostStatLine := reflect.ValueOf(&hostStat).Elem()
@@ -242,6 +275,17 @@ func (d *MongodbData) flush(acc telegraf.Accumulator) {
 		)
 		db.Fields = make(map[string]interface{})
 	}
+	for _, col := range d.ColData {
+		d.Tags["collection"] = col.Name
+		d.Tags["db_name"] = col.DbName
+		acc.AddFields(
+			"mongodb_col_stats",
+			col.Fields,
+			d.Tags,
+			d.StatLine.Time,
+		)
+		col.Fields = make(map[string]interface{})
+	}
 	for _, host := range d.ShardHostData {
 		d.Tags["hostname"] = host.Name
 		acc.AddFields(

plugins/inputs/mongodb/mongodb_server.go

@@ -70,7 +70,7 @@ func (s *Server) gatherOplogStats() *OplogStats {
 	return stats
 }
 
-func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error {
+func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool, gatherColStats bool, colStatsDbs []string) error {
 	s.Session.SetMode(mgo.Eventual, true)
 	s.Session.SetSocketTimeout(0)
 	result_server := &ServerStatus{}
@@ -147,11 +147,48 @@ func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error
 		}
 	}
 
+	result_col_stats := &ColStats{}
+	if gatherColStats == true {
+		names := []string{}
+		names, err = s.Session.DatabaseNames()
+		if err != nil {
+			log.Println("E! Error getting database names (" + err.Error() + ")")
+		}
+		for _, db_name := range names {
+			if stringInSlice(db_name, colStatsDbs) || len(colStatsDbs) == 0 {
+				var colls []string
+				colls, err = s.Session.DB(db_name).CollectionNames()
+				if err != nil {
+					log.Println("E! Error getting collection names (" + err.Error() + ")")
+				}
+				for _, col_name := range colls {
+					col_stat_line := &ColStatsData{}
+					err = s.Session.DB(db_name).Run(bson.D{
+						{
+							Name:  "collStats",
+							Value: col_name,
+						},
+					}, col_stat_line)
+					if err != nil {
+						log.Println("E! Error getting col stats from " + col_name + "(" + err.Error() + ")")
+					}
+					collection := &Collection{
+						Name:         col_name,
+						DbName:       db_name,
+						ColStatsData: col_stat_line,
+					}
+					result_col_stats.Collections = append(result_col_stats.Collections, *collection)
+				}
+			}
+		}
+	}
+
 	result := &MongoStatus{
 		ServerStatus:  result_server,
 		ReplSetStatus: result_repl,
 		ClusterStatus: result_cluster,
 		DbStats:       result_db_stats,
+		ColStats:      result_col_stats,
 		ShardStats:    resultShards,
 		OplogStats:    oplogStats,
 	}
@@ -173,8 +210,18 @@ func (s *Server) gatherData(acc telegraf.Accumulator, gatherDbStats bool) error
 		)
 		data.AddDefaultStats()
 		data.AddDbStats()
+		data.AddColStats()
 		data.AddShardHostStats()
 		data.flush(acc)
 	}
 	return nil
 }
+
+func stringInSlice(a string, list []string) bool {
+	for _, b := range list {
+		if b == a {
+			return true
+		}
+	}
+	return false
+}

plugins/inputs/mongodb/mongostat.go

@@ -34,6 +34,7 @@ type MongoStatus struct {
 	ReplSetStatus *ReplSetStatus
 	ClusterStatus *ClusterStatus
 	DbStats       *DbStats
+	ColStats      *ColStats
 	ShardStats    *ShardStats
 	OplogStats    *OplogStats
 }
@@ -92,6 +93,26 @@ type DbStatsData struct {
 	GleStats    interface{} `bson:"gleStats"`
 }
 
+type ColStats struct {
+	Collections []Collection
+}
+
+type Collection struct {
+	Name         string
+	DbName       string
+	ColStatsData *ColStatsData
+}
+
+type ColStatsData struct {
+	Collection     string  `bson:"ns"`
+	Count          int64   `bson:"count"`
+	Size           int64   `bson:"size"`
+	AvgObjSize     float64 `bson:"avgObjSize"`
+	StorageSize    int64   `bson:"storageSize"`
+	TotalIndexSize int64   `bson:"totalIndexSize"`
+	Ok             int64   `bson:"ok"`
+}
+
 // ClusterStatus stores information related to the whole cluster
 type ClusterStatus struct {
 	JumboChunksCount int64
@@ -541,6 +562,9 @@ type StatLine struct {
 	// DB stats field
 	DbStatsLines []DbStatLine
 
+	// Col Stats field
+	ColStatsLines []ColStatLine
+
 	// Shard stats
 	TotalInUse, TotalAvailable, TotalCreated, TotalRefreshing int64
 
@@ -560,6 +584,16 @@ type DbStatLine struct {
 	IndexSize   int64
 	Ok          int64
 }
+type ColStatLine struct {
+	Name           string
+	DbName         string
+	Count          int64
+	Size           int64
+	AvgObjSize     float64
+	StorageSize    int64
+	TotalIndexSize int64
+	Ok             int64
+}
 
 type ShardHostStatLine struct {
 	InUse      int64
@@ -918,6 +952,26 @@ func NewStatLine(oldMongo, newMongo MongoStatus, key string, all bool, sampleSec
 		returnVal.DbStatsLines = append(returnVal.DbStatsLines, *dbStatLine)
 	}
 
+	newColStats := *newMongo.ColStats
+	for _, col := range newColStats.Collections {
+		colStatsData := col.ColStatsData
+		// mongos doesn't have the db key, so setting the db name
+		if colStatsData.Collection == "" {
+			colStatsData.Collection = col.Name
+		}
+		colStatLine := &ColStatLine{
+			Name:           colStatsData.Collection,
+			DbName:         col.DbName,
+			Count:          colStatsData.Count,
+			Size:           colStatsData.Size,
+			AvgObjSize:     colStatsData.AvgObjSize,
+			StorageSize:    colStatsData.StorageSize,
+			TotalIndexSize: colStatsData.TotalIndexSize,
+			Ok:             colStatsData.Ok,
+		}
+		returnVal.ColStatsLines = append(returnVal.ColStatsLines, *colStatLine)
+	}
+
 	// Set shard stats
 	newShardStats := *newMongo.ShardStats
 	returnVal.TotalInUse = newShardStats.TotalInUse