From 12df909ec8b0d8f2faedd660098d43baa5870118 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=96=D0=B5=D1=80=D0=BE=D0=BC=D1=81=D0=BA=D0=B8=D0=B9=20?= =?UTF-8?q?=D0=9C=D0=B0=D0=BA=D1=81=D0=B8=D0=BC?= Date: Thu, 4 Jun 2026 16:42:34 +0300 Subject: [PATCH 1/2] [wip]: filter deleted mappings in metadata (from file) --- .../statshouse-metadata.go | 12 ++- internal/metadata/binlog_event.go | 3 +- internal/metadata/dbv2.go | 92 ++++++++++++++++++- internal/metadata/dbv2_test.go | 2 +- 4 files changed, 102 insertions(+), 7 deletions(-) diff --git a/cmd/statshouse-metadata/statshouse-metadata.go b/cmd/statshouse-metadata/statshouse-metadata.go index b48733618..19b985625 100644 --- a/cmd/statshouse-metadata/statshouse-metadata.go +++ b/cmd/statshouse-metadata/statshouse-metadata.go @@ -62,6 +62,8 @@ var argv struct { version bool help bool secureMode bool + + deletionCandidateMappingsPath string } type Logger struct{} @@ -103,6 +105,8 @@ func parseArgs() { pflag.Int64Var(&argv.budgetBonus, "global-budget", metadata.GlobalBudget, "create mapping budget. After spent this budget meta will use step system") pflag.Var(&argv.trustedSubnetGroupsFlag, "trusted-subnet-groups", "trusted subnet groups; format: group1,group1b;group2 (CIDR list, groups split by ';')") + pflag.StringVar(&argv.deletionCandidateMappingsPath, "deletion-candidate-mappings-path", "", "path to file with deletion candidate mappings") + pflag.Parse() } @@ -271,7 +275,7 @@ func run() error { } host := srvfunc.Hostname() log.Println("[debug] opening db and reread binlog") - db, err := metadata.OpenDB(argv.dbPath, metadata.Options{ + db, err := metadata.OpenDB(argv.dbPath, argv.deletionCandidateMappingsPath, metadata.Options{ Host: host, MaxBudget: argv.maxBudget, StepSec: argv.stepSec, @@ -282,6 +286,12 @@ func run() error { if err != nil { return fmt.Errorf("db-path: %s, failed to open db: %w", argv.dbPath, err) } + loadedLen, err := db.DeletionCandidatesLen() + if err != nil { + return err + } + log.Printf("[debug]: loaded %d delection candidate mapping ids from %q", loadedLen, argv.deletionCandidateMappingsPath) + defer func() { err := db.Close() if err != nil { diff --git a/internal/metadata/binlog_event.go b/internal/metadata/binlog_event.go index d54b08542..ab4c35397 100644 --- a/internal/metadata/binlog_event.go +++ b/internal/metadata/binlog_event.go @@ -10,10 +10,9 @@ import ( "fmt" "time" - "github.com/VKCOM/statshouse/internal/sqlite" - "github.com/VKCOM/statshouse/internal/data_model/gen2/tlmetadata" "github.com/VKCOM/statshouse/internal/data_model/gen2/tlstatshouse" + "github.com/VKCOM/statshouse/internal/sqlite" "github.com/VKCOM/statshouse/internal/vkgo/basictl" binlog2 "github.com/VKCOM/statshouse/internal/vkgo/binlog" diff --git a/internal/metadata/dbv2.go b/internal/metadata/dbv2.go index 2dca6d6d9..1effc43b6 100644 --- a/internal/metadata/dbv2.go +++ b/internal/metadata/dbv2.go @@ -8,7 +8,11 @@ package metadata import ( "context" + "encoding/binary" "fmt" + "io" + "os" + "sort" "time" "github.com/VKCOM/statshouse/internal/data_model" @@ -17,6 +21,7 @@ import ( "github.com/VKCOM/statshouse/internal/format" "github.com/VKCOM/statshouse/internal/sqlite" "github.com/VKCOM/statshouse/internal/vkgo/binlog/fsbinlog" + "go.uber.org/atomic" ) type DBV2 struct { @@ -37,6 +42,8 @@ type DBV2 struct { globalBudget int64 lastMappingIDToInsert int32 + + deletionCandidateMappings *atomic.Pointer[[]int32] } type Options struct { @@ -160,6 +167,7 @@ const mappingCountReadLimit int64 = 50000 // we want to catch metricBytesReadLim func OpenDB( path string, + deletionCandidateMappingsPath string, opt Options, binlog fsbinlog.BinlogReadWrite) (*DBV2, error) { if opt.Now == nil { @@ -170,6 +178,17 @@ func OpenDB( return nil } } + + deletionCandidateMappingIds := atomic.NewPointer[[]int32](nil) + if deletionCandidateMappingsPath != "" { + if ids, err := loadDeletionCandidatesFromBinFile(deletionCandidateMappingsPath); err != nil { + return nil, fmt.Errorf("error loading mappings from file %s: %v", deletionCandidateMappingsPath, err) + } else { + deletionCandidateMappingIds.Store(&ids) + } + } + + // TODO: implement filtering deleted mappings in binlog eng, err := sqlite.OpenEngine(sqlite.Options{ Path: path, APPID: appId, @@ -190,8 +209,9 @@ func OpenDB( maxBudget: opt.MaxBudget, globalBudget: opt.GlobalBudget, - now: opt.Now, - lastTimeCommit: opt.Now(), + now: opt.Now, + lastTimeCommit: opt.Now(), + deletionCandidateMappings: deletionCandidateMappingIds, } return db, nil @@ -306,6 +326,10 @@ func (db *DBV2) GetNewMappings(ctx context.Context, fromID int32, page int32) ([ return cache, err } + if db.IsDeletionCandidate(int32(id)) { + continue + } + result = append(result, tlstatshouse.Mapping{ Value: int32(id), Str: name, @@ -323,6 +347,8 @@ func (db *DBV2) GetNewMappings(ctx context.Context, fromID int32, page int32) ([ if row.Next() { maxID64, _ := row.ColumnInt64(0) maxID = int32(maxID64) + // NOTE: maxID logic preserved as-is because deletion candidates have smaller ids than existing max id + // breaking this invariant can lead to an endless request cycle } return cache, nil }) @@ -344,6 +370,10 @@ func (db *DBV2) GetLastNMappings(ctx context.Context, n int) ([]tlstatshouse.Map return cache, err } + if db.IsDeletionCandidate(int32(id)) { + continue + } + result = append(result, tlstatshouse.Mapping{ Value: int32(id), Str: name, @@ -533,7 +563,11 @@ func (db *DBV2) GetMappingByValue(ctx context.Context, value string) (int32, boo row := conn.Query("select_mapping_by_name", "SELECT id FROM mappings where name = $name", sqlite.BlobString("$name", value)) if row.Next() { id, _ := row.ColumnInt64(0) - res = int32(id) + if db.IsDeletionCandidate(int32(id)) { + notExists = true + } else { + res = int32(id) + } } else { notExists = true } @@ -801,3 +835,55 @@ func (db *DBV2) SaveEntityold(ctx context.Context, name string, id int64, oldVer }) return result, err } + +func loadDeletionCandidatesFromBinFile(path string) ([]int32, error) { + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("open mappings file %s: %w", path, err) + } + defer f.Close() + + st, err := f.Stat() + if err != nil { + return nil, fmt.Errorf("stat mappings file %s: %w", path, err) + } + + size := st.Size() + if size%4 != 0 { + return nil, fmt.Errorf("invalid mappings file size %d: not divisible by 4", size) + } + + buf := make([]byte, int(size)) + + if _, err := io.ReadFull(f, buf); err != nil { + return nil, fmt.Errorf("read mappings file %s: %w", path, err) + } + + // safe []int32 parse + sort check + ids := make([]int32, int(size/4)) + lastId := int32(-1) + for i := range ids { + id := int32(binary.LittleEndian.Uint32(buf[i*4:])) + if id <= lastId { + return nil, fmt.Errorf("invalid mappings file (parsing error or not sorted in ascending order) i: %d, ids[i]: %d, ids[i-1]: %d", i, id, lastId) + } + ids[i] = id + lastId = id + } + + return ids, nil +} + +func (db *DBV2) DeletionCandidatesLen() (int, error) { + ids := db.deletionCandidateMappings.Load() + if ids == nil { + return 0, fmt.Errorf("no delection candidates loaded") + } + return len(*ids), nil +} + +func (db *DBV2) IsDeletionCandidate(id int32) bool { + ids := *db.deletionCandidateMappings.Load() + i := sort.Search(len(ids), func(i int) bool { return ids[i] >= id }) + return i < len(ids) && ids[i] == id +} diff --git a/internal/metadata/dbv2_test.go b/internal/metadata/dbv2_test.go index 45ce0e19c..5c1cb3eff 100644 --- a/internal/metadata/dbv2_test.go +++ b/internal/metadata/dbv2_test.go @@ -68,7 +68,7 @@ func initD1b(t *testing.T, dir string, dbFile string, createBl bool, options *Op bl, err := fsbinlog.NewFsBinlog(&Logger{}, boptions) require.NoError(t, err) - db, err := OpenDB(dir+"/"+dbFile, *options, bl) + db, err := OpenDB(dir+"/"+dbFile, "", *options, bl) require.NoError(t, err) return db, bl } From 85a6ae6bea98fa2a9c93711dc09bb3d0e34d030c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=96=D0=B5=D1=80=D0=BE=D0=BC=D1=81=D0=BA=D0=B8=D0=B9=20?= =?UTF-8?q?=D0=9C=D0=B0=D0=BA=D1=81=D0=B8=D0=BC?= Date: Thu, 4 Jun 2026 18:18:07 +0300 Subject: [PATCH 2/2] add filtering deletion candidates logic to binlog events --- internal/metadata/binlog_event.go | 19 +++++++++++++++---- internal/metadata/dbv2.go | 21 +++++++++++++++------ 2 files changed, 30 insertions(+), 10 deletions(-) diff --git a/internal/metadata/binlog_event.go b/internal/metadata/binlog_event.go index ab4c35397..ea93ecbe6 100644 --- a/internal/metadata/binlog_event.go +++ b/internal/metadata/binlog_event.go @@ -8,6 +8,7 @@ package metadata import ( "fmt" + "sync/atomic" "time" "github.com/VKCOM/statshouse/internal/data_model/gen2/tlmetadata" @@ -19,7 +20,7 @@ import ( "github.com/VKCOM/statshouse/internal/vkgo/binlog/fsbinlog" ) -func applyScanEvent(scanOnly bool) func(conn sqlite.Conn, offset int64, data []byte) (int, error) { +func applyScanEvent(scanOnly bool, deletionCandidateIds *atomic.Pointer[[]int32]) func(conn sqlite.Conn, offset int64, data []byte) (int, error) { return func(conn sqlite.Conn, offset int64, data []byte) (int, error) { readCount := 0 var editMetricEvent tlmetadata.EditMetricEvent @@ -85,6 +86,7 @@ func applyScanEvent(scanOnly bool) func(conn sqlite.Conn, offset int64, data []b } } case putMappingEvent.TLTag(): + // NOTE: event only fires by an admin endpoint, so we don't filter out deletion candidates tail, err = putMappingEvent.ReadTL1(data) if err != nil { return fsbinlog.AddPadding(readCount), err @@ -96,6 +98,7 @@ func applyScanEvent(scanOnly bool) func(conn sqlite.Conn, offset int64, data []b } } case createMappingEvent.TLTag(): + // NOTE: the event only fires in getOrCreateMapping for mappings that aren't deletion candidates tail, err = createMappingEvent.ReadTL1(data) if err != nil { return fsbinlog.AddPadding(readCount), err @@ -112,7 +115,7 @@ func applyScanEvent(scanOnly bool) func(conn sqlite.Conn, offset int64, data []b return fsbinlog.AddPadding(readCount), err } if !scanOnly { - _, _, err := applyPutBootstrap(conn, nil, putBootstrapEvent.Mappings) + _, _, err := applyPutBootstrap(conn, nil, putBootstrapEvent.Mappings, deletionCandidateIds) if err != nil { return fsbinlog.AddPadding(readCount), fmt.Errorf("can't apply binlog event MetadataPutBootstrapEvent: %w", err) } @@ -184,6 +187,7 @@ func applyEditEntityEvent(conn sqlite.Conn, event tlmetadata.EditEntityEvent) er // todo support metric namespacing func applyCreateMappingEvent(conn sqlite.Conn, event tlmetadata.CreateMappingEvent) error { + // NOTE: filtering deletion candidates in usage _, err := conn.Exec("insert_flood_limit", "INSERT OR REPLACE INTO flood_limits (last_time_update, count_free, metric_name) VALUES ($t, $c, $name)", sqlite.Int64("$t", int64(event.UpdatedAt)), sqlite.Int64("$c", event.Budget), @@ -233,7 +237,7 @@ func applyCreateEntityEvent(conn sqlite.Conn, event tlmetadata.CreateEntityEvent return nil } -func getOrCreateMapping(conn sqlite.Conn, cache []byte, metricName, key string, now time.Time, globalBudget, maxBudget, budgetBonus int64, stepSec uint32, lastCreatedID int32) (tlmetadata.GetMappingResponse, []byte, error) { +func getOrCreateMapping(conn sqlite.Conn, cache []byte, metricName, key string, now time.Time, globalBudget, maxBudget, budgetBonus int64, stepSec uint32, lastCreatedID int32, deletionCandidateIds *atomic.Pointer[[]int32]) (tlmetadata.GetMappingResponse, []byte, error) { var id int32 row := conn.Query("select_mapping", "SELECT id FROM mappings where name = $name;", sqlite.BlobString("$name", key)) if row.Error() != nil { @@ -242,6 +246,9 @@ func getOrCreateMapping(conn sqlite.Conn, cache []byte, metricName, key string, if row.Next() { resp, _ := row.ColumnInt64(0) id := int32(resp) + if IsDeletionCandidate(id, deletionCandidateIds) { + return tlmetadata.GetMappingResponse{}, cache, fmt.Errorf("requested id %d is marked for deletion", id) + } return tlmetadata.GetMappingResponse0{Id: id}.AsUnion(), cache, nil } pred := roundTime(now, stepSec) @@ -302,6 +309,7 @@ func getOrCreateMapping(conn sqlite.Conn, cache []byte, metricName, key string, } func putMapping(conn sqlite.Conn, cache []byte, ks []string, vs []int32) ([]byte, error) { + // NOTE: filtering deletion candidates in usage for i := range ks { _, err := conn.Exec("upsert_mapping", "INSERT OR REPLACE INTO mappings(id, name) VALUES($id, $name);", sqlite.Int64("$id", int64(vs[i])), sqlite.BlobString("$name", ks[i])) if err != nil { @@ -317,9 +325,12 @@ func putMapping(conn sqlite.Conn, cache []byte, ks []string, vs []int32) ([]byte return cache, nil } -func applyPutBootstrap(conn sqlite.Conn, cache []byte, mappings []tlstatshouse.Mapping) (int32, []byte, error) { +func applyPutBootstrap(conn sqlite.Conn, cache []byte, mappings []tlstatshouse.Mapping, deletionCandidateIds *atomic.Pointer[[]int32]) (int32, []byte, error) { filteredMappings := make([]tlstatshouse.Mapping, 0, len(mappings)) for _, m := range mappings { + if IsDeletionCandidate(m.Value, deletionCandidateIds) { + continue + } k, isExists, err := getMappingByID(conn, m.Value) if err != nil { return 0, cache, err diff --git a/internal/metadata/dbv2.go b/internal/metadata/dbv2.go index 1effc43b6..487e3d9c3 100644 --- a/internal/metadata/dbv2.go +++ b/internal/metadata/dbv2.go @@ -21,7 +21,7 @@ import ( "github.com/VKCOM/statshouse/internal/format" "github.com/VKCOM/statshouse/internal/sqlite" "github.com/VKCOM/statshouse/internal/vkgo/binlog/fsbinlog" - "go.uber.org/atomic" + "sync/atomic" ) type DBV2 struct { @@ -179,7 +179,7 @@ func OpenDB( } } - deletionCandidateMappingIds := atomic.NewPointer[[]int32](nil) + deletionCandidateMappingIds := &atomic.Pointer[[]int32]{} if deletionCandidateMappingsPath != "" { if ids, err := loadDeletionCandidatesFromBinFile(deletionCandidateMappingsPath); err != nil { return nil, fmt.Errorf("error loading mappings from file %s: %v", deletionCandidateMappingsPath, err) @@ -193,7 +193,7 @@ func OpenDB( Path: path, APPID: appId, Scheme: scheme, - }, binlog, applyScanEvent(false), applyScanEvent(true)) + }, binlog, applyScanEvent(false, deletionCandidateMappingIds), applyScanEvent(true, deletionCandidateMappingIds)) if err != nil { return nil, fmt.Errorf("failed to open engine: %w", err) } @@ -593,6 +593,9 @@ func (db *DBV2) PrintAllMappings(ctx context.Context) error { func (db *DBV2) GetMappingByID(ctx context.Context, id int32) (string, bool, error) { var res string var isExists bool + if db.IsDeletionCandidate(id) { + return "", false, nil + } err := db.eng.Do(ctx, "get_mapping_by_key", func(conn sqlite.Conn, cache []byte) ([]byte, error) { var err error res, isExists, err = getMappingByID(conn, id) @@ -602,6 +605,7 @@ func (db *DBV2) GetMappingByID(ctx context.Context, id int32) (string, bool, err } func getMappingByID(conn sqlite.Conn, id int32) (k string, isExists bool, err error) { + // NOTE: deletion candidates filtered in 2 usages row := conn.Query("select_mapping_by_id", "SELECT name FROM mappings where id = $id", sqlite.Int64("$id", int64(id))) if row.Next() { k, err = row.ColumnBlobString(0) @@ -654,7 +658,7 @@ func (db *DBV2) GetOrCreateMapping(ctx context.Context, metricName, key string) now := db.now() err := db.eng.Do(ctx, "get_or_create_mapping", func(conn sqlite.Conn, cache []byte) ([]byte, error) { var err error - resp, cache, err = getOrCreateMapping(conn, cache, metricName, key, now, db.globalBudget, db.maxBudget, db.budgetBonus, db.stepSec, db.lastMappingIDToInsert) + resp, cache, err = getOrCreateMapping(conn, cache, metricName, key, now, db.globalBudget, db.maxBudget, db.budgetBonus, db.stepSec, db.lastMappingIDToInsert, db.deletionCandidateMappings) if resp.IsCreated() { created, _ := resp.AsCreated() db.lastMappingIDToInsert = created.Id @@ -668,6 +672,7 @@ func (db *DBV2) GetOrCreateMapping(ctx context.Context, metricName, key string) } func (db *DBV2) PutMapping(ctx context.Context, ks []string, vs []int32) error { + // NOTE: admin endpoint, filtering deletion candidates not implemented if len(ks) != len(vs) { return fmt.Errorf("can't match keys size and values size") } @@ -703,7 +708,7 @@ func (db *DBV2) PutBootstrap(ctx context.Context, mappings []tlstatshouse.Mappin var count int32 err := db.eng.Do(ctx, "put_bootstrap", func(conn sqlite.Conn, cache []byte) ([]byte, error) { var err error - count, cache, err = applyPutBootstrap(conn, cache, mappings) + count, cache, err = applyPutBootstrap(conn, cache, mappings, db.deletionCandidateMappings) return cache, err }) return count, err @@ -883,7 +888,11 @@ func (db *DBV2) DeletionCandidatesLen() (int, error) { } func (db *DBV2) IsDeletionCandidate(id int32) bool { - ids := *db.deletionCandidateMappings.Load() + return IsDeletionCandidate(id, db.deletionCandidateMappings) +} + +func IsDeletionCandidate(id int32, deletionCandidateIds *atomic.Pointer[[]int32]) bool { + ids := *deletionCandidateIds.Load() i := sort.Search(len(ids), func(i int) bool { return ids[i] >= id }) return i < len(ids) && ids[i] == id }