Add Prometheus Metrics

WORKSPACE

@@ -1,6 +1,6 @@
 workspace(name = "com_github_buildbarn_bb_remote_asset")
 
-load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive", "http_file")
+load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive")
 
 http_archive(
     name = "bazel_gomock",
@@ -21,9 +21,9 @@ http_archive(
 
 http_archive(
     name = "io_bazel_rules_docker",
-    sha256 = "14ac30773fdb393ddec90e158c9ec7ebb3f8a4fd533ec2abbfd8789ad81a284b",
-    strip_prefix = "rules_docker-0.12.1",
-    urls = ["https://github.com/bazelbuild/rules_docker/releases/download/v0.12.1/rules_docker-v0.12.1.tar.gz"],
+    sha256 = "4521794f0fba2e20f3bf15846ab5e01d5332e587e9ce81629c7f96c793bb7036",
+    strip_prefix = "rules_docker-0.14.4",
+    urls = ["https://github.com/bazelbuild/rules_docker/releases/download/v0.14.4/rules_docker-v0.14.4.tar.gz"],
 )
 
 http_archive(
@@ -44,10 +44,21 @@ http_archive(
     ],
 )
 
-load("@io_bazel_rules_docker//repositories:repositories.bzl", container_repositories = "repositories")
+load(
+    "@io_bazel_rules_docker//repositories:repositories.bzl",
+    container_repositories = "repositories",
+)
 
 container_repositories()
 
+load("@io_bazel_rules_docker//repositories:deps.bzl", container_deps = "deps")
+
+container_deps()
+
+load("@io_bazel_rules_docker//repositories:pip_repositories.bzl", container_pip_deps = "pip_deps")
+
+container_pip_deps()
+
 load("@io_bazel_rules_go//go:deps.bzl", "go_register_toolchains", "go_rules_dependencies")
 
 go_rules_dependencies()

cmd/bb_remote_asset/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/storage/blobstore:go_default_library",
         "@com_github_bazelbuild_remote_apis//build/bazel/remote/asset/v1:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/blobstore/configuration:go_default_library",
+        "@com_github_buildbarn_bb_storage//pkg/clock:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/digest:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/global:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/grpc:go_default_library",

cmd/bb_remote_asset/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/buildbarn/bb-remote-asset/pkg/storage"
 	asset_configuration "github.com/buildbarn/bb-remote-asset/pkg/storage/blobstore"
 	blobstore_configuration "github.com/buildbarn/bb-storage/pkg/blobstore/configuration"
+	"github.com/buildbarn/bb-storage/pkg/clock"
 	bb_digest "github.com/buildbarn/bb-storage/pkg/digest"
 	"github.com/buildbarn/bb-storage/pkg/global"
 	bb_grpc "github.com/buildbarn/bb-storage/pkg/grpc"
@@ -72,6 +73,7 @@ func main() {
 	pushServer := push.NewAssetPushServer(
 		assetStore,
 		allowUpdatesForInstances)
+	metricsPushServer := push.NewMetricsAssetPushServer(pushServer, clock.SystemClock, "push")
 
 	// Spawn gRPC servers for client and worker traffic.
 	go func() {
@@ -82,7 +84,7 @@ func main() {
 				func(s *grpc.Server) {
 					// Register services
 					remoteasset.RegisterFetchServer(s, fetchServer)
-					remoteasset.RegisterPushServer(s, pushServer)
+					remoteasset.RegisterPushServer(s, metricsPushServer)
 				}))
 	}()
 

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/golang/mock v1.4.4-0.20200406172829-6d816de489c1
 	github.com/golang/protobuf v1.4.2
 	github.com/gorilla/mux v1.7.4
+	github.com/prometheus/client_golang v1.7.1
 	github.com/stretchr/testify v1.6.1
 	google.golang.org/genproto v0.0.0-20200726014623-da3ae01ef02d
 	google.golang.org/grpc v1.31.0

go.sum

@@ -102,6 +102,7 @@ github.com/buildbarn/bb-storage v0.0.0-20200802055539-f0281e269c07 h1:XkZNtqFiqf
 github.com/buildbarn/bb-storage v0.0.0-20200802055539-f0281e269c07/go.mod h1:1jAvzPmKXMQPMSY2cfwVtsPRIzRY0SjJGcTZjLFc+v8=
 github.com/buildbarn/bb-storage v0.0.0-20200810194216-94a85b49d206 h1:3iIeKIDawX2iCnx88L6rtJK/ijyeumAv9/OM1/Oq4pA=
 github.com/buildbarn/bb-storage v0.0.0-20200821164915-b920faec87ed h1:HNHZvHqXFkxlAK5U/FzW86pzFS0bH+Hns6AJj/f3t28=
+github.com/buildbarn/bb-storage v0.0.0-20200917193456-be1ddb93b029 h1:zhOzsA5cmzydLGXiFpwEippGOnY2n7caoCmKT6y1m2k=
 github.com/census-instrumentation/opencensus-proto v0.2.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.2.1 h1:glEXhBS5PSLLv4IXzLA5yPRVX4bilULVyxxbrfOtDAk=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=

pkg/configuration/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "//pkg/storage:go_default_library",
         "@com_github_bazelbuild_remote_apis//build/bazel/remote/asset/v1:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/blobstore/configuration:go_default_library",
+        "@com_github_buildbarn_bb_storage//pkg/clock:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/digest:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",

pkg/configuration/new_fetcher.go

@@ -8,6 +8,7 @@ import (
 	pb "github.com/buildbarn/bb-remote-asset/pkg/proto/configuration/bb_remote_asset/fetch"
 	"github.com/buildbarn/bb-remote-asset/pkg/storage"
 	blobstore_configuration "github.com/buildbarn/bb-storage/pkg/blobstore/configuration"
+	"github.com/buildbarn/bb-storage/pkg/clock"
 	bb_digest "github.com/buildbarn/bb-storage/pkg/digest"
 
 	remoteasset "github.com/bazelbuild/remote-apis/build/bazel/remote/asset/v1"
@@ -19,7 +20,7 @@ import (
 // NewFetcherFromConfiguration creates a new Remote Asset API Fetch server from
 // a jsonnet configuration.
 func NewFetcherFromConfiguration(configuration *pb.FetcherConfiguration,
-	assetStore *storage.AssetStore,
+	assetStore storage.AssetStore,
 	casBlobAccessCreator blobstore_configuration.BlobAccessCreator) (remoteasset.FetchServer, error) {
 	var fetcher remoteasset.FetchServer
 	switch backend := configuration.Backend.(type) {
@@ -57,5 +58,6 @@ func NewFetcherFromConfiguration(configuration *pb.FetcherConfiguration,
 		return nil, status.Errorf(codes.InvalidArgument, "Fetcher configuration is invalid as no supported Fetchers are defined.")
 	}
 
-	return fetch.NewValidatingFetcher(fetch.NewLoggingFetcher(fetcher)), nil
+	fetchServer := fetch.NewValidatingFetcher(fetch.NewLoggingFetcher(fetcher))
+	return fetch.NewMetricsFetcher(fetchServer, clock.SystemClock, "fetch"), nil
 }

pkg/fetch/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "error_fetcher.go",
         "http_fetcher.go",
         "logging_fetcher.go",
+        "metrics_fetcher.go",
         "validating_fetcher.go",
     ],
     importpath = "github.com/buildbarn/bb-remote-asset/pkg/fetch",
@@ -16,9 +17,11 @@ go_library(
         "@com_github_bazelbuild_remote_apis//build/bazel/remote/asset/v1:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/blobstore:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/blobstore/buffer:go_default_library",
+        "@com_github_buildbarn_bb_storage//pkg/clock:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/digest:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/util:go_default_library",
         "@com_github_golang_protobuf//ptypes:go_default_library_gen",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@go_googleapis//google/rpc:status_go_proto",
         "@io_bazel_rules_go//proto/wkt:timestamp_go_proto",
         "@org_golang_google_grpc//codes:go_default_library",

pkg/fetch/caching_fetcher.go

@@ -18,12 +18,12 @@ import (
 
 type cachingFetcher struct {
 	fetcher    remoteasset.FetchServer
-	assetStore *storage.AssetStore
+	assetStore storage.AssetStore
 }
 
 // NewCachingFetcher creates a decorator for remoteasset.FetchServer implementations to avoid having to fetch the
 // blob remotely multiple times
-func NewCachingFetcher(fetcher remoteasset.FetchServer, assetStore *storage.AssetStore) remoteasset.FetchServer {
+func NewCachingFetcher(fetcher remoteasset.FetchServer, assetStore storage.AssetStore) remoteasset.FetchServer {
 	return &cachingFetcher{
 		fetcher:    fetcher,
 		assetStore: assetStore,

pkg/fetch/http_fetcher.go

@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"encoding/base64"
 	"encoding/hex"
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"strings"
@@ -46,7 +47,7 @@ func (hf *httpFetcher) FetchBlob(ctx context.Context, req *remoteasset.FetchBlob
 	}
 
 	if hf.allowUpdatesForInstances[instanceName] == false {
-		return nil, status.Errorf(codes.PermissionDenied, "This instance is not permitted to update the CAS.")
+		return nil, status.Errorf(codes.PermissionDenied, fmt.Sprintf("This instance ('%s') is not permitted to update the CAS.", instanceName))
 	}
 
 	// TODO: Address the following fields

pkg/fetch/metrics_fetcher.go

@@ -0,0 +1,90 @@
+package fetch
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	remoteasset "github.com/bazelbuild/remote-apis/build/bazel/remote/asset/v1"
+	"github.com/buildbarn/bb-storage/pkg/clock"
+	"github.com/buildbarn/bb-storage/pkg/util"
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var (
+	httpFetcherOperationsPrometheusMetrics sync.Once
+
+	httpFetcherOperationsBlobSizeBytes = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "buildbarn",
+			Subsystem: "remote_asset",
+			Name:      "http_fetcher_blob_size_bytes",
+			Help:      "Size of blobs fetched using the http fetcher, in bytes",
+			Buckets:   util.DecimalExponentialBuckets(-3, 6, 2),
+		},
+		[]string{"name", "operation"})
+	blobAccessOperationsDurationSeconds = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "buildbarn",
+			Subsystem: "remote_asset",
+			Name:      "http_fetcher_duration_seconds",
+			Help:      "Amount of time spent per operation on fetching remote assets, in seconds.",
+			Buckets:   util.DecimalExponentialBuckets(-3, 6, 2),
+		},
+		[]string{"name", "operation", "status"})
+)
+
+type metricsFetcher struct {
+	fetcher remoteasset.FetchServer
+	clock   clock.Clock
+
+	fetchBlobBlobSizeBytes        prometheus.Observer
+	fetchBlobDurationSeconds      prometheus.ObserverVec
+	fetchDirectoryDurationSeconds prometheus.ObserverVec
+}
+
+// NewMetricsFetcher creates a fetcher which logs metrics to prometheus
+func NewMetricsFetcher(fetcher remoteasset.FetchServer, clock clock.Clock, name string) remoteasset.FetchServer {
+	httpFetcherOperationsPrometheusMetrics.Do(func() {
+		prometheus.MustRegister(httpFetcherOperationsBlobSizeBytes)
+		prometheus.MustRegister(blobAccessOperationsDurationSeconds)
+	})
+
+	return &metricsFetcher{
+		fetcher: fetcher,
+		clock:   clock,
+
+		fetchBlobBlobSizeBytes:        httpFetcherOperationsBlobSizeBytes.WithLabelValues(name, "Fetch Blob"),
+		fetchBlobDurationSeconds:      blobAccessOperationsDurationSeconds.MustCurryWith(map[string]string{"name": name, "operation": "FetchBlob"}),
+		fetchDirectoryDurationSeconds: blobAccessOperationsDurationSeconds.MustCurryWith(map[string]string{"name": name, "operation": "FetchDirectory"}),
+	}
+}
+
+func (mf *metricsFetcher) updateDurationSeconds(vec prometheus.ObserverVec, code codes.Code, timeStart time.Time) {
+	vec.WithLabelValues(code.String()).Observe(mf.clock.Now().Sub(timeStart).Seconds())
+}
+
+func (mf *metricsFetcher) FetchBlob(ctx context.Context, req *remoteasset.FetchBlobRequest) (*remoteasset.FetchBlobResponse, error) {
+	timeStart := mf.clock.Now()
+	resp, err := mf.fetcher.FetchBlob(ctx, req)
+	if err != nil {
+		mf.updateDurationSeconds(mf.fetchBlobDurationSeconds, status.Code(err), timeStart)
+		return nil, err
+	}
+	mf.updateDurationSeconds(mf.fetchBlobDurationSeconds, codes.Code(resp.Status.Code), timeStart)
+	mf.fetchBlobBlobSizeBytes.Observe(float64(resp.BlobDigest.SizeBytes))
+	return resp, err
+}
+
+func (mf *metricsFetcher) FetchDirectory(ctx context.Context, req *remoteasset.FetchDirectoryRequest) (*remoteasset.FetchDirectoryResponse, error) {
+	timeStart := mf.clock.Now()
+	resp, err := mf.fetcher.FetchDirectory(ctx, req)
+	if err != nil {
+		mf.updateDurationSeconds(mf.fetchDirectoryDurationSeconds, status.Code(err), timeStart)
+		return nil, err
+	}
+	mf.updateDurationSeconds(mf.fetchDirectoryDurationSeconds, codes.Code(resp.Status.Code), timeStart)
+	return resp, err
+}

pkg/push/BUILD.bazel

@@ -2,14 +2,19 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["push_server.go"],
+    srcs = [
+        "metrics_push_server.go",
+        "push_server.go",
+    ],
     importpath = "github.com/buildbarn/bb-remote-asset/pkg/push",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/storage:go_default_library",
         "@com_github_bazelbuild_remote_apis//build/bazel/remote/asset/v1:go_default_library",
+        "@com_github_buildbarn_bb_storage//pkg/clock:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/digest:go_default_library",
         "@com_github_buildbarn_bb_storage//pkg/util:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
     ],

pkg/push/metrics_push_server.go

@@ -0,0 +1,87 @@
+package push
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	remoteasset "github.com/bazelbuild/remote-apis/build/bazel/remote/asset/v1"
+	"github.com/buildbarn/bb-storage/pkg/clock"
+	"github.com/buildbarn/bb-storage/pkg/util"
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var (
+	assetStoreOperationsPrometheusMetrics sync.Once
+
+	pushServerOperationsBlobSizeBytes = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "buildbarn",
+			Subsystem: "push_server",
+			Name:      "push_server_blob_size_bytes",
+			Help:      "Size of blobs being pushed, in bytes.",
+			Buckets:   util.DecimalExponentialBuckets(-3, 6, 2),
+		},
+		[]string{"name", "operation"})
+	pushServerOperationsDurationSeconds = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "buildbarn",
+			Subsystem: "push_server",
+			Name:      "blob_access_duration_seconds",
+			Help:      "Amount of time spent per operation on pushing remote assets, in seconds.",
+			Buckets:   util.DecimalExponentialBuckets(-3, 6, 2),
+		},
+		[]string{"name", "operation", "grpc_code"})
+
+	// todo(arlyon): directory size?
+)
+
+type metricsAssetPushServer struct {
+	pushServer remoteasset.PushServer
+	clock      clock.Clock
+
+	pushBlobBlobSizeBytes        prometheus.Observer
+	pushBlobDurationSeconds      prometheus.ObserverVec
+	pushDirectoryDurationSeconds prometheus.ObserverVec
+}
+
+// NewMetricsAssetPushServer wraps the PushServer to
+// report prometheus metrics.
+func NewMetricsAssetPushServer(ps remoteasset.PushServer, clock clock.Clock, name string) remoteasset.PushServer {
+	assetStoreOperationsPrometheusMetrics.Do(func() {
+		prometheus.MustRegister(pushServerOperationsBlobSizeBytes)
+		prometheus.MustRegister(pushServerOperationsDurationSeconds)
+	})
+
+	return &metricsAssetPushServer{
+		pushServer: ps,
+		clock:      clock,
+
+		pushBlobBlobSizeBytes:        pushServerOperationsBlobSizeBytes.WithLabelValues(name, "PushBlob"),
+		pushBlobDurationSeconds:      pushServerOperationsDurationSeconds.MustCurryWith(map[string]string{"name": name, "operation": "PushBlob"}),
+		pushDirectoryDurationSeconds: pushServerOperationsDurationSeconds.MustCurryWith(map[string]string{"name": name, "operation": "PushDirectory"}),
+	}
+}
+
+func (s *metricsAssetPushServer) updateDurationSeconds(vec prometheus.ObserverVec, code codes.Code, timeStart time.Time) {
+	vec.WithLabelValues(code.String()).Observe(s.clock.Now().Sub(timeStart).Seconds())
+}
+
+func (s *metricsAssetPushServer) PushBlob(ctx context.Context, req *remoteasset.PushBlobRequest) (*remoteasset.PushBlobResponse, error) {
+	if req.BlobDigest != nil {
+		s.pushBlobBlobSizeBytes.Observe(float64(req.BlobDigest.SizeBytes))
+	}
+	timeStart := s.clock.Now()
+	resp, err := s.pushServer.PushBlob(ctx, req)
+	s.updateDurationSeconds(s.pushBlobDurationSeconds, status.Code(err), timeStart)
+	return resp, err
+}
+
+func (s *metricsAssetPushServer) PushDirectory(ctx context.Context, req *remoteasset.PushDirectoryRequest) (*remoteasset.PushDirectoryResponse, error) {
+	timeStart := s.clock.Now()
+	resp, err := s.pushServer.PushDirectory(ctx, req)
+	s.updateDurationSeconds(s.pushDirectoryDurationSeconds, status.Code(err), timeStart)
+	return resp, err
+}

pkg/push/push_server.go

@@ -12,13 +12,13 @@ import (
 )
 
 type assetPushServer struct {
-	assetStore               *storage.AssetStore
+	assetStore               storage.AssetStore
 	allowUpdatesForInstances map[digest.InstanceName]bool
 }
 
 // NewAssetPushServer creates a gRPC service for serving the contents
 // of a Remote Asset Push server.
-func NewAssetPushServer(AssetStore *storage.AssetStore, allowUpdatesForInstances map[digest.InstanceName]bool) remoteasset.PushServer {
+func NewAssetPushServer(AssetStore storage.AssetStore, allowUpdatesForInstances map[digest.InstanceName]bool) remoteasset.PushServer {
 	return &assetPushServer{
 		assetStore:               AssetStore,
 		allowUpdatesForInstances: allowUpdatesForInstances,