diff --git a/index/scorch/merge.go b/index/scorch/merge.go index 8d10078a1..c3fc592da 100644 --- a/index/scorch/merge.go +++ b/index/scorch/merge.go @@ -60,17 +60,8 @@ func (s *Scorch) mergerLoop() { var lastEpochMergePlanned uint64 var ctrlMsg *mergerCtrl - mergePlannerOptions, err := s.parseMergePlannerOptions() - if err != nil { - s.fireAsyncError(NewScorchError( - merger, - fmt.Sprintf("mergerPlannerOptions json parsing err: %v", err), - ErrOptionsParse, - )) - return - } ctrlMsgDflt := &mergerCtrl{ctx: context.Background(), - options: mergePlannerOptions, + options: s.mergePlannerOptions, doneCh: nil} OUTER: @@ -234,14 +225,9 @@ func (s *Scorch) ForceMerge(ctx context.Context, return nil } -func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions, +func (s *Scorch) parseMergePlannerOptions(po *persisterOptions) (*mergeplan.MergePlanOptions, error) { mergePlannerOptions := mergeplan.DefaultMergePlanOptions - - po, err := s.parsePersisterOptions() - if err != nil { - return nil, err - } // by default use the MaxSizeInMemoryMergePerWorker from the persister option // as the FloorSegmentFileSize for the merge planner which would be the // first tier size in the planning. If the value is 0, then we don't use the @@ -569,6 +555,7 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot, // is updated - which is valid, since the snapshot updated in bolt is // cleaned up only if its zero ref'd (MB-66163 for more details) s.markIneligibleForRemoval(filename) + newMergedSegmentFileNames[flushID] = filename // the newly merged segment is already flushed out to disk, just needs // to be opened using mmap. @@ -582,7 +569,6 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot, } newDocIDsSet[flushID] = newDocIDs newMergedSegmentIDs[flushID] = newSegmentID - newMergedSegmentFileNames[flushID] = filename newMergedSegments[flushID], err = s.segPlugin.OpenUsing(path, s.segmentConfig) if err != nil { em.Lock() diff --git a/index/scorch/persister.go b/index/scorch/persister.go index b38cdad85..99a9189bf 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -83,6 +83,31 @@ var ( // maximum size of data that a single worker is allowed to perform the in-memory // merge operation. DefaultMaxSizeInMemoryMergePerWorker int = 0 + + // NumSnapshotsToKeep represents how many recent, old snapshots to + // keep around per Scorch instance. Useful for apps that require + // rollback'ability. + NumSnapshotsToKeep int = 1 + + // RollbackSamplingInterval controls how far back we are looking + // in the history to get the rollback points. + // If we have to keep N snapshots ( = NumSnapshotsToKeep ), then upon + // including the very latest snapshot there should be N snapshots. + // If the timestamp of the latest snapshot is T, and the + // RollbackSamplingInterval is D, then the very last one would be T - (N - 1) * D + // A value of 0 means we will not sample, and will keep the very latest N snapshots. + // Example: + // - For N = 3, and the below timing diagram: + // a b c d + // |-----|-----|-----| --> time + // | D | D | D | + // X Y Z T + // - The rollback points would be b, c, and d. a would be too old. + RollbackSamplingInterval time.Duration = 0 + + // Controls what portion of the earlier rollback points to retain during + // a infrequent/sparse mutation scenario + RollbackRetentionFactor float64 = 0.5 ) type persisterOptions struct { @@ -133,16 +158,6 @@ func (s *Scorch) persisterLoop() { var unpersistedCallbacks []index.BatchCallback - po, err := s.parsePersisterOptions() - if err != nil { - s.fireAsyncError(NewScorchError( - persister, - fmt.Sprintf("persisterOptions json parsing err: %v", err), - ErrOptionsParse, - )) - return - } - OUTER: for { atomic.AddUint64(&s.stats.TotPersistLoopBeg, 1) @@ -158,7 +173,7 @@ OUTER: lastMergedEpoch = ew.epoch } lastMergedEpoch, persistWatchers = s.pausePersisterForMergerCatchUp(lastPersistedEpoch, - lastMergedEpoch, persistWatchers, po) + lastMergedEpoch, persistWatchers, s.persisterOptions) var ourSnapshot *IndexSnapshot var ourPersisted []chan error @@ -181,7 +196,7 @@ OUTER: if ourSnapshot != nil { startTime := time.Now() - err := s.persistSnapshot(ourSnapshot, po) + err := s.persistSnapshot(ourSnapshot, s.persisterOptions) for _, ch := range ourPersisted { if err != nil { ch <- err @@ -833,11 +848,8 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) { case s.persists <- persist: } - select { - case <-s.closeCh: - return segment.ErrClosed - case <-persist.applied: - } + // blockingly wait until the persist has been applied + <-persist.applied } err = tx.Commit() @@ -931,7 +943,9 @@ func (s *Scorch) loadFromBolt() error { if err != nil { return err } - + // we initialize the checkPoints from all the persisted snapshots, + // because if we have state to load, the previous process was already + // maintaining checkpoints, whatever survives in the bucket now is that state. persistedSnapshots, err := s.rootBoltSnapshotMetaData() if err != nil { return err @@ -1286,97 +1300,68 @@ func (s *Scorch) removeOldData() { } } -// NumSnapshotsToKeep represents how many recent, old snapshots to -// keep around per Scorch instance. Useful for apps that require -// rollback'ability. -var NumSnapshotsToKeep = 1 - -// RollbackSamplingInterval controls how far back we are looking -// in the history to get the rollback points. -// For example, a value of 10 minutes ensures that the -// protected snapshots (NumSnapshotsToKeep = 3) are: -// -// the very latest snapshot(ie the current one), -// the snapshot that was persisted 10 minutes before the current one, -// the snapshot that was persisted 20 minutes before the current one -// -// By default however, the timeseries way of protecting snapshots is -// disabled, and we protect the latest three contiguous snapshots -var RollbackSamplingInterval = 0 * time.Minute - -// Controls what portion of the earlier rollback points to retain during -// a infrequent/sparse mutation scenario -var RollbackRetentionFactor = float64(0.5) - func getTimeSeriesSnapshots(maxDataPoints int, interval time.Duration, - snapshots []*snapshotMetaData, -) (int, map[uint64]time.Time) { - if interval == 0 { - return len(snapshots), map[uint64]time.Time{} + snapshots []*snapshotMetaData) map[uint64]time.Time { + if interval == 0 || len(snapshots) == 0 || maxDataPoints <= 0 { + return map[uint64]time.Time{} } // the map containing the time series snapshots, i.e the timeseries of snapshots // each of which is separated by rollbackSamplingInterval - rv := make(map[uint64]time.Time) + rv := make(map[uint64]time.Time, maxDataPoints) // the last point in the "time series", i.e. the timeseries of snapshots // each of which is separated by rollbackSamplingInterval ptr := len(snapshots) - 1 rv[snapshots[ptr].epoch] = snapshots[ptr].timeStamp numSnapshotsProtected := 1 - // traverse the list in reverse order, older timestamps to newer ones. - for i := ptr - 1; i >= 0; i-- { - // If we find a timeStamp which is the next datapoint in our - // timeseries of snapshots, and newer by RollbackSamplingInterval duration - // (comparison in terms of minutes), which is the interval of our time - // series. In this case, add the epoch rv - if snapshots[i].timeStamp.Sub(snapshots[ptr].timeStamp).Minutes() > - interval.Minutes() { - if _, ok := rv[snapshots[i+1].epoch]; !ok { - rv[snapshots[i+1].epoch] = snapshots[i+1].timeStamp - ptr = i + 1 - numSnapshotsProtected++ + for i := ptr - 1; i >= 0 && numSnapshotsProtected < maxDataPoints; i-- { + sinceLast := snapshots[i].timeStamp.Sub(snapshots[ptr].timeStamp) + if sinceLast >= interval { + // capture the snapshot at the interval boundary: the exact match + // if there is one, otherwise the older neighbour (i+1), which was + // the last snapshot seen before the interval was crossed + idx := i + if sinceLast > interval { + idx = i + 1 } - } else if snapshots[i].timeStamp.Sub(snapshots[ptr].timeStamp).Minutes() == - interval.Minutes() { - if _, ok := rv[snapshots[i].epoch]; !ok { - rv[snapshots[i].epoch] = snapshots[i].timeStamp - ptr = i + if _, ok := rv[snapshots[idx].epoch]; !ok { + rv[snapshots[idx].epoch] = snapshots[idx].timeStamp + ptr = idx numSnapshotsProtected++ } } - - if numSnapshotsProtected >= maxDataPoints { - break - } } - return ptr, rv + return rv } // getProtectedSnapshots aims to fetch the epochs keep based on a timestamp basis. // It tries to get NumSnapshotsToKeep snapshots, each of which are separated // by a time duration of RollbackSamplingInterval. -func getProtectedSnapshots(rollbackSamplingInterval time.Duration, - numSnapshotsToKeep int, - persistedSnapshots []*snapshotMetaData, -) map[uint64]time.Time { +func (s *Scorch) getProtectedSnapshots(liveSnapshots []*snapshotMetaData) map[uint64]time.Time { // keep numSnapshotsToKeep - 1 worth of time series snapshots, because we always // must preserve the very latest snapshot in bolt as well to avoid accidental // deletes of bolt entries and cleanups by the purger code. - lastPoint, protectedEpochs := getTimeSeriesSnapshots(numSnapshotsToKeep-1, - rollbackSamplingInterval, persistedSnapshots) - if len(protectedEpochs) < numSnapshotsToKeep { - numSnapshotsNeeded := numSnapshotsToKeep - len(protectedEpochs) - // we protected the contiguous snapshots from the last point in time series - for i := 0; i < numSnapshotsNeeded && i < lastPoint; i++ { - protectedEpochs[persistedSnapshots[i].epoch] = persistedSnapshots[i].timeStamp + protectedEpochs := getTimeSeriesSnapshots(s.numSnapshotsToKeep-1, + s.rollbackSamplingInterval, liveSnapshots) + numProtected := len(protectedEpochs) + // always protect the latest snapshot + latestSnapshot := liveSnapshots[0] + if _, ok := protectedEpochs[latestSnapshot.epoch]; !ok { + protectedEpochs[latestSnapshot.epoch] = latestSnapshot.timeStamp + numProtected++ + } + // if we still have not protected enough snapshots, then protect the next most recent snapshots + for i := 1; i < len(liveSnapshots) && numProtected < s.numSnapshotsToKeep; i++ { + if _, ok := protectedEpochs[liveSnapshots[i].epoch]; !ok { + protectedEpochs[liveSnapshots[i].epoch] = liveSnapshots[i].timeStamp + numProtected++ } } - return protectedEpochs } func newCheckPoints(snapshots map[uint64]time.Time) []*snapshotMetaData { - rv := make([]*snapshotMetaData, 0) + rv := make([]*snapshotMetaData, 0, len(snapshots)) keys := make([]uint64, 0, len(snapshots)) for k := range snapshots { @@ -1384,7 +1369,7 @@ func newCheckPoints(snapshots map[uint64]time.Time) []*snapshotMetaData { } sort.SliceStable(keys, func(i, j int) bool { - return snapshots[keys[i]].Sub(snapshots[keys[j]]) > 0 + return snapshots[keys[i]].After(snapshots[keys[j]]) }) for _, key := range keys { @@ -1397,21 +1382,20 @@ func newCheckPoints(snapshots map[uint64]time.Time) []*snapshotMetaData { return rv } -// Removes enough snapshots from the rootBolt so that the -// s.eligibleForRemoval stays under the NumSnapshotsToKeep policy. func (s *Scorch) removeOldBoltSnapshots() (numRemoved int, err error) { - persistedSnapshots, err := s.rootBoltSnapshotMetaData() + // first get the set of live snapshots + liveSnapshots, err := s.getLiveSnapshots() if err != nil { return 0, err } - if len(persistedSnapshots) <= s.numSnapshotsToKeep { - // we need to keep everything + // if no live snapshots, then nothing to do + if len(liveSnapshots) == 0 { return 0, nil } - protectedSnapshots := getProtectedSnapshots(s.rollbackSamplingInterval, - s.numSnapshotsToKeep, persistedSnapshots) + // then get the set of protected snapshots, which are the ones we want to keep + protectedSnapshots := s.getProtectedSnapshots(liveSnapshots) var epochsToRemove []uint64 var newEligible []uint64 @@ -1520,26 +1504,19 @@ func (s *Scorch) removeOldZapFiles() error { return nil } -// In sparse mutation scenario, it can so happen that all protected -// snapshots are older than the numSnapshotsToKeep * rollbackSamplingInterval -// duration. This results in all of them being purged from the boltDB -// and the next iteration of the removeOldData() would end up protecting -// latest contiguous snapshot which is a poor pattern in the rollback checkpoints. -// Hence we try to retain at most retentionFactor portion worth of old snapshots -// in such a scenario using the following function -func getBoundaryCheckPoint(retentionFactor float64, - checkPoints []*snapshotMetaData, timeStamp time.Time, -) time.Time { - if checkPoints != nil { - boundary := checkPoints[int(math.Floor(float64(len(checkPoints))* - retentionFactor))] - if timeStamp.Sub(boundary.timeStamp) > 0 { - // return the extended boundary which will dictate the older snapshots - // to be retained - return boundary.timeStamp - } +func (s *Scorch) getBoundaryCheckPoint(timeStamp time.Time) time.Time { + numCheckpoints := float64(len(s.checkPoints)) + if numCheckpoints == 0 { + return timeStamp + } + checkpointIdx := int(math.Floor(numCheckpoints * s.rollbackRetentionFactor)) + if checkpointIdx >= len(s.checkPoints) { + checkpointIdx = len(s.checkPoints) - 1 + } + boundary := s.checkPoints[checkpointIdx] + if boundary.timeStamp.Before(timeStamp) { + return boundary.timeStamp } - return timeStamp } @@ -1548,35 +1525,21 @@ type snapshotMetaData struct { timeStamp time.Time } +// returns all the snapshots that are currently persisted +// in the rootBolt, sorted by latest to oldest epoch. func (s *Scorch) rootBoltSnapshotMetaData() ([]*snapshotMetaData, error) { var rv []*snapshotMetaData - currTime := time.Now() - // including the very latest snapshot there should be n snapshots, so the - // very last one would be tc - (n-1) * d - // for eg for n = 3 the checkpoints preserved should be tc, tc - d, tc - 2d - expirationDuration := time.Duration(s.numSnapshotsToKeep-1) * s.rollbackSamplingInterval - err := s.rootBolt.View(func(tx *util.BoltTxImpl) error { snapshots := tx.Bucket(util.BoltSnapshotsBucket) if snapshots == nil { return nil } sc := snapshots.Cursor() - var found bool - // traversal order - latest -> oldest epoch for sk, _ := sc.Last(); sk != nil; sk, _ = sc.Prev() { _, snapshotEpoch, err := decodeUvarintAscending(sk) if err != nil { continue } - - if expirationDuration == 0 { - rv = append(rv, &snapshotMetaData{ - epoch: snapshotEpoch, - }) - continue - } - snapshot := snapshots.GetBucket(sk) if snapshot == nil { continue @@ -1594,32 +1557,71 @@ func (s *Scorch) rootBoltSnapshotMetaData() ([]*snapshotMetaData, error) { if err != nil { continue } - // Don't keep snapshots older than - // expiration duration (numSnapshotsToKeep * - // rollbackSamplingInterval, by default) - if currTime.Sub(timeStamp) <= expirationDuration { - rv = append(rv, &snapshotMetaData{ - epoch: snapshotEpoch, - timeStamp: timeStamp, - }) - } else { - if !found { - found = true - boundary := getBoundaryCheckPoint(s.rollbackRetentionFactor, - s.checkPoints, timeStamp) - expirationDuration = currTime.Sub(boundary) - continue - } - k := encodeUvarintAscending(nil, snapshotEpoch) - err = snapshots.DeleteBucket(k) - if err == bolt.ErrBucketNotFound { - err = nil - } + meta := &snapshotMetaData{ + epoch: snapshotEpoch, + timeStamp: timeStamp, } + rv = append(rv, meta) } return nil }) - return rv, err + if err != nil { + return nil, err + } + return rv, nil +} + +func (s *Scorch) getLiveSnapshots() ([]*snapshotMetaData, error) { + // get all the snapshots that are currently persisted + meta, err := s.rootBoltSnapshotMetaData() + if err != nil { + return nil, err + } + // if none persisted, then nothing to do + if len(meta) == 0 { + return nil, nil + } + // check if we have a rollback sampling interval, if not, + // then we will just return the latest numSnapshotsToKeep snapshots + if s.rollbackSamplingInterval <= 0 { + if len(meta) <= s.numSnapshotsToKeep { + return meta, nil + } + return meta[:s.numSnapshotsToKeep], nil + } + // we consider a snapshot to be live if: + // 1. it is the latest snapshot + // 2. it is within our expiration duration + var liveSnapshots []*snapshotMetaData + // always keep the latest snapshot + liveSnapshots = append(liveSnapshots, meta[0]) + extraSnapshots := s.numSnapshotsToKeep - 1 + if extraSnapshots <= 0 { + return liveSnapshots, nil + } + // if we need extra snapshots, compute an expiration duration + // beyond which we will not consider snapshots to be live + currTime := time.Now() + expirationDuration := time.Duration(extraSnapshots) * s.rollbackSamplingInterval + cutoffTime := currTime.Add(-expirationDuration) + // if we have previous checkpoints, then we can extend + // the cutoff time based on the boundary checkpoint + for _, snapshot := range meta[1:] { + if snapshot.timeStamp.Before(cutoffTime) { + boundary := s.getBoundaryCheckPoint(snapshot.timeStamp) + if boundary.Before(snapshot.timeStamp) { + cutoffTime = boundary + } + break + } + } + // add all snapshots that are newer than the cutoff time + for _, snapshot := range meta[1:] { + if !snapshot.timeStamp.Before(cutoffTime) { + liveSnapshots = append(liveSnapshots, snapshot) + } + } + return liveSnapshots, nil } func (s *Scorch) RootBoltSnapshotEpochs() ([]uint64, error) { diff --git a/index/scorch/rollback_test.go b/index/scorch/rollback_test.go index db514c835..d111e6549 100644 --- a/index/scorch/rollback_test.go +++ b/index/scorch/rollback_test.go @@ -15,15 +15,19 @@ package scorch import ( - "fmt" "io" + "math/rand" "os" "path/filepath" + "strconv" + "sync/atomic" "testing" "time" "github.com/blevesearch/bleve/v2/document" + "github.com/blevesearch/bleve/v2/util" index "github.com/blevesearch/bleve_index_api" + segment "github.com/blevesearch/scorch_segment_api/v2" ) func TestIndexRollback(t *testing.T) { @@ -252,12 +256,40 @@ func TestIndexRollback(t *testing.T) { } } +func openTestBolt(t *testing.T, name string) *util.RootBoltImpl { + t.Helper() + cfg := CreateConfig(name) + if err := InitTest(cfg); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := DestroyTest(cfg); err != nil { + t.Error(err) + } + }) + path := cfg["path"].(string) + if err := os.MkdirAll(path, 0o700); err != nil { + t.Fatal(err) + } + rootBolt, err := util.OpenBolt(filepath.Join(path, "root.bolt"), 0o600, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := rootBolt.Close(); err != nil { + t.Error(err) + } + }) + return rootBolt +} + func TestGetProtectedSnapshots(t *testing.T) { - origRollbackSamplingInterval := RollbackSamplingInterval - defer func() { - RollbackSamplingInterval = origRollbackSamplingInterval - }() - RollbackSamplingInterval = 10 * time.Minute + rootBolt := openTestBolt(t, "TestGetProtectedSnapshots") + interval := 10 * time.Minute + s := &Scorch{ + rootBolt: rootBolt, + rollbackSamplingInterval: interval, + } currentTimeStamp := time.Now() tests := []struct { title string @@ -270,11 +302,11 @@ func TestGetProtectedSnapshots(t *testing.T) { title: "epochs that have exact timestamps as per expectation for protecting", metaData: []*snapshotMetaData{ {epoch: 100, timeStamp: currentTimeStamp}, - {epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))}, - {epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))}, - {epoch: 50, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval))}, - {epoch: 35, timeStamp: currentTimeStamp.Add(-(6 * RollbackSamplingInterval / 5))}, - {epoch: 10, timeStamp: currentTimeStamp.Add(-(2 * RollbackSamplingInterval))}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(interval / 6))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(interval))}, + {epoch: 35, timeStamp: currentTimeStamp.Add(-(6 * interval / 5))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(2 * interval))}, }, numSnapshotsToKeep: 3, expCount: 3, @@ -284,9 +316,9 @@ func TestGetProtectedSnapshots(t *testing.T) { title: "epochs that have exact timestamps as per expectation for protecting", metaData: []*snapshotMetaData{ {epoch: 100, timeStamp: currentTimeStamp}, - {epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))}, - {epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))}, - {epoch: 50, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval))}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(interval / 6))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(interval))}, }, numSnapshotsToKeep: 2, expCount: 2, @@ -297,26 +329,26 @@ func TestGetProtectedSnapshots(t *testing.T) { "always retain the latest one", metaData: []*snapshotMetaData{ {epoch: 100, timeStamp: currentTimeStamp}, - {epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))}, - {epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))}, - {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * RollbackSamplingInterval / 4))}, - {epoch: 35, timeStamp: currentTimeStamp.Add(-(6 * RollbackSamplingInterval / 5))}, - {epoch: 10, timeStamp: currentTimeStamp.Add(-(2 * RollbackSamplingInterval))}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(interval / 6))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * interval / 4))}, + {epoch: 35, timeStamp: currentTimeStamp.Add(-(6 * interval / 5))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(2 * interval))}, }, numSnapshotsToKeep: 3, expCount: 3, expEpochs: []uint64{100, 35, 10}, }, { - title: "protecting epochs when we don't have enough snapshots with RollbackSamplingInterval" + + title: "protecting epochs when we don't have enough snapshots with rollbackSamplingInterval" + " separated timestamps", metaData: []*snapshotMetaData{ {epoch: 100, timeStamp: currentTimeStamp}, - {epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))}, - {epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))}, - {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * RollbackSamplingInterval / 4))}, - {epoch: 35, timeStamp: currentTimeStamp.Add(-(5 * RollbackSamplingInterval / 6))}, - {epoch: 10, timeStamp: currentTimeStamp.Add(-(7 * RollbackSamplingInterval / 8))}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(interval / 6))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * interval / 4))}, + {epoch: 35, timeStamp: currentTimeStamp.Add(-(5 * interval / 6))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(7 * interval / 8))}, }, numSnapshotsToKeep: 4, expCount: 4, @@ -324,224 +356,349 @@ func TestGetProtectedSnapshots(t *testing.T) { }, { title: "epochs of which some are approximated to the expected timestamps, and" + - " we don't have enough snapshots with RollbackSamplingInterval separated timestamps", + " we don't have enough snapshots with rollbackSamplingInterval separated timestamps", metaData: []*snapshotMetaData{ {epoch: 100, timeStamp: currentTimeStamp}, - {epoch: 99, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 12))}, - {epoch: 88, timeStamp: currentTimeStamp.Add(-(RollbackSamplingInterval / 6))}, - {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * RollbackSamplingInterval / 4))}, - {epoch: 35, timeStamp: currentTimeStamp.Add(-(8 * RollbackSamplingInterval / 7))}, - {epoch: 10, timeStamp: currentTimeStamp.Add(-(6 * RollbackSamplingInterval / 5))}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(interval / 6))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(3 * interval / 4))}, + {epoch: 35, timeStamp: currentTimeStamp.Add(-(8 * interval / 7))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(6 * interval / 5))}, }, numSnapshotsToKeep: 3, expCount: 3, expEpochs: []uint64{100, 50, 10}, }, + { + title: "sparse time series with a recent snapshot must still retain " + + "numSnapshotsToKeep by filling the skipped middle snapshots", + metaData: []*snapshotMetaData{ + {epoch: 100, timeStamp: currentTimeStamp}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(6 * interval / 10))}, + {epoch: 88, timeStamp: currentTimeStamp.Add(-(7 * interval / 10))}, + {epoch: 77, timeStamp: currentTimeStamp.Add(-(8 * interval / 10))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(15 * interval / 10))}, + }, + numSnapshotsToKeep: 4, + expCount: 4, + expEpochs: []uint64{100, 99, 88, 10}, + }, + { + title: "latest snapshot exactly one interval from the oldest must " + + "still retain numSnapshotsToKeep", + metaData: []*snapshotMetaData{ + {epoch: 100, timeStamp: currentTimeStamp}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 2))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(interval))}, + }, + numSnapshotsToKeep: 3, + expCount: 3, + expEpochs: []uint64{100, 99, 10}, + }, + { + title: "numSnapshotsToKeep=2 with a boundary-spaced older snapshot must " + + "protect exactly the latest and the oldest", + metaData: []*snapshotMetaData{ + {epoch: 100, timeStamp: currentTimeStamp}, + {epoch: 99, timeStamp: currentTimeStamp.Add(-(interval / 12))}, + {epoch: 50, timeStamp: currentTimeStamp.Add(-(2 * interval))}, + {epoch: 10, timeStamp: currentTimeStamp.Add(-(3 * interval))}, + }, + numSnapshotsToKeep: 2, + expCount: 2, + expEpochs: []uint64{100, 10}, + }, } - for i, test := range tests { - protectedEpochs := getProtectedSnapshots(RollbackSamplingInterval, - test.numSnapshotsToKeep, test.metaData) + s.numSnapshotsToKeep = test.numSnapshotsToKeep + protectedEpochs := s.getProtectedSnapshots(test.metaData) if len(protectedEpochs) != test.expCount { - t.Errorf("%d test: %s, getProtectedSnapshots expected to return %d "+ + t.Errorf("test %d: %s, getProtectedSnapshots expected to return %d "+ "snapshots, but got: %d", i, test.title, test.expCount, len(protectedEpochs)) } for _, e := range test.expEpochs { if _, found := protectedEpochs[e]; !found { - t.Errorf("%d test: %s, %d epoch expected to be protected, "+ + t.Errorf("test %d: %s, %d epoch expected to be protected, "+ "but missing from protected list: %v", i, test.title, e, protectedEpochs) } } } } -func indexDummyData(t *testing.T, scorchi *Scorch, i int) { - // create a batch, insert 2 new documents - batch := index.NewBatch() - doc := document.NewDocument(fmt.Sprintf("%d", i)) - doc.AddField(document.NewTextField("name", []uint64{}, []byte("test1"))) - batch.Update(doc) - doc = document.NewDocument(fmt.Sprintf("%d", i+1)) - doc.AddField(document.NewTextField("name", []uint64{}, []byte("test2"))) - batch.Update(doc) - - err := scorchi.Batch(batch) +// updateRoot creates a new snapshot, updates the root +// to point to it, and returns the new snapshot. +// if empty is true, then the new snapshot will be a delete-only +// snapshot, otherwise it will contain a single document. +func updateRoot(t *testing.T, s *Scorch, empty bool) *IndexSnapshot { + t.Helper() + var seg segment.Segment + var ids []string + var err error + if !empty { + doc := genDoc(t) + docs := []index.Document{doc} + ids = []string{doc.ID()} + seg, _, err = s.segPlugin.New(docs) + if err != nil { + t.Fatalf("failed to create segment: %v", err) + } + } else { + seg = nil + ids = []string{strconv.Itoa(rand.Intn(1000000))} + } + intro := &segmentIntroduction{ + id: atomic.AddUint64(&s.nextSegmentID, 1), + data: seg, + ids: ids, + applied: make(chan error), + } + go func(intro *segmentIntroduction) { + s.introduceSegment(intro) + }(intro) + err = <-intro.applied if err != nil { - t.Fatal(err) + t.Fatalf("failed to apply segment introduction: %v", err) } + return s.root } -type testFSDirector string - -func (f testFSDirector) GetWriter(filePath string) (io.WriteCloser, - error) { - dir, file := filepath.Split(filePath) - if dir != "" { - err := os.MkdirAll(filepath.Join(string(f), dir), os.ModePerm) - if err != nil { - return nil, err +// persistToBolt persists the given snapshot to bolt, +// sets the timestamp for the snapshot in the bolt metadata and +// persists any unpersisted segments to the directory. +func persistToBolt(t *testing.T, snapshot *IndexSnapshot, s *Scorch, timeStamp time.Time) error { + t.Helper() + tx, err := s.rootBolt.Begin(true) + if err != nil { + return err + } + _, _, err = prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin, nil) + if err != nil { + _ = tx.Rollback() + return err + } + snapshotsBucket, err := tx.CreateBucketIfNotExists(util.BoltSnapshotsBucket) + if err != nil { + _ = tx.Rollback() + return err + } + newSnapshotKey := encodeUvarintAscending(nil, snapshot.epoch) + snapshotBucket, err := snapshotsBucket.CreateBucketIfNotExists(newSnapshotKey) + if err != nil { + _ = tx.Rollback() + return err + } + metaBucket, err := snapshotBucket.CreateBucketIfNotExists(util.BoltMetaDataKey) + if err != nil { + _ = tx.Rollback() + return err + } + timeStampBinary, err := timeStamp.MarshalText() + if err != nil { + _ = tx.Rollback() + return err + } + err = metaBucket.Put(util.BoltMetaDataTimeStamp, timeStampBinary, nil) + if err != nil { + _ = tx.Rollback() + return err + } + for _, segmentSnapshot := range snapshot.segment { + if seg, ok := segmentSnapshot.segment.(segment.UnpersistedSegment); ok { + filename := zapFileName(segmentSnapshot.id) + path := filepath.Join(s.path, filename) + err := persistToDirectory(seg, nil, path) + if err != nil { + _ = tx.Rollback() + return err + } + persistedSeg, err := s.segPlugin.OpenUsing(path, s.segmentConfig) + if err != nil { + _ = tx.Rollback() + return err + } + segmentSnapshot.segment = persistedSeg } } - - return os.OpenFile(filepath.Join(string(f), dir, file), - os.O_RDWR|os.O_CREATE, 0600) + err = tx.Commit() + if err != nil { + _ = tx.Rollback() + return err + } + err = s.rootBolt.Sync() + if err != nil { + _ = tx.Rollback() + return err + } + return nil } func TestLatestSnapshotProtected(t *testing.T) { cfg := CreateConfig("TestLatestSnapshotProtected") - numSnapshotsToKeepOrig := NumSnapshotsToKeep - NumSnapshotsToKeep = 3 - rollbackSamplingIntervalOrig := RollbackSamplingInterval - RollbackSamplingInterval = 10 * time.Second - err := InitTest(cfg) if err != nil { t.Fatal(err) } - defer func() { - NumSnapshotsToKeep = numSnapshotsToKeepOrig - RollbackSamplingInterval = rollbackSamplingIntervalOrig - err := DestroyTest(cfg) - if err != nil { - t.Log(err) - } - }() - - // disable merger and purger - RegistryEventCallbacks["test"] = func(e Event) bool { - if e.Kind == EventKindPreMergeCheck || e.Kind == EventKindPurgerCheck { - return false - } - return true + defer func() { _ = DestroyTest(cfg) }() + dirPath := cfg["path"].(string) + if err = os.Mkdir(dirPath, 0o755); err != nil { + t.Fatal(err) } - cfg["eventCallbackName"] = "test" analysisQueue := index.NewAnalysisQueue(1) idx, err := NewScorch(Name, cfg, analysisQueue) if err != nil { - t.Fatal(err) + t.Fatalf("failed to create Scorch: %v", err) } - - defer func() { - err := idx.Close() - if err != nil { - t.Fatal(err) - } - }() - - scorchi, ok := idx.(*Scorch) + s, ok := idx.(*Scorch) if !ok { - t.Fatalf("Not a scorch index?") + t.Fatalf("expected *Scorch, got %T", s) } - - err = scorchi.Open() + s.numSnapshotsToKeep = 3 + s.rollbackSamplingInterval = 10 * time.Second + s.path = dirPath + err = s.openBolt() if err != nil { - t.Fatal(err) + t.Fatalf("failed to open bolt: %v", err) } - // replicate the following scenario of persistence of snapshots - // tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5 + // tc - 9d/5, tc - 6d/5, tc - 3d/4, tc - d/6, tc - d/12, tc // approximate timestamps where there's a chance that the latest snapshot - // might not fit into the time-series - indexDummyData(t, scorchi, 1) - persistedSnapshots, err := scorchi.rootBoltSnapshotMetaData() + // might not fit into the time-series and get purged along with its segment + // files. + const ( + numSnapshots = 6 + ) + curTime := time.Now() + d := s.rollbackSamplingInterval + // old --> new + timeStamps := []time.Time{ + curTime.Add(-(9 * d / 5)), + curTime.Add(-(6 * d / 5)), + curTime.Add(-(3 * d / 4)), + curTime.Add(-(d / 6)), + curTime.Add(-(d / 12)), + curTime, + } + snapshots := make([]*IndexSnapshot, numSnapshots) + for i := 0; i < numSnapshots; i++ { + snapshots[i] = updateRoot(t, s, false) + err = persistToBolt(t, snapshots[i], s, timeStamps[i]) + if err != nil { + t.Fatalf("failed to persist snapshot %d: %v", i, err) + } + } + meta, err := s.getLiveSnapshots() if err != nil { - t.Fatal(err) + t.Fatalf("failed to get live snapshots: %v", err) } - - if len(persistedSnapshots) != 1 { - t.Fatalf("expected 1 persisted snapshot, got %d", len(persistedSnapshots)) + if len(meta) != numSnapshots { + t.Fatalf("expected %d live snapshots, got %d", numSnapshots, len(meta)) } - time.Sleep(4 * RollbackSamplingInterval / 5) - indexDummyData(t, scorchi, 3) - time.Sleep(9 * RollbackSamplingInterval / 20) - indexDummyData(t, scorchi, 5) - time.Sleep(7 * RollbackSamplingInterval / 12) - indexDummyData(t, scorchi, 7) - time.Sleep(1 * RollbackSamplingInterval / 12) - indexDummyData(t, scorchi, 9) - - persistedSnapshots, err = scorchi.rootBoltSnapshotMetaData() + for i, m := range meta { + expectIdx := numSnapshots - i - 1 + expectedEpoch := snapshots[expectIdx].epoch + if m.epoch != expectedEpoch { + t.Fatalf("expected epoch %d, got %d", expectedEpoch, m.epoch) + } + expectedTimeStamp := timeStamps[expectIdx] + if !m.timeStamp.Equal(expectedTimeStamp) { + t.Fatalf("expected timestamp %v, got %v", expectedTimeStamp, m.timeStamp) + } + } + protectedSnapshots := s.getProtectedSnapshots(meta) + if len(protectedSnapshots) != s.numSnapshotsToKeep { + t.Fatalf("expected %d protected snapshots, got %d", s.numSnapshotsToKeep, len(protectedSnapshots)) + } + expectedProtectedEpochs := []uint64{6, 2, 1} + for _, expectedEpoch := range expectedProtectedEpochs { + if _, found := protectedSnapshots[expectedEpoch]; !found { + t.Fatalf("expected epoch %d to be protected, but not found", expectedEpoch) + } + } + err = s.Close() if err != nil { - t.Fatal(err) + t.Fatalf("failed to close Scorch: %v", err) } +} - protectedSnapshots := getProtectedSnapshots(RollbackSamplingInterval, NumSnapshotsToKeep, persistedSnapshots) - if len(protectedSnapshots) != 3 { - t.Fatalf("expected %d protected snapshots, got %d", NumSnapshotsToKeep, len(protectedSnapshots)) - } - if _, ok := protectedSnapshots[persistedSnapshots[0].epoch]; !ok { - t.Fatalf("expected %d to be protected, but not found", persistedSnapshots[0].epoch) +// testFSDirectory implements index.Directory by writing files under a +// directory on the local filesystem, mirroring bleve.FileSystemDirectory. It +// is used as the backup target for CopyTo. +type testFSDirectory string + +func (d testFSDirectory) GetWriter(filePath string) (io.WriteCloser, error) { + dir, file := filepath.Split(filePath) + if dir != "" { + if err := os.MkdirAll(filepath.Join(string(d), dir), os.ModePerm); err != nil { + return nil, err + } } + return os.OpenFile(filepath.Join(string(d), dir, file), + os.O_RDWR|os.O_CREATE, 0o600) } func TestBackupRacingWithPurge(t *testing.T) { cfg := CreateConfig("TestBackupRacingWithPurge") - numSnapshotsToKeepOrig := NumSnapshotsToKeep - NumSnapshotsToKeep = 3 - rollbackSamplingIntervalOrig := RollbackSamplingInterval - RollbackSamplingInterval = 10 * time.Second err := InitTest(cfg) if err != nil { t.Fatal(err) } - defer func() { - NumSnapshotsToKeep = numSnapshotsToKeepOrig - RollbackSamplingInterval = rollbackSamplingIntervalOrig - err := DestroyTest(cfg) - if err != nil { - t.Log(err) - } - }() - - // disable merger and purger - RegistryEventCallbacks["test"] = func(e Event) bool { - if e.Kind == EventKindPreMergeCheck || e.Kind == EventKindPurgerCheck { - return false - } - return true + defer func() { _ = DestroyTest(cfg) }() + dirPath := cfg["path"].(string) + if err = os.Mkdir(dirPath, 0o755); err != nil { + t.Fatal(err) } - cfg["eventCallbackName"] = "test" analysisQueue := index.NewAnalysisQueue(1) idx, err := NewScorch(Name, cfg, analysisQueue) if err != nil { - t.Fatal(err) + t.Fatalf("failed to create Scorch: %v", err) } - defer func() { - err := idx.Close() - if err != nil { - t.Fatal(err) - } - }() - - scorchi, ok := idx.(*Scorch) + s, ok := idx.(*Scorch) if !ok { - t.Fatalf("Not a scorch index?") + t.Fatalf("expected *Scorch, got %T", s) } - - err = scorchi.Open() + s.numSnapshotsToKeep = 3 + s.rollbackSamplingInterval = 10 * time.Second + s.path = dirPath + err = s.openBolt() if err != nil { - t.Fatal(err) + t.Fatalf("failed to open bolt: %v", err) } - // replicate the following scenario of persistence of snapshots - // tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5 + // tc - 9d/5, tc - 6d/5, tc - 3d/4, tc - d/6, tc - d/12, tc // approximate timestamps where there's a chance that the latest snapshot - // might not fit into the time-series - indexDummyData(t, scorchi, 1) - time.Sleep(4 * RollbackSamplingInterval / 5) - indexDummyData(t, scorchi, 3) - time.Sleep(9 * RollbackSamplingInterval / 20) - indexDummyData(t, scorchi, 5) - time.Sleep(7 * RollbackSamplingInterval / 12) - indexDummyData(t, scorchi, 7) - time.Sleep(1 * RollbackSamplingInterval / 12) - indexDummyData(t, scorchi, 9) - + // might not fit into the time-series and get purged along with its segment + // files. + const ( + numSnapshots = 6 + ) + curTime := time.Now() + d := s.rollbackSamplingInterval + // old --> new + timeStamps := []time.Time{ + curTime.Add(-(9 * d / 5)), + curTime.Add(-(6 * d / 5)), + curTime.Add(-(3 * d / 4)), + curTime.Add(-(d / 6)), + curTime.Add(-(d / 12)), + curTime, + } + for i := 0; i < numSnapshots; i++ { + snapshot := updateRoot(t, s, false) + if err = persistToBolt(t, snapshot, s, timeStamps[i]); err != nil { + t.Fatalf("failed to persist snapshot %d: %v", i, err) + } + } + // now update the root again, but with an empty snapshot, + // so that the latest root now references a segment that + // belongs to the snapshot which was just persisted to bolt. + updateRoot(t, s, true) // now if the purge code is invoked, there's a possibility of the latest snapshot // being removed from bolt and the corresponding file segment getting cleaned up. - scorchi.removeOldData() - - copyReader := scorchi.CopyReader() + s.removeOldData() + // acquire the copy reader which will now refer to + // the latest in-memory snapshot which was the + // empty snapshot we just introduced. + copyReader := s.CopyReader() defer func() { copyReader.CloseCopyReader() }() backupidxConfig := CreateConfig("backup-directory") @@ -555,81 +712,299 @@ func TestBackupRacingWithPurge(t *testing.T) { t.Log(err) } }() - // if the latest snapshot was purged, the following will return error - err = copyReader.CopyTo(testFSDirector(backupidxConfig["path"].(string))) + err = copyReader.CopyTo(testFSDirectory(backupidxConfig["path"].(string))) if err != nil { t.Fatalf("error copying the index: %v", err) } + err = s.Close() + if err != nil { + t.Fatalf("failed to close Scorch: %v", err) + } } func TestSparseMutationCheckpointing(t *testing.T) { cfg := CreateConfig("TestSparseMutationCheckpointing") - numSnapshotsToKeepOrig := NumSnapshotsToKeep - NumSnapshotsToKeep = 3 - rollbackSamplingIntervalOrig := RollbackSamplingInterval - RollbackSamplingInterval = 2 * time.Second - err := InitTest(cfg) if err != nil { t.Fatal(err) } - defer func() { - NumSnapshotsToKeep = numSnapshotsToKeepOrig - RollbackSamplingInterval = rollbackSamplingIntervalOrig - err := DestroyTest(cfg) - if err != nil { - t.Log(err) - } - }() - - // disable merger and purger - RegistryEventCallbacks["test"] = func(e Event) bool { - if e.Kind == EventKindPreMergeCheck { - return false - } - return true + defer func() { _ = DestroyTest(cfg) }() + dirPath := cfg["path"].(string) + if err = os.Mkdir(dirPath, 0o755); err != nil { + t.Fatal(err) } - cfg["eventCallbackName"] = "test" + analysisQueue := index.NewAnalysisQueue(1) idx, err := NewScorch(Name, cfg, analysisQueue) + if err != nil { + t.Fatalf("failed to create Scorch: %v", err) + } + s, ok := idx.(*Scorch) + if !ok { + t.Fatalf("expected *Scorch, got %T", s) + } + s.numSnapshotsToKeep = 3 + s.rollbackSamplingInterval = 2 * time.Second + s.rollbackRetentionFactor = 0.5 + s.path = dirPath + err = s.openBolt() + if err != nil { + t.Fatalf("failed to open bolt: %v", err) + } + const ( + numSnapshots = 5 + ) + curTime := time.Now() + d := s.rollbackSamplingInterval + // old --> new + timeStamps := []time.Time{ + curTime.Add(-(6 * d)), + curTime.Add(-(5 * d)), + curTime.Add(-(4 * d)), + curTime.Add(-(3 * d)), + curTime, + } + for i := 0; i < numSnapshots; i++ { + snapshot := updateRoot(t, s, false) + if err = persistToBolt(t, snapshot, s, timeStamps[i]); err != nil { + t.Fatalf("failed to persist snapshot %d: %v", i, err) + } + } + s.checkPoints = newCheckPoints(map[uint64]time.Time{ + 2: timeStamps[1], + 3: timeStamps[2], + 4: timeStamps[3], + }) + meta, err := s.getLiveSnapshots() if err != nil { t.Fatal(err) } + if len(meta) != s.numSnapshotsToKeep { + t.Fatalf("expected %d live snapshots, got %d", s.numSnapshotsToKeep, len(meta)) + } + for i, m := range meta { + expectIdx := numSnapshots - i - 1 + expectedEpoch := uint64(expectIdx + 1) + if m.epoch != expectedEpoch { + t.Fatalf("expected epoch %d, got %d", expectedEpoch, m.epoch) + } + expectedTimeStamp := timeStamps[expectIdx] + if !m.timeStamp.Equal(expectedTimeStamp) { + t.Fatalf("expected timestamp %v, got %v", expectedTimeStamp, m.timeStamp) + } + } + protectedSnapshots := s.getProtectedSnapshots(meta) + if len(protectedSnapshots) != s.numSnapshotsToKeep { + t.Fatalf("expected %d protected snapshots, got %d", s.numSnapshotsToKeep, len(protectedSnapshots)) + } + expectedProtectedEpochs := []uint64{5, 4, 3} + for _, expectedEpoch := range expectedProtectedEpochs { + if _, found := protectedSnapshots[expectedEpoch]; !found { + t.Fatalf("expected epoch %d to be protected, but not found", expectedEpoch) + } + } + err = s.Close() + if err != nil { + t.Fatalf("failed to close Scorch: %v", err) + } +} - scorchi, ok := idx.(*Scorch) +func TestRollbackCheckpointsOnRestart(t *testing.T) { + cfg := CreateConfig("TestRollbackCheckpointsOnRestart") + err := InitTest(cfg) + if err != nil { + t.Fatal(err) + } + defer func() { _ = DestroyTest(cfg) }() + dirPath := cfg["path"].(string) + if err = os.Mkdir(dirPath, 0o755); err != nil { + t.Fatal(err) + } + analysisQueue := index.NewAnalysisQueue(1) + idx, err := NewScorch(Name, cfg, analysisQueue) + if err != nil { + t.Fatalf("failed to create Scorch: %v", err) + } + s, ok := idx.(*Scorch) if !ok { - t.Fatalf("Not a scorch index?") + t.Fatalf("expected *Scorch, got %T", s) + } + s.numSnapshotsToKeep = 3 + s.rollbackSamplingInterval = 2 * time.Second + s.rollbackRetentionFactor = 0.5 + s.path = dirPath + err = s.openBolt() + if err != nil { + t.Fatalf("failed to open bolt: %v", err) + } + const ( + numSnapshots = 6 + ) + curTime := time.Now() + d := s.rollbackSamplingInterval + // old --> new + timeStamps := []time.Time{ + curTime.Add(-(10 * d)), + curTime.Add(-(9 * d)), + curTime.Add(-(8 * d)), + + curTime.Add(-(2 * d)), + curTime.Add(-(1 * d)), + curTime, + } + for i := 0; i < 3; i++ { + snapshot := updateRoot(t, s, false) + if err = persistToBolt(t, snapshot, s, timeStamps[i]); err != nil { + t.Fatalf("failed to persist snapshot %d: %v", i, err) + } } - - err = scorchi.Open() + err = s.Close() + if err != nil { + t.Fatalf("failed to close Scorch: %v", err) + } + idx, err = NewScorch(Name, cfg, analysisQueue) + if err != nil { + t.Fatalf("failed to create Scorch: %v", err) + } + s, ok = idx.(*Scorch) + if !ok { + t.Fatalf("expected *Scorch, got %T", s) + } + s.numSnapshotsToKeep = 3 + s.rollbackSamplingInterval = 2 * time.Second + s.rollbackRetentionFactor = 0.5 + s.path = dirPath + err = s.openBolt() + if err != nil { + t.Fatalf("failed to open bolt: %v", err) + } + for i := 3; i < numSnapshots; i++ { + snapshot := updateRoot(t, s, false) + if err = persistToBolt(t, snapshot, s, timeStamps[i]); err != nil { + t.Fatalf("failed to persist snapshot %d: %v", i, err) + } + } + meta, err := s.getLiveSnapshots() if err != nil { t.Fatal(err) } + if len(meta) != 5 { + t.Fatalf("expected %d live snapshots, got %d", 5, len(meta)) + } + for i, m := range meta { + expectIdx := numSnapshots - i - 1 + expectedEpoch := uint64(expectIdx + 1) + if m.epoch != expectedEpoch { + t.Fatalf("expected epoch %d, got %d", expectedEpoch, m.epoch) + } + expectedTimeStamp := timeStamps[expectIdx] + if !m.timeStamp.Equal(expectedTimeStamp) { + t.Fatalf("expected timestamp %v, got %v", expectedTimeStamp, m.timeStamp) + } + } + protectedSnapshots := s.getProtectedSnapshots(meta) + if len(protectedSnapshots) != s.numSnapshotsToKeep { + t.Fatalf("expected %d protected snapshots, got %d", s.numSnapshotsToKeep, len(protectedSnapshots)) + } + expectedProtectedEpochs := []uint64{6, 3, 2} + for _, expectedEpoch := range expectedProtectedEpochs { + if _, found := protectedSnapshots[expectedEpoch]; !found { + t.Fatalf("expected epoch %d to be protected, but not found", expectedEpoch) + } + } + err = s.Close() + if err != nil { + t.Fatalf("failed to close Scorch: %v", err) + } +} - // create 4 snapshots every 2 seconds - indexDummyData(t, scorchi, 1) - time.Sleep(RollbackSamplingInterval) - indexDummyData(t, scorchi, 3) - time.Sleep(RollbackSamplingInterval) - indexDummyData(t, scorchi, 5) - time.Sleep(RollbackSamplingInterval) - indexDummyData(t, scorchi, 7) - - // now the another snapshot is persisted outside of the window of checkpointing - // and we should be able to retain some older checkpoints as well along with - // the latest one - time.Sleep(time.Duration(NumSnapshotsToKeep) * RollbackSamplingInterval) - indexDummyData(t, scorchi, 9) - - persistedSnapshots, err := scorchi.rootBoltSnapshotMetaData() +func TestRollackOptions(t *testing.T) { + cfg := CreateConfig("TestRollackOptions") + err := InitTest(cfg) if err != nil { t.Fatal(err) } - - // should have more than 1 snapshots - protectedSnapshots := getProtectedSnapshots(RollbackSamplingInterval, NumSnapshotsToKeep, persistedSnapshots) - if len(protectedSnapshots) <= 1 { - t.Fatalf("expected %d protected snapshots, got %d", NumSnapshotsToKeep, len(protectedSnapshots)) + defer func() { _ = DestroyTest(cfg) }() + dirPath := cfg["path"].(string) + if err = os.Mkdir(dirPath, 0o755); err != nil { + t.Fatal(err) + } + type testCase struct { + cfg map[string]interface{} + expectErr bool + } + tests := []testCase{ + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 3, + "rollbackSamplingInterval": "10m", + "rollbackRetentionFactor": 0.5, + }, + expectErr: false, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 0.5, + "rollbackSamplingInterval": "10a", + "rollbackRetentionFactor": 1, + }, + expectErr: true, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 3.0, + "rollbackSamplingInterval": 10, + "rollbackRetentionFactor": 1.5, + }, + expectErr: true, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 3.0, + "rollbackSamplingInterval": "10s", + "rollbackRetentionFactor": 0, + }, + expectErr: false, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": "3.0", + "rollbackSamplingInterval": "10s", + "rollbackRetentionFactor": 2, + }, + expectErr: true, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 3.0, + "rollbackSamplingInterval": "10s", + "rollbackRetentionFactor": "2", + }, + expectErr: true, + }, + { + cfg: map[string]interface{}{ + "numSnapshotsToKeep": 3.0, + "rollbackSamplingInterval": "10s", + "rollbackRetentionFactor": 2, + }, + expectErr: true, + }, + } + for i, test := range tests { + test.cfg["path"] = dirPath + s, err := NewScorch(Name, test.cfg, index.NewAnalysisQueue(1)) + gotErr := err != nil + wantErr := test.expectErr + if gotErr != wantErr { + t.Fatalf("test %d: expected error: %v, got error: %v", i, wantErr, gotErr) + } + if err == nil { + err = s.Close() + if err != nil { + t.Fatalf("test %d: failed to close Scorch: %v", i, err) + } + } } } diff --git a/index/scorch/scorch.go b/index/scorch/scorch.go index 6ea5295a4..d63c66a08 100644 --- a/index/scorch/scorch.go +++ b/index/scorch/scorch.go @@ -70,6 +70,9 @@ type Scorch struct { // must be accessed within the rootLock as it is accessed by the asynchronous cleanup routine. copyScheduled map[string]int + persisterOptions *persisterOptions + mergePlannerOptions *mergeplan.MergePlanOptions + numSnapshotsToKeep int rollbackRetentionFactor float64 checkPoints []*snapshotMetaData @@ -130,10 +133,9 @@ func (t ScorchErrorType) Error() string { // ErrType values for ScorchError const ( - ErrAsyncPanic = ScorchErrorType("async panic error") - ErrPersist = ScorchErrorType("persist error") - ErrCleanup = ScorchErrorType("cleanup error") - ErrOptionsParse = ScorchErrorType("options parse error") + ErrAsyncPanic = ScorchErrorType("async panic error") + ErrPersist = ScorchErrorType("persist error") + ErrCleanup = ScorchErrorType("cleanup error") ) // ScorchError is passed to onAsyncError when errors are @@ -232,18 +234,53 @@ func NewScorch(storeName string, if ok { rv.onAsyncError = RegistryAsyncErrorCallbacks[aecbName] } - // validate any custom persistor options to - // prevent an async error in the persistor routine - _, err = rv.parsePersisterOptions() + + rv.numSnapshotsToKeep = NumSnapshotsToKeep + if v, ok := rv.config["numSnapshotsToKeep"]; ok { + var n int + var err error + if n, err = parseToInteger(v); err != nil { + return nil, fmt.Errorf("numSnapshotsToKeep parse err: %v", err) + } + if n > 0 { + rv.numSnapshotsToKeep = n + } + } + + rv.rollbackSamplingInterval = RollbackSamplingInterval + if v, ok := rv.config["rollbackSamplingInterval"]; ok { + var d time.Duration + var err error + if d, err = parseToTimeDuration(v); err != nil { + return nil, fmt.Errorf("rollbackSamplingInterval parse err: %v", err) + } + rv.rollbackSamplingInterval = d + } + + rv.rollbackRetentionFactor = RollbackRetentionFactor + if v, ok := rv.config["rollbackRetentionFactor"]; ok { + var r float64 + var err error + if r, err = parseToFloat(v); err != nil { + return nil, fmt.Errorf("rollbackRetentionFactor parse err: %v", err) + } + if r < 0 || r > 1 { + return nil, fmt.Errorf("rollbackRetentionFactor must be between 0 and 1") + } + rv.rollbackRetentionFactor = r + } + + po, err := rv.parsePersisterOptions() if err != nil { return nil, err } - // validate any custom merge planner options to - // prevent an async error in the merger routine - _, err = rv.parseMergePlannerOptions() + rv.persisterOptions = po + + mpo, err := rv.parseMergePlannerOptions(po) if err != nil { return nil, err } + rv.mergePlannerOptions = mpo if trainer := initTrainer(rv, config); trainer != nil { rv.trainer = trainer @@ -328,6 +365,7 @@ func (s *Scorch) openBolt() error { s.unsafeBatch = true } + // bolt options rootBoltOpt := *bolt.DefaultOptions if s.readOnly { rootBoltOpt.ReadOnly = true @@ -392,42 +430,6 @@ func (s *Scorch) openBolt() error { } } - s.numSnapshotsToKeep = NumSnapshotsToKeep - if v, ok := s.config["numSnapshotsToKeep"]; ok { - var t int - if t, err = parseToInteger(v); err != nil { - return fmt.Errorf("numSnapshotsToKeep parse err: %v", err) - } - if t > 0 { - s.numSnapshotsToKeep = t - } - } - - s.rollbackSamplingInterval = RollbackSamplingInterval - if v, ok := s.config["rollbackSamplingInterval"]; ok { - var t time.Duration - if t, err = parseToTimeDuration(v); err != nil { - return fmt.Errorf("rollbackSamplingInterval parse err: %v", err) - } - s.rollbackSamplingInterval = t - } - - s.rollbackRetentionFactor = RollbackRetentionFactor - if v, ok := s.config["rollbackRetentionFactor"]; ok { - var r float64 - if r, ok = v.(float64); ok { - return fmt.Errorf("rollbackRetentionFactor parse err: %v", err) - } - s.rollbackRetentionFactor = r - } - - typ, ok := s.config["spatialPlugin"].(string) - if ok { - if err := s.loadSpatialAnalyzerPlugin(typ); err != nil { - return err - } - } - return nil } @@ -1003,6 +1005,20 @@ func parseToInteger(i interface{}) (int, error) { } } +func parseToFloat(i interface{}) (float64, error) { + switch v := i.(type) { + case float64: + return v, nil + case float32: + return float64(v), nil + case int: + return float64(v), nil + + default: + return 0, fmt.Errorf("expects float64, float32 or int value") + } +} + // Holds Zap's field level stats at a segment level type fieldStats struct { // StatName -> FieldName -> value @@ -1365,10 +1381,6 @@ func (s *Scorch) DropFileWriterIDs(ids map[string]struct{}) error { var mergePlanner mergePlanFunc = func(ourSnapshot *IndexSnapshot) (*mergeplan.MergePlan, error) { // Create a merge plan with the filtered segments and force a merge // to remove the callback from the segments. - mergePlannerOptions, err := s.parseMergePlannerOptions() - if err != nil { - return nil, fmt.Errorf("mergePlannerOption json parsing err: %v", err) - } atomic.AddUint64(&s.stats.TotFileMergePlan, 1) // filter all segments that have callbacks that need to be removed @@ -1387,7 +1399,7 @@ func (s *Scorch) DropFileWriterIDs(ids map[string]struct{}) error { } // attempt a merge plan with the default merge planner options - mergePlan, err := mergeplan.Plan(segsToCompact, mergePlannerOptions) + mergePlan, err := mergeplan.Plan(segsToCompact, s.mergePlannerOptions) if err != nil { atomic.AddUint64(&s.stats.TotFileMergePlanErr, 1) return nil, fmt.Errorf("merge plan creation err: %v", err)