feat(Flags): [Breaking] Add flag for snapshot duration frequency

compose/compose.go

@@ -311,7 +311,7 @@ func getAlpha(idx int, raft string) service {
 	}
 
 	if opts.SnapshotAfter != "" {
-		raft = fmt.Sprintf("%s; snapshot-after=%s", raft, opts.SnapshotAfter)
+		raft = fmt.Sprintf("%s; snapshot-after-entries=%s", raft, opts.SnapshotAfter)
 	}
 	svc.Command += fmt.Sprintf(` --raft "%s"`, raft)
 

dgraph/cmd/alpha/run.go

@@ -158,9 +158,15 @@ they form a Raft group and provide synchronous replication.
 		Flag("learner",
 			`Make this Alpha a "learner" node. In learner mode, this Alpha will not participate `+
 				"in Raft elections. This can be used to achieve a read-only replica.").
-		Flag("snapshot-after",
+		Flag("snapshot-after-entries",
 			"Create a new Raft snapshot after N number of Raft entries. The lower this number, "+
-				"the more frequent snapshot creation will be.").
+				"the more frequent snapshot creation will be. Snapshots are created if either "+
+				"snapshot-after-duration or snapshot-after-entries threshold is crossed.").
+		Flag("snapshot-after-duration",
+			"Frequency at which we should create a new raft snapshots. Set "+
+				"to 0 to disable duration based snapshot. Snapshots are "+
+				"created if either snapshot-after-duration or "+
+				"snapshot-after-entries threshold is crossed.").
 		Flag("pending-proposals",
 			"Number of pending mutation proposals. Useful for rate limiting.").
 		String())

worker/draft.go

@@ -997,10 +997,13 @@ func (n *node) updateRaftProgress() error {
 
 func (n *node) checkpointAndClose(done chan struct{}) {
 	slowTicker := time.NewTicker(time.Minute)
+	lastSnapshotTime := time.Now()
 	defer slowTicker.Stop()
 
-	snapshotAfter := x.WorkerConfig.Raft.GetUint64("snapshot-after")
-	x.AssertTruef(snapshotAfter > 10, "raft.snapshot-after must be a number greater than 10")
+	snapshotAfterEntries := x.WorkerConfig.Raft.GetUint64("snapshot-after-entries")
+	x.AssertTruef(snapshotAfterEntries > 10, "raft.snapshot-after must be a number greater than 10")
+
+	snapshotAfterDuration := x.WorkerConfig.Raft.GetDuration("snapshot-after-duration")
 
 	for {
 		select {
@@ -1026,16 +1029,19 @@ func (n *node) checkpointAndClose(done chan struct{}) {
 
 				// If we don't have a snapshot, or if there are too many log files in Raft,
 				// calculate a new snapshot.
-				calculate := raft.IsEmptySnap(snap) || n.Store.NumLogFiles() > 4
+				// For snapshotAfterDuration, 0 is a special value used to
+				// disable time based snapshots.
+				calculate := raft.IsEmptySnap(snap) || n.Store.NumLogFiles() > 4 ||
+					(snapshotAfterDuration != 0 && time.Since(lastSnapshotTime) > snapshotAfterDuration)
 
 				if chk, err := n.Store.Checkpoint(); err == nil {
 					if first, err := n.Store.FirstIndex(); err == nil {
 						// Save some cycles by only calculating snapshot if the checkpoint has gone
 						// quite a bit further than the first index.
-						calculate = calculate || chk >= first+snapshotAfter
+						calculate = calculate || chk >= first+snapshotAfterEntries
 						glog.V(3).Infof("Evaluating snapshot first:%d chk:%d (chk-first:%d) "+
-							"snapshotAfter:%d snap:%v", first, chk, chk-first,
-							snapshotAfter, calculate)
+							"snapshotAfterEntries:%d snap:%v", first, chk, chk-first,
+							snapshotAfterEntries, calculate)
 					}
 				}
 				// We keep track of the applied index in the p directory. Even if we don't take
@@ -1056,6 +1062,7 @@ func (n *node) checkpointAndClose(done chan struct{}) {
 					if err := n.proposeSnapshot(); err != nil {
 						glog.Errorf("While calculating and proposing snapshot: %v", err)
 					}
+					lastSnapshotTime = time.Now()
 				}
 				go n.abortOldTransactions()
 			}

worker/server_state.go

@@ -38,10 +38,11 @@ const (
 	//       For easy readability, keep the options without default values (if any) at the end of
 	//       the *Defaults string. Also, since these strings are printed in --help text, avoid line
 	//       breaks.
-	AclDefaults       = `access-ttl=6h; refresh-ttl=30d; secret-file=;`
-	AuditDefaults     = `compress=false; days=10; size=100; dir=; output=; encrypt-file=;`
-	BadgerDefaults    = `compression=snappy; goroutines=8; max-retries=-1;`
-	RaftDefaults      = `learner=false; snapshot-after=10000; pending-proposals=256; idx=; group=;`
+	AclDefaults    = `access-ttl=6h; refresh-ttl=30d; secret-file=;`
+	AuditDefaults  = `compress=false; days=10; size=100; dir=; output=; encrypt-file=;`
+	BadgerDefaults = `compression=snappy; goroutines=8; max-retries=-1;`
+	RaftDefaults   = `learner=false; snapshot-after-entries=10000; ` +
+		`snapshot-after-duration=30m; pending-proposals=256; idx=; group=;`
 	SecurityDefaults  = `token=; whitelist=;`
 	LudicrousDefaults = `enabled=false; concurrency=2000;`
 	CDCDefaults       = `file=; kafka=; sasl_user=; sasl_password=; ca_cert=; client_cert=; ` +