Avoid repeated calls of NewIterator which might be slow (#718 & #722)

pump/storage/metrics.go

@@ -13,6 +13,22 @@ var (
 			Help:      "gc ts of storage",
 		})
 
+	doneGcTSGauge = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: "binlog",
+			Subsystem: "pump_storage",
+			Name:      "done_gc_ts",
+			Help:      "the metadata and vlog after this gc ts has been collected",
+		})
+
+	deletedKv = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "binlog",
+			Subsystem: "pump_storage",
+			Name:      "deleted_kv_total",
+			Help:      "deleted kv number",
+		})
+
 	storageSizeGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "binlog",
@@ -67,6 +83,8 @@ var (
 // InitMetircs register the metrics to registry
 func InitMetircs(registry *prometheus.Registry) {
 	registry.MustRegister(gcTSGauge)
+	registry.MustRegister(doneGcTSGauge)
+	registry.MustRegister(deletedKv)
 	registry.MustRegister(maxCommitTSGauge)
 	registry.MustRegister(tikvQueryCount)
 	registry.MustRegister(errorCount)

pump/storage/storage.go

@@ -1,3 +1,16 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package storage
 
 import (
@@ -577,6 +590,12 @@ func (a *Append) doGCTS(ts int64) {
 
 	deleteNum := 0
 
+	irange := &util.Range{
+		Start: encodeTSKey(0),
+		Limit: encodeTSKey(ts + 1),
+	}
+	iter := a.metadata.NewIterator(irange, nil)
+
 	for {
 		nStr, err := a.metadata.GetProperty("leveldb.num-files-at-level0")
 		if err != nil {
@@ -596,12 +615,6 @@ func (a *Append) doGCTS(ts int64) {
 			continue
 		}
 
-		irange := &util.Range{
-			Start: encodeTSKey(0),
-			Limit: encodeTSKey(ts + 1),
-		}
-		iter := a.metadata.NewIterator(irange, nil)
-
 		deleteBatch := 0
 		var lastKey []byte
 
@@ -615,32 +628,36 @@ func (a *Append) doGCTS(ts int64) {
 				if err != nil {
 					log.Error(err)
 				}
+				deletedKv.Add(float64(batch.Len()))
 				batch.Reset()
 				deleteBatch++
 			}
 		}
 
-		iter.Release()
-
 		if deleteBatch < 100 {
 			if batch.Len() > 0 {
 				err := a.metadata.Write(batch, nil)
 				if err != nil {
 					log.Error(err)
 				}
+				deletedKv.Add(float64(batch.Len()))
 				batch.Reset()
 			}
 			break
 		}
 
 		if len(lastKey) > 0 {
 			a.vlog.gcTS(decodeTSKey(lastKey))
+			doneGcTSGauge.Set(float64(oracle.ExtractPhysical(uint64(decodeTSKey(lastKey)))))
 		}
 
 		log.Infof("has delete %d number", deleteNum)
 	}
 
+	iter.Release()
+
 	a.vlog.gcTS(ts)
+	doneGcTSGauge.Set(float64(oracle.ExtractPhysical(uint64(ts))))
 	log.Infof("finish gc, ts: %d delete num: %d", ts, deleteNum)
 }