Make the KV-Encode and Deliver-to-Importer concurrent; Turn the old m…

…etrics into real Prometheus metrics (pingcap#88)

* *: replace the ad-hoc common.metrics by prometheus

For now we do not distinguish between tables. The 4 old metrics are
replaced by 6 new histograms.

* restore: move the delivery into its own goroutine to parallelize it

This should increase the performance by 30%.

* tests: retain the logs after running

* restore: apply backpressure on KV-Encoding to prevent OOM

* restore: print the duration needed for import + cleanup

* metric, restore: record checksum durations

* restore: limits the size of each Put() message to 31 MB

Prevents the gRPC ResourceExhausted error on caused by importer's
MaxRecvSize.

* restore: don't print time taken when import failed

* restore: fix missing error check

lightning/metric/metric.go

@@ -74,6 +74,63 @@ var (
 	//  - running
 	//  - finished
 	//  - failed
+
+	ImportSecondsHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "import_seconds",
+			Help:      "time needed to import a table",
+			Buckets:   prometheus.ExponentialBuckets(0.125, 2, 6),
+		},
+	)
+	BlockReadSecondsHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "block_read_seconds",
+			Help:      "time needed to read a block",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 3.1622776601683795, 7),
+		},
+	)
+	BlockReadBytesHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "block_read_bytes",
+			Help:      "number of bytes being read out from data source",
+			Buckets:   prometheus.ExponentialBuckets(1024, 2, 8),
+		},
+	)
+	BlockEncodeSecondsHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "block_encode_seconds",
+			Help:      "time needed to encode a block",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 3.1622776601683795, 10),
+		},
+	)
+	BlockDeliverSecondsHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "block_deliver_seconds",
+			Help:      "time needed to deliver a block",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 3.1622776601683795, 10),
+		},
+	)
+	BlockDeliverBytesHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "block_deliver_bytes",
+			Help:      "number of bytes being sent out to importer",
+			Buckets:   prometheus.ExponentialBuckets(512, 2, 10),
+		},
+	)
+	ChecksumSecondsHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "lightning",
+			Name:      "checksum_seconds",
+			Help:      "time needed to complete the checksum stage",
+			Buckets:   prometheus.ExponentialBuckets(1, 2.2679331552660544, 10),
+		},
+	)
 )
 
 func init() {
@@ -82,6 +139,13 @@ func init() {
 	prometheus.MustRegister(KvEncoderCounter)
 	prometheus.MustRegister(TableCounter)
 	prometheus.MustRegister(ChunkCounter)
+	prometheus.MustRegister(ImportSecondsHistogram)
+	prometheus.MustRegister(BlockReadSecondsHistogram)
+	prometheus.MustRegister(BlockReadBytesHistogram)
+	prometheus.MustRegister(BlockEncodeSecondsHistogram)
+	prometheus.MustRegister(BlockDeliverSecondsHistogram)
+	prometheus.MustRegister(BlockDeliverBytesHistogram)
+	prometheus.MustRegister(ChecksumSecondsHistogram)
 }
 
 func RecordTableCount(status string, err error) {