add blobstore metric

Signed-off-by: Christian Richter <crichter@owncloud.com>

go.mod

@@ -54,12 +54,14 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/xattr v0.4.7
 	github.com/prometheus/alertmanager v0.24.0
+	github.com/prometheus/client_golang v1.12.1
 	github.com/rs/cors v1.8.2
 	github.com/rs/zerolog v1.27.0
 	github.com/sciencemesh/meshdirectory-web v1.0.4
 	github.com/sethvargo/go-password v0.2.0
 	github.com/stretchr/testify v1.7.2
 	github.com/studio-b12/gowebdav v0.0.0-20220128162035-c7b1ff8a5e62
+	github.com/test-go/testify v1.1.4
 	github.com/thanhpk/randstr v1.0.4
 	github.com/tus/tusd v1.9.0
 	github.com/wk8/go-ordered-map v1.0.0
@@ -151,7 +153,6 @@ require (
 	github.com/pkg/term v1.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/pquerna/cachecontrol v0.1.0 // indirect
-	github.com/prometheus/client_golang v1.12.1 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
@@ -165,7 +166,6 @@ require (
 	github.com/sirupsen/logrus v1.8.1 // indirect
 	github.com/smartystreets/assertions v1.13.0 // indirect
 	github.com/stretchr/objx v0.4.0 // indirect
-	github.com/test-go/testify v1.1.4 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/urfave/cli/v2 v2.3.0 // indirect
 	github.com/xanzy/ssh-agent v0.3.0 // indirect

pkg/storage/fs/ocis/blobstore/blobstore.go

@@ -48,7 +48,7 @@ func New(root string) (*Blobstore, error) {
 }
 
 // Upload stores some data in the blobstore under the given key
-func (bs *Blobstore) Upload(node *node.Node, data io.Reader) error {
+func (bs *Blobstore) Upload(node *node.Node, data io.ReadCloser) error {
 
 	// ensure parent path exists
 	if err := os.MkdirAll(filepath.Dir(bs.path(node)), 0700); err != nil {

pkg/storage/fs/s3ng/blobstore/blobstore.go

@@ -30,6 +30,7 @@ import (
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // Blobstore provides an interface to an s3 compatible blobstore
@@ -39,6 +40,13 @@ type Blobstore struct {
 	bucket string
 }
 
+type PrometheusAwareReadCloser struct {
+	r io.ReadCloser
+	m *prometheus.CounterVec
+}
+
+var metrics = NewMetrics()
+
 // New returns a new Blobstore
 func New(endpoint, region, bucket, accessKey, secretKey string) (*Blobstore, error) {
 	u, err := url.Parse(endpoint)
@@ -62,8 +70,22 @@ func New(endpoint, region, bucket, accessKey, secretKey string) (*Blobstore, err
 	}, nil
 }
 
+func (p *PrometheusAwareReadCloser) Read(b []byte) (n int, err error) {
+	n, err = p.r.Read(b)
+	p.m.WithLabelValues().Add(float64(n))
+	return
+}
+
+func (p *PrometheusAwareReadCloser) Close() error {
+	return p.r.Close()
+}
+
 // Upload stores some data in the blobstore under the given key
-func (bs *Blobstore) Upload(node *node.Node, reader io.Reader) error {
+func (bs *Blobstore) Upload(node *node.Node, reader io.ReadCloser) error {
+	reader = &PrometheusAwareReadCloser{
+		r: reader,
+		m: metrics.Tx,
+	}
 	size := int64(-1)
 	if file, ok := reader.(*os.File); ok {
 		info, err := file.Stat()
@@ -87,7 +109,10 @@ func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not download object '%s' from bucket '%s'", bs.path(node), bs.bucket)
 	}
-	return reader, nil
+	return &PrometheusAwareReadCloser{
+		r: reader,
+		m: metrics.Rx,
+	}, nil
 }
 
 // Delete deletes a blob from the blobstore

pkg/storage/fs/s3ng/blobstore/metrics.go

@@ -0,0 +1,39 @@
+package blobstore
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	// Namespace defines the namespace for the defines metrics.
+	Namespace = "ocis"
+
+	// Subsystem defines the subsystem for the defines metrics.
+	Subsystem = "thumbnails"
+)
+
+// Metrics defines the available metrics of this service.
+type Metrics struct {
+	Rx *prometheus.CounterVec
+	Tx *prometheus.CounterVec
+}
+
+// New initializes the available metrics.
+func NewMetrics() *Metrics {
+	m := &Metrics{
+		Rx: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "Rx",
+			Help:      "Storage access rx",
+		}, []string{}),
+		Tx: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "Tx",
+			Help:      "Storage access tx",
+		}, []string{}),
+	}
+	_ = prometheus.Register(m.Rx)
+	_ = prometheus.Register(m.Tx)
+
+	return m
+}

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -88,7 +88,7 @@ type Tree interface {
 	RestoreRecycleItemFunc(ctx context.Context, spaceid, key, trashPath string, target *node.Node) (*node.Node, *node.Node, func() error, error)
 	PurgeRecycleItemFunc(ctx context.Context, spaceid, key, purgePath string) (*node.Node, func() error, error)
 
-	WriteBlob(node *node.Node, reader io.Reader) error
+	WriteBlob(node *node.Node, reader io.ReadCloser) error
 	ReadBlob(node *node.Node) (io.ReadCloser, error)
 	DeleteBlob(node *node.Node) error
 

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -47,7 +47,7 @@ import (
 
 // Blobstore defines an interface for storing blobs in a blobstore
 type Blobstore interface {
-	Upload(node *node.Node, reader io.Reader) error
+	Upload(node *node.Node, reader io.ReadCloser) error
 	Download(node *node.Node) (io.ReadCloser, error)
 	Delete(node *node.Node) error
 }
@@ -859,7 +859,7 @@ func calculateTreeSize(ctx context.Context, nodePath string) (uint64, error) {
 }
 
 // WriteBlob writes a blob to the blobstore
-func (t *Tree) WriteBlob(node *node.Node, reader io.Reader) error {
+func (t *Tree) WriteBlob(node *node.Node, reader io.ReadCloser) error {
 	return t.blobstore.Upload(node, reader)
 }
 

pkg/storage/utils/decomposedfs/upload/upload.go

@@ -64,7 +64,7 @@ type Tree interface {
 	RestoreRecycleItemFunc(ctx context.Context, spaceid, key, trashPath string, target *node.Node) (*node.Node, *node.Node, func() error, error)
 	PurgeRecycleItemFunc(ctx context.Context, spaceid, key, purgePath string) (*node.Node, func() error, error)
 
-	WriteBlob(node *node.Node, reader io.Reader) error
+	WriteBlob(node *node.Node, reader io.ReadCloser) error
 	ReadBlob(node *node.Node) (io.ReadCloser, error)
 	DeleteBlob(node *node.Node) error