Index fields count metric (#6)

* bump golang version to 1.18

* fix panic on empty indexing field

* add fields count metric

* separate collectors, add fields limit metric

.github/workflows/go.yml

@@ -16,10 +16,10 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v2
       with:
-        go-version: 1.16
+        go-version: 1.18
 
     - name: Install promu package
-      run: go get -v github.com/prometheus/promu
+      run: go install -v github.com/prometheus/promu@latest
 
     - name: Build
       run: |

Dockerfile

@@ -1,10 +1,10 @@
-FROM golang:1.16-buster as builder
+FROM golang:1.18.5-buster as builder
 
 WORKDIR /src
 
 COPY . .
 
-RUN go get -v github.com/prometheus/promu \
+RUN go install -v github.com/prometheus/promu@latest \
     && promu build -v --prefix build
 
 

client.go → collector/client.go

@@ -1,4 +1,4 @@
-package main
+package collector
 
 import (
 	"crypto/tls"
@@ -7,13 +7,15 @@ import (
 	"net/http"
 
 	elasticsearch "github.com/elastic/go-elasticsearch/v7"
+	"github.com/sirupsen/logrus"
 )
 
 type Client struct {
-	es *elasticsearch.Client
+	es     *elasticsearch.Client
+	logger *logrus.Logger
 }
 
-func NewClient(addresses []string, tlsClientConfig *tls.Config) (*Client, error) {
+func NewClient(logger *logrus.Logger, addresses []string, tlsClientConfig *tls.Config) (*Client, error) {
 	cfg := elasticsearch.Config{
 		Addresses: addresses,
 		Transport: &http.Transport{
@@ -26,12 +28,12 @@ func NewClient(addresses []string, tlsClientConfig *tls.Config) (*Client, error)
 		return nil, err
 	}
 
-	return &Client{es}, nil
+	return &Client{es: es, logger: logger}, nil
 
 }
 
 func (c *Client) GetIndices(s []string) (map[string]interface{}, error) {
-	log.Debug("Getting indices stats: ", s)
+	c.logger.Debug("Getting indices stats: ", s)
 	resp, err := c.es.Indices.Stats(
 		c.es.Indices.Stats.WithIndex(s...),
 	)
@@ -53,7 +55,7 @@ func (c *Client) GetIndices(s []string) (map[string]interface{}, error) {
 }
 
 func (c *Client) GetSnapshots(sr string) (map[string][]interface{}, error) {
-	log.Debug("Getting snapshots in: ", sr)
+	c.logger.Debug("Getting snapshots in: ", sr)
 	resp, err := c.es.Snapshot.Get(sr, []string{"*"})
 	if err != nil {
 		return nil, fmt.Errorf("error getting response: %s", err)
@@ -73,7 +75,7 @@ func (c *Client) GetSnapshots(sr string) (map[string][]interface{}, error) {
 }
 
 func (c *Client) GetInfo() (map[string]interface{}, error) {
-	log.Debug("Getting cluster info")
+	c.logger.Debug("Getting cluster info")
 	resp, err := c.es.Info()
 	if err != nil {
 		return nil, fmt.Errorf("error getting response: %s", err)
@@ -91,3 +93,48 @@ func (c *Client) GetInfo() (map[string]interface{}, error) {
 
 	return r, nil
 }
+
+func (c *Client) GetMapping(s []string) (map[string]interface{}, error) {
+	c.logger.Debug("Getting indices mapping: ", s)
+	resp, err := c.es.Indices.GetMapping(
+		c.es.Indices.GetMapping.WithIndex(s...),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("error getting response: %s", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("request failed: %v", resp.String())
+	}
+
+	var r map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}
+
+func (c *Client) GetSettings(s []string) (map[string]interface{}, error) {
+	c.logger.Debug("Getting indices settings: ", s)
+	resp, err := c.es.Indices.GetSettings(
+		c.es.Indices.GetSettings.WithIndex(s...),
+		c.es.Indices.GetSettings.WithIncludeDefaults(true),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("error getting response: %s", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("request failed: %v", resp.String())
+	}
+
+	var r map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+		return nil, err
+	}
+
+	return r, nil
+}

collector/collector.go

@@ -0,0 +1,115 @@
+package collector
+
+import (
+	"crypto/tls"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	namespace = "oneday_elasticsearch"
+)
+
+var (
+	labels       = []string{"index", "index_group"}
+	labels_group = []string{"index_group"}
+	slabels      = []string{"repository"}
+)
+
+func NewCollector(logger *logrus.Logger, address, project string, repo string, tlsClientConfig *tls.Config) error {
+	client, err := NewClient(logger, []string{address}, tlsClientConfig)
+	if err != nil {
+		return fmt.Errorf("error creating the client: %v", err)
+	}
+
+	info, err := client.GetInfo()
+	if err != nil {
+		return fmt.Errorf("error getting cluster info: %v", err)
+	}
+	logger.Infof("Cluster info: %v", info)
+
+	cluster := info["cluster_name"].(string)
+
+	constLabels := prometheus.Labels{
+		"cluster": cluster,
+		"project": project,
+	}
+
+	err = prometheus.Register(NewFieldsCollector(logger, client, labels, labels_group, constLabels))
+	if err != nil {
+		return fmt.Errorf("error registering index fields count collector: %v", err)
+	}
+
+	err = prometheus.Register(NewIndicesCollector(logger, client, labels, labels_group, constLabels))
+	if err != nil {
+		return fmt.Errorf("error registering indices stats collector: %v", err)
+	}
+
+	err = prometheus.Register(NewSettingsCollector(logger, client, labels, labels_group, constLabels))
+	if err != nil {
+		return fmt.Errorf("error registering indices settings collector: %v", err)
+	}
+
+	if repo != "" {
+		err = prometheus.Register(NewSnapshotCollector(logger, client, repo, slabels, constLabels))
+		if err != nil {
+			return fmt.Errorf("error registering snapshots stats collector: %v", err)
+		}
+	}
+
+	return nil
+}
+
+func todayFunc() string { return time.Now().Format("2006.01.02") }
+
+func indicesPatternFunc(today string) string { return fmt.Sprintf("*-%s", today) }
+
+// Find date -Y.m.d (-2021.12.01) and replace
+func indexGroupLabelFunc(index, today string) string {
+	return strings.ToLower(strings.TrimSuffix(index, "-"+today))
+}
+
+// Walk over the json map by path `f1.f2.f3` and return the last field
+func walk(m map[string]interface{}, path string) (interface{}, bool) {
+	p := strings.Split(path, ".")
+	for _, v := range p[:len(p)-1] {
+		if v, ok := m[v]; ok {
+			switch v := v.(type) {
+			case map[string]interface{}:
+				m = v
+			default:
+				return v, false
+			}
+		}
+	}
+
+	v, ok := m[p[len(p)-1]]
+
+	return v, ok
+}
+
+// Count mapping fields by `type` field
+// https://github.com/elastic/elasticsearch/issues/68947#issue-806860754
+func countFields(m map[string]interface{}, count float64) float64 {
+	if v, ok := m["type"]; ok {
+		switch v := v.(type) {
+		case map[string]interface{}:
+			count = countFields(v, count)
+		case string:
+			return count + 1
+		}
+	}
+
+	for _, v := range m {
+		switch v := v.(type) {
+		case map[string]interface{}:
+			count = countFields(v, count)
+		}
+	}
+
+	return count
+}