TrueCloudLab/frostfs-node
20
1
Fork 28
Code Issues 107 Pull requests 4 Projects Releases 2 Packages 4 Activity Actions

[#312] metrics: Add writecache metrcis

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-05-19 11:17:19 +03:00 • committed by Evgenii Stratonikov
parent d212d908b5
commit 2ce43935f9
14 changed files with 415 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

4
pkg/local_object_storage/engine/metrics.go
View file

@ -2,6 +2,8 @@ package engine
import (
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/metrics"
)
type MetricRegister interface {
@ -24,6 +26,8 @@ type MetricRegister interface {
AddToContainerSize(cnrID string, size int64)
AddToPayloadCounter(shardID string, size int64)
WriteCache() metrics.WriteCacheMetrics
}
func elapsed(addFunc func(d time.Duration)) func() {

7
pkg/local_object_storage/engine/shards.go
View file

@ -8,6 +8,7 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/hrw"
"github.com/google/uuid"
@ -98,6 +99,12 @@ func (e *StorageEngine) createShard(opts []shard.Option) (*shard.Shard, error) {
mw: e.metrics,
},
))
opts = append(opts, shard.WithExtraWriteCacheOptions(writecache.WithMetrics(
&writeCacheMetrics{
shardID: id.String(),
metrics: e.metrics.WriteCache(),
},
)))
}
e.mtx.RUnlock()

53
pkg/local_object_storage/engine/writecache.go
View file

@ -2,9 +2,13 @@ package engine
import (
"context"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/metrics"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
@ -52,3 +56,52 @@ func (e *StorageEngine) FlushWriteCache(ctx context.Context, p FlushWriteCachePr
return FlushWriteCacheRes{}, sh.FlushWriteCache(ctx, prm)
}
type writeCacheMetrics struct {
shardID string
metrics metrics.WriteCacheMetrics
}
func (m *writeCacheMetrics) Get(d time.Duration, success bool, st writecache.StorageType) {
m.metrics.AddGetDuration(m.shardID, success, d)
m.metrics.IncGetCounter(m.shardID, success, st.String())
}
func (m *writeCacheMetrics) Delete(d time.Duration, success bool, st writecache.StorageType) {
m.metrics.AddDeleteDuration(m.shardID, success, d)
m.metrics.IncDeleteCounter(m.shardID, success, st.String())
if success {
m.metrics.DecActualCount(m.shardID, st.String())
}
}
func (m *writeCacheMetrics) Put(d time.Duration, success bool, st writecache.StorageType) {
m.metrics.AddPutDuration(m.shardID, success, d)
m.metrics.IncPutCounter(m.shardID, success, st.String())
if success {
m.metrics.IncActualCount(m.shardID, st.String())
}
}
func (m *writeCacheMetrics) SetEstimateSize(db, fstree uint64) {
m.metrics.SetEstimateSize(m.shardID, db, writecache.StorageTypeDB.String())
m.metrics.SetEstimateSize(m.shardID, fstree, writecache.StorageTypeFSTree.String())
}
func (m *writeCacheMetrics) SetMode(mode mode.Mode) {
m.metrics.SetMode(m.shardID, mode.String())
}
func (m *writeCacheMetrics) SetActualCounters(db, fstree uint64) {
m.metrics.SetActualCount(m.shardID, db, writecache.StorageTypeDB.String())
m.metrics.SetActualCount(m.shardID, fstree, writecache.StorageTypeFSTree.String())
}
func (m *writeCacheMetrics) Flush(success bool, st writecache.StorageType) {
m.metrics.IncFlushCounter(m.shardID, success, st.String())
}
func (m *writeCacheMetrics) Evict(st writecache.StorageType) {
m.metrics.DecActualCount(m.shardID, st.String())
m.metrics.IncEvictCounter(m.shardID, st.String())
}

7
pkg/local_object_storage/shard/shard.go
View file

@ -187,6 +187,13 @@ func WithWriteCacheOptions(opts ...writecache.Option) Option {
}
}
// WithExtraWriteCacheOptions returns option to add extra write cache options.
func WithExtraWriteCacheOptions(opts ...writecache.Option) Option {
return func(c *cfg) {
c.writeCacheOpts = append(c.writeCacheOpts, opts...)
}
}
// WithPiloramaOptions returns option to set internal write cache options.
func WithPiloramaOptions(opts ...pilorama.Option) Option {
return func(c *cfg) {

6
pkg/local_object_storage/writecache/delete.go
View file

@ -24,7 +24,7 @@ func (c *cache) Delete(ctx context.Context, addr oid.Address) error {
defer span.End()
deleted := false
storageType := storageTypeUndefined
storageType := StorageTypeUndefined
startedAt := time.Now()
defer func() {
c.metrics.Delete(time.Since(startedAt), deleted, storageType)
@ -46,7 +46,7 @@ func (c *cache) Delete(ctx context.Context, addr oid.Address) error {
})
if dataSize > 0 {
storageType = storageTypeDB
storageType = StorageTypeDB
err := c.db.Update(func(tx *bbolt.Tx) error {
b := tx.Bucket(defaultBucket)
err := b.Delete([]byte(saddr))
@ -65,7 +65,7 @@ func (c *cache) Delete(ctx context.Context, addr oid.Address) error {
return nil
}
storageType = storageTypeFSTree
storageType = StorageTypeFSTree
_, err := c.fsTree.Delete(ctx, common.DeletePrm{Address: addr})
if err == nil {
storagelog.Write(c.log,

8
pkg/local_object_storage/writecache/flush.go
View file

@ -199,7 +199,7 @@ func (c *cache) flushFSTree(ctx context.Context, ignoreErrors bool) error {
return err
}
err = c.flushObject(ctx, &obj, data, storageTypeFSTree)
err = c.flushObject(ctx, &obj, data, StorageTypeFSTree)
if err != nil {
if ignoreErrors {
return nil
@ -228,7 +228,7 @@ func (c *cache) workerFlushSmall() {
return
}
err := c.flushObject(context.TODO(), obj, nil, storageTypeDB)
err := c.flushObject(context.TODO(), obj, nil, StorageTypeDB)
if err != nil {
// Error is handled in flushObject.
continue
@ -239,7 +239,7 @@ func (c *cache) workerFlushSmall() {
}
// flushObject is used to write object directly to the main storage.
func (c *cache) flushObject(ctx context.Context, obj *object.Object, data []byte, st storageType) error {
func (c *cache) flushObject(ctx context.Context, obj *object.Object, data []byte, st StorageType) error {
var err error
defer func() {
@ -319,7 +319,7 @@ func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
return err
}
if err := c.flushObject(ctx, &obj, data, storageTypeDB); err != nil {
if err := c.flushObject(ctx, &obj, data, StorageTypeDB); err != nil {
return err
}
}

6
pkg/local_object_storage/writecache/get.go
View file

@ -33,7 +33,7 @@ func (c *cache) Get(ctx context.Context, addr oid.Address) (*objectSDK.Object, e
func (c *cache) getInternal(ctx context.Context, saddr string, addr oid.Address) (*objectSDK.Object, error) {
found := false
storageType := storageTypeUndefined
storageType := StorageTypeUndefined
startedAt := time.Now()
defer func() {
c.metrics.Get(time.Since(startedAt), found, storageType)
@ -43,7 +43,7 @@ func (c *cache) getInternal(ctx context.Context, saddr string, addr oid.Address)
if err == nil {
obj := objectSDK.New()
found = true
storageType = storageTypeDB
storageType = StorageTypeDB
return obj, obj.Unmarshal(value)
}
@ -53,7 +53,7 @@ func (c *cache) getInternal(ctx context.Context, saddr string, addr oid.Address)
}
found = true
storageType = storageTypeFSTree
storageType = StorageTypeFSTree
return res.Object, nil
}

39
pkg/local_object_storage/writecache/metrics.go
View file

@ -6,37 +6,44 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
)
type storageType string
type StorageType string
func (t StorageType) String() string {
return string(t)
}
const (
storageTypeUndefined storageType = "null"
storageTypeDB storageType = "db"
storageTypeFSTree storageType = "fstree"
StorageTypeUndefined StorageType = "null"
StorageTypeDB StorageType = "db"
StorageTypeFSTree StorageType = "fstree"
)
type Metrics interface {
Get(d time.Duration, success bool, st storageType)
Delete(d time.Duration, success bool, st storageType)
Put(d time.Duration, success bool, st storageType)
Flush(success bool, st storageType)
Evict(st storageType)
Get(d time.Duration, success bool, st StorageType)
Delete(d time.Duration, success bool, st StorageType)
Put(d time.Duration, success bool, st StorageType)
Flush(success bool, st StorageType)
Evict(st StorageType)
Estimate(db, fstree uint64)
SetEstimateSize(db, fstree uint64)
SetMode(m mode.Mode)
SetActualCounters(db, fstree uint64)
}
type metricsStub struct{}
func (s *metricsStub) Get(time.Duration, bool, storageType) {}
func (s *metricsStub) Get(time.Duration, bool, StorageType) {}
func (s *metricsStub) Delete(time.Duration, bool, storageType) {}
func (s *metricsStub) Delete(time.Duration, bool, StorageType) {}
func (s *metricsStub) Put(time.Duration, bool, storageType) {}
func (s *metricsStub) Put(time.Duration, bool, StorageType) {}
func (s *metricsStub) Estimate(uint64, uint64) {}
func (s *metricsStub) SetEstimateSize(uint64, uint64) {}
func (s *metricsStub) SetMode(mode.Mode) {}
func (s *metricsStub) Flush(bool, storageType) {}
func (s *metricsStub) SetActualCounters(uint64, uint64) {}
func (s *metricsStub) Evict(storageType) {}
func (s *metricsStub) Flush(bool, StorageType) {}
func (s *metricsStub) Evict(StorageType) {}

6
pkg/local_object_storage/writecache/put.go
View file

@ -36,7 +36,7 @@ func (c *cache) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, erro
startedAt := time.Now()
added := false
storageType := storageTypeUndefined
storageType := StorageTypeUndefined
defer func() {
c.metrics.Put(time.Since(startedAt), added, storageType)
}()
@ -59,7 +59,7 @@ func (c *cache) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, erro
}
if sz <= c.smallObjectSize {
storageType = storageTypeDB
storageType = StorageTypeDB
err := c.putSmall(oi)
if err == nil {
added = true
@ -67,7 +67,7 @@ func (c *cache) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, erro
return common.PutRes{}, err
}
storageType = storageTypeFSTree
storageType = StorageTypeFSTree
err := c.putBig(ctx, oi.addr, prm)
if err == nil {
added = true

3
pkg/local_object_storage/writecache/state.go
View file

@ -11,7 +11,7 @@ import (
func (c *cache) estimateCacheSize() uint64 {
db := c.objCounters.DB() * c.smallObjectSize
fstree := c.objCounters.FS() * c.maxObjectSize
c.metrics.Estimate(db, fstree)
c.metrics.SetEstimateSize(db, fstree)
return db + fstree
}
@ -71,6 +71,7 @@ func (c *cache) initCounters() error {
c.objCounters.cDB.Store(inDB)
c.objCounters.cFS.Store(inFS)
c.metrics.SetActualCounters(inDB, inFS)
return nil
}

4
pkg/local_object_storage/writecache/storage.go
View file

@ -79,7 +79,7 @@ func (c *cache) deleteFromDB(keys []string) []string {
})
for i := 0; i < errorIndex; i++ {
c.objCounters.DecDB()
c.metrics.Evict(storageTypeDB)
c.metrics.Evict(StorageTypeDB)
storagelog.Write(c.log,
storagelog.AddressField(keys[i]),
storagelog.StorageTypeField(wcStorageType),
@ -122,7 +122,7 @@ func (c *cache) deleteFromDisk(ctx context.Context, keys []string) []string {
storagelog.StorageTypeField(wcStorageType),
storagelog.OpField("fstree DELETE"),
)
c.metrics.Evict(storageTypeFSTree)
c.metrics.Evict(StorageTypeFSTree)
c.objCounters.DecFS()
}
}