support go time package format gc time

pkg/util/util.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strconv"
 	"strings"
 	"time"
 
@@ -301,3 +302,16 @@ func WaitUntilTimeout(name string, fn func(), timeout time.Duration) {
 	case <-exited:
 	}
 }
+
+// ParseGCDuration parses gc durations. The default unit is day.
+func ParseGCDuration(gc string) (time.Duration, error) {
+	d, err := strconv.ParseUint(gc, 10, 64)
+	if err == nil {
+		return time.Duration(d) * 24 * time.Hour, nil
+	}
+	gcDuration, err := time.ParseDuration(gc)
+	if err != nil {
+		return 0, errors.Errorf("unsupported gc time %s, please use 7 or 8h(max unit is hour) format gc time. err: %s", gc, err)
+	}
+	return gcDuration, nil
+}

pkg/util/util_test.go

@@ -110,6 +110,24 @@ func (s *utilSuite) TestStdLogger(c *C) {
 	c.Assert(entrys[2].Message, Matches, ".*hola:Goodbye!.*")
 }
 
+func (s *utilSuite) TestParseGCDuration(c *C) {
+	gc := "7"
+	expectDuration := 7 * 24 * time.Hour
+	duration, err := ParseGCDuration(gc)
+	c.Assert(err, IsNil)
+	c.Assert(duration, Equals, expectDuration)
+
+	gc = "30m"
+	expectDuration = 30 * time.Minute
+	duration, err = ParseGCDuration(gc)
+	c.Assert(err, IsNil)
+	c.Assert(duration, Equals, expectDuration)
+
+	gc = "7d"
+	_, err = ParseGCDuration(gc)
+	c.Assert(err, ErrorMatches, `unsupported gc time 7d, please use 7 or 8h\(max unit is hour\) format gc time.*`)
+}
+
 type getAddrIPSuite struct{}
 
 var _ = Suite(&getAddrIPSuite{})

pump/config.go

@@ -38,7 +38,7 @@ const (
 	defaultListenAddr        = "127.0.0.1:8250"
 	defautMaxMsgSize         = math.MaxInt32 // max grpc message size
 	defaultHeartbeatInterval = 2
-	defaultGC                = 7
+	defaultGC                = "7"
 	defaultDataDir           = "data.pump"
 
 	// default interval time to generate fake binlog, the unit is second
@@ -65,8 +65,8 @@ type Config struct {
 	EtcdDialTimeout   time.Duration
 	DataDir           string `toml:"data-dir" json:"data-dir"`
 	HeartbeatInterval int    `toml:"heartbeat-interval" json:"heartbeat-interval"`
-	// pump only stores binlog events whose ts >= current time - GC(day)
-	GC       int             `toml:"gc" json:"gc"`
+	// pump only stores binlog events whose ts >= current time - GC Time. The default unit is day
+	GC       string          `toml:"gc" json:"gc"`
 	LogFile  string          `toml:"log-file" json:"log-file"`
 	Security security.Config `toml:"security" json:"security"`
 
@@ -76,6 +76,7 @@ type Config struct {
 	MetricsInterval int
 	configFile      string
 	printVersion    bool
+	GCDuration      time.Duration `tome:"-" json:"-"`
 	tls             *tls.Config
 	Storage         storage.Config `toml:"storage" json:"storage"`
 }
@@ -100,7 +101,7 @@ func NewConfig() *Config {
 	fs.StringVar(&cfg.EtcdURLs, "pd-urls", defaultEtcdURLs, "a comma separated list of the PD endpoints")
 	fs.StringVar(&cfg.DataDir, "data-dir", "", "the path to store binlog data")
 	fs.IntVar(&cfg.HeartbeatInterval, "heartbeat-interval", defaultHeartbeatInterval, "number of seconds between heartbeat ticks")
-	fs.IntVar(&cfg.GC, "gc", defaultGC, "recycle binlog files older than gc days")
+	fs.StringVar(&cfg.GC, "gc", defaultGC, "recycle binlog files older than gc time. default unit is day. also accept 8h format time(max unit is hour)")
 	fs.StringVar(&cfg.LogLevel, "L", "info", "log level: debug, info, warn, error, fatal")
 	fs.StringVar(&cfg.MetricsAddr, "metrics-addr", "", "prometheus pushgateway address, leaves it empty will disable prometheus push")
 	fs.IntVar(&cfg.MetricsInterval, "metrics-interval", 15, "prometheus client push interval in second, set \"0\" to disable prometheus push")
@@ -176,6 +177,10 @@ func (cfg *Config) Parse(arguments []string) error {
 	util.AdjustString(&cfg.DataDir, defaultDataDir)
 	util.AdjustInt(&cfg.HeartbeatInterval, defaultHeartbeatInterval)
 
+	cfg.GCDuration, err = util.ParseGCDuration(cfg.GC)
+	if err != nil {
+		return err
+	}
 	return cfg.validate()
 }
 
@@ -186,8 +191,8 @@ func (cfg *Config) configFromFile(path string) error {
 // validate checks whether the configuration is valid
 func (cfg *Config) validate() error {
 	// check GC
-	if cfg.GC <= 0 {
-		return errors.Errorf("GC is %d, must bigger than 0", cfg.GC)
+	if cfg.GCDuration <= 0 {
+		return errors.Errorf("GC is %s, must bigger than 0", cfg.GC)
 	}
 
 	// check ListenAddr

pump/config_test.go

@@ -18,6 +18,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"time"
 
 	"github.com/BurntSushi/toml"
 	. "github.com/pingcap/check"
@@ -29,7 +30,7 @@ type testConfigSuite struct{}
 
 func (s *testConfigSuite) TestValidate(c *C) {
 	cfg := Config{}
-	cfg.GC = 1
+	cfg.GCDuration = 24 * time.Hour
 	cfg.ListenAddr = "http://:8250"
 	cfg.EtcdURLs = "http://192.168.10.23:7777"
 

pump/server.go

@@ -178,7 +178,7 @@ func NewServer(cfg *Config) (*Server, error) {
 		cancel:        cancel,
 		metrics:       metrics,
 		tiStore:       tiStore,
-		gcDuration:    time.Duration(cfg.GC) * 24 * time.Hour,
+		gcDuration:    cfg.GCDuration,
 		pdCli:         pdCli,
 		cfg:           cfg,
 		triggerGC:     make(chan time.Time),

pump/server_test.go

@@ -719,7 +719,7 @@ func (s *startServerSuite) TestStartPumpServer(c *C) {
 		ctx:        ctx,
 		cancel:     cancel,
 		tiStore:    nil,
-		gcDuration: time.Duration(cfg.GC) * 24 * time.Hour,
+		gcDuration: 24 * time.Hour,
 		pdCli:      nil,
 		cfg:        cfg,
 		triggerGC:  make(chan time.Time),