learning_go/go_exporter/main.go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	_ "os/exec"
	"strings"
	"time"

	prometheus "github.com/prometheus/client_golang/prometheus"
	promhttp "github.com/prometheus/client_golang/prometheus/promhttp"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	pipe "gopkg.in/pipe.v2"
)

var logger *zap.SugaredLogger

// Here I store all images info in scope of one pool
type RBDUsage struct {
	Images []RBDImageUsage
}

// Here I do collect info about images itself
type RBDImageUsage struct {
	Name          string `json:"name"`
	id            int    `json: "id"`
	RequestedSize uint64 `json:"provisioned_size"`
	UsedSize      uint64 `json: "used_size"`
}

var total_rbd_requested_size_per_pool = prometheus.NewGaugeVec(prometheus.GaugeOpts{
	Name: "total_rbd_requested_size_per_pool",
	Help: "total size of all requested RBDs in a specific pool",
	},
	[]string{"poolname"},
)

var total_rbd_requested_size = prometheus.NewGauge(prometheus.GaugeOpts{
	Name: "total_rbd_requested_size",
	Help: "total size of all RBDs in the cluster",
})

// Here I initialize logger and set some custom settings
func loggerInit() *zap.SugaredLogger {
	config := zap.NewDevelopmentConfig()
	config.EncoderConfig.TimeKey = "timestamp"
	config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
	logger, err := config.Build()
	if err != nil {
		panic(fmt.Sprintf("Logger set up failed: %v", err))
	}
	defer logger.Sync()
	return logger.Sugar()
}

// This func runs before main() to get all the things set up before main execution
func init() {
	logger = loggerInit()
	logger.Info("Setting up logger is complete successfully")
	logger.Info("Registering prom metrics")
	
	prometheus.MustRegister(
		total_rbd_requested_size_per_pool,
		total_rbd_requested_size,
	)

}

// List pools with application rbd enabled
func listPools() ([]string, error) {
	results := []string{}
	command1 := "ceph"
	args1 := []string{"osd", "pool", "ls", "detail"}
	command2 := "grep"
	args2 := []string{"application rbd"}
	command3 := "awk"
	args3 := []string{"{print $3}"}

	logger.Infof("Listing pools")

	// This is a pipe conjuction to execute "ceph | grep | awk" pipe
	p := pipe.Line(
		pipe.Exec(command1, args1...),
		pipe.Exec(command2, args2...),
		pipe.Exec(command3, args3...),
	)

	output, err := pipe.CombinedOutput(p)
	if err != nil {
		panic(err)
	}

	//Returns iterator
	lines := strings.SplitSeq(string(output), "\n")

	for v := range lines {
		v = strings.TrimSpace(v)
		// awk return result surrounded by single quotes so here I delete it
		v = strings.Trim(v, "'")
		// Sometimes here I have an empty string, that's why I check if it is not
		if v != "" {
			results = append(results, v)
			continue
		}
	}
	return results, nil
}

// List rbd of each pool
func getRBD(poolList []string) map[string][]string {
	RBDmap := make(map[string][]string)
	args := []string{"ls", "-p"}

	// Here I iterate over pool names
	for _, v := range poolList {
		var results []string
		//...and start a command rbd ls -p <pool_name>
		new_args := append(args, v)
		p := pipe.Line(
			pipe.Exec("rbd", new_args...),
		)
		output, err := pipe.CombinedOutput(p)
		if err != nil {
			panic(err)
		}

		//Returns iterator
		lines := strings.SplitSeq(string(output), "\n")

		for i := range lines {
			//delete whitespaces around
			i = strings.TrimSpace(i)
			// check if there is no empty entries
			if i != "" {
				results = append(results, i)
				continue
			}
		}
		RBDmap[v] = results
	}
	return RBDmap
}

// Here I check total provisioned size of each RBD image in a pool
func RbdChecker(rbdMap map[string][]string) map[string][]RBDUsage{
	total := make(map[string][]RBDUsage)

	for pool, rbdlist := range rbdMap {
		logger.Infof("Processing pool %s", pool)

		for _, rbdName := range rbdlist {
			total[pool] = append(total[pool],GetRBDStats(pool, rbdName))

		}
	}
	logger.Debugf("Final map is %v",total)
	return total
}

// Grabbing info about specific image
func GetRBDStats(pool string, rbdname string) RBDUsage {
	rbdPath := fmt.Sprintf("%s/%s", pool, rbdname)
	args := []string{"du", "--format", "json", rbdPath}

	p := pipe.Line(
		pipe.Exec("rbd", args...),
	)
	output, err := pipe.CombinedOutput(p)
	if err != nil {
		logger.Fatalf("Error in processing RBD %v", err)
	}

	var usage RBDUsage
	if err := json.Unmarshal(output, &usage); err != nil {
		logger.Fatalf("Error in unmarshaling %v", err)
	}
	return usage

}

func FormMetrirs(rbdStats map[string][]RBDUsage) {
	// this is requested size overall cluster
	var totalSize uint64 = 0

	// Iterate over pools
	for poolName := range rbdStats {
		var totalSizePerPool uint64 = 0
		logger.Debugf("Forming metrics for pool %s",poolName)
		
		//Iterate over all RBDs in the pool
		for _,rbdName := range rbdStats[poolName] {
			logger.Debugf("Processings rbd %v",rbdName)
			totalSizePerPool = totalSizePerPool + rbdName.Images[0].RequestedSize	
			// logger.Debugf("RBD name is %s and its size is %d",rbdName.Images[0].Name,rbdName.Images[0].RequestedSize)
		}
		logger.Debugf("Total size requested by RBDs of a pool %s is %d bytes",poolName,totalSizePerPool)

		total_rbd_requested_size_per_pool.WithLabelValues(
			poolName,
		).Set(float64(totalSizePerPool))
		totalSize = totalSize + totalSizePerPool
	}	
	logger.Debugf("Total size of all RBDs in a cluster is %d",totalSize)
	total_rbd_requested_size.Set(float64(totalSize))
}

// the main loop for monitoting
func startCheking() {

	//Ticks every 5 seconds
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	// Every tick I start listPools function
	for range ticker.C {
		poolList, err := listPools()
		if err != nil {
			logger.Fatalf("Error in listing pools %v", err)
		}
		// Get the map of all RBDs in all pools
		rbdMap := getRBD(poolList)
		//Get all RBDs info
		RBDStats := RbdChecker(rbdMap)
		//Fill out metrics
		FormMetrirs(RBDStats)
	}

}

func main() {
	defer logger.Sync()
	http.Handle("/metrics", promhttp.Handler())

	// HTTP runs in separate thread cuz it blocks futher execution of main
	go func() {
		logger.Info("Starting http server")
		// Here I check for errors if HTTP fails
		if err := http.ListenAndServe(":9040", nil); err != nil {
			logger.Fatalf("HTTP server failed to start %v", err)
		}
		logger.Info("HTTP server started")
	}()

	go func() {
		logger.Info("Start checking")
		startCheking()
	}()

	select {}
}