[tests-only] Add blobstore metrics

Makefile

@@ -113,7 +113,7 @@ check-changelog: release-deps
 	go run tools/check-changelog/main.go -branch edge
 
 check-changelog-drone:
-	go run tools/check-changelog/main.go -repo origin -branch edge -pr "$(PR)"
+	go run tools/check-changelog/main.go -repo origin -branch experimental -pr "$(PR)"
 
 # to be run in CI platform
 ci: build-ci test  lint-ci

go.mod

@@ -58,12 +58,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.13.0
 	github.com/rs/cors v1.8.2
 	github.com/rs/zerolog v1.28.0
 	github.com/sciencemesh/meshdirectory-web v1.0.4
 	github.com/sethvargo/go-password v0.2.0
 	github.com/stretchr/testify v1.8.0
 	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.2
 	github.com/wk8/go-ordered-map v1.0.0
@@ -171,7 +173,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.13.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.37.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
@@ -184,7 +185,6 @@ require (
 	github.com/shurcooL/vfsgen v0.0.0-20200824052919-0d455de96546 // indirect
 	github.com/sirupsen/logrus v1.9.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.16.3 // indirect
 	github.com/xanzy/ssh-agent v0.3.2 // indirect

go.sum

@@ -824,7 +824,6 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.2 h1:4jaiDzPyXQvSd7D0EjG45355tLlV3VOECpq10pLC+8s=
 github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=

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,20 @@ type Blobstore struct {
 	bucket string
 }
 
+// PrometheusAwareReader provides an interface to an prometheus aware Reader
+type PrometheusAwareReader struct {
+	r io.Reader
+	m *prometheus.CounterVec
+}
+
+// PrometheusAwareReadSeekCloser provides an interface to a prometheus aware ReadCloser
+type PrometheusAwareReadSeekCloser struct {
+	r io.ReadSeekCloser
+	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 +77,37 @@ func New(endpoint, region, bucket, accessKey, secretKey string) (*Blobstore, err
 	}, nil
 }
 
+// Read implements the read function of the PrometheusAwareReader
+func (p *PrometheusAwareReader) Read(b []byte) (n int, err error) {
+	n, err = p.r.Read(b)
+	p.m.WithLabelValues().Add(float64(n))
+	return
+}
+
+// Read implements the read function of the PrometheusAwareReadSeekCloser
+func (p *PrometheusAwareReadSeekCloser) Read(b []byte) (n int, err error) {
+	n, err = p.r.Read(b)
+	p.m.WithLabelValues().Add(float64(n))
+	return
+}
+
+// Seek implements the seek function of the PrometheusAwareReadSeekCloser
+func (p *PrometheusAwareReadSeekCloser) Seek(offset int64, whence int) (int64, error) {
+	return p.r.Seek(offset, whence)
+
+}
+
+// Close implements the close function of the PrometheusAwareReadCloser
+func (p *PrometheusAwareReadSeekCloser) 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 {
+	reader = &PrometheusAwareReader{
+		r: reader,
+		m: metrics.Tx,
+	}
 	size := int64(-1)
 	if file, ok := reader.(*os.File); ok {
 		info, err := file.Stat()
@@ -87,7 +131,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 &PrometheusAwareReadSeekCloser{
+		r: reader,
+		m: metrics.Rx,
+	}, nil
 }
 
 // Delete deletes a blob from the blobstore

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

@@ -0,0 +1,66 @@
+// Copyright 2018-2021 CERN
+//
+// 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// In applying this license, CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+package blobstore
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"go-micro.dev/v4/util/log"
+)
+
+var (
+	// Namespace defines the namespace for the defines metrics.
+	Namespace = "ocis"
+
+	// Subsystem defines the subsystem for the defines metrics.
+	Subsystem = "s3ng"
+)
+
+// Metrics defines the available metrics of this service.
+type Metrics struct {
+	Rx *prometheus.CounterVec
+	Tx *prometheus.CounterVec
+}
+
+// NewMetrics 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{}),
+	}
+	err := prometheus.Register(m.Rx)
+	if err != nil {
+		log.Errorf("Failed to register prometheus storage rx Counter (%s)", err)
+	}
+	err = prometheus.Register(m.Tx)
+	if err != nil {
+		log.Errorf("Failed to register prometheus storage tx Counter (%s)", err)
+	}
+
+	return m
+}