From eaad2605d16f086f54c2380a6c5b3fc02f39baec Mon Sep 17 00:00:00 2001
From: Injun Song <ijsong@gmail.com>
Date: Fri, 14 Jul 2023 09:47:22 +0900
Subject: [PATCH] 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         |  1 +
 cmd/varlogsn/flags.go       | 32 ++++++++++++++++++++++++++++++++
 cmd/varlogsn/varlogsn.go    |  3 +++
 internal/storage/config.go  | 10 ++++++++++
 internal/storage/storage.go |  1 +
 5 files changed, 47 insertions(+)

diff --git a/cmd/varlogsn/cli.go b/cmd/varlogsn/cli.go
index ce11b5b53..49c15c406 100644
--- a/cmd/varlogsn/cli.go
+++ b/cmd/varlogsn/cli.go
@@ -73,6 +73,7 @@ func newStartCommand() *cli.Command {
 			flagStorageMaxConcurrentCompaction,
 			flagStorageMetricsLogInterval,
 			flagStorageVerbose.BoolFlag(),
+			flagStorageTrimDelay,
 
 			// logger options
 			flags.LogDir,
diff --git a/cmd/varlogsn/flags.go b/cmd/varlogsn/flags.go
index 14c98037b..18359b191 100644
--- a/cmd/varlogsn/flags.go
+++ b/cmd/varlogsn/flags.go
@@ -195,4 +195,36 @@ 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{
+		Name:  "exporter-type",
+		Usage: "exporter type: stdout, otlp or noop",
+		Envs:  []string{"EXPORTER_TYPE"},
+	}
+	flagExporterStopTimeout = flags.FlagDesc{
+		Name:  "expoter-stop-timeout",
+		Usage: "timeout for stopping exporter",
+		Envs:  []string{"EXPORTER_STOP_TIMEOUT"},
+	}
+	flagStdoutExporterPrettyPrint = flags.FlagDesc{
+		Name:  "exporter-pretty-print",
+		Usage: "pretty print when using stdout exporter",
+		Envs:  []string{"EXPORTER_PRETTY_PRINT"},
+	}
+	flagOTLPExporterInsecure = flags.FlagDesc{
+		Name:  "exporter-otlp-insecure",
+		Usage: "disable client transport security for the OTLP exporter",
+		Envs:  []string{"EXPORTER_OTLP_INSECURE"},
+	}
+	flagOTLPExporterEndpoint = flags.FlagDesc{
+		Name:  "exporter-otlp-endpoint",
+		Usage: "the endpoint that exporter connects",
+		Envs:  []string{"EXPORTER_OTLP_ENDPOINT"},
+	}
 )
diff --git a/cmd/varlogsn/varlogsn.go b/cmd/varlogsn/varlogsn.go
index 78e98fc70..bbab5640f 100644
--- a/cmd/varlogsn/varlogsn.go
+++ b/cmd/varlogsn/varlogsn.go
@@ -277,6 +277,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
 }
 
diff --git a/internal/storage/config.go b/internal/storage/config.go
index 4b6096020..d5cf05440 100644
--- a/internal/storage/config.go
+++ b/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 {
diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index e9e34d846..8ffda3ba0 100644
--- a/internal/storage/storage.go
+++ b/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++ {