[WIP]: filter deleted mappings in metadata

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 {

internal/metadata/binlog_event.go

@@ -8,19 +8,19 @@ package metadata
 
 import (
 	"fmt"
+	"sync/atomic"
 	"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"
 	"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
@@ -86,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
@@ -97,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
@@ -113,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)
 					}
@@ -185,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),
@@ -234,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 {
@@ -243,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)
@@ -303,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 {
@@ -318,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

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"
+	"sync/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,11 +178,22 @@ func OpenDB(
 			return 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)
+		} else {
+			deletionCandidateMappingIds.Store(&ids)
+		}
+	}
+
+	// TODO: implement filtering deleted mappings in binlog
 	eng, err := sqlite.OpenEngine(sqlite.Options{
 		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)
 	}
@@ -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
 		}
@@ -559,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)
@@ -568,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)
@@ -620,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
@@ -634,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")
 	}
@@ -669,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
@@ -801,3 +840,59 @@ 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 {
+	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
+}

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
 }