Add Flush method to sink interface

metric-collector.go

@@ -269,6 +269,10 @@ func main() {
 						tmpPoints = tmpPoints[1:]
 					}
 				}
+
+				if err := sink.Flush(); err != nil {
+					log.Printf("sink error: %s\n", err)
+				}
 			}
 		}
 	}()

sinks/influxSink.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
 	lp "github.com/influxdata/line-protocol"
@@ -57,8 +58,8 @@ func (s *InfluxSink) Init(config SinkConfig) error {
 }
 
 func (s *InfluxSink) Write(point lp.MutableMetric) error {
-	var tags map[string]string
+	tags := map[string]string{}
-	var fields map[string]interface{}
+	fields := map[string]interface{}{}
 	for _, t := range point.TagList() {
 		tags[t.Key] = t.Value
 	}
@@ -70,6 +71,10 @@ func (s *InfluxSink) Write(point lp.MutableMetric) error {
 	return err
 }
 
+func (s *InfluxSink) Flush() error {
+	return nil
+}
+
 func (s *InfluxSink) Close() {
 	log.Print("Closing InfluxDB connection")
 	s.client.Close()

sinks/metricSink.go

@@ -29,5 +29,6 @@ type Sink struct {
 type SinkFuncs interface {
 	Init(config SinkConfig) error
 	Write(point lp.MutableMetric) error
+	Flush() error
 	Close()
 }

sinks/natsSink.go

@@ -79,6 +79,10 @@ func (s *NatsSink) Write(point lp.MutableMetric) error {
 	return nil
 }
 
+func (s *NatsSink) Flush() error {
+	return nil
+}
+
 func (s *NatsSink) Close() {
 	log.Print("Closing Nats connection")
 	if s.client != nil {

sinks/stdoutSink.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"strings"
+
 	//	"time"
 	lp "github.com/influxdata/line-protocol"
 )
@@ -54,6 +55,10 @@ func (s *StdoutSink) Write(point lp.MutableMetric) error {
 	return nil
 }
 
+func (s *StdoutSink) Flush() error {
+	return nil
+}
+
 func (s *StdoutSink) Close() {
 	return
 }