diff --git a/cli/observability_flags_test.go b/cli/observability_flags_test.go index 2c26d59e1..9eb58c01a 100644 --- a/cli/observability_flags_test.go +++ b/cli/observability_flags_test.go @@ -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", diff --git a/go.mod b/go.mod index 1828250ad..c4268c3a7 100644 --- a/go.mod +++ b/go.mod @@ -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 ( diff --git a/go.sum b/go.sum index 3e490d27f..3ac13d52d 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/cache/cache_metrics.go b/internal/cache/cache_metrics.go index fc0a68ae2..faf6bbb1a 100644 --- a/internal/cache/cache_metrics.go +++ b/internal/cache/cache_metrics.go @@ -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) } diff --git a/internal/cache/content_cache.go b/internal/cache/content_cache.go index 9b077de9b..74b5a5f04 100644 --- a/internal/cache/content_cache.go +++ b/internal/cache/content_cache.go @@ -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") } diff --git a/internal/cache/content_cache_concurrency_test.go b/internal/cache/content_cache_concurrency_test.go index a2b749121..9b42cf099 100644 --- a/internal/cache/content_cache_concurrency_test.go +++ b/internal/cache/content_cache_concurrency_test.go @@ -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) { diff --git a/internal/cache/content_cache_data_test.go b/internal/cache/content_cache_data_test.go index 64c693854..7e29660dc 100644 --- a/internal/cache/content_cache_data_test.go +++ b/internal/cache/content_cache_data_test.go @@ -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{})) diff --git a/internal/cache/content_cache_metadata_test.go b/internal/cache/content_cache_metadata_test.go index 069c2cd37..ffc142405 100644 --- a/internal/cache/content_cache_metadata_test.go +++ b/internal/cache/content_cache_metadata_test.go @@ -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{})) diff --git a/internal/cache/content_cache_test.go b/internal/cache/content_cache_test.go index 28aa91900..2a546b561 100644 --- a/internal/cache/content_cache_test.go +++ b/internal/cache/content_cache_test.go @@ -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) diff --git a/internal/cache/persistent_lru_cache.go b/internal/cache/persistent_lru_cache.go index 474c31e1a..240dbd91b 100644 --- a/internal/cache/persistent_lru_cache.go +++ b/internal/cache/persistent_lru_cache.go @@ -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) diff --git a/internal/cache/persistent_lru_cache_test.go b/internal/cache/persistent_lru_cache_test.go index 8dad3a3fd..24dec1373 100644 --- a/internal/cache/persistent_lru_cache_test.go +++ b/internal/cache/persistent_lru_cache_test.go @@ -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) diff --git a/internal/metrics/metric_test.go b/internal/metrics/metric_test.go new file mode 100644 index 000000000..5c667044e --- /dev/null +++ b/internal/metrics/metric_test.go @@ -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 +} diff --git a/internal/metrics/metrics_counter.go b/internal/metrics/metrics_counter.go new file mode 100644 index 000000000..b6a19a64a --- /dev/null +++ b/internal/metrics/metrics_counter.go @@ -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 +} diff --git a/internal/metrics/metrics_counter_test.go b/internal/metrics/metrics_counter_test.go new file mode 100644 index 000000000..6ac56d5cc --- /dev/null +++ b/internal/metrics/metrics_counter_test.go @@ -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()) +} diff --git a/internal/metrics/metrics_distribution.go b/internal/metrics/metrics_distribution.go new file mode 100644 index 000000000..dff44b39d --- /dev/null +++ b/internal/metrics/metrics_distribution.go @@ -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 +} diff --git a/internal/metrics/metrics_distribution_test.go b/internal/metrics/metrics_distribution_test.go new file mode 100644 index 000000000..d38393a30 --- /dev/null +++ b/internal/metrics/metrics_distribution_test.go @@ -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)) +} diff --git a/internal/metrics/metrics_duration_distribution_test.go b/internal/metrics/metrics_duration_distribution_test.go new file mode 100644 index 000000000..0bba97e53 --- /dev/null +++ b/internal/metrics/metrics_duration_distribution_test.go @@ -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()) +} diff --git a/internal/metrics/metrics_gauge.go b/internal/metrics/metrics_gauge.go new file mode 100644 index 000000000..dcccee761 --- /dev/null +++ b/internal/metrics/metrics_gauge.go @@ -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 +} diff --git a/internal/metrics/metrics_gauge_test.go b/internal/metrics/metrics_gauge_test.go new file mode 100644 index 000000000..4be217be5 --- /dev/null +++ b/internal/metrics/metrics_gauge_test.go @@ -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()) +} diff --git a/internal/metrics/metrics_registry.go b/internal/metrics/metrics_registry.go new file mode 100644 index 000000000..1a61c39b7 --- /dev/null +++ b/internal/metrics/metrics_registry.go @@ -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, ";") + "]" +} diff --git a/internal/metrics/metrics_registry_test.go b/internal/metrics/metrics_registry_test.go new file mode 100644 index 000000000..1197ffaf8 --- /dev/null +++ b/internal/metrics/metrics_registry_test.go @@ -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) +} diff --git a/internal/metrics/metrics_thresholds.go b/internal/metrics/metrics_thresholds.go new file mode 100644 index 000000000..f56d3a5b0 --- /dev/null +++ b/internal/metrics/metrics_thresholds.go @@ -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 +} diff --git a/internal/metrics/metrics_throughput.go b/internal/metrics/metrics_throughput.go new file mode 100644 index 000000000..db2a517a0 --- /dev/null +++ b/internal/metrics/metrics_throughput.go @@ -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 +} diff --git a/internal/metrics/metrics_throughput_test.go b/internal/metrics/metrics_throughput_test.go new file mode 100644 index 000000000..80618e153 --- /dev/null +++ b/internal/metrics/metrics_throughput_test.go @@ -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()) +} diff --git a/internal/metrics/prom_cache.go b/internal/metrics/prom_cache.go new file mode 100644 index 000000000..69a1121a3 --- /dev/null +++ b/internal/metrics/prom_cache.go @@ -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)...) +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 75a865d4b..753ad5ca7 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -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") } } diff --git a/repo/api_server_repository.go b/repo/api_server_repository.go index cc54ccdc9..e86cb93be 100644 --- a/repo/api_server_repository.go +++ b/repo/api_server_repository.go @@ -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") } diff --git a/repo/blob/storagemetrics/storage_metrics.go b/repo/blob/storagemetrics/storage_metrics.go new file mode 100644 index 000000000..6fe1de017 --- /dev/null +++ b/repo/blob/storagemetrics/storage_metrics.go @@ -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"), + } +} diff --git a/repo/blob/storagemetrics/storage_metrics_test.go b/repo/blob/storagemetrics/storage_metrics_test.go new file mode 100644 index 000000000..14bf0c52a --- /dev/null +++ b/repo/blob/storagemetrics/storage_metrics_test.go @@ -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) +} diff --git a/repo/content/caching_options.go b/repo/content/caching_options.go index 36657077f..f456d9755 100644 --- a/repo/content/caching_options.go +++ b/repo/content/caching_options.go @@ -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) +} diff --git a/repo/content/committed_read_manager.go b/repo/content/committed_read_manager.go index 4ce633cfe..a04a45e81 100644 --- a/repo/content/committed_read_manager.go +++ b/repo/content/committed_read_manager.go @@ -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") } diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go index 2459882f2..9c0ac09fe 100644 --- a/repo/content/content_manager.go +++ b/repo/content/content_manager.go @@ -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), } diff --git a/repo/content/content_manager_lock_free.go b/repo/content/content_manager_lock_free.go index ba8cfa711..8e769f07f 100644 --- a/repo/content/content_manager_lock_free.go +++ b/repo/content/content_manager_lock_free.go @@ -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 } diff --git a/repo/content/content_manager_metrics.go b/repo/content/content_manager_metrics.go index c979532bb..20d625e82 100644 --- a/repo/content/content_manager_metrics.go +++ b/repo/content/content_manager_metrics.go @@ -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), + } } diff --git a/repo/grpc_repository_client.go b/repo/grpc_repository_client.go index f5f8a7a64..cb87d8aec 100644 --- a/repo/grpc_repository_client.go +++ b/repo/grpc_repository_client.go @@ -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") } diff --git a/repo/manifest/manifest_manager.go b/repo/manifest/manifest_manager.go index cfd2638b0..ba01dc823 100644 --- a/repo/manifest/manifest_manager.go +++ b/repo/manifest/manifest_manager.go @@ -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 diff --git a/repo/manifest/manifest_manager_test.go b/repo/manifest/manifest_manager_test.go index da4123609..b31ffa8e0 100644 --- a/repo/manifest/manifest_manager_test.go +++ b/repo/manifest/manifest_manager_test.go @@ -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 diff --git a/repo/object/object_manager.go b/repo/object/object_manager.go index 10078ada0..ce3341ccd 100644 --- a/repo/object/object_manager.go +++ b/repo/object/object_manager.go @@ -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, diff --git a/repo/object/object_manager_test.go b/repo/object/object_manager_test.go index d19458096..b5d0fea8e 100644 --- a/repo/object/object_manager_test.go +++ b/repo/object/object_manager_test.go @@ -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) } diff --git a/repo/open.go b/repo/open.go index f299e1ef6..a57751ee5 100644 --- a/repo/open.go +++ b/repo/open.go @@ -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{}), } diff --git a/repo/repository.go b/repo/repository.go index 42e371fde..f2d932b35 100644 --- a/repo/repository.go +++ b/repo/repository.go @@ -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 diff --git a/repo/repository_test.go b/repo/repository_test.go index f4f703ced..22d04284e 100644 --- a/repo/repository_test.go +++ b/repo/repository_test.go @@ -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