haminer: change buffered mode from max size to time interval

Previously, we forward the logs only if total collected logs in buffer
is greater or equal to 10.

This commit change the model into using time interval, where the logs
will be send every N seconds (default to 15 seconds).

1 files changed, 30 insertions, 6 deletions

diff --git a/config.go b/config.go
index 4bdd57b..d8c66e3 100644
--- a/config.go
+++ b/config.go
@@ -8,6 +8,7 @@ import (
 	"log"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/shuLhan/share/lib/ini"
 )
@@ -16,6 +17,7 @@ import (
 const (
 	ConfigKeyAcceptBackend        = "accept_backend"
 	ConfigKeyCaptureRequestHeader = "capture_request_header"
+	ConfigKeyForwardInterval      = "forward_interval"
 	ConfigKeyInfluxAPIWrite       = "influxdb_api_write"
 	ConfigKeyListen               = "listen"
 )
@@ -24,7 +26,7 @@ const (
 const (
 	defListenAddr      = "127.0.0.1"
 	defListenPort      = 5140
-	defMaxBufferedLogs = 10
+	defForwardInterval = 15 * time.Second
 )
 
 //
@@ -39,6 +41,9 @@ type Config struct {
 	// AcceptBackend list of backend to be filtered.
 	AcceptBackend []string
 
+	// ForwardInterval define an interval where logs will be forwarded.
+	ForwardInterval time.Duration
+
 	// List of request headers to be parsed and mapped as keys in halog
 	// output.
 	RequestHeaders []string
@@ -46,10 +51,6 @@ type Config struct {
 	// InfluxAPIWrite define HTTP API to write to Influxdb.
 	InfluxAPIWrite string
 
-	// MaxBufferedLogs define a number of logs that will be keep in buffer
-	// before being forwarded.
-	MaxBufferedLogs int
-
 	// retags contains list of pre-processing rules for tag.
 	retags []*tagPreprocessor
 }
@@ -62,7 +63,7 @@ func NewConfig() (cfg *Config) {
 	return &Config{
 		ListenAddr:      defListenAddr,
 		ListenPort:      defListenPort,
-		MaxBufferedLogs: defMaxBufferedLogs,
+		ForwardInterval: defForwardInterval,
 	}
 }
 
@@ -94,12 +95,35 @@ func (cfg *Config) Load(path string) {
 		cfg.InfluxAPIWrite = v
 	}
 
+	v, _ = in.Get("haminer", "", ConfigKeyForwardInterval)
+	cfg.SetForwardInterval(v)
+
 	sec := in.GetSection("preprocess", "tag")
 
 	cfg.parsePreprocessTag(sec)
 }
 
 //
+// SetForwardInterval set forward interval using string formatted, e.g. "20s"
+// where "s" represent unit time in "second".
+//
+func (cfg *Config) SetForwardInterval(v string) {
+	if len(v) == 0 {
+		return
+	}
+
+	var err error
+
+	cfg.ForwardInterval, err = time.ParseDuration(v)
+	if err != nil {
+		log.Println("SetForwardInterval: ", err)
+	}
+	if cfg.ForwardInterval < defForwardInterval {
+		cfg.ForwardInterval = defForwardInterval
+	}
+}
+
+//
 // SetListen will parse `v` value as "addr:port", and set config address and
 // port based on it.
 //