MongoDB cloud watch metrics

mongo-cloud-watch.go

/* User must have privileges to run: 1. db.stats() 2. db.serverStatus() 3. collStats on every column in the db 4. listCollections on relevant DB */
package main

import (
	"context"
	"fmt"
	"log"
	"math"
	"os"
	"reflect"
	"time"

	awsSession "github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/cloudwatch"
	"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 dbName string
var dbReplicaSetName string
var machineName string
var db *mongo.Database
var cw *cloudwatch.CloudWatch

var namespace = "EC2/MongoCS"
var namespace = "EC2/MongoS1"
var unit = "Count"

func toFloat64(val interface{}) float64 {
	floatType := reflect.TypeOf(float64(0))
	v := reflect.ValueOf(val)
	v = reflect.Indirect(v)
	if v.Type().ConvertibleTo(floatType) {
		return v.Convert(floatType).Float()
	}
	fmt.Println("Fail", v.Kind())
	return 0
}

func toGB(val interface{}) float64 {
	floatVal := toFloat64(val)
	v := floatVal / (1024 * 1024 * 1024)
	return math.Floor(v*100) / 100
}

func makeMetricDatum(name string, value interface{}, dims map[string]string) *cloudwatch.MetricDatum {
	// Additional dimensions
	dims["Machine"] = machineName
	dims["ReplicaSet"] = dbReplicaSetName
	var dimensions []*cloudwatch.Dimension
	getDim := func(dimName string, dimVal string) *cloudwatch.Dimension {
		return &cloudwatch.Dimension{Name: &dimName, Value: &dimVal}
	}
	for dimName, dimVal := range dims {
		dimensions = append(dimensions, getDim(dimName, dimVal))
	}
	valToSend := toFloat64(value)
	return &cloudwatch.MetricDatum{
		MetricName: &name,
		Dimensions: dimensions,
		Value:      &valToSend,
		Unit:       &unit,
	}
}

func getDBMetrics() {
	v := db.RunCommand(context.Background(), bson.D{primitive.E{Key: "dbStats"}})
	dbStats := make(map[string]interface{})
	HandleError(v.Err())
	v.Decode(&dbStats)

	metricsDB := map[string]interface{}{
		"collections": dbStats["collections"],
		"objects":     dbStats["objects"],
		"dataSize":    toGB(dbStats["dataSize"]),
		"storageSize": toGB(dbStats["storageSize"]),
		"indexSize":   toGB(dbStats["indexSize"]),
	}

	metricData := []*cloudwatch.MetricDatum{}
	for key, value := range metricsDB {
		metricData = append(metricData, makeMetricDatum(key, value, map[string]string{"Collection": "all"}))
	}
	fmt.Println(metricData)
	cw.PutMetricData(&cloudwatch.PutMetricDataInput{Namespace: &namespace, MetricData: metricData})
}

func getSeverMetrics() {
	serverStatus := make(map[string]interface{})
	v := db.RunCommand(context.Background(), bson.D{primitive.E{Key: "serverStatus"}})
	HandleError(v.Err())
	v.Decode(&serverStatus)

	serverMetrics := make(map[string]interface{})
	serverConnections := serverStatus["connections"].(map[string]interface{})
	serverMetrics["connections"] = serverConnections["current"]

	metricData := []*cloudwatch.MetricDatum{}
	for key, value := range serverMetrics {
		metricData = append(metricData, makeMetricDatum(key, value, map[string]string{}))
	}
	fmt.Println(metricData)
	cw.PutMetricData(&cloudwatch.PutMetricDataInput{Namespace: &namespace, MetricData: metricData})
}

func getCollectionMetrics() {
	singleCollectionMetrics := func(collectionName string) {
		collectionMetrics := make(map[string]interface{})
		v := db.RunCommand(
			context.Background(),
			bson.D{
				primitive.E{Key: "collStats", Value: collectionName},
			},
		)
		HandleError(v.Err())
		v.Decode(&collectionMetrics)
		metricsCollection := make(map[string]interface{})
		metricsCollection["size"] = toGB(collectionMetrics["size"])
		metricsCollection["storageSize"] = toGB(collectionMetrics["storageSize"])
		metricsCollection["totalIndexSize"] = toGB(collectionMetrics["totalIndexSize"])
		metricsCollection["totalIndexes"] = collectionMetrics["nindexes"]
		metricsCollection["count"] = collectionMetrics["count"]

		metricData := []*cloudwatch.MetricDatum{}
		for key, value := range metricsCollection {
			metricData = append(metricData, makeMetricDatum(key, value, map[string]string{"Collection": collectionName}))
		}
		fmt.Println(metricData)
		cw.PutMetricData(&cloudwatch.PutMetricDataInput{Namespace: &namespace, MetricData: metricData})
	}

	collections, err := db.ListCollectionNames(context.Background(), bson.D{primitive.E{Key:"database", Value: dbName}})
	HandleError(err)
	for _, collection := range collections {
		singleCollectionMetrics(collection)
	}
}

func getOrDefault(key string, defVal string) string {
	val := os.Getenv(key)
	if val == "" {
		return defVal
	}
	return val
}

func connectWithAuth(uri string, username string, password string) *mongo.Client {
	clientOpts := options.Client()
	clientOpts.SetAuth(options.Credential{Username: username, Password: password})
	client, err := mongo.NewClient(clientOpts.ApplyURI(uri))
	HandleError(err)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	err = client.Connect(ctx)
	HandleError(err)
	return client
}

func HandleError(err error) {
	if err != nil {
		log.Fatalln(err)
	}
}

func main() {
	dbName = getOrDefault("MONGO_DB_NAME", "MONGO_DB_NAME")
	dbReplicaSetName = getOrDefault("MONGO_REPLICA_SET", "MONGO_REPLICA_SET")
	machineName = getOrDefault("MACHINE_NAME", "MACHINE_NAME")

	cw = cloudwatch.New(awsSession.Must(awsSession.NewSession()))

	pwd := "password"
	username := "username"
	client := connectWithAuth("mongodb://localhost:27017", username, pwd)

	db = client.Database(dbName)

	getDBMetrics()
	getCollectionMetrics()
	getSeverMetrics()
}