mvcc: use GaugeFunc metric to load db size when requested

integration/metrics_test.go

@@ -15,13 +15,17 @@
 package integration
 
 import (
+	"context"
+	"strconv"
 	"testing"
+	"time"
 
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"github.com/coreos/etcd/pkg/testutil"
 )
 
-// TestMetricDbSize checks that the db size metric is set on boot.
-func TestMetricDbSize(t *testing.T) {
+// TestMetricDbSizeBoot checks that the db size metric is set on boot.
+func TestMetricDbSizeBoot(t *testing.T) {
 	defer testutil.AfterTest(t)
 	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
 	defer clus.Terminate(t)
@@ -35,3 +39,59 @@ func TestMetricDbSize(t *testing.T) {
 		t.Fatalf("expected non-zero, got %q", v)
 	}
 }
+
+// TestMetricDbSizeDefrag checks that the db size metric is set after defrag.
+func TestMetricDbSizeDefrag(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	kvc := toGRPC(clus.Client(0)).KV
+	mc := toGRPC(clus.Client(0)).Maintenance
+
+	// expand the db size
+	numPuts := 10
+	putreq := &pb.PutRequest{Key: []byte("k"), Value: make([]byte, 4096)}
+	for i := 0; i < numPuts; i++ {
+		if _, err := kvc.Put(context.TODO(), putreq); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// wait for backend txn sync
+	time.Sleep(500 * time.Millisecond)
+
+	beforeDefrag, err := clus.Members[0].Metric("etcd_debugging_mvcc_db_total_size_in_bytes")
+	if err != nil {
+		t.Fatal(err)
+	}
+	bv, err := strconv.Atoi(beforeDefrag)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if expected := numPuts * len(putreq.Value); bv < expected {
+		t.Fatalf("expected db size greater than %d, got %d", expected, bv)
+	}
+
+	// clear out historical keys
+	creq := &pb.CompactionRequest{Revision: int64(numPuts), Physical: true}
+	if _, err := kvc.Compact(context.TODO(), creq); err != nil {
+		t.Fatal(err)
+	}
+
+	// defrag should give freed space back to fs
+	mc.Defragment(context.TODO(), &pb.DefragmentRequest{})
+	afterDefrag, err := clus.Members[0].Metric("etcd_debugging_mvcc_db_total_size_in_bytes")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	av, err := strconv.Atoi(afterDefrag)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if bv <= av {
+		t.Fatalf("expected less than %d, got %d after defrag", bv, av)
+	}
+}

mvcc/kvstore.go

@@ -251,6 +251,11 @@ func (s *store) Restore(b backend.Backend) error {
 }
 
 func (s *store) restore() error {
+	reportDbTotalSizeInBytesMu.Lock()
+	b := s.b
+	reportDbTotalSizeInBytes = func() float64 { return float64(b.Size()) }
+	reportDbTotalSizeInBytesMu.Unlock()
+
 	min, max := newRevBytes(), newRevBytes()
 	revToBytes(revision{main: 1}, min)
 	revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)
@@ -261,8 +266,6 @@ func (s *store) restore() error {
 	tx := s.b.BatchTx()
 	tx.Lock()
 
-	dbTotalSize.Set(float64(s.b.Size()))
-
 	_, finishedCompactBytes := tx.UnsafeRange(metaBucketName, finishedCompactKeyName, nil, 0)
 	if len(finishedCompactBytes) != 0 {
 		s.compactMainRev = bytesToRev(finishedCompactBytes[0]).main

mvcc/kvstore_txn.go

@@ -105,7 +105,6 @@ func (tw *storeTxnWrite) End() {
 	if len(tw.changes) != 0 {
 		tw.s.revMu.Unlock()
 	}
-	dbTotalSize.Set(float64(tw.s.b.Size()))
 	tw.s.mu.RUnlock()
 }
 

mvcc/metrics.go

@@ -15,6 +15,8 @@
 package mvcc
 
 import (
+	"sync"
+
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -129,12 +131,21 @@ var (
 			Buckets: prometheus.ExponentialBuckets(100, 2, 14),
 		})
 
-	dbTotalSize = prometheus.NewGauge(prometheus.GaugeOpts{
+	dbTotalSize = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
 		Namespace: "etcd_debugging",
 		Subsystem: "mvcc",
 		Name:      "db_total_size_in_bytes",
 		Help:      "Total size of the underlying database in bytes.",
-	})
+	},
+		func() float64 {
+			reportDbTotalSizeInBytesMu.RLock()
+			defer reportDbTotalSizeInBytesMu.RUnlock()
+			return reportDbTotalSizeInBytes()
+		},
+	)
+	// overridden by mvcc initialization
+	reportDbTotalSizeInBytesMu sync.RWMutex
+	reportDbTotalSizeInBytes   func() float64 = func() float64 { return 0 }
 )
 
 func init() {