diff --git a/kvrocks.conf b/kvrocks.conf
index b58827ea48d..e8bd2a50b29 100644
--- a/kvrocks.conf
+++ b/kvrocks.conf
@@ -97,6 +97,16 @@ dir /tmp/kvrocks
 #
 log-dir stdout
 
+# You can configure log-retention-days to control whether to enable the log cleaner
+# and the maximum retention days that the INFO level logs will be kept.
+#
+# if set to -1, that means to disable the log cleaner.
+# if set to 0, all previous INFO level logs will be immediately removed.
+# if set to between 0 to INT_MAX, that means it will retent latest N(log-retention-days) day logs. 
+
+# By default the log-retention-days is -1.
+log-retention-days -1
+
 # When running in daemonize mode, kvrocks writes a PID file in ${CONFIG_DIR}/kvrocks.pid by
 # default. You can specify a custom pid file location here.
 # pidfile /var/run/kvrocks.pid
diff --git a/src/config/config.cc b/src/config/config.cc
index 5618517a090..8551bddae6a 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -159,6 +159,7 @@ Config::Config() {
       {"migrate-sequence-gap", false, new IntField(&sequence_gap, 10000, 1, INT_MAX)},
       {"unixsocket", true, new StringField(&unixsocket, "")},
       {"unixsocketperm", true, new OctalField(&unixsocketperm, 0777, 1, INT_MAX)},
+      {"log-retention-days", false, new IntField(&log_retention_days, -1, -1, INT_MAX)},
 
       /* rocksdb options */
       {"rocksdb.compression", false, new EnumField(&RocksDB.compression, compression_type_enum, 0)},
@@ -454,6 +455,20 @@ void Config::initFieldCallback() {
          if (cluster_enabled) srv->slot_migrate_->SetSequenceGapSize(sequence_gap);
          return Status::OK();
        }},
+      {"log-retention-days",
+       [this](Server *srv, const std::string &k, const std::string &v) -> Status {
+         if (!srv) return Status::OK();
+         if (Util::ToLower(log_dir) == "stdout") {
+           return {Status::NotOK, "can't set the 'log-retention-days' when the log dir is stdout"};
+         }
+
+         if (log_retention_days != -1) {
+           google::EnableLogCleaner(log_retention_days);
+         } else {
+           google::DisableLogCleaner();
+         }
+         return Status::OK();
+       }},
       {"rocksdb.target_file_size_base",
        [this](Server *srv, const std::string &k, const std::string &v) -> Status {
          if (!srv) return Status::OK();
diff --git a/src/config/config.h b/src/config/config.h
index 144ae192deb..2871f551b85 100644
--- a/src/config/config.h
+++ b/src/config/config.h
@@ -139,6 +139,7 @@ struct Config {
   int pipeline_size;
   int sequence_gap;
 
+  int log_retention_days;
   // profiling
   int profiling_sample_ratio = 0;
   int profiling_sample_record_threshold_ms = 0;
diff --git a/src/main.cc b/src/main.cc
index 356b49c9c6f..86d8e33c89a 100644
--- a/src/main.cc
+++ b/src/main.cc
@@ -177,7 +177,10 @@ static void initGoogleLog(const Config *config) {
     FLAGS_stderrthreshold = google::ERROR;
     FLAGS_logtostdout = true;
   } else {
-    FLAGS_log_dir = config->log_dir;
+    FLAGS_log_dir = config->log_dir + "/";
+    if (config->log_retention_days != -1) {
+      google::EnableLogCleaner(config->log_retention_days);
+    }
   }
 }
 
diff --git a/tests/gocase/unit/log/logclean_test.go b/tests/gocase/unit/log/logclean_test.go
new file mode 100644
index 00000000000..993ee42d03e
--- /dev/null
+++ b/tests/gocase/unit/log/logclean_test.go
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package log
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/apache/incubator-kvrocks/tests/gocase/util"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLogClean(t *testing.T) {
+	err := os.Mkdir("/tmp/kvrocks/logfile", os.ModePerm)
+	if err != nil {
+		return
+	}
+	srv1 := util.StartServer(t, map[string]string{
+		"log-dir":            "/tmp/kvrocks/logfile",
+		"log-retention-days": "0",
+	})
+	srv1.Close()
+	files1, _ := os.ReadDir("/tmp/kvrocks/logfile")
+	time.Sleep(2000 * time.Millisecond)
+	srv2 := util.StartServer(t, map[string]string{
+		"log-dir":            "/tmp/kvrocks/logfile",
+		"log-retention-days": "0",
+	})
+	srv2.Close()
+	files2, _ := os.ReadDir("/tmp/kvrocks/logfile")
+	islogclear := false
+	for _, f1 := range files1 {
+		ishave := false
+		for _, f2 := range files2 {
+			if f1.Name() == f2.Name() {
+				ishave = true
+				break
+			}
+		}
+		if !ishave {
+			islogclear = true
+			break
+		}
+	}
+	require.Equal(t, true, islogclear)
+	err = os.RemoveAll("/tmp/kvrocks/logfile")
+	if err != nil {
+		return
+	}
+}