refactor(repository): refactored Prometheus metrics (#2532)

This may be a breaking change for users who rely on particular kopia metrics (unlikely): - introduced blob-level metrics: * `kopia_blob_download_full_blob_bytes_total` * `kopia_blob_download_partial_blob_bytes_total` * `kopia_blob_upload_bytes_total` * `kopia_blob_storage_latency_ms` - per-method latency distribution * `kopia_blob_errors_total` - per-method error counter - updated cache metrics to indicate particular cache * `kopia_cache_hit_bytes_total{cache="CACHE_TYPE"}` * `kopia_cache_hit_total{cache="CACHE_TYPE"}` * `kopia_cache_malformed_total{cache="CACHE_TYPE"}` * `kopia_cache_miss_total{cache="CACHE_TYPE"}` * `kopia_cache_miss_errors_total{cache="CACHE_TYPE"}` * `kopia_cache_miss_bytes_total{cache="CACHE_TYPE"}` * `kopia_cache_store_errors_total{cache="CACHE_TYPE"}` where `CACHE_TYPE` is one of `contents`, `metadata` or `index-blobs` - reorganized and unified content-level metrics: * `kopia_content_write_bytes_total` * `kopia_content_write_duration_nanos_total` * `kopia_content_compression_attempted_bytes_total` * `kopia_content_compression_attempted_duration_nanos_total` * `kopia_content_compression_savings_bytes_total` * `kopia_content_compressible_bytes_total` * `kopia_content_non_compressible_bytes_total` * `kopia_content_after_compression_bytes_total` * `kopia_content_decompressed_bytes_total` * `kopia_content_decompressed_duration_nanos_total` * `kopia_content_encrypted_bytes_total` * `kopia_content_encrypted_duration_nanos_total` * `kopia_content_hashed_bytes_total` * `kopia_content_hashed_duration_nanos_total` * `kopia_content_deduplicated_bytes_total` * `kopia_content_read_bytes_total` * `kopia_content_read_duration_nanos_total` * `kopia_content_decrypted_bytes_total` * `kopia_content_decrypted_duration_nanos_total` * `kopia_content_uploaded_bytes_total` Also introduced `internal/metrics` framework which constructs Prometheus metrics in a uniform way and will allow us to include some of these metrics in telemetry report in future PRs.
2026-10-27 16:48:59 -04:00 · 2022-11-10 05:30:06 +00:00
1 parent 3ff7b5a840
commit 78edd92692
@@ -69,7 +69,7 @@ func TestMetricsPushFlags(t *testing.T) {

 	for _, b := range bodies {
 		// make sure bodies include some kopia metrics, don't need more
-		require.Contains(t, b, "kopia_content_cache_hit_bytes")
+		require.Contains(t, b, "kopia_cache_hit_bytes_total")
 	}

 	env.RunAndExpectFailure(t, "repo", "status",
@@ -49,13 +49,13 @@ require (
 	github.com/zeebo/blake3 v0.2.3
 	go.opencensus.io v0.24.0 // indirect
 	go.uber.org/zap v1.23.0
-	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa
-	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4
-	golang.org/x/net v0.0.0-20221014081412-f15817d10f9b
+	golang.org/x/crypto v0.1.0
+	golang.org/x/mod v0.6.0
+	golang.org/x/net v0.1.0
 	golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783
 	golang.org/x/sync v0.1.0
-	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec
-	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
+	golang.org/x/sys v0.1.0
+	golang.org/x/term v0.1.0
 	golang.org/x/text v0.4.0
 	google.golang.org/api v0.103.0
 	google.golang.org/grpc v1.50.1
@@ -91,7 +91,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.37.0
 	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/rs/xid v1.4.0 // indirect
@@ -108,10 +107,12 @@ require (
 	github.com/edsrzf/mmap-go v1.1.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/klauspost/reedsolomon v1.11.1
+	github.com/prometheus/client_model v0.3.0
 	go.opentelemetry.io/otel v1.11.1
 	go.opentelemetry.io/otel/exporters/jaeger v1.11.1
 	go.opentelemetry.io/otel/sdk v1.11.1
 	go.opentelemetry.io/otel/trace v1.11.1
+	golang.org/x/exp v0.0.0-20221028150844-83b7d23a625f
 )

 require (
@@ -448,8 +448,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa h1:zuSxTR4o9y82ebqCUJYNGJbGPo6sKVl54f/TVDObg1c=
-golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -460,6 +460,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
+golang.org/x/exp v0.0.0-20221028150844-83b7d23a625f h1:Al51T6tzvuh3oiwX11vex3QgJ2XTedFPGmbEVh8cdoc=
+golang.org/x/exp v0.0.0-20221028150844-83b7d23a625f/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
@@ -480,8 +482,8 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 h1:6zppjxzCulZykYSLyVDYbneBfbaBIQPYMevg0bEwv2s=
-golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.6.0 h1:b9gGHsz9/HhJ3HF5DHQytPpuwocVTChQJK3AvoLRD5I=
+golang.org/x/mod v0.6.0/go.mod h1:4mET923SAdbXp2ki8ey+zGs1SLqsuM2Y0uvdZR/fUNI=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -519,8 +521,8 @@ golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
-golang.org/x/net v0.0.0-20221014081412-f15817d10f9b h1:tvrvnPFcdzp294diPnrdZZZ8XUt2Tyj7svb7X52iDuU=
-golang.org/x/net v0.0.0-20221014081412-f15817d10f9b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
+golang.org/x/net v0.1.0 h1:hZ/3BUoy5aId7sCpA/Tc5lt8DkFgdVS2onTpJsZ/fl0=
+golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -588,11 +590,12 @@ golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.1.0 h1:g6Z6vPFA9dYBAF7DWcH6sCcOntplXsDKcliusYijMlw=
+golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -1,66 +1,71 @@
 package cache

-import (
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-)
+import "github.com/kopia/kopia/internal/metrics"

-//nolint:gochecknoglobals,promlinter
-var (
-	metricHitCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_hit_count",
-		Help: "Number of time content was retrieved from the cache",
-	})
-
-	metricHitBytes = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_hit_bytes",
-		Help: "Number of bytes retrieved from the cache",
-	})
-
-	metricMissCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_miss_count",
-		Help: "Number of time content was not found in the cache and fetched from the storage",
-	})
-
-	metricMalformedCacheDataCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_malformed",
-		Help: "Number of times malformed content was read from the cache",
-	})
-
-	metricMissBytes = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_missed_bytes",
-		Help: "Number of bytes retrieved from the underlying storage",
-	})
-
-	metricMissErrors = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_miss_error_count",
-		Help: "Number of time content could not be found in the underlying storage",
-	})
-
-	metricStoreErrors = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_cache_store_error_count",
-		Help: "Number of time content could not be saved in the cache",
-	})
-)
-
-func reportMissError() {
-	metricMissErrors.Inc()
+type metricsStruct struct {
+	metricHitCount                *metrics.Counter
+	metricHitBytes                *metrics.Counter
+	metricMissCount               *metrics.Counter
+	metricMalformedCacheDataCount *metrics.Counter
+	metricMissBytes               *metrics.Counter
+	metricMissErrors              *metrics.Counter
+	metricStoreErrors             *metrics.Counter
 }

-func reportMissBytes(length int64) {
-	metricMissCount.Inc()
-	metricMissBytes.Add(float64(length))
+func initMetricsStruct(mr *metrics.Registry, cacheID string) metricsStruct {
+	labels := map[string]string{
+		"cache": cacheID,
+	}
+
+	return metricsStruct{
+		metricHitCount: mr.CounterInt64(
+			"cache_hit",
+			"Number of time content was retrieved from the cache", labels),
+
+		metricHitBytes: mr.CounterInt64(
+			"cache_hit_bytes",
+			"Number of bytes retrieved from the cache", labels),
+
+		metricMissCount: mr.CounterInt64(
+			"cache_miss",
+			"Number of time content was not found in the cache and fetched from the storage", labels),
+
+		metricMalformedCacheDataCount: mr.CounterInt64(
+			"cache_malformed",
+			"Number of times malformed content was read from the cache", labels),
+
+		metricMissBytes: mr.CounterInt64(
+			"cache_miss_bytes",
+			"Number of bytes retrieved from the underlying storage", labels),
+
+		metricMissErrors: mr.CounterInt64(
+			"cache_miss_errors",
+			"Number of time content could not be found in the underlying storage", labels),
+
+		metricStoreErrors: mr.CounterInt64(
+			"cache_store_errors",
+			"Number of time content could not be saved in the cache", labels),
+	}
 }

-func reportHitBytes(length int64) {
-	metricHitCount.Inc()
-	metricHitBytes.Add(float64(length))
+func (s *metricsStruct) reportMissError() {
+	s.metricMissErrors.Add(1)
 }

-func reportMalformedData() {
-	metricMalformedCacheDataCount.Inc()
+func (s *metricsStruct) reportMissBytes(length int64) {
+	s.metricMissCount.Add(1)
+	s.metricMissBytes.Add(length)
 }

-func reportStoreError() {
-	metricStoreErrors.Inc()
+func (s *metricsStruct) reportHitBytes(length int64) {
+	s.metricHitCount.Add(1)
+	s.metricHitBytes.Add(length)
+}
+
+func (s *metricsStruct) reportMalformedData() {
+	s.metricMalformedCacheDataCount.Add(1)
+}
+
+func (s *metricsStruct) reportStoreError() {
+	s.metricStoreErrors.Add(1)
 }
@@ -7,6 +7,7 @@ import (

 	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/impossible"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/repo/blob"
 )

@@ -95,12 +96,12 @@ func (c *contentCacheImpl) getContentFromFullBlob(ctx context.Context, blobID bl
 func (c *contentCacheImpl) fetchBlobInternal(ctx context.Context, blobID blob.ID, blobData *gather.WriteBuffer) error {
 	// read the entire blob
 	if err := c.st.GetBlob(ctx, blobID, 0, -1, blobData); err != nil {
-		reportMissError()
+		c.pc.reportMissError()

 		return errors.Wrapf(err, "failed to get blob with ID %s", blobID)
 	}

-	reportMissBytes(int64(blobData.Length()))
+	c.pc.reportMissBytes(int64(blobData.Length()))

 	// store the whole blob in the cache.
 	c.pc.Put(ctx, BlobIDCacheKey(blobID), blobData.Bytes())
@@ -131,12 +132,12 @@ func (c *contentCacheImpl) getContentFromFullOrPartialBlob(ctx context.Context,
 	}

 	if err := c.st.GetBlob(ctx, blobID, offset, length, output); err != nil {
-		reportMissError()
+		c.pc.reportMissError()

 		return errors.Wrapf(err, "failed to get blob with ID %s", blobID)
 	}

-	reportMissBytes(int64(output.Length()))
+	c.pc.reportMissBytes(int64(output.Length()))

 	c.pc.Put(ctx, ContentIDCacheKey(contentID), output.Bytes())

@@ -173,7 +174,7 @@ func (c *contentCacheImpl) CacheStorage() Storage {
 }

 // NewContentCache creates new content cache for data contents.
-func NewContentCache(ctx context.Context, st blob.Storage, opt Options) (ContentCache, error) {
+func NewContentCache(ctx context.Context, st blob.Storage, opt Options, mr *metrics.Registry) (ContentCache, error) {
 	cacheStorage := opt.Storage
 	if cacheStorage == nil {
 		if opt.BaseCacheDirectory == "" {
@@ -188,7 +189,7 @@ func NewContentCache(ctx context.Context, st blob.Storage, opt Options) (Content
 		}
 	}

-	pc, err := NewPersistentCache(ctx, opt.CacheSubDir, cacheStorage, ChecksumProtection(opt.HMACSecret), opt.Sweep)
+	pc, err := NewPersistentCache(ctx, opt.CacheSubDir, cacheStorage, ChecksumProtection(opt.HMACSecret), opt.Sweep, mr)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to create base cache")
 	}
@@ -27,7 +27,7 @@ func newContentDataCache(ctx context.Context, st blob.Storage, cacheStorage cach
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: 100,
 		},
-	})
+	}, nil)
 }

 func newContentMetadataCache(ctx context.Context, st blob.Storage, cacheStorage cache.Storage) (cache.ContentCache, error) {
@@ -38,7 +38,7 @@ func newContentMetadataCache(ctx context.Context, st blob.Storage, cacheStorage
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: 100,
 		},
-	})
+	}, nil)
 }

 func TestPrefetchBlocksGetContent_DataCache(t *testing.T) {
@@ -27,7 +27,7 @@ func TestContentCacheForData(t *testing.T) {
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: 100,
 		},
-	})
+	}, nil)
 	require.NoError(t, err)

 	var tmp gather.WriteBuffer
@@ -78,7 +78,7 @@ func TestContentCacheForData_Passthrough(t *testing.T) {

 	ctx := testlogging.Context(t)

-	dataCache, err := cache.NewContentCache(ctx, underlying, cache.Options{})
+	dataCache, err := cache.NewContentCache(ctx, underlying, cache.Options{}, nil)

 	require.NoError(t, err)
 	require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{}))
@@ -29,7 +29,7 @@ func TestContentCacheForMetadata(t *testing.T) {
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: 100,
 		},
-	})
+	}, nil)
 	require.NoError(t, err)

 	cacheStorage := metadataCache.CacheStorage()
@@ -80,7 +80,7 @@ func TestContentCacheForMetadata_Passthrough(t *testing.T) {
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: 100,
 		},
-	})
+	}, nil)

 	require.NoError(t, err)
 	require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{}))
@@ -62,7 +62,7 @@ func TestCacheExpiration(t *testing.T) {
 			SweepFrequency: 500 * time.Millisecond,
 			TouchThreshold: -1,
 		},
-	})
+	}, nil)

 	require.NoError(t, err)

@@ -121,7 +121,7 @@ func TestDiskContentCache(t *testing.T) {
 		Sweep: cache.SweepSettings{
 			MaxSizeBytes: maxBytes,
 		},
-	})
+	}, nil)
 	require.NoError(t, err)

 	defer cc.Close(ctx)
@@ -208,7 +208,7 @@ func TestCacheFailureToOpen(t *testing.T) {
 	_, err := cache.NewContentCache(testlogging.Context(t), underlyingStorage, cache.Options{
 		Storage: withoutTouchBlob{faultyCache},
 		Sweep:   cache.SweepSettings{MaxSizeBytes: 10000},
-	})
+	}, nil)
 	require.Error(t, err)
 	require.ErrorContains(t, err, someError.Error())

@@ -218,7 +218,7 @@ func TestCacheFailureToOpen(t *testing.T) {
 	cc, err := cache.NewContentCache(ctx, underlyingStorage, cache.Options{
 		Storage: withoutTouchBlob{faultyCache},
 		Sweep:   cache.SweepSettings{MaxSizeBytes: 10000},
-	})
+	}, nil)
 	require.NoError(t, err)

 	cc.Close(ctx)
@@ -235,7 +235,7 @@ func TestCacheFailureToWrite(t *testing.T) {
 	cc, err := cache.NewContentCache(testlogging.Context(t), underlyingStorage, cache.Options{
 		Storage: withoutTouchBlob{faultyCache},
 		Sweep:   cache.SweepSettings{MaxSizeBytes: 10000},
-	})
+	}, nil)
 	require.NoError(t, err)

 	ctx := testlogging.Context(t)
@@ -270,7 +270,7 @@ func TestCacheFailureToRead(t *testing.T) {
 	cc, err := cache.NewContentCache(testlogging.Context(t), underlyingStorage, cache.Options{
 		Storage: withoutTouchBlob{faultyCache},
 		Sweep:   cache.SweepSettings{MaxSizeBytes: 10000},
-	})
+	}, nil)
 	require.NoError(t, err)

 	ctx := testlogging.Context(t)
@@ -14,6 +14,7 @@ import (
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/ctxutil"
 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/internal/releasable"
 	"github.com/kopia/kopia/internal/timetrack"
 	"github.com/kopia/kopia/repo/blob"
@@ -51,6 +52,8 @@ type PersistentCache struct {
 	periodicSweepClosed  chan struct{}

 	mutexCache *lru.Cache
+
+	metricsStruct
 }

 // CacheStorage returns cache storage.
@@ -104,12 +107,12 @@ func (c *PersistentCache) GetOrLoad(ctx context.Context, key string, fetch func(
 	}

 	if err := fetch(output); err != nil {
-		reportMissError()
+		c.reportMissError()

 		return err
 	}

-	reportMissBytes(int64(output.Length()))
+	c.reportMissBytes(int64(output.Length()))

 	c.Put(ctx, key, output.Bytes())

@@ -140,7 +143,7 @@ func (c *PersistentCache) GetPartial(ctx context.Context, key string, offset, le

 		if err := prot.Verify(key, tmp.Bytes(), output); err == nil {
 			// cache hit
-			reportHitBytes(int64(output.Length()))
+			c.reportHitBytes(int64(output.Length()))

 			// cache hit
 			c.cacheStorage.TouchBlob(ctx, blob.ID(key), c.sweep.TouchThreshold) //nolint:errcheck
@@ -149,7 +152,7 @@ func (c *PersistentCache) GetPartial(ctx context.Context, key string, offset, le
 		}

 		// delete invalid blob
-		reportMalformedData()
+		c.reportMalformedData()

 		if err := c.cacheStorage.DeleteBlob(ctx, blob.ID(key)); err != nil && !errors.Is(err, blob.ErrBlobNotFound) {
 			log(ctx).Errorf("unable to delete %v entry %v: %v", c.description, key, err)
@@ -162,7 +165,7 @@ func (c *PersistentCache) GetPartial(ctx context.Context, key string, offset, le
 		l = 0
 	}

-	reportMissBytes(l)
+	c.reportMissBytes(l)

 	return false
 }
@@ -181,7 +184,7 @@ func (c *PersistentCache) Put(ctx context.Context, key string, data gather.Bytes
 	c.storageProtection.Protect(key, data, &protected)

 	if err := c.cacheStorage.PutBlob(ctx, blob.ID(key), protected.Bytes(), blob.PutOptions{}); err != nil {
-		reportStoreError()
+		c.reportStoreError()

 		log(ctx).Errorf("unable to add %v to %v: %v", key, c.description, err)
 	}
@@ -331,7 +334,7 @@ func (s SweepSettings) applyDefaults() SweepSettings {
 }

 // NewPersistentCache creates the persistent cache in the provided storage.
-func NewPersistentCache(ctx context.Context, description string, cacheStorage Storage, storageProtection StorageProtection, sweep SweepSettings) (*PersistentCache, error) {
+func NewPersistentCache(ctx context.Context, description string, cacheStorage Storage, storageProtection StorageProtection, sweep SweepSettings, mr *metrics.Registry) (*PersistentCache, error) {
 	if cacheStorage == nil {
 		return nil, nil
 	}
@@ -348,6 +351,7 @@ func NewPersistentCache(ctx context.Context, description string, cacheStorage St
 		periodicSweepClosed: make(chan struct{}),
 		description:         description,
 		storageProtection:   storageProtection,
+		metricsStruct:       initMetricsStruct(mr, description),
 	}

 	c.mutexCache, _ = lru.New(mutexCacheSize)
@@ -29,7 +29,7 @@ func TestPersistentLRUCache(t *testing.T) {
 		MaxSizeBytes:   maxSizeBytes,
 		TouchThreshold: cache.DefaultTouchThreshold,
 		SweepFrequency: cache.DefaultSweepFrequency,
-	})
+	}, nil)
 	require.NoError(t, err)

 	var tmp gather.WriteBuffer
@@ -71,7 +71,7 @@ func TestPersistentLRUCache(t *testing.T) {
 		MaxSizeBytes:   maxSizeBytes,
 		TouchThreshold: cache.DefaultTouchThreshold,
 		SweepFrequency: cache.DefaultSweepFrequency,
-	})
+	}, nil)
 	require.NoError(t, err)

 	verifyCached(ctx, t, pc, "key1", nil)
@@ -85,7 +85,7 @@ func TestPersistentLRUCache(t *testing.T) {
 		MaxSizeBytes:   maxSizeBytes,
 		TouchThreshold: cache.DefaultTouchThreshold,
 		SweepFrequency: cache.DefaultSweepFrequency,
-	})
+	}, nil)
 	require.NoError(t, err)

 	someError := errors.Errorf("some error")
@@ -135,7 +135,7 @@ func TestPersistentLRUCache_Invalid(t *testing.T) {

 	fs.AddFault(blobtesting.MethodGetMetadata).ErrorInstead(someError)

-	pc, err := cache.NewPersistentCache(ctx, "test", fc, nil, cache.SweepSettings{})
+	pc, err := cache.NewPersistentCache(ctx, "test", fc, nil, cache.SweepSettings{}, nil)
 	require.ErrorIs(t, err, someError)
 	require.Nil(t, pc)
 }
@@ -153,7 +153,7 @@ func TestPersistentLRUCache_GetDeletesInvalidBlob(t *testing.T) {
 	fs := blobtesting.NewFaultyStorage(st)
 	fc := faultyCache{fs}

-	pc, err := cache.NewPersistentCache(ctx, "test", fc, cache.ChecksumProtection([]byte{1, 2, 3}), cache.SweepSettings{})
+	pc, err := cache.NewPersistentCache(ctx, "test", fc, cache.ChecksumProtection([]byte{1, 2, 3}), cache.SweepSettings{}, nil)
 	require.NoError(t, err)

 	pc.Put(ctx, "key", gather.FromSlice([]byte{1, 2, 3}))
@@ -183,7 +183,7 @@ func TestPersistentLRUCache_PutIgnoresStorageFailure(t *testing.T) {
 	fs := blobtesting.NewFaultyStorage(st)
 	fc := faultyCache{fs}

-	pc, err := cache.NewPersistentCache(ctx, "test", fc, cache.ChecksumProtection([]byte{1, 2, 3}), cache.SweepSettings{})
+	pc, err := cache.NewPersistentCache(ctx, "test", fc, cache.ChecksumProtection([]byte{1, 2, 3}), cache.SweepSettings{}, nil)
 	require.NoError(t, err)

 	fs.AddFault(blobtesting.MethodPutBlob).ErrorInstead(someError)
@@ -213,7 +213,7 @@ func TestPersistentLRUCache_SweepMinSweepAge(t *testing.T) {
 		SweepFrequency: 100 * time.Millisecond,
 		MaxSizeBytes:   1000,
 		MinSweepAge:    10 * time.Second,
-	})
+	}, nil)
 	require.NoError(t, err)
 	pc.Put(ctx, "key", gather.FromSlice([]byte{1, 2, 3}))
 	pc.Put(ctx, "key2", gather.FromSlice(bytes.Repeat([]byte{1, 2, 3}, 1e6)))
@@ -241,7 +241,7 @@ func TestPersistentLRUCache_SweepIgnoresErrors(t *testing.T) {
 	pc, err := cache.NewPersistentCache(ctx, "test", fc, cache.ChecksumProtection([]byte{1, 2, 3}), cache.SweepSettings{
 		SweepFrequency: 100 * time.Millisecond,
 		MaxSizeBytes:   1000,
-	})
+	}, nil)
 	require.NoError(t, err)

 	// ignore delete errors forever
@@ -275,7 +275,7 @@ func TestPersistentLRUCache_Sweep1(t *testing.T) {
 		SweepFrequency: 10 * time.Millisecond,
 		MaxSizeBytes:   1,
 		MinSweepAge:    0 * time.Second,
-	})
+	}, nil)
 	require.NoError(t, err)
 	pc.Put(ctx, "key", gather.FromSlice([]byte{1, 2, 3}))
 	pc.Put(ctx, "key", gather.FromSlice(bytes.Repeat([]byte{1, 2, 3}, 1e6)))
@@ -329,7 +329,7 @@ func TestPersistentLRUCache_Defaults(t *testing.T) {

 	pc, err := cache.NewPersistentCache(ctx, "testing", cs, nil, cache.SweepSettings{
 		MaxSizeBytes: maxSizeBytes,
-	})
+	}, nil)
 	require.NoError(t, err)

 	defer pc.Close(ctx)
@@ -0,0 +1,54 @@
+package metrics_test
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	io_prometheus_client "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+)
+
+func mustFindMetric(t *testing.T, wantName string, wantType io_prometheus_client.MetricType, wantLabels map[string]string) *io_prometheus_client.Metric {
+	t.Helper()
+
+	mf, err := prometheus.DefaultGatherer.Gather()
+	require.NoError(t, err)
+
+	for _, f := range mf {
+		if f.GetName() != wantName {
+			continue
+		}
+
+		if f.GetType() != wantType {
+			continue
+		}
+
+		for _, l := range f.Metric {
+			if len(l.Label) != len(wantLabels) {
+				continue
+			}
+
+			found := true
+
+			for _, lab := range l.Label {
+				if wantLabels[lab.GetName()] != lab.GetValue() {
+					found = false
+				}
+			}
+
+			if found {
+				return l
+			}
+		}
+	}
+
+	for _, f := range mf {
+		for _, l := range f.Metric {
+			t.Logf("  %v %v %v", f.GetName(), f.GetType(), l.Label)
+		}
+	}
+
+	require.Failf(t, "metric %v not found", wantName)
+
+	return nil
+}
@@ -0,0 +1,65 @@
+package metrics
+
+import (
+	"sync/atomic"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Counter represents a monotonically increasing int64 counter value.
+type Counter struct {
+	state atomic.Int64
+
+	prom prometheus.Counter
+}
+
+// Add adds a value to a counter.
+func (c *Counter) Add(v int64) {
+	if c == nil {
+		return
+	}
+
+	c.prom.Add(float64(v))
+	c.state.Add(v)
+}
+
+// newState initializes counter state and returns previous state or nil.
+func (c *Counter) newState() int64 {
+	return c.state.Swap(0)
+}
+
+// Snapshot captures the momentary state of a counter.
+func (c *Counter) Snapshot() int64 {
+	if c == nil {
+		return 0
+	}
+
+	return c.state.Load()
+}
+
+// CounterInt64 gets a persistent int64 counter with the provided name.
+func (r *Registry) CounterInt64(name, help string, labels map[string]string) *Counter {
+	if r == nil {
+		return nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	fullName := name + labelsSuffix(labels)
+
+	c := r.allCounters[fullName]
+	if c == nil {
+		c = &Counter{
+			prom: getPrometheusCounter(prometheus.CounterOpts{
+				Name: prometheusPrefix + name + prometheusCounterSuffix,
+				Help: help,
+			}, labels),
+		}
+
+		c.newState()
+		r.allCounters[fullName] = c
+	}
+
+	return c
+}
@@ -0,0 +1,66 @@
+package metrics_test
+
+import (
+	"testing"
+
+	prommodel "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/metrics"
+)
+
+func TestCounter_Nil(t *testing.T) {
+	var e *metrics.Registry
+	cnt := e.CounterInt64("aaa", "bbb", nil)
+	require.Nil(t, cnt)
+	cnt.Add(33)
+	require.Equal(t, int64(0), cnt.Snapshot())
+}
+
+func TestCounter_NoLabels(t *testing.T) {
+	e := metrics.NewRegistry()
+	cnt := e.CounterInt64("some_gauge", "some-help", nil)
+
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+	cnt.Add(33)
+	require.Equal(t, 33.0,
+		mustFindMetric(t, "kopia_some_gauge_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+	cnt.Add(100)
+	require.Equal(t, 133.0,
+		mustFindMetric(t, "kopia_some_gauge_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+
+	require.Equal(t, int64(133), cnt.Snapshot())
+}
+
+func TestCounter_WithLabels(t *testing.T) {
+	e := metrics.NewRegistry()
+	cnt1 := e.CounterInt64("some_gauge2", "some-help", map[string]string{"key1": "label1"})
+	cnt2 := e.CounterInt64("some_gauge2", "some-help", map[string]string{"key1": "label2"})
+
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label1"}).
+			GetCounter().GetValue())
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label2"}).
+			GetCounter().GetValue())
+	cnt1.Add(33)
+	cnt2.Add(44)
+	require.Equal(t, 44.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label2"}).
+			GetCounter().GetValue())
+	require.Equal(t, 33.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label1"}).
+			GetCounter().GetValue())
+	cnt1.Add(100)
+	cnt2.Add(100)
+	require.Equal(t, 133.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label1"}).
+			GetCounter().GetValue())
+	require.Equal(t, 144.0,
+		mustFindMetric(t, "kopia_some_gauge2_total", prommodel.MetricType_COUNTER, map[string]string{"key1": "label2"}).
+			GetCounter().GetValue())
+}
@@ -0,0 +1,186 @@
+package metrics
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/exp/constraints"
+)
+
+// DistributionState captures momentary state of a Distribution.
+type DistributionState[T constraints.Float | constraints.Integer] struct {
+	Min              T
+	Max              T
+	Sum              T
+	Count            int64
+	BucketCounters   []int64
+	BucketThresholds []T
+}
+
+// Mean returns arithmetic mean value captured in the distribution.
+func (s *DistributionState[T]) Mean() T {
+	if s.Count == 0 {
+		return 0
+	}
+
+	return s.Sum / T(s.Count)
+}
+
+// Distribution measures distribution/summary of values.
+type Distribution[T constraints.Integer | constraints.Float] struct {
+	mu               sync.Mutex
+	state            atomic.Pointer[DistributionState[T]]
+	bucketThresholds []T
+
+	prom            prometheus.Observer
+	prometheusScale float64
+}
+
+// Observe adds the provided observation value to the summary.
+func (d *Distribution[T]) Observe(value T) {
+	if d == nil {
+		return
+	}
+
+	st := d.state.Load()
+	b := bucketForThresholds(d.bucketThresholds, value)
+
+	d.prom.Observe(float64(value) / d.prometheusScale)
+
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	st.Sum += value
+	st.Count++
+
+	if st.Count == 1 {
+		st.Max = value
+		st.Min = value
+	} else {
+		if value > st.Max {
+			st.Max = value
+		}
+
+		if value < st.Min {
+			st.Min = value
+		}
+	}
+
+	st.BucketCounters[b]++
+}
+
+// Snapshot returns a snapshot of the distribution state.
+func (d *Distribution[T]) Snapshot() DistributionState[T] {
+	if d == nil {
+		return DistributionState[T]{}
+	}
+
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	s := d.state.Load()
+
+	return DistributionState[T]{
+		Sum:            s.Sum,
+		Count:          s.Count,
+		Min:            s.Min,
+		Max:            s.Max,
+		BucketCounters: append([]int64(nil), s.BucketCounters...),
+	}
+}
+
+func (d *Distribution[T]) newState() *DistributionState[T] {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	return d.state.Swap(&DistributionState[T]{
+		BucketCounters:   make([]int64, len(d.bucketThresholds)+1),
+		BucketThresholds: d.bucketThresholds,
+	})
+}
+
+// DurationDistribution gets a persistent duration distribution with the provided name.
+func (r *Registry) DurationDistribution(name, help string, thresholds *Thresholds[time.Duration], labels map[string]string) *Distribution[time.Duration] {
+	if r == nil {
+		return nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	fullName := name + labelsSuffix(labels)
+	h := r.allDurationDistributions[fullName]
+
+	if h == nil {
+		var promBuckets []float64
+
+		// convert from nanoseconds to seconds
+		promScale := thresholds.promScale
+		for _, v := range thresholds.values {
+			promBuckets = append(promBuckets, float64(v)/promScale)
+		}
+
+		h = &Distribution[time.Duration]{
+			bucketThresholds: thresholds.values,
+			prometheusScale:  promScale,
+			prom: getPrometheusHistogram(
+				prometheus.HistogramOpts{
+					Name:    prometheusPrefix + name + thresholds.prometheusSuffix,
+					Help:    help,
+					Buckets: promBuckets,
+				},
+				labels,
+			),
+		}
+
+		h.newState()
+
+		r.allDurationDistributions[fullName] = h
+	}
+
+	return h
+}
+
+// SizeDistribution gets a persistent size distribution with the provided name.
+func (r *Registry) SizeDistribution(name, help string, thresholds *Thresholds[int64], labels map[string]string) *Distribution[int64] {
+	if r == nil {
+		return nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	fullName := name + labelsSuffix(labels)
+	h := r.allSizeDistributions[fullName]
+
+	if h == nil {
+		var promBuckets []float64
+
+		// convert from nanoseconds to seconds
+		promScale := thresholds.promScale
+		for _, v := range thresholds.values {
+			promBuckets = append(promBuckets, float64(v)/promScale)
+		}
+
+		h = &Distribution[int64]{
+			bucketThresholds: thresholds.values,
+			prometheusScale:  promScale,
+			prom: getPrometheusHistogram(
+				prometheus.HistogramOpts{
+					Name:    prometheusPrefix + name + thresholds.prometheusSuffix,
+					Help:    help,
+					Buckets: promBuckets,
+				},
+				labels,
+			),
+		}
+
+		h.newState()
+
+		r.allSizeDistributions[fullName] = h
+	}
+
+	return h
+}
@@ -0,0 +1,23 @@
+package metrics
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBucketForThresholds(t *testing.T) {
+	buckets := IOLatencyThresholds.values
+	n := len(buckets)
+
+	assert.Equal(t, 0, bucketForThresholds(buckets, buckets[0]-1))
+
+	for i := 0; i < n; i++ {
+		assert.Equal(t, i, bucketForThresholds(buckets, buckets[i]-1))
+		assert.Equal(t, i, bucketForThresholds(buckets, buckets[i]))
+		assert.Equal(t, i+1, bucketForThresholds(buckets, buckets[i]+1), "looking for %v", buckets[i]+1)
+	}
+
+	assert.Equal(t, n, bucketForThresholds(buckets, math.MaxInt64))
+}
@@ -0,0 +1,142 @@
+package metrics_test
+
+import (
+	"testing"
+	"time"
+
+	prommodel "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/metrics"
+)
+
+func TestDurationDistribution_Nil(t *testing.T) {
+	var e *metrics.Registry
+	dist := e.DurationDistribution("aaa", "bbb", metrics.IOLatencyThresholds, nil)
+	require.Nil(t, dist)
+	dist.Observe(time.Second)
+	require.Equal(t, metrics.DistributionState[time.Duration]{}, dist.Snapshot())
+}
+
+func TestSizeDistribution_Nil(t *testing.T) {
+	var e *metrics.Registry
+	cnt := e.SizeDistribution("aaa", "bbb", metrics.ISOBytesThresholds, nil)
+	require.Nil(t, cnt)
+	cnt.Observe(333)
+}
+
+func TestDurationDistribution_NoLabels(t *testing.T) {
+	e := metrics.NewRegistry()
+	cnt := e.DurationDistribution("some_dur_dist", "some-help", metrics.IOLatencyThresholds, nil)
+	cnt.Observe(time.Second)
+
+	h1 := mustFindMetric(t, "kopia_some_dur_dist_ms", prommodel.MetricType_HISTOGRAM, nil)
+	require.Equal(t, uint64(1), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 1000.0, h1.GetHistogram().GetSampleSum())
+
+	cnt.Observe(30 * time.Second)
+
+	h1 = mustFindMetric(t, "kopia_some_dur_dist_ms", prommodel.MetricType_HISTOGRAM, nil)
+	require.Equal(t, uint64(2), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 31000.0, h1.GetHistogram().GetSampleSum())
+}
+
+func TestDurationDistribution_WithLabels(t *testing.T) {
+	e := metrics.NewRegistry()
+	cnt1 := e.DurationDistribution("some_dur_dist2", "some-help", metrics.IOLatencyThresholds, map[string]string{"key1": "label1"})
+	cnt2 := e.DurationDistribution("some_dur_dist2", "some-help", metrics.IOLatencyThresholds, map[string]string{"key1": "label2"})
+
+	snap0 := cnt1.Snapshot()
+
+	require.Equal(t, int64(0), snap0.Count)
+	require.Equal(t, 0*time.Second, snap0.Min)
+	require.Equal(t, 0*time.Second, snap0.Max)
+	require.Equal(t, 0*time.Second, snap0.Sum)
+	require.Equal(t, 0*time.Second, snap0.Mean())
+
+	cnt1.Observe(time.Second)
+	cnt2.Observe(time.Hour)
+
+	h1 := mustFindMetric(t, "kopia_some_dur_dist2_ms", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label1"})
+	h2 := mustFindMetric(t, "kopia_some_dur_dist2_ms", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label2"})
+	require.Equal(t, uint64(1), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 1000.0, h1.GetHistogram().GetSampleSum())
+	require.Equal(t, uint64(1), h2.GetHistogram().GetSampleCount())
+	require.Equal(t, 3600000.0, h2.GetHistogram().GetSampleSum())
+
+	cnt1.Observe(30 * time.Second)
+	cnt2.Observe(50 * time.Second)
+
+	h1 = mustFindMetric(t, "kopia_some_dur_dist2_ms", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label1"})
+	h2 = mustFindMetric(t, "kopia_some_dur_dist2_ms", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label2"})
+	require.Equal(t, uint64(2), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 31000.0, h1.GetHistogram().GetSampleSum())
+	require.Equal(t, uint64(2), h2.GetHistogram().GetSampleCount())
+	require.Equal(t, 3650000.0, h2.GetHistogram().GetSampleSum())
+
+	snap1 := cnt1.Snapshot()
+
+	require.Equal(t, int64(2), snap1.Count)
+	require.Equal(t, time.Second, snap1.Min)
+	require.Equal(t, 30*time.Second, snap1.Max)
+	require.Equal(t, 31*time.Second, snap1.Sum)
+	require.Equal(t, 15500*time.Millisecond, snap1.Mean())
+
+	snap2 := cnt2.Snapshot()
+
+	require.Equal(t, int64(2), snap2.Count)
+	require.Equal(t, 50*time.Second, snap2.Min)
+	require.Equal(t, time.Hour, snap2.Max)
+	require.Equal(t, time.Hour+50*time.Second, snap2.Sum)
+	require.Equal(t, 30*time.Minute+25*time.Second, snap2.Mean())
+}
+
+func TestSizeDistribution_WithLabels(t *testing.T) {
+	e := metrics.NewRegistry()
+	cnt1 := e.SizeDistribution("some_size_dist", "some-help", metrics.ISOBytesThresholds, map[string]string{"key1": "label1"})
+	cnt2 := e.SizeDistribution("some_size_dist", "some-help", metrics.ISOBytesThresholds, map[string]string{"key1": "label2"})
+
+	snap0 := cnt1.Snapshot()
+
+	require.Equal(t, int64(0), snap0.Count)
+	require.Equal(t, int64(0), snap0.Min)
+	require.Equal(t, int64(0), snap0.Max)
+	require.Equal(t, int64(0), snap0.Sum)
+	require.Equal(t, int64(0), snap0.Mean())
+
+	cnt1.Observe(1000)
+	cnt2.Observe(1e6)
+
+	h1 := mustFindMetric(t, "kopia_some_size_dist", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label1"})
+	h2 := mustFindMetric(t, "kopia_some_size_dist", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label2"})
+	require.Equal(t, uint64(1), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 1000.0, h1.GetHistogram().GetSampleSum())
+	require.Equal(t, uint64(1), h2.GetHistogram().GetSampleCount())
+	require.Equal(t, 1.0e6, h2.GetHistogram().GetSampleSum())
+
+	cnt1.Observe(300)
+	cnt2.Observe(50e6)
+
+	h1 = mustFindMetric(t, "kopia_some_size_dist", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label1"})
+	h2 = mustFindMetric(t, "kopia_some_size_dist", prommodel.MetricType_HISTOGRAM, map[string]string{"key1": "label2"})
+	require.Equal(t, uint64(2), h1.GetHistogram().GetSampleCount())
+	require.Equal(t, 1300.0, h1.GetHistogram().GetSampleSum())
+	require.Equal(t, uint64(2), h2.GetHistogram().GetSampleCount())
+	require.Equal(t, 51.0e6, h2.GetHistogram().GetSampleSum())
+
+	snap1 := cnt1.Snapshot()
+
+	require.Equal(t, int64(2), snap1.Count)
+	require.Equal(t, int64(300), snap1.Min)
+	require.Equal(t, int64(1000), snap1.Max)
+	require.Equal(t, int64(1300), snap1.Sum)
+	require.Equal(t, int64(650), snap1.Mean())
+
+	snap2 := cnt2.Snapshot()
+
+	require.Equal(t, int64(2), snap2.Count)
+	require.Equal(t, int64(1000000), snap2.Min)
+	require.Equal(t, int64(50000000), snap2.Max)
+	require.Equal(t, int64(51000000), snap2.Sum)
+	require.Equal(t, int64(25500000), snap2.Mean())
+}
@@ -0,0 +1,68 @@
+package metrics
+
+import (
+	"sync/atomic"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Gauge is a metric representing momentary state.
+type Gauge struct {
+	state atomic.Int64
+
+	prom prometheus.Gauge
+}
+
+// Set sets the value of a gauge.
+func (c *Gauge) Set(v int64) {
+	if c == nil {
+		return
+	}
+
+	if c.prom != nil {
+		c.prom.Set(float64(v))
+	}
+
+	c.state.Store(v)
+}
+
+// newState initializes counter state and returns previous state or 0.
+func (c *Gauge) newState() int64 {
+	return c.state.Swap(0)
+}
+
+// Snapshot captures a momentary state of the gauge.
+func (c *Gauge) Snapshot() int64 {
+	if c == nil {
+		return 0
+	}
+
+	return c.state.Load()
+}
+
+// Gauge gets a persistent gauge with the provided name.
+func (r *Registry) Gauge(name, help string, labels map[string]string) *Gauge {
+	if r == nil {
+		return nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	fullName := name + labelsSuffix(labels)
+
+	c := r.allGauges[fullName]
+	if c == nil {
+		c = &Gauge{
+			prom: getPrometheusGauge(prometheus.GaugeOpts{
+				Name: prometheusPrefix + name,
+				Help: help,
+			}, labels),
+		}
+		c.newState()
+
+		r.allGauges[fullName] = c
+	}
+
+	return c
+}
@@ -0,0 +1,66 @@
+package metrics_test
+
+import (
+	"testing"
+
+	prommodel "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/metrics"
+)
+
+func TestGauge_Nil(t *testing.T) {
+	var mr *metrics.Registry
+	g := mr.Gauge("aaa", "bbb", nil)
+	require.Nil(t, g)
+	g.Set(33)
+	require.Equal(t, int64(0), g.Snapshot())
+}
+
+func TestGauge_NoLabels(t *testing.T) {
+	mr := metrics.NewRegistry()
+	g := mr.Gauge("some_gauge", "some-help", nil)
+
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge", prommodel.MetricType_GAUGE, nil).
+			GetGauge().GetValue())
+	g.Set(33)
+	require.Equal(t, 33.0,
+		mustFindMetric(t, "kopia_some_gauge", prommodel.MetricType_GAUGE, nil).
+			GetGauge().GetValue())
+	g.Set(133)
+	require.Equal(t, 133.0,
+		mustFindMetric(t, "kopia_some_gauge", prommodel.MetricType_GAUGE, nil).
+			GetGauge().GetValue())
+
+	require.Equal(t, int64(133), g.Snapshot())
+}
+
+func TestGauge_WithLabels(t *testing.T) {
+	mr := metrics.NewRegistry()
+	g1 := mr.Gauge("some_gauge2", "some-help", map[string]string{"key1": "label1"})
+	g2 := mr.Gauge("some_gauge2", "some-help", map[string]string{"key1": "label2"})
+
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label1"}).
+			GetGauge().GetValue())
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label2"}).
+			GetGauge().GetValue())
+	g1.Set(33)
+	g2.Set(44)
+	require.Equal(t, 44.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label2"}).
+			GetGauge().GetValue())
+	require.Equal(t, 33.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label1"}).
+			GetGauge().GetValue())
+	g1.Set(133)
+	g2.Set(144)
+	require.Equal(t, 133.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label1"}).
+			GetGauge().GetValue())
+	require.Equal(t, 144.0,
+		mustFindMetric(t, "kopia_some_gauge2", prommodel.MetricType_GAUGE, map[string]string{"key1": "label2"}).
+			GetGauge().GetValue())
+}
@@ -0,0 +1,119 @@
+// Package metrics provides unified way of emitting metrics inside Kopia.
+package metrics
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/kopia/kopia/repo/logging"
+)
+
+var log = logging.Module("metrics")
+
+// Registry groups together all metrics stored in the repository and provides ways of accessing them.
+type Registry struct {
+	mu                       sync.Mutex
+	allCounters              map[string]*Counter
+	allGauges                map[string]*Gauge
+	allThroughput            map[string]*Throughput
+	allDurationDistributions map[string]*Distribution[time.Duration]
+	allSizeDistributions     map[string]*Distribution[int64]
+}
+
+// Snapshot captures the state of all metrics.
+type Snapshot struct {
+	Counters              map[string]int64
+	Gauges                map[string]int64
+	DurationDistributions map[string]DistributionState[time.Duration]
+	SizeDistributions     map[string]DistributionState[int64]
+}
+
+// Snapshot captures the snapshot of all metrics.
+func (r *Registry) Snapshot() Snapshot {
+	s := Snapshot{
+		Counters:              map[string]int64{},
+		Gauges:                map[string]int64{},
+		DurationDistributions: map[string]DistributionState[time.Duration]{},
+		SizeDistributions:     map[string]DistributionState[int64]{},
+	}
+
+	for k, c := range r.allCounters {
+		s.Counters[k] = c.Snapshot()
+	}
+
+	for k, c := range r.allGauges {
+		s.Gauges[k] = c.Snapshot()
+	}
+
+	for k, c := range r.allDurationDistributions {
+		s.DurationDistributions[k] = c.Snapshot()
+	}
+
+	for k, c := range r.allSizeDistributions {
+		s.SizeDistributions[k] = c.Snapshot()
+	}
+
+	return s
+}
+
+// Close closes the metrics registry.
+func (r *Registry) Close(ctx context.Context) error {
+	return nil
+}
+
+// Log logs all metrics in the registry.
+func (r *Registry) Log(ctx context.Context) error {
+	if r == nil {
+		return nil
+	}
+
+	s := r.Snapshot()
+
+	for n, val := range s.Counters {
+		log(ctx).Debugw("COUNTER", "name", n, "value", val)
+	}
+
+	for n, val := range s.Gauges {
+		log(ctx).Debugw("GAUGE", "name", n, "value", val)
+	}
+
+	for n, st := range s.DurationDistributions {
+		log(ctx).Debugw("DURATION-DISTRIBUTION", "name", n, "counters", st.BucketCounters, "cnt", st.Count, "sum", st.Sum, "min", st.Min, "avg", st.Mean(), "max", st.Max)
+	}
+
+	for n, st := range s.SizeDistributions {
+		if st.Count > 0 {
+			log(ctx).Debugw("SIZE-DISTRIBUTION", "name", n, "counters", st.BucketCounters, "cnt", st.Count, "sum", st.Sum, "min", st.Min, "avg", st.Mean(), "max", st.Max)
+		}
+	}
+
+	return nil
+}
+
+// NewRegistry returns new metrics registry.
+func NewRegistry() *Registry {
+	e := &Registry{
+		allCounters:              map[string]*Counter{},
+		allGauges:                map[string]*Gauge{},
+		allDurationDistributions: map[string]*Distribution[time.Duration]{},
+		allSizeDistributions:     map[string]*Distribution[int64]{},
+		allThroughput:            map[string]*Throughput{},
+	}
+
+	return e
+}
+
+func labelsSuffix(l map[string]string) string {
+	if len(l) == 0 {
+		return ""
+	}
+
+	var params []string
+	for k, v := range l {
+		params = append(params, k+":"+v)
+	}
+
+	return "[" + strings.Join(params, ";") + "]"
+}
@@ -0,0 +1,49 @@
+package metrics_test
+
+import (
+	"bytes"
+	"context"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/metrics"
+	"github.com/kopia/kopia/repo/logging"
+)
+
+func TestMetricEmitter_Nil(t *testing.T) {
+	var m *metrics.Registry
+
+	m.Log(context.Background())
+	m.Close(context.Background())
+}
+
+func TestMetricEmitter_NotNil(t *testing.T) {
+	var buf bytes.Buffer
+
+	ctx := logging.WithLogger(context.Background(), logging.ToWriter(&buf))
+
+	r := metrics.NewRegistry()
+	r.CounterInt64("c1", "h1", nil).Add(33)
+	r.Gauge("g1", "h1", nil).Set(44)
+	r.Throughput("t1", "h1", nil).Observe(44, time.Second)
+	r.DurationDistribution("d1", "h1", metrics.IOLatencyThresholds, nil).Observe(33 * time.Second)
+	r.SizeDistribution("s1", "h1", metrics.ISOBytesThresholds, nil).Observe(333)
+	require.NoError(t, r.Log(ctx))
+	require.NoError(t, r.Close(ctx))
+
+	lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
+	sort.Strings(lines)
+
+	require.Equal(t, []string{
+		"COUNTER\t{\"name\":\"c1\",\"value\":33}",
+		"COUNTER\t{\"name\":\"t1_bytes\",\"value\":44}",
+		"COUNTER\t{\"name\":\"t1_duration_nanos\",\"value\":1000000000}",
+		"DURATION-DISTRIBUTION\t{\"name\":\"d1\",\"counters\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0],\"cnt\":1,\"sum\":\"33s\",\"min\":\"33s\",\"avg\":\"33s\",\"max\":\"33s\"}",
+		"GAUGE\t{\"name\":\"g1\",\"value\":44}",
+		"SIZE-DISTRIBUTION\t{\"name\":\"s1\",\"counters\":[0,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],\"cnt\":1,\"sum\":333,\"min\":333,\"avg\":333,\"max\":333}",
+	}, lines)
+}
@@ -0,0 +1,129 @@
+package metrics
+
+import (
+	"time"
+
+	"golang.org/x/exp/constraints"
+)
+
+// Thresholds encapsulates a set of bucket thresholds used in Summary[T].
+type Thresholds[T constraints.Float | constraints.Integer] struct {
+	values           []T
+	promScale        float64
+	prometheusSuffix string
+}
+
+// ISOBytesThresholds is a set of thresholds for sizes.
+//
+//nolint:gochecknoglobals
+var ISOBytesThresholds = &Thresholds[int64]{
+	[]int64{
+		100,
+		250,
+		500,
+		1_000,
+		2_500,
+		5_000,
+		10_000,
+		25_000,
+		50_000,
+		100_000,
+		250_000,
+		500_000,
+		1_000_000,
+		2_500_000,
+		5_000_000,
+		10_000_000,
+		25_000_000,
+		50_000_000,
+		100_000_000,
+		250_000_000,
+		500_000_000,
+	},
+	1,
+	"",
+}
+
+// IOLatencyThresholds is a set of thresholds that can represent IO latencies from 500us to 50s.
+//
+//nolint:gochecknoglobals
+var IOLatencyThresholds = &Thresholds[time.Duration]{
+	[]time.Duration{
+		500 * time.Microsecond,
+
+		1000 * time.Microsecond,
+		2500 * time.Microsecond,
+		5000 * time.Microsecond,
+
+		10 * time.Millisecond,
+		25 * time.Millisecond,
+		50 * time.Millisecond,
+
+		100 * time.Millisecond,
+		250 * time.Millisecond,
+		500 * time.Millisecond,
+
+		1000 * time.Millisecond,
+		2500 * time.Millisecond,
+		5000 * time.Millisecond,
+
+		10 * time.Second,
+		25 * time.Second,
+		50 * time.Second,
+	},
+	1e6, // store as milliseconds
+	"_ms",
+}
+
+// CPULatencyThresholds is a set of thresholds that can represent CPU latencies from 10us to 5s.
+//
+//nolint:gochecknoglobals
+var CPULatencyThresholds = &Thresholds[time.Duration]{
+	[]time.Duration{
+		10 * time.Microsecond,
+		25 * time.Microsecond,
+		50 * time.Microsecond,
+
+		100 * time.Microsecond,
+		250 * time.Microsecond,
+		500 * time.Microsecond,
+
+		1000 * time.Microsecond,
+		2500 * time.Microsecond,
+		5000 * time.Microsecond,
+
+		10 * time.Millisecond,
+		25 * time.Millisecond,
+		50 * time.Millisecond,
+
+		100 * time.Millisecond,
+		250 * time.Millisecond,
+		500 * time.Millisecond,
+
+		1000 * time.Millisecond,
+		2500 * time.Millisecond,
+		5000 * time.Millisecond,
+	},
+	1, // export to Prometheus as nanoseconds
+	"_ns",
+}
+
+func bucketForThresholds[T constraints.Integer | constraints.Float](thresholds []T, d T) int {
+	l, r := 0, len(thresholds)-1
+
+	for l <= r {
+		m := (l + r) >> 1
+		mid := thresholds[m]
+
+		switch {
+		case mid < d:
+			l = m + 1
+		case mid > d:
+			r = m - 1
+		default:
+			return m
+		}
+	}
+
+	return l
+}
@@ -0,0 +1,42 @@
+package metrics
+
+import (
+	"time"
+)
+
+// Throughput measures throughput by keeping track of total value and total duration.
+type Throughput struct {
+	totalBytes    *Counter
+	totalDuration *Counter
+}
+
+// Observe increases duration and total value counters based on the amount of time to process particular amount of data.
+func (c *Throughput) Observe(size int64, dt time.Duration) {
+	if c == nil {
+		return
+	}
+
+	c.totalBytes.Add(size)
+	c.totalDuration.Add(dt.Nanoseconds())
+}
+
+// Throughput gets a persistent counter with the provided name.
+func (r *Registry) Throughput(name, help string, labels map[string]string) *Throughput {
+	if r == nil {
+		return nil
+	}
+
+	fullName := name + labelsSuffix(labels)
+
+	c := r.allThroughput[fullName]
+	if c == nil {
+		c = &Throughput{
+			totalBytes:    r.CounterInt64(name+"_bytes", help, labels),
+			totalDuration: r.CounterInt64(name+"_duration_nanos", help, labels),
+		}
+
+		r.allThroughput[fullName] = c
+	}
+
+	return c
+}
@@ -0,0 +1,41 @@
+package metrics_test
+
+import (
+	"testing"
+	"time"
+
+	prommodel "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/metrics"
+)
+
+func TestThroughput_Nil(t *testing.T) {
+	var r *metrics.Registry
+
+	th := r.Throughput("some_throughput", "some-help", nil)
+	th.Observe(1, time.Second)
+}
+
+func TestThroughput_NotNil(t *testing.T) {
+	r := metrics.NewRegistry()
+
+	th := r.Throughput("some_throughput2", "some-help", nil)
+
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_throughput2_bytes_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+	require.Equal(t, 0.0,
+		mustFindMetric(t, "kopia_some_throughput2_duration_nanos_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+
+	th.Observe(500, 500*time.Millisecond)
+	th.Observe(1, time.Second)
+
+	require.Equal(t, 501.0,
+		mustFindMetric(t, "kopia_some_throughput2_bytes_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+	require.Equal(t, 1.5e9,
+		mustFindMetric(t, "kopia_some_throughput2_duration_nanos_total", prommodel.MetricType_COUNTER, nil).
+			GetCounter().GetValue())
+}
@@ -0,0 +1,64 @@
+package metrics
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"golang.org/x/exp/maps"
+)
+
+const (
+	prometheusCounterSuffix = "_total"
+	prometheusPrefix        = "kopia_"
+)
+
+//nolint:gochecknoglobals
+var (
+	promCacheMutex sync.Mutex
+	promCounters   = map[string]*prometheus.CounterVec{}
+	promGauges     = map[string]*prometheus.GaugeVec{}
+	promHistograms = map[string]*prometheus.HistogramVec{}
+)
+
+func getPrometheusCounter(opts prometheus.CounterOpts, labels map[string]string) prometheus.Counter {
+	promCacheMutex.Lock()
+	defer promCacheMutex.Unlock()
+
+	prom := promCounters[opts.Name]
+	if prom == nil {
+		prom = promauto.NewCounterVec(opts, maps.Keys(labels))
+
+		promCounters[opts.Name] = prom
+	}
+
+	return prom.WithLabelValues(maps.Values(labels)...)
+}
+
+func getPrometheusGauge(opts prometheus.GaugeOpts, labels map[string]string) prometheus.Gauge {
+	promCacheMutex.Lock()
+	defer promCacheMutex.Unlock()
+
+	prom := promGauges[opts.Name]
+	if prom == nil {
+		prom = promauto.NewGaugeVec(opts, maps.Keys(labels))
+
+		promGauges[opts.Name] = prom
+	}
+
+	return prom.WithLabelValues(maps.Values(labels)...)
+}
+
+func getPrometheusHistogram(opts prometheus.HistogramOpts, labels map[string]string) prometheus.Observer {
+	promCacheMutex.Lock()
+	defer promCacheMutex.Unlock()
+
+	prom := promHistograms[opts.Name]
+	if prom == nil {
+		prom = promauto.NewHistogramVec(opts, maps.Keys(labels))
+
+		promHistograms[opts.Name] = prom
+	}
+
+	return prom.WithLabelValues(maps.Values(labels)...)
+}
@@ -140,7 +140,7 @@ func TestGPRServer_AuthenticationError(t *testing.T) {
 	if _, err := repo.OpenGRPCAPIRepository(ctx, apiServerInfo, repo.ClientOptions{
 		Username: "bad-username",
 		Hostname: "bad-hostname",
-	}, nil, "bad-password"); err == nil {
+	}, nil, "bad-password", nil); err == nil {
 		t.Fatal("unexpected success when connecting with invalid username")
 	}
 }
@@ -13,6 +13,7 @@ import (
 	"github.com/kopia/kopia/internal/cache"
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/internal/remoterepoapi"
 	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content"
@@ -39,6 +40,7 @@ type apiServerRepository struct {
 	cliOpts                          ClientOptions
 	omgr                             *object.Manager
 	wso                              WriteSessionOptions
+	metricsRegistry                  *metrics.Registry

 	isSharedReadOnlySession bool
 	contentCache            *cache.PersistentCache
@@ -155,7 +157,7 @@ func (r *apiServerRepository) NewWriter(ctx context.Context, opt WriteSessionOpt
 	w := &r2

 	// create object manager using a remote repo as contentManager implementation.
-	omgr, err := object.NewObjectManager(ctx, w, r.objectFormat)
+	omgr, err := object.NewObjectManager(ctx, w, r.objectFormat, r.metricsRegistry)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "error initializing object manager")
 	}
@@ -273,10 +275,15 @@ func (r *apiServerRepository) PrefetchContents(ctx context.Context, contentIDs [
 	return resp.ContentIDs
 }

+// Metrics provides access to the metrics registry.
+func (r *apiServerRepository) Metrics() *metrics.Registry {
+	return r.metricsRegistry
+}
+
 var _ Repository = (*apiServerRepository)(nil)

 // openRestAPIRepository connects remote repository over Kopia API.
-func openRestAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts ClientOptions, contentCache *cache.PersistentCache, password string) (Repository, error) {
+func openRestAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts ClientOptions, contentCache *cache.PersistentCache, password string, mr *metrics.Registry) (Repository, error) {
 	cli, err := apiclient.NewKopiaAPIClient(apiclient.Options{
 		BaseURL:                             si.BaseURL,
 		TrustedServerCertificateFingerprint: si.TrustedServerCertificateFingerprint,
@@ -296,6 +303,7 @@ func openRestAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts Clien
 			OnUpload: func(i int64) {},
 		},
 		isSharedReadOnlySession: true,
+		metricsRegistry:         mr,
 	}

 	var p remoterepoapi.Parameters
@@ -314,7 +322,7 @@ func openRestAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts Clien
 	rr.serverSupportsContentCompression = p.SupportsContentCompression

 	// create object manager using rr as contentManager implementation.
-	omgr, err := object.NewObjectManager(ctx, rr, rr.objectFormat)
+	omgr, err := object.NewObjectManager(ctx, rr, rr.objectFormat, mr)
 	if err != nil {
 		return nil, errors.Wrap(err, "error initializing object manager")
 	}
@@ -0,0 +1,228 @@
+// Package storagemetrics implements wrapper around Storage that adds metrics around all activity.
+package storagemetrics
+
+import (
+	"context"
+	"time"
+
+	"github.com/kopia/kopia/internal/metrics"
+	"github.com/kopia/kopia/internal/timetrack"
+	"github.com/kopia/kopia/repo/blob"
+)
+
+type blobMetrics struct {
+	base blob.Storage
+
+	downloadedBytesPartial *metrics.Counter
+	downloadedBytesFull    *metrics.Counter
+	uploadedBytes          *metrics.Counter
+	listBlobItems          *metrics.Counter
+
+	getBlobPartialDuration *metrics.Distribution[time.Duration]
+	getBlobFullDuration    *metrics.Distribution[time.Duration]
+	putBlobDuration        *metrics.Distribution[time.Duration]
+	getCapacityDuration    *metrics.Distribution[time.Duration]
+	getMetadataDuration    *metrics.Distribution[time.Duration]
+	deleteBlobDuration     *metrics.Distribution[time.Duration]
+	listBlobsDuration      *metrics.Distribution[time.Duration]
+	closeDuration          *metrics.Distribution[time.Duration]
+	flushCachesDuration    *metrics.Distribution[time.Duration]
+
+	getBlobErrors     *metrics.Counter
+	getCapacityErrors *metrics.Counter
+	getMetadataErrors *metrics.Counter
+	putBlobErrors     *metrics.Counter
+	deleteBlobErrors  *metrics.Counter
+	listBlobsErrors   *metrics.Counter
+	closeErrors       *metrics.Counter
+	flushCachesErrors *metrics.Counter
+}
+
+func (s *blobMetrics) GetBlob(ctx context.Context, id blob.ID, offset, length int64, output blob.OutputBuffer) error {
+	timer := timetrack.StartTimer()
+	err := s.base.GetBlob(ctx, id, offset, length, output)
+	dt := timer.Elapsed()
+
+	if length < 0 {
+		s.downloadedBytesFull.Add(int64(output.Length()))
+		s.getBlobFullDuration.Observe(dt)
+	} else {
+		s.downloadedBytesPartial.Add(int64(output.Length()))
+		s.getBlobPartialDuration.Observe(dt)
+	}
+
+	if err != nil {
+		s.getBlobErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+func (s *blobMetrics) GetCapacity(ctx context.Context) (blob.Capacity, error) {
+	timer := timetrack.StartTimer()
+	c, err := s.base.GetCapacity(ctx)
+	dt := timer.Elapsed()
+
+	s.getCapacityDuration.Observe(dt)
+
+	if err != nil {
+		s.getCapacityErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return c, err
+}
+
+func (s *blobMetrics) GetMetadata(ctx context.Context, id blob.ID) (blob.Metadata, error) {
+	timer := timetrack.StartTimer()
+	result, err := s.base.GetMetadata(ctx, id)
+	dt := timer.Elapsed()
+
+	s.getMetadataDuration.Observe(dt)
+
+	if err != nil {
+		s.getMetadataErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return result, err
+}
+
+func (s *blobMetrics) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes, opts blob.PutOptions) error {
+	timer := timetrack.StartTimer()
+	err := s.base.PutBlob(ctx, id, data, opts)
+	dt := timer.Elapsed()
+
+	s.putBlobDuration.Observe(dt)
+
+	if err != nil {
+		s.putBlobErrors.Add(1)
+	} else {
+		s.uploadedBytes.Add(int64(data.Length()))
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+func (s *blobMetrics) DeleteBlob(ctx context.Context, id blob.ID) error {
+	timer := timetrack.StartTimer()
+	err := s.base.DeleteBlob(ctx, id)
+	dt := timer.Elapsed()
+
+	s.deleteBlobDuration.Observe(dt)
+
+	if err != nil {
+		s.deleteBlobErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+func (s *blobMetrics) ListBlobs(ctx context.Context, prefix blob.ID, callback func(blob.Metadata) error) error {
+	timer := timetrack.StartTimer()
+	cnt := int64(0)
+	err := s.base.ListBlobs(ctx, prefix, func(bi blob.Metadata) error {
+		cnt++
+		return callback(bi)
+	})
+	dt := timer.Elapsed()
+
+	s.listBlobItems.Add(cnt)
+	s.listBlobsDuration.Observe(dt)
+
+	if err != nil {
+		s.listBlobsErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+func (s *blobMetrics) Close(ctx context.Context) error {
+	timer := timetrack.StartTimer()
+	err := s.base.Close(ctx)
+	dt := timer.Elapsed()
+
+	s.closeDuration.Observe(dt)
+
+	if err != nil {
+		s.closeErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+func (s *blobMetrics) ConnectionInfo() blob.ConnectionInfo {
+	return s.base.ConnectionInfo()
+}
+
+func (s *blobMetrics) DisplayName() string {
+	return s.base.DisplayName()
+}
+
+func (s *blobMetrics) FlushCaches(ctx context.Context) error {
+	timer := timetrack.StartTimer()
+	err := s.base.FlushCaches(ctx)
+	dt := timer.Elapsed()
+
+	s.flushCachesDuration.Observe(dt)
+
+	if err != nil {
+		s.flushCachesErrors.Add(1)
+	}
+
+	//nolint:wrapcheck
+	return err
+}
+
+// NewWrapper returns a Storage wrapper that logs all storage commands.
+func NewWrapper(wrapped blob.Storage, mr *metrics.Registry) blob.Storage {
+	durationSummaryForMethod := func(m string) *metrics.Distribution[time.Duration] {
+		return mr.DurationDistribution(
+			"blob_storage_latency",
+			"Latency of blob storage operation by method",
+			metrics.IOLatencyThresholds,
+			map[string]string{"method": m},
+		)
+	}
+
+	errorCounterForMethod := func(m string) *metrics.Counter {
+		return mr.CounterInt64(
+			"blob_errors",
+			"Number of storage operation errors by method",
+			map[string]string{"method": m},
+		)
+	}
+
+	return &blobMetrics{
+		base: wrapped,
+
+		downloadedBytesPartial: mr.CounterInt64("blob_download_partial_blob_bytes", "Number of bytes downloaded as partial blobs", nil),
+		downloadedBytesFull:    mr.CounterInt64("blob_download_full_blob_bytes", "Number of bytes downloaded as full blobs", nil),
+		uploadedBytes:          mr.CounterInt64("blob_upload_bytes", "Number of bytes uploaded", nil),
+		listBlobItems:          mr.CounterInt64("blob_list_items", "Number of list items returned", nil),
+
+		getBlobPartialDuration: durationSummaryForMethod("GetBlob-partial"),
+		getBlobFullDuration:    durationSummaryForMethod("GetBlob-full"),
+		getCapacityDuration:    durationSummaryForMethod("GetCapacity"),
+		getMetadataDuration:    durationSummaryForMethod("GetMetadata"),
+		putBlobDuration:        durationSummaryForMethod("PutBlob"),
+		deleteBlobDuration:     durationSummaryForMethod("DeleteBlob"),
+		listBlobsDuration:      durationSummaryForMethod("ListBlobs"),
+		closeDuration:          durationSummaryForMethod("Close"),
+		flushCachesDuration:    durationSummaryForMethod("FlushCaches"),
+
+		getBlobErrors:     errorCounterForMethod("GetBlob"),
+		getCapacityErrors: errorCounterForMethod("GetCapacity"),
+		getMetadataErrors: errorCounterForMethod("GetMetadata"),
+		putBlobErrors:     errorCounterForMethod("PutBlob"),
+		deleteBlobErrors:  errorCounterForMethod("DeleteBlob"),
+		listBlobsErrors:   errorCounterForMethod("ListBlobs"),
+		closeErrors:       errorCounterForMethod("Close"),
+		flushCachesErrors: errorCounterForMethod("FlushCaches"),
+	}
+}
@@ -0,0 +1,288 @@
+package storagemetrics_test
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/metrics"
+	"github.com/kopia/kopia/internal/testlogging"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/storagemetrics"
+)
+
+func TestStorageMetrics_PutBlob(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodPutBlob).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_upload_bytes", 0)
+	requireCounterValue(t, snap, "blob_errors[method:PutBlob]", 0)
+
+	require.ErrorIs(t, ms.PutBlob(ctx, "someBlob1", gather.FromSlice([]byte{1, 2, 3, 4}), blob.PutOptions{}), someError)
+	require.NoError(t, ms.PutBlob(ctx, "someBlob", gather.FromSlice([]byte{1, 2, 3}), blob.PutOptions{}))
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:PutBlob]", 1)
+	requireCounterValue(t, snap, "blob_upload_bytes", 3)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:PutBlob]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_GetBlob(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	require.NoError(t, st.PutBlob(ctx, "someBlob", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodGetBlob).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_download_full_blob_bytes", 0)
+	requireCounterValue(t, snap, "blob_download_partial_blob_bytes", 0)
+	requireCounterValue(t, snap, "blob_errors[method:GetBlob]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	require.ErrorIs(t, ms.GetBlob(ctx, "someBlob", 0, -1, &tmp), someError)
+	require.NoError(t, ms.GetBlob(ctx, "someBlob", 0, -1, &tmp))
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:GetBlob]", 1)
+	requireCounterValue(t, snap, "blob_download_full_blob_bytes", 5)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:GetBlob-full]"]
+	require.EqualValues(t, 2, d.Count)
+
+	require.NoError(t, ms.GetBlob(ctx, "someBlob", 2, 2, &tmp))
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:GetBlob]", 1)
+	requireCounterValue(t, snap, "blob_download_full_blob_bytes", 5)
+	require.Equal(t, 2, tmp.Length())
+
+	d = snap.DurationDistributions["blob_storage_latency[method:GetBlob-partial]"]
+	require.EqualValues(t, 1, d.Count)
+}
+
+func TestStorageMetrics_GetMetadata(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	require.NoError(t, st.PutBlob(ctx, "someBlob", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodGetMetadata).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:GetMetadata]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	_, err := ms.GetMetadata(ctx, "someBlob")
+	require.ErrorIs(t, err, someError)
+
+	_, err = ms.GetMetadata(ctx, "someBlob")
+	require.NoError(t, err)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:GetMetadata]", 1)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:GetMetadata]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_GetCapacity(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodGetCapacity).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:GetCapacity]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	_, err := ms.GetCapacity(ctx)
+	require.ErrorIs(t, err, someError)
+
+	_, err = ms.GetCapacity(ctx)
+	require.ErrorIs(t, err, blob.ErrNotAVolume)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:GetCapacity]", 2)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:GetCapacity]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_DeleteBlob(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	require.NoError(t, st.PutBlob(ctx, "someBlob", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodDeleteBlob).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:DeleteBlob]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	err := ms.DeleteBlob(ctx, "someBlob")
+	require.ErrorIs(t, err, someError)
+
+	err = ms.DeleteBlob(ctx, "someBlob")
+	require.NoError(t, err)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:DeleteBlob]", 1)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:DeleteBlob]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_Close(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodClose).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:Close]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	err := ms.Close(ctx)
+	require.ErrorIs(t, err, someError)
+
+	err = ms.Close(ctx)
+	require.NoError(t, err)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:Close]", 1)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:Close]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_FlushCaches(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodFlushCaches).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:FlushCaches]", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	err := ms.FlushCaches(ctx)
+	require.ErrorIs(t, err, someError)
+
+	err = ms.FlushCaches(ctx)
+	require.NoError(t, err)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:FlushCaches]", 1)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:FlushCaches]"]
+	require.EqualValues(t, 2, d.Count)
+}
+
+func TestStorageMetrics_ListBlobs(t *testing.T) {
+	ctx := testlogging.Context(t)
+	someError := errors.Errorf("foo")
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+	require.NoError(t, st.PutBlob(ctx, "someBlob1", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+	require.NoError(t, st.PutBlob(ctx, "someBlob2", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+	require.NoError(t, st.PutBlob(ctx, "someBlob3", gather.FromSlice([]byte{1, 2, 3, 4, 5}), blob.PutOptions{}))
+
+	fs := blobtesting.NewFaultyStorage(st)
+	fs.AddFault(blobtesting.MethodListBlobs).ErrorInstead(someError)
+
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(fs, mr)
+	snap := mr.Snapshot()
+
+	requireCounterValue(t, snap, "blob_errors[method:ListBlobs]", 0)
+	requireCounterValue(t, snap, "blob_list_items", 0)
+
+	var tmp gather.WriteBuffer
+	defer tmp.Close()
+
+	err := ms.ListBlobs(ctx, "", func(bm blob.Metadata) error { return nil })
+	require.ErrorIs(t, err, someError)
+
+	err = ms.ListBlobs(ctx, "", func(bm blob.Metadata) error { return nil })
+	require.NoError(t, err)
+
+	snap = mr.Snapshot()
+	requireCounterValue(t, snap, "blob_errors[method:ListBlobs]", 1)
+
+	d := snap.DurationDistributions["blob_storage_latency[method:ListBlobs]"]
+	require.EqualValues(t, 2, d.Count)
+	requireCounterValue(t, snap, "blob_list_items", 3)
+}
+
+func TestStorageMetrics_Misc(t *testing.T) {
+	st := blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil)
+	mr := metrics.NewRegistry()
+	ms := storagemetrics.NewWrapper(st, mr)
+	require.Equal(t, st.ConnectionInfo(), ms.ConnectionInfo())
+	require.Equal(t, st.DisplayName(), ms.DisplayName())
+}
+
+func requireCounterValue(t *testing.T, snap metrics.Snapshot, key string, want int64) {
+	t.Helper()
+
+	v, ok := snap.Counters[key]
+	require.True(t, ok)
+	require.EqualValues(t, want, v)
+}
@@ -1,6 +1,9 @@
 package content

-import "time"
+import (
+	"path/filepath"
+	"time"
+)

 // DurationSeconds represents the duration in seconds.
 type DurationSeconds float64
@@ -36,3 +39,16 @@ func (c *CachingOptions) CloneOrDefault() *CachingOptions {

 	return &c2
 }
+
+// CacheSubdirOrEmpty returns path to a cache subdirectory or empty string if cache is disabled.
+func (c *CachingOptions) CacheSubdirOrEmpty(subdir string) string {
+	if c == nil {
+		return ""
+	}
+
+	if c.CacheDirectory == "" {
+		return ""
+	}
+
+	return filepath.Join(c.CacheDirectory, subdir)
+}
@@ -16,7 +16,9 @@ import (
 	"github.com/kopia/kopia/internal/epoch"
 	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/listcache"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/internal/ownwrites"
+	"github.com/kopia/kopia/internal/timetrack"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/blob/filesystem"
 	"github.com/kopia/kopia/repo/blob/sharded"
@@ -113,6 +115,8 @@ type SharedManager struct {
 	contextLogger      logging.Logger
 	internalLogManager *internalLogManager
 	internalLogger     *zap.SugaredLogger // backing logger for 'sharedBaseLogger'
+
+	metricsStruct
 }

 func (sm *SharedManager) readPackFileLocalIndex(ctx context.Context, packFile blob.ID, packFileLength int64, output *gather.WriteBuffer) error {
@@ -284,19 +288,26 @@ func (sm *SharedManager) decryptContentAndVerify(payload gather.Bytes, bi Info,
 		return errors.Errorf("unsupported compressor %x", h)
 	}

+	t0 := timetrack.StartTimer()
+
 	if err := c.Decompress(output, tmp.Bytes().Reader(), true); err != nil {
 		return errors.Wrap(err, "error decompressing")
 	}

+	sm.decompressedBytes.Observe(int64(tmp.Length()), t0.Elapsed())
+
 	return nil
 }

 func (sm *SharedManager) decryptAndVerify(encrypted gather.Bytes, iv []byte, output *gather.WriteBuffer) error {
+	t0 := timetrack.StartTimer()
+
 	if err := sm.format.Encryptor().Decrypt(encrypted, iv, output); err != nil {
 		sm.Stats.foundInvalidContent()
 		return errors.Wrap(err, "decrypt")
 	}

+	sm.decryptedBytes.Observe(int64(encrypted.Length()), t0.Elapsed())
 	sm.Stats.foundValidContent()
 	sm.Stats.decrypted(output.Length())

@@ -384,7 +395,7 @@ func (sm *SharedManager) namedLogger(n string) logging.Logger {
 	return sm.contextLogger
 }

-func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *CachingOptions) error {
+func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *CachingOptions, mr *metrics.Registry) error {
 	dataCache, err := cache.NewContentCache(ctx, sm.st, cache.Options{
 		BaseCacheDirectory: caching.CacheDirectory,
 		CacheSubDir:        "contents",
@@ -393,7 +404,7 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca
 			MaxSizeBytes: caching.MaxCacheSizeBytes,
 			MinSweepAge:  caching.MinContentSweepAge.DurationOrDefault(DefaultDataCacheSweepAge),
 		},
-	})
+	}, mr)
 	if err != nil {
 		return errors.Wrap(err, "unable to initialize content cache")
 	}
@@ -412,7 +423,7 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca
 			MaxSizeBytes: metadataCacheSize,
 			MinSweepAge:  caching.MinMetadataSweepAge.DurationOrDefault(DefaultMetadataCacheSweepAge),
 		},
-	})
+	}, mr)
 	if err != nil {
 		return errors.Wrap(err, "unable to initialize metadata cache")
 	}
@@ -422,10 +433,10 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca
 		return errors.Wrap(err, "unable to initialize index blob cache storage")
 	}

-	indexBlobCache, err := cache.NewPersistentCache(ctx, "index blob cache", indexBlobStorage, cache.ChecksumProtection(caching.HMACSecret), cache.SweepSettings{
+	indexBlobCache, err := cache.NewPersistentCache(ctx, "index-blobs", indexBlobStorage, cache.ChecksumProtection(caching.HMACSecret), cache.SweepSettings{
 		MaxSizeBytes: metadataCacheSize,
 		MinSweepAge:  caching.MinMetadataSweepAge.DurationOrDefault(DefaultMetadataCacheSweepAge),
-	})
+	}, mr)
 	if err != nil {
 		return errors.Wrap(err, "unable to create index blob cache")
 	}
@@ -552,7 +563,15 @@ func (sm *SharedManager) release(ctx context.Context) error {

 	sm.indexBlobManagerV1.epochMgr.Flush()

-	return errors.Wrap(sm.st.Close(ctx), "error closing storage")
+	if err := sm.st.Close(ctx); err != nil {
+		return errors.Wrap(err, "error closing storage")
+	}
+
+	if err := sm.metricsEmitter.Close(ctx); err != nil {
+		return errors.Wrap(err, "error closing metrics")
+	}
+
+	return nil
 }

 // AlsoLogToContentLog wraps the provided content so that all logs are also sent to
@@ -571,7 +590,7 @@ func (sm *SharedManager) shouldRefreshIndexes() bool {
 }

 // NewSharedManager returns SharedManager that is used by SessionWriteManagers on top of a repository.
-func NewSharedManager(ctx context.Context, st blob.Storage, prov format.Provider, caching *CachingOptions, opts *ManagerOptions) (*SharedManager, error) {
+func NewSharedManager(ctx context.Context, st blob.Storage, prov format.Provider, caching *CachingOptions, opts *ManagerOptions, mr *metrics.Registry) (*SharedManager, error) {
 	opts = opts.CloneOrDefault()
 	if opts.TimeNow == nil {
 		opts.TimeNow = clock.Now
@@ -601,6 +620,8 @@ func NewSharedManager(ctx context.Context, st blob.Storage, prov format.Provider
 		internalLogManager:      ilm,
 		internalLogger:          internalLog,
 		contextLogger:           logging.Module(FormatLogModule)(ctx),
+
+		metricsStruct: initMetricsStruct(mr),
 	}

 	// remember logger defined for the context.
@@ -608,7 +629,7 @@ func NewSharedManager(ctx context.Context, st blob.Storage, prov format.Provider

 	caching = caching.CloneOrDefault()

-	if err := sm.setupReadManagerCaches(ctx, caching); err != nil {
+	if err := sm.setupReadManagerCaches(ctx, caching, mr); err != nil {
 		return nil, errors.Wrap(err, "error setting up read manager caches")
 	}

@@ -16,6 +16,7 @@ import (
 	"github.com/kopia/kopia/internal/cache"
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/timetrack"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content/index"
@@ -787,6 +788,11 @@ func (bm *WriteManager) SupportsContentCompression() (bool, error) {
 // WriteContent saves a given content of data to a pack group with a provided name and returns a contentID
 // that's based on the contents of data written.
 func (bm *WriteManager) WriteContent(ctx context.Context, data gather.Bytes, prefix index.IDPrefix, comp compression.HeaderID) (ID, error) {
+	t0 := timetrack.StartTimer()
+	defer func() {
+		bm.writeContentBytes.Observe(int64(data.Length()), t0.Elapsed())
+	}()
+
 	mp, mperr := bm.format.GetMutableParameters()
 	if mperr != nil {
 		return EmptyID, errors.Wrap(mperr, "mutable parameters")
@@ -796,8 +802,6 @@ func (bm *WriteManager) WriteContent(ctx context.Context, data gather.Bytes, pre
 		return EmptyID, err
 	}

-	reportContentWriteBytes(int64(data.Length()))
-
 	if err := prefix.ValidateSingle(); err != nil {
 		return EmptyID, errors.Wrap(err, "invalid prefix")
 	}
@@ -824,6 +828,9 @@ func (bm *WriteManager) WriteContent(ctx context.Context, data gather.Bytes, pre
 	// content already tracked
 	if err == nil {
 		if !bi.GetDeleted() {
+			bm.deduplicatedContents.Add(1)
+			bm.deduplicatedBytes.Add(int64(data.Length()))
+
 			return contentID, nil
 		}

@@ -840,14 +847,16 @@ func (bm *WriteManager) WriteContent(ctx context.Context, data gather.Bytes, pre

 // GetContent gets the contents of a given content. If the content is not found returns ErrContentNotFound.
 func (bm *WriteManager) GetContent(ctx context.Context, contentID ID) (v []byte, err error) {
+	t0 := timetrack.StartTimer()
+
 	defer func() {
 		switch {
 		case err == nil:
-			reportContentGetBytes(int64(len(v)))
+			bm.getContentBytes.Observe(int64(len(v)), t0.Elapsed())
 		case errors.Is(err, ErrContentNotFound):
-			reportContentGetNotFound()
+			bm.getContentNotFoundCount.Add(1)
 		default:
-			reportContentGetError()
+			bm.getContentErrorCount.Add(1)
 		}
 	}()

@@ -970,7 +979,7 @@ func NewManagerForTesting(ctx context.Context, st blob.Storage, f format.Provide
 		options.TimeNow = clock.Now
 	}

-	sharedManager, err := NewSharedManager(ctx, st, f, caching, options)
+	sharedManager, err := NewSharedManager(ctx, st, f, caching, options, nil)
 	if err != nil {
 		return nil, errors.Wrap(err, "error initializing read manager")
 	}
@@ -1001,7 +1010,10 @@ func NewWriteManager(ctx context.Context, sm *SharedManager, options SessionOpti
 		packIndexBuilder:      make(index.Builder),
 		sessionUser:           options.SessionUser,
 		sessionHost:           options.SessionHost,
-		onUpload:              options.OnUpload,
+		onUpload: func(numBytes int64) {
+			options.OnUpload(numBytes)
+			sm.uploadedBytes.Add(numBytes)
+		},

 		log: sm.namedLogger(writeManagerID),
 	}
@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/otel/trace"

 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/timetrack"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content/index"
@@ -50,22 +51,36 @@ func (sm *SharedManager) maybeCompressAndEncryptDataForPacking(data gather.Bytes
 			return NoCompression, errors.Errorf("unsupported compressor %x", comp)
 		}

+		t0 := timetrack.StartTimer()
+
 		if err := c.Compress(&tmp, data.Reader()); err != nil {
 			return NoCompression, errors.Wrap(err, "compression error")
 		}

+		sm.compressionAttemptedBytes.Observe(int64(data.Length()), t0.Elapsed())
+
 		if cd := tmp.Length(); cd >= data.Length() {
 			// data was not compressible enough.
 			comp = NoCompression
+
+			sm.nonCompressibleBytes.Add(int64(data.Length()))
 		} else {
+			sm.compressionSavings.Add(int64(data.Length()) - int64(cd))
+			sm.compressibleBytes.Add(int64(data.Length()))
 			data = tmp.Bytes()
 		}
 	}

+	sm.afterCompressionBytes.Add(int64(data.Length()))
+
+	t1 := timetrack.StartTimer()
+
 	if err := sm.format.Encryptor().Encrypt(data, iv, output); err != nil {
 		return NoCompression, errors.Wrap(err, "unable to encrypt")
 	}

+	sm.encryptedBytes.Observe(int64(output.Length()), t1.Elapsed())
+
 	sm.Stats.encrypted(data.Length())

 	return comp, nil
@@ -181,8 +196,11 @@ func (sm *SharedManager) writePackFileNotLocked(ctx context.Context, packFile bl

 func (sm *SharedManager) hashData(output []byte, data gather.Bytes) []byte {
 	// Hash the content and compute encryption key.
+	t0 := timetrack.StartTimer()
 	contentID := sm.format.HashFunc()(output, data)
 	sm.Stats.hashedContent(data.Length())

+	sm.hashedBytes.Observe(int64(data.Length()), t0.Elapsed())
+
 	return contentID
 }
@@ -1,54 +1,62 @@
 package content

 import (
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/kopia/kopia/internal/metrics"
 )

-// content cache metrics.
-//
-//nolint:gochecknoglobals,promlinter
-var (
-	metricContentGetCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_get_count",
-		Help: "Number of time GetContent() was called",
-	})
-	metricContentGetNotFoundCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_get_not_found_count",
-		Help: "Number of time GetContent() was called and the result was not found",
-	})
-	metricContentGetErrorCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_get_error_count",
-		Help: "Number of time GetContent() was called and the result was an error",
-	})
-	metricContentGetBytes = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_get_bytes",
-		Help: "Number of bytes retrieved using GetContent",
-	})
-	metricContentWriteContentCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_write_count",
-		Help: "Number of time WriteContent() was called",
-	})
-	metricContentWriteContentBytes = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "kopia_content_write_bytes",
-		Help: "Number of bytes passed to WriteContent()",
-	})
-)
+type metricsStruct struct {
+	metricsEmitter *metrics.Registry

-func reportContentWriteBytes(length int64) {
-	metricContentWriteContentCount.Inc()
-	metricContentWriteContentBytes.Add(float64(length))
+	uploadedBytes *metrics.Counter
+
+	getContentBytes         *metrics.Throughput
+	getContentErrorCount    *metrics.Counter
+	getContentNotFoundCount *metrics.Counter
+
+	// number of bytes written
+	writeContentBytes *metrics.Throughput
+
+	// number of bytes after compression was applied (if any).
+	afterCompressionBytes *metrics.Counter
+
+	nonCompressibleBytes *metrics.Counter
+
+	compressibleBytes  *metrics.Counter
+	compressionSavings *metrics.Counter
+
+	hashedBytes               *metrics.Throughput
+	encryptedBytes            *metrics.Throughput
+	decryptedBytes            *metrics.Throughput
+	compressionAttemptedBytes *metrics.Throughput
+	decompressedBytes         *metrics.Throughput
+
+	deduplicatedBytes    *metrics.Counter
+	deduplicatedContents *metrics.Counter
 }

-func reportContentGetBytes(length int64) {
-	metricContentGetCount.Inc()
-	metricContentGetBytes.Add(float64(length))
-}
+func initMetricsStruct(mr *metrics.Registry) metricsStruct {
+	return metricsStruct{
+		metricsEmitter: mr,

-func reportContentGetError() {
-	metricContentGetErrorCount.Inc()
-}
+		uploadedBytes:           mr.CounterInt64("content_uploaded_bytes", "Number of bytes uploaded from content manager.", nil),
+		getContentErrorCount:    mr.CounterInt64("content_get_error_count", "Number of time GetContent() was called and the result was an error", nil),
+		getContentNotFoundCount: mr.CounterInt64("content_get_not_found_count", "Number of time GetContent() was called and the result was not found", nil),
+		deduplicatedContents:    mr.CounterInt64("content_deduplicated", "Number of contents deduplicated.", nil),
+		deduplicatedBytes:       mr.CounterInt64("content_deduplicated_bytes", "Number of bytes deduplicated.", nil),

-func reportContentGetNotFound() {
-	metricContentGetNotFoundCount.Inc()
+		writeContentBytes: mr.Throughput("content_write", "WriteContent throughput (before deduplication)", nil),
+		hashedBytes:       mr.Throughput("content_hashed", "Hashing throughput.", nil),
+
+		afterCompressionBytes: mr.CounterInt64("content_after_compression_bytes", "Number of bytes after deduplication and compression", nil),
+		nonCompressibleBytes:  mr.CounterInt64("content_non_compressible_bytes", "Number of bytes that were found to be non-compressible stage.", nil),
+		compressibleBytes:     mr.CounterInt64("content_compressible_bytes", "Number of bytes that were found to be compressible.", nil),
+		compressionSavings:    mr.CounterInt64("content_compression_savings_bytes", "Number of bytes that were saved due to compression.", nil),
+
+		encryptedBytes:            mr.Throughput("content_encrypted", "Encryption throughput.", nil),
+		compressionAttemptedBytes: mr.Throughput("content_compression_attempted", "Compression throughput.", nil),
+
+		getContentBytes:   mr.Throughput("content_read", "Number of bytes read", nil),
+		decryptedBytes:    mr.Throughput("content_decrypted", "Decryption throughput.", nil),
+		decompressedBytes: mr.Throughput("content_decompressed", "Decompression throughput.", nil),
+	}
 }
@@ -22,6 +22,7 @@ import (
 	"github.com/kopia/kopia/internal/ctxutil"
 	"github.com/kopia/kopia/internal/gather"
 	apipb "github.com/kopia/kopia/internal/grpcapi"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/internal/retry"
 	"github.com/kopia/kopia/internal/tlsutil"
 	"github.com/kopia/kopia/repo/blob"
@@ -85,6 +86,8 @@ type grpcRepositoryClient struct {
 	contentCache *cache.PersistentCache

 	recent recentlyRead
+
+	metricsRegistry *metrics.Registry
 }

 type grpcInnerSession struct {
@@ -458,7 +461,7 @@ func (r *grpcInnerSession) Flush(ctx context.Context) error {
 }

 func (r *grpcRepositoryClient) NewWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, RepositoryWriter, error) {
-	w, err := newGRPCAPIRepositoryForConnection(ctx, r.conn, r.connRefCount, r.cliOpts, opt, r.contentCache, false)
+	w, err := newGRPCAPIRepositoryForConnection(ctx, r.conn, r.connRefCount, r.cliOpts, opt, r.contentCache, false, r.metricsRegistry)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -770,7 +773,7 @@ func (c grpcCreds) RequireTransportSecurity() bool {

 // OpenGRPCAPIRepository opens the Repository based on remote GRPC server.
 // The APIServerInfo must have the address of the repository as 'https://host:port'
-func OpenGRPCAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts ClientOptions, contentCache *cache.PersistentCache, password string) (Repository, error) {
+func OpenGRPCAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts ClientOptions, contentCache *cache.PersistentCache, password string, mr *metrics.Registry) (Repository, error) {
 	var transportCreds credentials.TransportCredentials

 	if si.TrustedServerCertificateFingerprint != "" {
@@ -801,7 +804,7 @@ func OpenGRPCAPIRepository(ctx context.Context, si *APIServerInfo, cliOpts Clien
 		return nil, errors.Wrap(err, "dial error")
 	}

-	rep, err := newGRPCAPIRepositoryForConnection(ctx, conn, new(int32), cliOpts, WriteSessionOptions{}, contentCache, true)
+	rep, err := newGRPCAPIRepositoryForConnection(ctx, conn, new(int32), cliOpts, WriteSessionOptions{}, contentCache, true, mr)
 	if err != nil {
 		return nil, err
 	}
@@ -868,8 +871,22 @@ func (r *grpcRepositoryClient) killInnerSession() {
 	}
 }

+// Metrics provides access to the metrics registry.
+func (r *grpcRepositoryClient) Metrics() *metrics.Registry {
+	return r.metricsRegistry
+}
+
 // newGRPCAPIRepositoryForConnection opens GRPC-based repository connection.
-func newGRPCAPIRepositoryForConnection(ctx context.Context, conn *grpc.ClientConn, connRefCount *int32, cliOpts ClientOptions, opt WriteSessionOptions, contentCache *cache.PersistentCache, transparentRetries bool) (*grpcRepositoryClient, error) {
+func newGRPCAPIRepositoryForConnection(
+	ctx context.Context,
+	conn *grpc.ClientConn,
+	connRefCount *int32,
+	cliOpts ClientOptions,
+	opt WriteSessionOptions,
+	contentCache *cache.PersistentCache,
+	transparentRetries bool,
+	mr *metrics.Registry,
+) (*grpcRepositoryClient, error) {
 	if opt.OnUpload == nil {
 		opt.OnUpload = func(i int64) {}
 	}
@@ -883,6 +900,7 @@ func newGRPCAPIRepositoryForConnection(ctx context.Context, conn *grpc.ClientCon
 		isReadOnly:         cliOpts.ReadOnly,
 		contentCache:       contentCache,
 		asyncWritesWG:      new(errgroup.Group),
+		metricsRegistry:    mr,
 	}

 	return inSessionWithoutRetry(ctx, rr, func(ctx context.Context, sess *grpcInnerSession) (*grpcRepositoryClient, error) {
@@ -900,7 +918,7 @@ func newGRPCAPIRepositoryForConnection(ctx context.Context, conn *grpc.ClientCon

 		rr.serverSupportsContentCompression = p.SupportsContentCompression

-		rr.omgr, err = object.NewObjectManager(ctx, rr, rr.objectFormat)
+		rr.omgr, err = object.NewObjectManager(ctx, rr, rr.objectFormat, rr.metricsRegistry)
 		if err != nil {
 			return nil, errors.Wrap(err, "unable to initialize object manager")
 		}
@@ -14,6 +14,7 @@ import (

 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/logging"
@@ -266,7 +267,7 @@ type ManagerOptions struct {
 }

 // NewManager returns new manifest manager for the provided content manager.
-func NewManager(ctx context.Context, b contentManager, options ManagerOptions) (*Manager, error) {
+func NewManager(ctx context.Context, b contentManager, options ManagerOptions, mr *metrics.Registry) (*Manager, error) {
 	timeNow := options.TimeNow
 	if timeNow == nil {
 		timeNow = clock.Now
@@ -169,7 +169,7 @@ func TestManifestInitCorruptedBlock(t *testing.T) {

 	t.Cleanup(func() { bm0.Close(ctx) })

-	mgr, err := NewManager(ctx, bm, ManagerOptions{})
+	mgr, err := NewManager(ctx, bm, ManagerOptions{}, nil)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -195,7 +195,7 @@ func TestManifestInitCorruptedBlock(t *testing.T) {

 	t.Cleanup(func() { bm.Close(ctx) })

-	mgr, err = NewManager(ctx, bm, ManagerOptions{})
+	mgr, err = NewManager(ctx, bm, ManagerOptions{}, nil)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -327,7 +327,7 @@ func newManagerForTesting(ctx context.Context, t *testing.T, data blobtesting.Da

 	t.Cleanup(func() { bm.Close(ctx) })

-	mm, err := NewManager(ctx, bm, ManagerOptions{})
+	mm, err := NewManager(ctx, bm, ManagerOptions{}, nil)
 	require.NoError(t, err)

 	return mm
@@ -9,6 +9,7 @@ import (
 	"github.com/pkg/errors"

 	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/repo/compression"
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/format"
@@ -197,7 +198,7 @@ func PrefetchBackingContents(ctx context.Context, contentMgr contentManager, obj
 }

 // NewObjectManager creates an ObjectManager with the specified content manager and format.
-func NewObjectManager(ctx context.Context, bm contentManager, f format.ObjectFormat) (*Manager, error) {
+func NewObjectManager(ctx context.Context, bm contentManager, f format.ObjectFormat, mr *metrics.Registry) (*Manager, error) {
 	om := &Manager{
 		contentMgr: bm,
 		Format:     f,
@@ -112,7 +112,7 @@ func setupTest(t *testing.T, compressionHeaderID map[content.ID]compression.Head

 	r, err := NewObjectManager(testlogging.Context(t), fcm, format.ObjectFormat{
 		Splitter: "FIXED-1M",
-	})
+	}, nil)
 	if err != nil {
 		t.Fatalf("can't create object manager: %v", err)
 	}
@@ -283,7 +283,7 @@ func TestObjectWriterRaceBetweenCheckpointAndResult(t *testing.T) {

 	om, err := NewObjectManager(testlogging.Context(t), fcm, format.ObjectFormat{
 		Splitter: "FIXED-1M",
-	})
+	}, nil)
 	if err != nil {
 		t.Fatalf("can't create object manager: %v", err)
 	}
@@ -15,11 +15,13 @@ import (
 	"github.com/kopia/kopia/internal/cache"
 	"github.com/kopia/kopia/internal/epoch"
 	"github.com/kopia/kopia/internal/feature"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/internal/retry"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/blob/beforeop"
 	loggingwrapper "github.com/kopia/kopia/repo/blob/logging"
 	"github.com/kopia/kopia/repo/blob/readonly"
+	"github.com/kopia/kopia/repo/blob/storagemetrics"
 	"github.com/kopia/kopia/repo/blob/throttling"
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/format"
@@ -123,7 +125,7 @@ func Open(ctx context.Context, configFile, password string, options *Options) (r
 	return openDirect(ctx, configFile, lc, password, options)
 }

-func getContentCacheOrNil(ctx context.Context, opt *content.CachingOptions, password string) (*cache.PersistentCache, error) {
+func getContentCacheOrNil(ctx context.Context, opt *content.CachingOptions, password string, mr *metrics.Registry) (*cache.PersistentCache, error) {
 	opt = opt.CloneOrDefault()

 	cs, err := cache.NewStorageOrNil(ctx, opt.CacheDirectory, opt.MaxCacheSizeBytes, "server-contents")
@@ -149,7 +151,7 @@ func getContentCacheOrNil(ctx context.Context, opt *content.CachingOptions, pass
 	pc, err := cache.NewPersistentCache(ctx, "cache-storage", cs, prot, cache.SweepSettings{
 		MaxSizeBytes: opt.MaxCacheSizeBytes,
 		MinSweepAge:  opt.MinContentSweepAge.DurationOrDefault(content.DefaultDataCacheSweepAge),
-	})
+	}, mr)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to open persistent cache")
 	}
@@ -159,16 +161,18 @@ func getContentCacheOrNil(ctx context.Context, opt *content.CachingOptions, pass

 // OpenAPIServer connects remote repository over Kopia API.
 func OpenAPIServer(ctx context.Context, si *APIServerInfo, cliOpts ClientOptions, cachingOptions *content.CachingOptions, password string) (Repository, error) {
-	contentCache, err := getContentCacheOrNil(ctx, cachingOptions, password)
+	me := metrics.NewRegistry()
+
+	contentCache, err := getContentCacheOrNil(ctx, cachingOptions, password, me)
 	if err != nil {
 		return nil, errors.Wrap(err, "error opening content cache")
 	}

 	if si.DisableGRPC {
-		return openRestAPIRepository(ctx, si, cliOpts, contentCache, password)
+		return openRestAPIRepository(ctx, si, cliOpts, contentCache, password, me)
 	}

-	return OpenGRPCAPIRepository(ctx, si, cliOpts, contentCache, password)
+	return OpenGRPCAPIRepository(ctx, si, cliOpts, contentCache, password, me)
 }

 // openDirect opens the repository that directly manipulates blob storage..
@@ -211,6 +215,9 @@ func openWithConfig(ctx context.Context, st blob.Storage, cliOpts ClientOptions,
 		DisableInternalLog: options.DisableInternalLog,
 	}

+	me := metrics.NewRegistry()
+	st = storagemetrics.NewWrapper(st, me)
+
 	fmgr, ferr := format.NewManager(ctx, st, cacheOpts.CacheDirectory, cliOpts.FormatBlobCacheDuration, password, cmOpts.TimeNow)
 	if ferr != nil {
 		return nil, errors.Wrap(ferr, "unable to create format manager")
@@ -279,7 +286,7 @@ func openWithConfig(ctx context.Context, st blob.Storage, cliOpts ClientOptions,
 	// background/interleaving upgrade lock storage monitor
 	st = upgradeLockMonitor(fmgr, options.UpgradeOwnerID, st, cmOpts.TimeNow, options.OnFatalError, options.TestOnlyIgnoreMissingRequiredFeatures)

-	scm, ferr := content.NewSharedManager(ctx, st, fmgr, cacheOpts, cmOpts)
+	scm, ferr := content.NewSharedManager(ctx, st, fmgr, cacheOpts, cmOpts, me)
 	if ferr != nil {
 		return nil, errors.Wrap(ferr, "unable to create shared content manager")
 	}
@@ -289,12 +296,12 @@ func openWithConfig(ctx context.Context, st blob.Storage, cliOpts ClientOptions,
 		SessionHost: cliOpts.Hostname,
 	}, "")

-	om, ferr := object.NewObjectManager(ctx, cm, fmgr.ObjectFormat())
+	om, ferr := object.NewObjectManager(ctx, cm, fmgr.ObjectFormat(), me)
 	if ferr != nil {
 		return nil, errors.Wrap(ferr, "unable to open object manager")
 	}

-	manifests, ferr := manifest.NewManager(ctx, cm, manifest.ManagerOptions{TimeNow: cmOpts.TimeNow})
+	manifests, ferr := manifest.NewManager(ctx, cm, manifest.ManagerOptions{TimeNow: cmOpts.TimeNow}, me)
 	if ferr != nil {
 		return nil, errors.Wrap(ferr, "unable to open manifests")
 	}
@@ -306,13 +313,14 @@ func openWithConfig(ctx context.Context, st blob.Storage, cliOpts ClientOptions,
 		mmgr:  manifests,
 		sm:    scm,
 		directRepositoryParameters: directRepositoryParameters{
-			cachingOptions: *cacheOpts,
-			fmgr:           fmgr,
-			timeNow:        cmOpts.TimeNow,
-			cliOpts:        cliOpts,
-			configFile:     configFile,
-			nextWriterID:   new(int32),
-			throttler:      throttler,
+			cachingOptions:  *cacheOpts,
+			fmgr:            fmgr,
+			timeNow:         cmOpts.TimeNow,
+			cliOpts:         cliOpts,
+			configFile:      configFile,
+			nextWriterID:    new(int32),
+			throttler:       throttler,
+			metricsRegistry: me,
 		},
 		closed: make(chan struct{}),
 	}
@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/otel"

 	"github.com/kopia/kopia/internal/clock"
+	"github.com/kopia/kopia/internal/metrics"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/blob/throttling"
 	"github.com/kopia/kopia/repo/content"
@@ -37,6 +38,7 @@ type Repository interface {
 	UpdateDescription(d string)
 	Refresh(ctx context.Context) error
 	Close(ctx context.Context) error
+	Metrics() *metrics.Registry
 }

 // RepositoryWriter provides methods to write to a repository.
@@ -87,13 +89,14 @@ type DirectRepositoryWriter interface {
 }

 type directRepositoryParameters struct {
-	configFile     string
-	cachingOptions content.CachingOptions
-	cliOpts        ClientOptions
-	timeNow        func() time.Time
-	fmgr           *format.Manager
-	nextWriterID   *int32
-	throttler      throttling.SettableThrottler
+	configFile      string
+	cachingOptions  content.CachingOptions
+	cliOpts         ClientOptions
+	timeNow         func() time.Time
+	fmgr            *format.Manager
+	nextWriterID    *int32
+	throttler       throttling.SettableThrottler
+	metricsRegistry *metrics.Registry
 }

 // directRepository is an implementation of repository that directly manipulates underlying storage.
@@ -248,12 +251,12 @@ func (r *directRepository) NewDirectWriter(ctx context.Context, opt WriteSession

 	mmgr, err := manifest.NewManager(ctx, cmgr, manifest.ManagerOptions{
 		TimeNow: r.timeNow,
-	})
+	}, r.metricsRegistry)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "error creating manifest manager")
 	}

-	omgr, err := object.NewObjectManager(ctx, cmgr, r.omgr.Format)
+	omgr, err := object.NewObjectManager(ctx, cmgr, r.omgr.Format, r.metricsRegistry)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "error creating object manager")
 	}
@@ -300,6 +303,11 @@ func (r *directRepository) Flush(ctx context.Context) error {
 	return errors.Wrap(r.cmgr.Flush(ctx), "error flushing contents")
 }

+// Metrics provides access to metrics registry.
+func (r *directRepository) Metrics() *metrics.Registry {
+	return r.metricsRegistry
+}
+
 // ObjectFormat returns the object format.
 func (r *directRepository) ObjectFormat() format.ObjectFormat {
 	return r.omgr.Format
@@ -647,6 +647,72 @@ func (s *formatSpecificTestSuite) TestChangePassword(t *testing.T) {
 	}
 }

+func TestMetrics_CompressibleData(t *testing.T) {
+	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
+	_ = ctx
+	ms := env.Repository.Metrics().Snapshot()
+
+	require.EqualValues(t, 0, ensureMapEntry(t, ms.Counters, "content_write_bytes"))
+
+	var (
+		inputData = bytes.Repeat([]byte{1, 2, 3, 4}, 100)
+		count     = 0
+		oid       object.ID
+	)
+
+	for ensureMapEntry(t, env.Repository.Metrics().Snapshot().Counters, "content_write_duration_nanos") < 5e6 {
+		w := env.RepositoryWriter.NewObjectWriter(ctx, object.WriterOptions{
+			Compressor: "gzip",
+		})
+		w.Write(inputData)
+
+		var err error
+
+		oid, err = w.Result()
+		require.NoError(t, err)
+
+		count++
+	}
+
+	ms = env.Repository.Metrics().Snapshot()
+	require.EqualValues(t, count*len(inputData), ensureMapEntry(t, ms.Counters, "content_write_bytes"))
+	require.EqualValues(t, count*len(inputData), ensureMapEntry(t, ms.Counters, "content_hashed_bytes"))
+	require.EqualValues(t, len(inputData), ensureMapEntry(t, ms.Counters, "content_compression_attempted_bytes"))
+
+	// this is what 100x{1,2,3,4} compresses down to using gzip, it's also
+	// the number of bytes that go into encryption.
+	const compressedByteCount = 36
+
+	const encryptionOverhead = 28
+
+	require.EqualValues(t, compressedByteCount, ensureMapEntry(t, ms.Counters, "content_after_compression_bytes"))
+	require.EqualValues(t, len(inputData), ensureMapEntry(t, ms.Counters, "content_compressible_bytes"))
+	require.EqualValues(t, 0, ensureMapEntry(t, ms.Counters, "content_non_compressible_bytes"))
+	require.EqualValues(t, len(inputData)-compressedByteCount, ensureMapEntry(t, ms.Counters, "content_compression_savings_bytes"))
+	require.EqualValues(t, compressedByteCount+encryptionOverhead, ensureMapEntry(t, ms.Counters, "content_encrypted_bytes"))
+
+	r, err := env.RepositoryWriter.OpenObject(ctx, oid)
+	require.NoError(t, err)
+
+	data, err := io.ReadAll(r)
+	require.NoError(t, err)
+	require.Equal(t, inputData, data)
+
+	ms = env.Repository.Metrics().Snapshot()
+	require.EqualValues(t, len(inputData), ensureMapEntry(t, ms.Counters, "content_read_bytes"))
+	require.EqualValues(t, compressedByteCount, ensureMapEntry(t, ms.Counters, "content_decompressed_bytes"))
+	require.EqualValues(t, compressedByteCount+encryptionOverhead, ensureMapEntry(t, ms.Counters, "content_decrypted_bytes"))
+}
+
+func ensureMapEntry[T any](t *testing.T, m map[string]T, key string) T {
+	t.Helper()
+
+	actual, got := m[key]
+	require.True(t, got, key)
+
+	return actual
+}
+
 func TestDeriveKey(t *testing.T) {
 	testPurpose := []byte{0, 0, 0, 0}
 	testKeyLength := 8