From b391ab91531c4d6da4521b0b5e0c46b0cfee8c16 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 31 Aug 2021 14:31:35 -0500 Subject: [PATCH] mutex sanity-check This implements a fairly straightforward sanity-check for mutexes, implemented as a bitmapfilter at the fragment level, and with higher levels combining results. There's two endpoints, an internal endpoint which only checks the local node's shards, and an external one which forwards requests (using the internal endpoint) to all the other nodes. The internal endpoint does not do key translation, the external one does. The transmission format is a probably-inefficient JSON blob, and returns data separated per-shard so we don't have as much merging work to do. This introduces a horrifying monstrosity function which tries to sneakily corrupt mutex fields and which has to be exported (EWWWWW) but which is only present in _test code (!??!! THIS WORKS WHY). Also one typo fix in unrelated code caused by not wanting to keep fighting with gofmt about this. --- api.go | 273 +++++++++++++++++++++++++++ api_test.go | 314 ++++++++++++++++++++++++++++++++ client.go | 5 + field.go | 19 +- field_internal_test.go | 33 ++++ fragment.go | 11 ++ http/client.go | 29 +++ http/handler.go | 56 +++++- roaring/filter.go | 69 +++++++ roaring/filter_internal_test.go | 48 ++++- view.go | 44 +++++ 11 files changed, 898 insertions(+), 3 deletions(-) diff --git a/api.go b/api.go index aa7cdc79a..ff9ce8dc3 100644 --- a/api.go +++ b/api.go @@ -2653,6 +2653,277 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 return nil } +func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) { + index := api.holder.Index(indexName) + if index == nil { + return nil, newNotFoundError(ErrIndexNotFound, indexName) + } + field := index.Field(fieldName) + if field == nil { + return nil, newNotFoundError(ErrFieldNotFound, fieldName) + } + return field.MutexCheck(ctx, qcx) +} + +// mergeIDLists merges a list of numeric IDs into another list, removing +// duplicates. +func mergeIDLists(dst []uint64, src []uint64) []uint64 { + dst = append(dst, src...) + sort.Slice(dst, func(i, j int) bool { + return dst[i] < dst[j] + }) + // dedup. + n := 0 + prev := dst[0] + for i := 0; i < len(dst); i++ { + if dst[i] != prev { + dst[n] = dst[i] + n++ + } + prev = dst[i] + } + return dst[:n] +} + +// mergeKeyLists merges a list of string IDs into another list, removing +// duplicates. +func mergeKeyLists(dst []string, src []string) []string { + dst = append(dst, src...) + sort.Slice(dst, func(i, j int) bool { + return dst[i] < dst[j] + }) + // dedup. + n := 0 + prev := dst[0] + for i := 0; i < len(dst); i++ { + if dst[i] != prev { + dst[n] = dst[i] + n++ + } + prev = dst[i] + } + return dst[:n] +} + +// MutexCheckNode checks for collisions in a given mutex field. The response is +// a map[shard]map[column]values, not translated. +func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) { + if err := api.validate(apiMutexCheck); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) +} + +// MutexCheck checks a named field for mutex violations, returning a +// map of record IDs to values for records that have multiple values in the +// field. The return will be one of: +// map[uint64][]uint64 // unkeyed index, unkeyed field +// map[uint64][]string // unkeyed index, keyed field +// map[string][]uint64 // keyed index, unkeyed field +// map[string][]string // keyed index, keyed field +func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (result interface{}, err error) { + if err = api.validate(apiMutexCheck); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + index, err := api.Index(ctx, indexName) + if err != nil { + return nil, err + } + field, err := api.Field(ctx, indexName, fieldName) + if err != nil { + return nil, err + } + if field.Type() != FieldTypeMutex { + return nil, errors.New("can only check mutex state for mutex fields") + } + // request data from other nodes as well + snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) + eg, _ := errgroup.WithContext(ctx) + myID := api.NodeID() + results := make([]map[uint64]map[uint64][]uint64, len(snap.Nodes)) + for i, node := range snap.Nodes { + i := i // loop variable shadowing is a war crime + if node.ID != myID { + node := node // loop variable shadowing again + eg.Go(func() (err error) { + results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName) + return err + }) + } else { + eg.Go(func() (err error) { + results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) + return err + }) + } + } + err = eg.Wait() + if err != nil { + return nil, err + } + // We now have a series of maps from shards to maps of record IDs to + // values. But wait! Either the field, or the index, might be using keys, + // and want those translated. So we have to translate those. We'll create + // some tables. + useIndexKeys := index.Keys() + useFieldKeys := field.Keys() + var indexKeys = map[uint64]string{} + var fieldKeys = map[uint64]string{} + var indexIDs []uint64 + var fieldIDs []uint64 + // We'll use the string "untranslated" as our default value and overwrite + // it with translations. We do check for missing translation values in + // our returns, but just in case, you know? + untranslated := "untranslated" + // We don't know which of four map types we want to be working with, + // but what we can do is make a function which works with that map type + // given the raw integer values, and is a closure with an already-created + // map which has already been stashed in `result`. Because maps are + // reference-y, this should actually work. + var process func(uint64, []uint64) + if useIndexKeys || useFieldKeys { + for _, nodeResults := range results { + for _, shardResults := range nodeResults { + for record, values := range shardResults { + if useIndexKeys { + if _, ok := indexKeys[record]; !ok { + indexKeys[record] = untranslated + indexIDs = append(indexIDs, record) + } + } + if useFieldKeys { + for _, value := range values { + if _, ok := fieldKeys[value]; !ok { + fieldKeys[value] = untranslated + fieldIDs = append(fieldIDs, value) + } + } + } + } + } + } + untranslatedKeys := 0 + // Obtain translation tables for the keys. + if useIndexKeys { + indexKeyList, err := api.cluster.translateIndexIDs(ctx, indexName, indexIDs) + if err != nil { + return nil, errors.Wrap(err, "translating index keys") + } + if len(indexKeyList) != len(indexIDs) { + return nil, fmt.Errorf("translating %d record IDs, got %d keys", len(indexIDs), len(indexKeyList)) + } + for i := range indexIDs { + if indexKeyList[i] != "" { + indexKeys[indexIDs[i]] = indexKeyList[i] + } else { + untranslatedKeys++ + } + } + } + if useFieldKeys { + fieldKeyList, err := api.cluster.translateFieldListIDs(field, fieldIDs) + if err != nil { + return nil, errors.Wrap(err, "translating index keys") + } + if len(fieldKeyList) != len(fieldIDs) { + return nil, fmt.Errorf("translating %d IDs, got %d keys", len(indexIDs), len(fieldKeyList)) + } + for i := range fieldIDs { + if fieldKeyList[i] != "" { + fieldKeys[fieldIDs[i]] = fieldKeyList[i] + } else { + untranslatedKeys++ + } + } + } + if untranslatedKeys > 0 { + api.server.logger.Warnf("translating mutex check results: %d key(s) untranslated", untranslatedKeys) + } + } + + // define the process functions. separated from above code just to make + // it easier to follow/compare them. + if useIndexKeys { + if useFieldKeys { + outMap := make(map[string][]string) + var valueKeys []string + result = outMap + process = func(recordID uint64, valueIDs []uint64) { + valueKeys = valueKeys[:0] + for _, id := range valueIDs { + valueKeys = append(valueKeys, fieldKeys[id]) + } + record := indexKeys[recordID] + if existing, ok := outMap[record]; ok { + outMap[record] = mergeKeyLists(existing, valueKeys) + } else { + // The append is so we can reuse this buffer safely, + // which matters if there's replication, because many + // cases won't need to copy the buffer, they'll just + // copy individual things from it. + outMap[record] = append([]string{}, valueKeys...) + } + } + } else { + outMap := make(map[string][]uint64) + result = outMap + process = func(recordID uint64, values []uint64) { + record := indexKeys[recordID] + if existing, ok := outMap[record]; ok { + outMap[record] = mergeIDLists(existing, values) + } else { + outMap[record] = values + } + } + } + } else { + if useFieldKeys { + outMap := make(map[uint64][]string) + var valueKeys []string + result = outMap + process = func(record uint64, valueIDs []uint64) { + valueKeys = valueKeys[:0] + for _, id := range valueIDs { + valueKeys = append(valueKeys, fieldKeys[id]) + } + if existing, ok := outMap[record]; ok { + outMap[record] = mergeKeyLists(existing, valueKeys) + } else { + // The append is so we can reuse this buffer safely, + // which matters if there's replication, because many + // cases won't need to copy the buffer, they'll just + // copy individual things from it. + outMap[record] = append([]string{}, valueKeys...) + } + } + } else { + outMap := make(map[uint64][]uint64) + result = outMap + process = func(record uint64, values []uint64) { + if existing, ok := outMap[record]; ok { + outMap[record] = mergeIDLists(existing, values) + } else { + outMap[record] = values + } + } + } + } + + for _, nodeResults := range results { + if len(nodeResults) == 0 { + continue + } + for _, v := range nodeResults { + if len(v) == 0 { + continue + } + for record, values := range v { + process(record, values) + } + } + } + return result, nil +} + type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` ReplicaN int `json:"replicaN"` @@ -2713,6 +2984,7 @@ const ( apiIDReset apiPartitionNodes apiIngestOperations + apiMutexCheck ) var methodsCommon = map[apiMethod]struct{}{ @@ -2781,4 +3053,5 @@ var methodsNormal = map[apiMethod]struct{}{ apiIDReset: {}, apiPartitionNodes: {}, apiIngestOperations: {}, + apiMutexCheck: {}, } diff --git a/api_test.go b/api_test.go index 7e34f5be4..bb2764aa2 100644 --- a/api_test.go +++ b/api_test.go @@ -30,6 +30,7 @@ import ( "github.com/molecula/featurebase/v2/boltdb" "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" + "github.com/molecula/featurebase/v2/shardwidth" "github.com/molecula/featurebase/v2/test" . "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck ) @@ -844,3 +845,316 @@ func TestAPI_IDAlloc(t *testing.T) { } }) } + +type mutexCheckIndex struct { + index *pilosa.Index + indexName string + createdAt int64 + fields map[bool]mutexCheckField +} + +type mutexCheckField struct { + fieldName string + field *pilosa.Field + createdAt int64 +} + +func TestAPI_MutexCheck(t *testing.T) { + c := test.MustRunCluster(t, 3) + defer c.Close() + + m0 := c.GetNode(0) + nodesByID := make(map[string]*test.Command, 3) + qcxsByID := make(map[string]*pilosa.Qcx, 3) + for i := 0; i < 3; i++ { + node := c.GetNode(i) + id := node.API.NodeID() + nodesByID[id] = node + } + + indexes := make(map[bool]mutexCheckIndex) + + ctx := context.Background() + for _, keyedIndex := range []bool{false, true} { + indexName := fmt.Sprintf("i%t", keyedIndex) + index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: keyedIndex, TrackExistence: true}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + if index.CreatedAt() == 0 { + t.Fatal("index createdAt is empty") + } + indexData := mutexCheckIndex{indexName: indexName, index: index, fields: make(map[bool]mutexCheckField), createdAt: index.CreatedAt()} + for _, keyedField := range []bool{false, true} { + fieldName := fmt.Sprintf("f%t", keyedField) + var field *pilosa.Field + if keyedField { + field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys()) + } else { + field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0)) + } + if err != nil { + t.Fatalf("creating field: %v", err) + } + if field.CreatedAt() == 0 { + t.Fatal("field createdAt is empty") + } + indexData.fields[keyedField] = mutexCheckField{fieldName: fieldName, field: field, createdAt: field.CreatedAt()} + } + indexes[keyedIndex] = indexData + } + + rowIDs := []uint64{0, 1, 2, 3} + colIDs := []uint64{0, 1, 2, 3} + rowKeysBase := []string{"v0", "v1", "v2", "v3"} + colKeysBase := []string{"c0", "c1", "c2", "c3"} + + const nShards = 10 + + // now, try the same thing for each combination of keyed/unkeyed. we + // share code between keyed/unkeyed fields, but for indexes, the logic + // is fundamentally different because we can't know shards in advance. + indexData := indexes[false] + for keyedField, fieldData := range indexData.fields { + t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) { + for id, node := range nodesByID { + qcxsByID[id] = node.API.Txf().NewQcx() + } + for shard := uint64(0); shard < nShards; shard++ { + // restore row/col ID values which can get altered by imports + for i := range rowIDs { + rowIDs[i] = uint64(i) + colIDs[i] = (shard << shardwidth.Exponent) + uint64(i) + (shard % 4) + } + req := &pilosa.ImportRequest{ + Index: indexData.indexName, + IndexCreatedAt: indexData.createdAt, + Field: fieldData.fieldName, + FieldCreatedAt: fieldData.createdAt, + Shard: shard, + ColumnIDs: colIDs, + } + if keyedField { + req.RowKeys = rowKeysBase + } else { + req.RowIDs = rowIDs + } + nodesForShard, err := m0.API.ShardNodes(ctx, indexData.indexName, shard) + if err != nil { + t.Fatalf("obtaining shard list: %v", err) + } + if len(nodesForShard) < 1 { + t.Fatalf("no nodes for shard %d", shard) + } + node := nodesByID[nodesForShard[0].ID] + if err := node.API.Import(ctx, qcxsByID[nodesForShard[0].ID], req); err != nil { + t.Fatalf("importing data: %v", err) + } + } + // and then we break the mutex and close the Qcxs + for id, node := range nodesByID { + field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName) + if err != nil { + t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err) + } + pilosa.CorruptAMutex(t, field, qcxsByID[id]) + err = qcxsByID[id].Finish() + if err != nil { + t.Fatalf("closing out transaction on node %s: %v", id, err) + } + } + qcx := m0.API.Txf().NewQcx() + defer qcx.Abort() + + results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName) + if err != nil { + t.Fatalf("checking mutexes: %v", err) + } + // first two shards of each group of 4 should have a collision in + // position 1 + expected := map[uint64]bool{ + (0 << shardwidth.Exponent) + 1: true, + (1 << shardwidth.Exponent) + 1: true, + (4 << shardwidth.Exponent) + 1: true, + (5 << shardwidth.Exponent) + 1: true, + (8 << shardwidth.Exponent) + 1: true, + (9 << shardwidth.Exponent) + 1: true, + } + if keyedField { + mapped, ok := results.(map[uint64][]string) + if !ok { + t.Fatalf("expected map[uint64][]string, got %T", results) + } + seen := 0 + for k, v := range mapped { + seen++ + if !expected[k] { + t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k) + } + if len(v) != 2 { + t.Fatalf("expected exactly two collisions") + } + } + if seen != len(expected) { + t.Fatalf("expected exactly %d records to have collisions", len(expected)) + } + } else { + mapped, ok := results.(map[uint64][]uint64) + if !ok { + t.Fatalf("expected map[uint64][]uint64, got %T", results) + } + seen := 0 + for k, v := range mapped { + seen++ + if !expected[k] { + t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k) + } + if len(v) != 2 { + t.Fatalf("expected exactly two collisions") + } + } + if seen != len(expected) { + t.Fatalf("expected exactly %d records to have collisions", len(expected)) + } + } + }) + } + indexData = indexes[true] + for keyedField, fieldData := range indexData.fields { + t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) { + for id, node := range nodesByID { + qcxsByID[id] = node.API.Txf().NewQcx() + } + req := &pilosa.ImportRequest{ + Index: indexData.indexName, + IndexCreatedAt: indexData.createdAt, + Field: fieldData.fieldName, + FieldCreatedAt: fieldData.createdAt, + Shard: 0, // ignored when using keys + } + rowKeys := make([]string, 0, len(rowKeysBase)*nShards) + colKeys := make([]string, 0, len(rowKeysBase)*nShards) + rowIDs = rowIDs[:0] + for shard := uint64(0); shard < nShards; shard++ { + for i := range rowKeysBase { + colKeys = append(colKeys, fmt.Sprintf("s%d-%s", shard, colKeysBase[i])) + if keyedField { + rowKeys = append(rowKeys, rowKeysBase[i]) + } else { + rowIDs = append(rowIDs, uint64(i)) + } + } + } + req.ColumnKeys = colKeys + if keyedField { + req.RowKeys = rowKeys + } else { + req.RowIDs = rowIDs + } + var id string + var node *test.Command + for id, node = range nodesByID { + break + } + if err := node.API.Import(ctx, qcxsByID[id], req); err != nil { + t.Fatalf("importing data: %v", err) + } + expected, err := node.API.FindIndexKeys(ctx, indexData.indexName, colKeys...) + if err != nil { + t.Fatalf("looking up index keys: %v", err) + } + for key, id := range expected { + // CorruptAMutex should only corrupt things in position 1 of their + // shards... + if id%(1< 0 { + return key.NeedData() + } + return key.RejectOne() +} + +func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) FilterResult { + value, basePos := uint64(key)>>rowExponent, uint64(key&keyMask)<<16 + containerCallback(data, func(u uint16) { + pos := basePos + uint64(u) + if b.first[pos] != ^uint64(0) { + b.extra[pos+b.base] = append(b.extra[pos+b.base], value) + } else { + b.first[pos] = value + } + }) + return key.MatchOne() +} + +// Report returns the set of duplicate values identified. +func (b *BitmapMutexDupFilter) Report() map[uint64][]uint64 { + // copy values into extra, and remove them from first, so calling + // Report() again won't cause double-appends. + for k, v := range b.extra { + kpos := k % (1 << shardwidth.Exponent) + if b.first[kpos] != ^uint64(0) { + v = append(v, 0) + // prepend so the lowest value goes at the beginning + copy(v[1:], v[:]) + v[0] = b.first[kpos] + b.first[kpos] = ^uint64(0) + b.extra[k] = v + } + } + return b.extra +} + // ApplyFilterToIterator is a simplistic implementation that applies a bitmap // filter to a ContainerIterator, returning an error if it encounters an error. // diff --git a/roaring/filter_internal_test.go b/roaring/filter_internal_test.go index 40abd7e1e..6de5cb7c6 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -347,5 +347,51 @@ func TestFilterWithRows(t *testing.T) { } }) } - +} + +func TestMutexDupFilter(t *testing.T) { + tests := []struct{ + pairs [][2]uint64 + expect map[uint64][]uint64 + }{ + { + pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}}, + expect: map[uint64][]uint64{0: {0, 1}}, + }, + { + pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}}, + expect: map[uint64][]uint64{0: {0, 1, 2}}, + }, + } + for num, test := range tests { + t.Run(fmt.Sprintf("case%d", num), func(t *testing.T) { + b := NewSliceBitmap() + for _, p := range test.pairs { + v := (p[1] << shardwidth.Exponent) | p[0] + b.DirectAdd(v) + } + dup := NewBitmapMutexDupFilter(0) + iter, _ := b.Containers.Iterator(0) + err := ApplyFilterToIterator(dup, iter) + if err != nil { + t.Fatalf("applying filter: %v", err) + } + expected := test.expect + got := dup.Report() + if len(expected) != len(got) { + t.Fatalf("expected %d entries in duplicate map, got %d", len(expected), len(got)) + } + for k, v := range expected { + gv := got[k] + if len(v) != len(gv) { + t.Fatalf("for id %d, expected %d (len %d), got %d (len %d)", k, v, len(v), gv, len(gv)) + } + for j := range v { + if gv[j] != v[j] { + t.Fatalf("for id %d, expected %d, got %d", k, v[j], gv[j]) + } + } + } + }) + } } diff --git a/view.go b/view.go index c38bd8572..e60105ceb 100644 --- a/view.go +++ b/view.go @@ -442,6 +442,50 @@ func (v *view) row(txOrig Tx, rowID uint64) (*Row, error) { } +// mutexCheck checks all available fragments for duplicate values. The return +// is map[column]map[shard][]values for collisions only. +func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) { + // We don't need the context, we just want the context-awareness on the error groups. + // It would be nice if the inner functions could use this too... + eg, _ := errgroup.WithContext(ctx) + throttle := make(chan struct{}, runtime.NumCPU()) + frags := v.allFragments() + results := make([]map[uint64][]uint64, len(frags)) + for i, frag := range frags { + // local copies for the goroutine to use + i, frag := i, frag + eg.Go(func() error { + // limit simultaneous parallel goroutines associated with this + throttle <- struct{}{} + defer func() { + <-throttle + }() + tx, finisher, err := qcx.GetTx(Txo{Index: v.idx, Shard: frag.shard}) + if err != nil { + return err + } + defer finisher(&err) + results[i], err = frag.mutexCheck(tx) + if err != nil { + return err + } + return nil + }) + } + err := eg.Wait() + if err != nil { + return nil, err + } + out := map[uint64]map[uint64][]uint64{} + for i, result := range results { + if len(result) == 0 { + continue + } + out[frags[i].shard] = result + } + return out, nil +} + // setBit sets a bit within the view. func (v *view) setBit(txOrig Tx, rowID, columnID uint64) (changed bool, err error) { shard := columnID / ShardWidth