Merge pull request #68 from rchicoli/improve-pipeline

improve pipeline

Author: rchicoli

.gitattributes

@@ -0,0 +1,5 @@
+*.go linguist-language=Go
+tests/* linguist-vendored
+docker/* linguist-vendored
+config/* linguist-vendored
+scripts/* linguist-vendored

.travis.yml

@@ -59,7 +59,7 @@ matrix:
     - stage: release
       # env: MAJOR_RELEASE=true
       # it would be nice to map github labels instead
-      env: RELEASE_TAG=0.3.0
+      # env: RELEASE_TAG=0.3.0
       if: type IN (push)
       script:
         - ./scripts/git-release.sh

README.md

@@ -127,7 +127,7 @@ Before creating a docker container, a healthy instance of Elasticsearch service
   - *examples*: 2
 
 ###### elasticsearch-bulk-size ######
-  - *bulk-size* specifies when to flush based on the size (in bytes) of the actions currently added. Defaults to 5 MB and can be set to -1 to be disabled.
+  - *bulk-size* specifies when to flush based on the size (in bytes) of the actions currently added. Set to -1 to disable it.
   - *examples*: 1024, -1
 
 ###### elasticsearch-bulk-flush-interval ######
@@ -171,6 +171,11 @@ $ docker run --rm -ti \
     --log-opt elasticsearch-timeout=10 \
     --log-opt elasticsearch-version=5 \
     --log-opt elasticsearch-fields=containerID,containerName,containerImageID,containerImageName,containerCreated \
+    --log-opt elasticsearch-bulk-workers=1 \
+    --log-opt elasticsearch-bulk-actions=1000 \
+    --log-opt elasticsearch-bulk-size=1024 \
+    --log-opt elasticsearch-bulk-flush-interval=1s \
+    --log-opt elasticsearch-bulk-stats=false \
         alpine echo this is a test logging message
 ```
 

main.go

@@ -10,7 +10,7 @@ func main() {
 
 	h := sdk.NewHandler(`{"Implements": ["LoggingDriver"]}`)
 	d := docker.NewDriver()
-	docker.Handlers(&h, d)
+	docker.Handlers(&h, &d)
 	if err := h.ServeUnix(d.Name(), 0); err != nil {
 		panic(err)
 	}

pkg/docker/driver.go

@@ -8,7 +8,6 @@ import (
 	"io"
 	"log"
 	"os"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -33,13 +32,18 @@ var l = log.New(os.Stderr, "", 0)
 
 // Driver ...
 type Driver struct {
-	mu     *sync.Mutex
-	logs   map[string]*container
-	logger logger.Logger
-
-	esClient elasticsearch.Client
+	mu        *sync.Mutex
+	container *container
+	pipeline  pipeline
+	esClient  elasticsearch.Client
+}
 
-	groker *grok.Grok
+type pipeline struct {
+	group    *errgroup.Group
+	ctx      context.Context
+	outputCh chan LogMessage
+	inputCh  chan logdriver.LogEntry
+	stopCh   chan struct{}
 }
 
 type container struct {
@@ -115,33 +119,25 @@ func (l LogMessage) timeOmityEmpty() *time.Time {
 // NewDriver ...
 func NewDriver() Driver {
 	return Driver{
-		logs: make(map[string]*container),
-		mu:   new(sync.Mutex),
+		mu: new(sync.Mutex),
 	}
 }
 
 // StartLogging ...
-func (d Driver) StartLogging(file string, info logger.Info) error {
-	d.mu.Lock()
-	if _, exists := d.logs[file]; exists {
-		d.mu.Unlock()
-		return fmt.Errorf("error: logger for %q already exists", file)
+func (d *Driver) StartLogging(file string, info logger.Info) error {
 
-	}
-	d.mu.Unlock()
+	// log.Printf("info: starting log: %s\n", file)
 
 	ctx := context.Background()
 
-	// log.Printf("info: starting log: %s\n", file)
-
 	f, err := fifo.OpenFifo(ctx, file, syscall.O_RDONLY, 0700)
 	if err != nil {
 		return fmt.Errorf("error: opening logger fifo: %q", file)
 	}
 
 	d.mu.Lock()
 	c := &container{stream: f, info: info}
-	d.logs[file] = c
+	d.container = c
 	d.mu.Unlock()
 
 	cfg := defaultLogOpt()
@@ -154,17 +150,13 @@ func (d Driver) StartLogging(file string, info logger.Info) error {
 		return fmt.Errorf("error: cannot create an elasticsearch client: %v", err)
 	}
 
-	d.groker, err = grok.NewGrok(cfg.grokMatch, cfg.grokPattern, cfg.grokPatternFrom, cfg.grokPatternSplitter, cfg.grokNamedCapture)
-	if err != nil {
-		return err
-	}
-
-	msgCh := make(chan LogMessage)
-	logCh := make(chan logdriver.LogEntry)
+	d.pipeline.group, d.pipeline.ctx = errgroup.WithContext(ctx)
+	d.pipeline.inputCh = make(chan logdriver.LogEntry)
+	d.pipeline.outputCh = make(chan LogMessage)
+	// d.pipeline.stopCh = make(chan struct{})
 
-	g, ectx := errgroup.WithContext(ctx)
+	d.pipeline.group.Go(func() error {
 
-	g.Go(func() error {
 		dec := protoio.NewUint32DelimitedReader(c.stream, binary.BigEndian, 1e6)
 		defer dec.Close()
 
@@ -182,28 +174,34 @@ func (d Driver) StartLogging(file string, info logger.Info) error {
 			}
 
 			select {
-			case logCh <- buf:
-			case <-ectx.Done():
-				return ectx.Err()
+			case d.pipeline.inputCh <- buf:
+			// case <-d.pipeline.stopCh:
+			// 	return nil
+			case <-d.pipeline.ctx.Done():
+				return d.pipeline.ctx.Err()
 			}
 			buf.Reset()
 		}
 	})
 
-	g.Go(func() error {
+	d.pipeline.group.Go(func() error {
 
 		var logMessage string
 
 		// custom log message fields
 		msg := getLostashFields(cfg.fields, c.info)
 
-		for m := range logCh {
+		groker, err := grok.NewGrok(cfg.grokMatch, cfg.grokPattern, cfg.grokPatternFrom, cfg.grokPatternSplitter, cfg.grokNamedCapture)
+		if err != nil {
+			return err
+		}
+
+		for m := range d.pipeline.inputCh {
 
 			logMessage = string(m.Line)
 
 			// BUG: (17.09.0~ce-0~debian) docker run command throws lots empty line messages
-			// TODO: profile: check for resource consumption
-			if len(strings.TrimSpace(logMessage)) == 0 {
+			if len(m.Line) == 0 {
 				// TODO: add log debug level
 				continue
 			}
@@ -212,40 +210,63 @@ func (d Driver) StartLogging(file string, info logger.Info) error {
 			msg.Partial = m.Partial
 			msg.TimeNano = m.TimeNano
 
-			msg.GrokLine, msg.Line, err = d.groker.ParseLine(cfg.grokMatch, logMessage, m.Line)
+			// TODO: create a PR to grok upstream for parsing bytes
+			// so that we avoid having to convert the message to string
+			msg.GrokLine, msg.Line, err = groker.ParseLine(cfg.grokMatch, logMessage, m.Line)
 			if err != nil {
 				l.Printf("error: [%v] parsing log message: %v\n", c.info.ID(), err)
 			}
 
+			// l.Printf("INFO: grokline: %v\n", msg.GrokLine)
+			// l.Printf("INFO: line: %v\n", string(msg.Line))
+
 			select {
-			case msgCh <- msg:
-			case <-ectx.Done():
-				return ectx.Err()
+			case d.pipeline.outputCh <- msg:
+			case <-d.pipeline.ctx.Done():
+				return d.pipeline.ctx.Err()
 			}
+
 		}
 
 		return nil
 	})
 
-	g.Go(func() error {
+	d.pipeline.group.Go(func() error {
 
-		err := d.esClient.NewBulkProcessorService(ectx, cfg.Bulk.workers, cfg.Bulk.actions, cfg.Bulk.size, cfg.Bulk.flushInterval, cfg.Bulk.stats)
+		err := d.esClient.NewBulkProcessorService(d.pipeline.ctx, cfg.Bulk.workers, cfg.Bulk.actions, cfg.Bulk.size, cfg.Bulk.flushInterval, cfg.Bulk.stats)
 		if err != nil {
-			l.Printf("error creating bulk processor: %v", err)
+			l.Printf("error creating bulk processor: %v\n", err)
 		}
 
-		for {
+		// l.Printf("receving from output\n")
+
+		for doc := range d.pipeline.outputCh {
+
+			// l.Printf("sending doc: %#v\n", doc.GrokLine)
+			d.esClient.Add(cfg.index, cfg.tzpe, doc)
+
 			select {
-			case doc := <-msgCh:
-				d.esClient.Add(cfg.index, cfg.tzpe, doc)
-			case <-ectx.Done():
-				return ectx.Err()
+			case <-d.pipeline.ctx.Done():
+				return d.pipeline.ctx.Err()
+			default:
 			}
 		}
+		// for {
+		// 	select {
+		// 	case doc := <-d.pipeline.outputCh:
+		// 		l.Printf("sending doc: %#v\n", doc.GrokLine)
+		// 		d.esClient.Add(cfg.index, cfg.tzpe, doc)
+
+		// 	case <-d.pipeline.ctx.Done():
+		// 		return d.pipeline.ctx.Err()
+		// 	}
+		// }
+
+		return nil
 	})
 
 	// TODO: create metrics from stats
-	// g.Go(func() error {
+	// d.pipeline.group.Go(func() error {
 	// 	stats := d.esClient.Stats()
 
 	// 	fields := log.Fields{
@@ -266,32 +287,51 @@ func (d Driver) StartLogging(file string, info logger.Info) error {
 	// 	}
 	// })
 
-	// Check whether any goroutines failed.
-	// if err := g.Wait(); err != nil {
-	// 	panic(err)
-	// }
-
 	return nil
 }
 
 // StopLogging ...
-func (d Driver) StopLogging(file string) error {
+// TODO: change api interface
+func (d *Driver) StopLogging(file string) error {
 
 	// log.Infof("info: stopping log: %s\n", file)
 
-	d.mu.Lock()
-	c, ok := d.logs[file]
-	if ok {
-		c.stream.Close()
-		delete(d.logs, file)
+	// TODO: count how many docs are in the queue before shutting down
+	// alternative: sleep flush interval time
+	time.Sleep(10 * time.Second)
+
+	if d.container != nil {
+		// l.Printf("INFO container: %v", d.container)
+		if err := d.container.stream.Close(); err != nil {
+			l.Printf("error: [%v] closing container stream: %v", d.container.info.ID(), err)
+		}
 	}
-	d.mu.Unlock()
 
 	if d.esClient != nil {
-		d.esClient.Close()
+		// l.Printf("INFO client: %v", d.esClient)
+		if err := d.esClient.Close(); err != nil {
+			l.Printf("error: closing client connection: %v", err)
+		}
 		d.esClient.Stop()
 	}
 
+	if d.pipeline.group != nil {
+		// l.Printf("INFO with pipeline: %v", d.pipeline)
+
+		// close(d.pipeline.inputCh)
+		// close(d.pipeline.outputCh)
+		// d.pipeline.stopCh <- struct{}{}
+
+		// TODO: close channels gracefully
+		// sadly the errgroup does not export cancel function
+		// create a PR to golang team with Stop() function
+		// Check whether any goroutines failed.
+		// if err := d.pipeline.group.Wait(); err != nil {
+		// 	l.Printf("error with pipeline: %v", err)
+		// }
+	}
+	// l.Printf("INFO done stop logging")
+
 	return nil
 }
 

pkg/extension/grok/grok.go

@@ -47,7 +47,7 @@ func NewGrok(grokMatch, grokPattern, grokPatternFrom, grokPatternSplitter string
 }
 
 // ParseLine ...
-func (g Grok) ParseLine(pattern, logMessage string, line []byte) (map[string]string, []byte, error) {
+func (g *Grok) ParseLine(pattern, logMessage string, line []byte) (map[string]string, []byte, error) {
 
 	if g.Grok == nil {
 		return nil, line, nil