add Elasticsearch support

config.go

@@ -29,6 +29,7 @@ type Config struct {
 	GRPC              GRPCConfig    `json:"grpc"`
 	TLS               TLSConfig     `json:"tls"`
 	Influx            InfluxConfig  `json:"influx"`
+	Es                EsConfig      `json:"es"`
 	Kafka             *KafkaConfig  `json:"kafka"`
 	Paths             []PathsConfig `json:"paths"`
 	Log               LogConfig     `json:"log"`
@@ -130,6 +131,24 @@ func fillupDefaults(config *Config) {
 	if config.Influx.AccumulatorFrequency == 0 {
 		config.Influx.AccumulatorFrequency = DefaultIDBAccumulatorFreq
 	}
+	if config.Es.BatchFrequency == 0 {
+		config.Es.BatchFrequency = DefaultESDBBatchFreq
+	}
+	if config.Es.BatchSize == 0 {
+		config.Es.BatchSize = DefaultESDBBatchSize
+	}
+	if config.Es.WriteTimeout == 0 {
+		config.Es.WriteTimeout = DefaultESDBTimeout
+	}
+	if config.Es.FlushBytes == 0 {
+		config.Es.FlushBytes = DefaultESDBFlushBytes
+	}
+	if config.Es.FlushInterval == 0 {
+		config.Es.FlushInterval = DefaultESDBFlushInterval
+	}
+	if config.Es.AccumulatorFrequency == 0 {
+		config.Es.AccumulatorFrequency = DefaultESDBAccumulatorFreq
+	}
 }
 
 // ParseJSONConfigFileList parses file list config
@@ -357,6 +376,9 @@ func ConfigRead(jctx *JCtx, init bool, restart *bool) error {
 
 		go periodicStats(jctx)
 		influxInit(jctx)
+		if jctx.config.Es.Server != "" {
+			esInit(jctx)
+		}
 		if err := KafkaInit(jctx); err != nil {
 			log.Printf("KafkaInit error : %v", err)
 		}

defaults.go

@@ -12,6 +12,18 @@ const (
 	DefaultIDBAccumulatorFreq = 2000
 	//DefaultIDBTimeout is 30 seconds
 	DefaultIDBTimeout = 30
+	//DefaultESDBBatchSize to use if user has not provided in the config
+	DefaultESDBBatchSize = 1024 * 100
+	//DefaultESDBBatchFreq is 2 seconds
+	DefaultESDBBatchFreq = 2000
+	//DefaultESDBAccumulatorFreq is 2 seconds
+	DefaultESDBAccumulatorFreq = 2000
+	//DefaultESDBTimeout is 30 seconds
+	DefaultESDBTimeout = 30
+	//DefaultESDBFlushBytes is 500 bytes
+	DefaultESDBFlushBytes = 500
+	//DefaultESDBFlushInterval is 30 seconds
+	DefaultESDBFlushInterval = 30
 
 	// MatchExpressionXpath is for the pattern matching the xpath and key-value pairs
 	MatchExpressionXpath = "\\/([^\\/]*)\\[(.*?)+?(?:\\])"