forked from FeatureBaseDB/featurebase
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- had to make sure we don't snapshot until directive is fully applied on a computer... otherwise there's races between loading the files and truncating the write log. - added a dirty bit to resources and a bool return to incrementing the write log... don't snapshot if it returns false because that means there's been no writes. (but make sure you close the storage transaction!) - added the actually snapshotting routine which just fires every <timeout> and serially snapshots everything. - tweaked some logging - added ability to get all tables in an org/db or literally all. I think I just needed the "literally all", but it was natural to allow it to be scoped to org or DB as well. (cherry picked from commit b8b08bc)
- Loading branch information
Showing
19 changed files
with
231 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/molecula/featurebase/v3/dax" | ||
| ) | ||
|
|
||
| func (c *Controller) snappingTurtleRoutine(period time.Duration) { | ||
| if period == 0 { | ||
| return // disable automatic snapshotting | ||
| } | ||
| ticker := time.NewTicker(period) | ||
| for { | ||
| select { | ||
| case <-c.stopping: | ||
| ticker.Stop() | ||
| c.logger.Debugf("TURTLE: Stopping Snapping Turtle") | ||
| return | ||
| case <-ticker.C: | ||
| c.snapAll() | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| func (c *Controller) snapAll() { | ||
| c.logger.Debugf("TURTLE: snapAll") | ||
| ctx := context.Background() | ||
| computeNodes, err := c.ComputeBalancer.CurrentState(ctx) | ||
| if err != nil { | ||
| c.logger.Printf("Error getting compute balancer state for snapping turtle: %v", err) | ||
| } | ||
|
|
||
| // Weird nested loop for snapshotting shard data. The reason for | ||
| // this is to avoid hotspotting each node in turn and spread the | ||
| // snapshotting load across all nodes rather than snapshotting all | ||
| // jobs on one node and then moving onto the next one. | ||
| i := 0 | ||
| stillWorking := true | ||
| for stillWorking { | ||
| stillWorking = false | ||
| for _, workerInfo := range computeNodes { | ||
| if len(workerInfo.Jobs) <= i { | ||
| continue | ||
| } | ||
| stillWorking = true | ||
| j, err := decodeShard(workerInfo.Jobs[i]) | ||
| if err != nil { | ||
| c.logger.Printf("couldn't decode a shard out of the job: '%s', err: %v", workerInfo.Jobs[i], err) | ||
| } | ||
| c.SnapshotShardData(ctx, j.t.QualifiedTableID(), j.shardNum()) | ||
| } | ||
| i++ | ||
| } | ||
|
|
||
| // Get all tables across all orgs/dbs so we can snapshot all keyed | ||
| // fields and look up whether a table is keyed to snapshot it's | ||
| // partitions. | ||
| tables, err := c.Schemar.Tables(ctx, dax.TableQualifier{}) | ||
| if err != nil { | ||
| c.logger.Printf("Couldn't get schema for snapshotting keys: %v", err) | ||
| return | ||
| } | ||
| // snapshot keyed fields | ||
| tableMap := make(map[dax.TableKey]*dax.QualifiedTable) | ||
| for _, table := range tables { | ||
| tableMap[table.Key()] = table | ||
| for _, f := range table.Fields { | ||
| if f.StringKeys() && !f.IsPrimaryKey() { | ||
| err := c.SnapshotFieldKeys(ctx, table.QualifiedID(), f.Name) | ||
| if err != nil { | ||
| c.logger.Printf("Couldn't snapshot table: %s, field: %s, error: %v", table, f.Name, err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Get all partition jobs from balancer and snapshot table keys | ||
| // for any partition that goes with a keyed table. Doing the same | ||
| // weird nested loop thing to avoid doing all jobs on one node | ||
| // back to back. | ||
| translateNodes, err := c.TranslateBalancer.CurrentState(ctx) | ||
| if err != nil { | ||
| c.logger.Printf("Error getting translate balancer state for snapping turtle: %v", err) | ||
| } | ||
|
|
||
| i = 0 | ||
| stillWorking = true | ||
| for stillWorking { | ||
| stillWorking = false | ||
| for _, workerInfo := range translateNodes { | ||
| if len(workerInfo.Jobs) <= i { | ||
| continue | ||
| } | ||
| stillWorking = true | ||
| j, err := decodePartition(workerInfo.Jobs[i]) | ||
| if err != nil { | ||
| table := tableMap[j.table()] | ||
| if table.StringKeys() { | ||
| c.SnapshotTableKeys(ctx, table.QualifiedID(), j.partitionNum()) | ||
| } | ||
| c.logger.Printf("couldn't decode a partition out of the job: '%s', err: %v", workerInfo.Jobs[i], err) | ||
| } | ||
| } | ||
| i++ | ||
| } | ||
| c.logger.Debugf("TURTLE: snapAll complete") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.