Merge pull request #1987 from hunchback/exporters-store-refactor

exporters: split storer and writer functions

Author: safchain

contrib/exporters/allinone/allinone.yml.default

@@ -18,19 +18,20 @@ pipeline:
     sa:
       # exclude_started_flows: true
   store:
+    type: buffered
+    buffered:
+      object_prefix: logs
+      bucket: bucket
+      max_flows_per_object: 6000
+      max_seconds_per_object: 60
+      max_seconds_per_stream: 86400
+      max_flow_array_size: 100000
+  write:
     type: s3
     s3:
-      # -- client parames --
       endpoint: http://127.0.0.1:9000
       region: local
-      bucket: bucket
       access_key: user
       secret_key: password
       # api_key: key
       # iam_endpoint: https://iam.cloud.ibm.com/identity/token
-      object_prefix: logs
-      # -- bulk store params --
-      # max_flows_per_object: 6000
-      # max_seconds_per_object: 60
-      # max_seconds_per_stream: 86400
-      # max_flow_array_size: 100000

contrib/exporters/core/Makefile

@@ -5,11 +5,12 @@ SRCS := \
 	filter.go \
 	pipeline.go \
 	main.go \
-	s3_client.go \
-	store_stdout.go \
-	store_s3.go \
+	store_buffered.go \
+	store_direct.go \
 	subscriber.go \
 	transform.go \
+	write_s3.go \
+	write_stdout.go \
 
 .PHONY: all
 all:

contrib/exporters/core/pipeline.go

@@ -65,6 +65,11 @@ type Storer interface {
 	SetPipeline(p *Pipeline)
 }
 
+// Writer allows uploading objects to an object storage service
+type Writer interface {
+	Write(bucket, objectKey, data, contentType, contentEncoding string, metadata map[string]*string) error
+}
+
 // Handler used for creating a phase handler from configuration
 type Handler = func(cfg *viper.Viper) (interface{}, error)
 
@@ -79,6 +84,7 @@ var (
 	EncoderHandlers     HandlersMap
 	CompressorHandlers  HandlersMap
 	StorerHandlers      HandlersMap
+	WriterHandlers      HandlersMap
 )
 
 // Register associates a handler with its' label
@@ -119,8 +125,12 @@ func init() {
 	CompressorHandlers.Register("gzip", NewCompressGzip, false)
 
 	StorerHandlers = make(HandlersMap)
-	StorerHandlers.Register("stdout", NewStoreStdout, true)
-	StorerHandlers.Register("s3", NewStoreS3, false)
+	StorerHandlers.Register("buffered", NewStoreBuffered, true)
+	StorerHandlers.Register("direct", NewStoreDirect, false)
+
+	WriterHandlers = make(HandlersMap)
+	WriterHandlers.Register("s3", NewWriteS3, true)
+	WriterHandlers.Register("stdout", NewWriteStdout, false)
 }
 
 // Pipeline manager
@@ -133,6 +143,7 @@ type Pipeline struct {
 	Encoder     Encoder
 	Compressor  Compressor
 	Storer      Storer
+	Writer      Writer
 }
 
 // NewPipeline defines the pipeline elements
@@ -167,13 +178,19 @@ func NewPipeline(cfg *viper.Viper) (*Pipeline, error) {
 		return nil, err
 	}
 
+	writer, err := WriterHandlers.Init(cfg, "write")
+	if err != nil {
+		return nil, err
+	}
+
 	p := &Pipeline{
 		Transformer: transformer.(Transformer),
 		Classifier:  classifier.(Classifier),
 		Filterer:    filterer.(Filterer),
 		Encoder:     encoder.(Encoder),
 		Compressor:  compressor.(Compressor),
 		Storer:      storer.(Storer),
+		Writer:      writer.(Writer),
 	}
 	storer.(Storer).SetPipeline(p)