asyncreader: add prometheus metrics

fs/asyncreader/asyncreader.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ncw/rclone/lib/pool"
 	"github.com/ncw/rclone/lib/readers"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
@@ -89,6 +90,7 @@ func (a *AsyncReader) init(rd io.ReadCloser, buffers int) {
 				}
 				err := b.read(a.in)
 				a.ready <- b
+				promBlocksReady.Inc()
 				if err != nil {
 					return
 				}
@@ -97,6 +99,7 @@ func (a *AsyncReader) init(rd io.ReadCloser, buffers int) {
 			}
 		}
 	}()
+	promOpen.Inc()
 }
 
 // bufferPool is a global pool of buffers
@@ -136,13 +139,22 @@ func (a *AsyncReader) fill() (err error) {
 			}
 			return a.err
 		}
+		promBlocksReady.Dec()
 		a.cur = b
 	}
 	return nil
 }
 
 // Read will return the next available data.
 func (a *AsyncReader) Read(p []byte) (n int, err error) {
+	now := time.Now()
+	defer func() {
+		promReadKibiBytes.Observe(float64(n) / 1024.)
+		promReadTimes.Observe(time.Since(now).Seconds())
+		if err != nil && err != io.EOF {
+			promReadErrors.Add(1)
+		}
+	}()
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
@@ -232,6 +244,7 @@ func (a *AsyncReader) SkipBytes(skip int) (ok bool) {
 					return false
 				}
 				a.cur = b
+				promBlocksReady.Dec()
 			default:
 				return false
 			}
@@ -280,6 +293,8 @@ func (a *AsyncReader) Abandon() {
 		a.putBuffer(a.cur)
 		a.cur = nil
 	}
+	promOpen.Dec()
+	promBlocksReady.Sub(float64(len(a.ready)))
 	for b := range a.ready {
 		a.putBuffer(b)
 	}
@@ -337,3 +352,42 @@ func (b *buffer) buffer() []byte {
 func (b *buffer) increment(n int) {
 	b.offset += n
 }
+
+var (
+	promReadKibiBytes = prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    "rclone_asyncreader_read_kibi_bytes",
+		Help:    "Amount of kibi bytes read in one Read call",
+		Buckets: prometheus.ExponentialBuckets(1., 4, 6),
+	})
+	promReadErrors = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "rclone_asyncreader_read_errors_total",
+		Help: "Number of errors during Read calls",
+	})
+	promReadTimes = prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name: "rclone_asyncreader_read_histogram_seconds",
+		Help: "Time spent in one Read call",
+	})
+	promBlocksReady = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "rclone_asyncreader_blocks_ready",
+		Help: "Number of buffer blocks filled at the same time",
+	})
+	promOpen = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "rclone_asyncreader_open",
+		Help: "Number of readers open at the same time",
+	})
+	promOpens = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "rclone_asyncreader_open_total",
+		Help: "Number of readers opened",
+	})
+)
+
+func init() {
+	prometheus.MustRegister(
+		promReadKibiBytes,
+		promReadErrors,
+		promReadTimes,
+		promBlocksReady,
+		promOpen,
+		promOpens,
+	)
+}