open-consul/agent/consul/state/graveyard.go
Dhia Ayachi 17190c0076
KV state store refactoring and partitioning (#11510)
* state: port KV and Tombstone tables to new pattern

* go fmt'ed

* handle wildcards for tombstones

* Fix graveyard ent vs oss

* fix oss compilation error

* add partition to tombstones and kv state store indexes

* refactor to use `indexWithEnterpriseIndexable`

* partition kvs indexID table

* add `partitionedIndexEntryName` in oss for test purpose

* Apply suggestions from code review

Co-authored-by: Chris S. Kim <ckim@hashicorp.com>
Co-authored-by: R.B. Boyer <4903+rboyer@users.noreply.github.com>

* add `singleValueID` implementation assertions

* remove entmeta reference from oss

Co-authored-by: R.B. Boyer <4903+rboyer@users.noreply.github.com>

Co-authored-by: Daniel Nephin <dnephin@hashicorp.com>
Co-authored-by: Chris S. Kim <ckim@hashicorp.com>
Co-authored-by: R.B. Boyer <4903+rboyer@users.noreply.github.com>
2021-11-08 09:35:56 -05:00

100 lines
2.6 KiB
Go

package state
import (
"fmt"
"github.com/hashicorp/go-memdb"
"github.com/hashicorp/consul/agent/structs"
)
// Tombstone is the internal type used to track tombstones.
type Tombstone struct {
Key string
Index uint64
structs.EnterpriseMeta
}
func (t Tombstone) IDValue() string {
return t.Key
}
// Graveyard manages a set of tombstones.
type Graveyard struct {
// GC is when we create tombstones to track their time-to-live.
// The GC is consumed upstream to manage clearing of tombstones.
gc *TombstoneGC
}
// NewGraveyard returns a new graveyard.
func NewGraveyard(gc *TombstoneGC) *Graveyard {
return &Graveyard{gc: gc}
}
// InsertTxn adds a new tombstone.
func (g *Graveyard) InsertTxn(tx WriteTxn, key string, idx uint64, entMeta *structs.EnterpriseMeta) error {
stone := &Tombstone{
Key: key,
Index: idx,
}
if entMeta != nil {
stone.EnterpriseMeta = *entMeta
}
// Insert the tombstone.
if err := g.insertTombstoneWithTxn(tx, "tombstones", stone, false); err != nil {
return fmt.Errorf("failed inserting tombstone: %s", err)
}
// If GC is configured, then we hint that this index requires reaping.
if g.gc != nil {
tx.Defer(func() { g.gc.Hint(idx) })
}
return nil
}
// DumpTxn returns all the tombstones.
func (g *Graveyard) DumpTxn(tx ReadTxn) (memdb.ResultIterator, error) {
return tx.Get(tableTombstones, indexID)
}
// RestoreTxn is used when restoring from a snapshot. For general inserts, use
// InsertTxn.
func (g *Graveyard) RestoreTxn(tx WriteTxn, stone *Tombstone) error {
if err := g.insertTombstoneWithTxn(tx, "tombstones", stone, true); err != nil {
return fmt.Errorf("failed inserting tombstone: %s", err)
}
return nil
}
// ReapTxn cleans out all tombstones whose index values are less than or equal
// to the given idx. This prevents unbounded storage growth of the tombstones.
func (g *Graveyard) ReapTxn(tx WriteTxn, idx uint64) error {
// This does a full table scan since we currently can't index on a
// numeric value. Since this is all in-memory and done infrequently
// this pretty reasonable.
stones, err := tx.Get(tableTombstones, indexID)
if err != nil {
return fmt.Errorf("failed querying tombstones: %s", err)
}
// Find eligible tombstones.
var objs []interface{}
for stone := stones.Next(); stone != nil; stone = stones.Next() {
if stone.(*Tombstone).Index <= idx {
objs = append(objs, stone)
}
}
// Delete the tombstones in a separate loop so we don't trash the
// iterator.
for _, obj := range objs {
if err := tx.Delete("tombstones", obj); err != nil {
return fmt.Errorf("failed deleting tombstone: %s", err)
}
}
return nil
}