feat(storagenode): add --storage-trim-delay to set a delay before the…

… deletion of log entries

This change adds a new flag --storage-trim-delay to the storage node. It sets the delay before
storage removes logs. If the value is zero, Trim will delay the removal of prefix log entries until
writing additional log entries.

cmd/varlogsn/cli.go

@@ -75,6 +75,7 @@ func newStartCommand() *cli.Command {
 			flagStorageMaxConcurrentCompaction,
 			flagStorageMetricsLogInterval,
 			flagStorageVerbose.BoolFlag(),
+			flagStorageTrimDelay,
 
 			// logger options
 			flags.LogDir,

cmd/varlogsn/flags.go

@@ -195,6 +195,11 @@ var (
 		EnvVars: []string{"STORAGE_METRICS_LOG_INTERVAL"},
 		Value:   storage.DefaultMetricsLogInterval,
 	}
+	flagStorageTrimDelay = &cli.DurationFlag{
+		Name:    "storage-trim-delay",
+		EnvVars: []string{"STORAGE_TRIM_DELAY"},
+		Usage:   "Delay before deletion of log entries caused by Trim operation. If zero, lazy deletion waits for other log entries to be appended.",
+	}
 
 	// flags for telemetry.
 	flagExporterType = flags.FlagDesc{

cmd/varlogsn/varlogsn.go

@@ -334,6 +334,9 @@ func parseStorageOptions(c *cli.Context) (opts []storage.Option, err error) {
 	if c.Bool(flagStorageVerbose.Name) {
 		opts = append(opts, storage.WithVerboseLogging())
 	}
+	if name := flagStorageTrimDelay.Name; c.IsSet(name) {
+		opts = append(opts, storage.WithTrimDelay(c.Duration(name)))
+	}
 	return opts, nil
 }
 

internal/storage/config.go

@@ -143,6 +143,7 @@ type config struct {
 	commitDBConfig     dbConfig
 	verbose            bool
 	metricsLogInterval time.Duration
+	trimDelay          time.Duration
 	logger             *zap.Logger
 	readOnly           bool
 }
@@ -246,6 +247,15 @@ func WithLogger(logger *zap.Logger) Option {
 	})
 }
 
+// WithTrimDelay sets the delay before storage removes logs. If the value is
+// zero, Trim will delay the removal of prefix log entries until writing
+// additional log entries.
+func WithTrimDelay(trimDelay time.Duration) Option {
+	return newFuncOption(func(cfg *config) {
+		cfg.trimDelay = trimDelay
+	})
+}
+
 // ReadOnly makes storage read-only. It is helpful only for testing. Usually,
 // users do not have to call it.
 func ReadOnly() Option {

internal/storage/storage.go

@@ -112,6 +112,7 @@ func (s *Storage) newDB(path string, cfg *dbConfig) (*pebble.DB, error) {
 		MaxConcurrentCompactions:    func() int { return cfg.maxConcurrentCompaction },
 		Levels:                      make([]pebble.LevelOptions, 7),
 		ErrorIfExists:               false,
+		FlushDelayDeleteRange:       s.trimDelay,
 	}
 	pebbleOpts.Levels[0].TargetFileSize = cfg.l0TargetFileSize
 	for i := 0; i < len(pebbleOpts.Levels); i++ {