Merge pull request #88 from rchicoli/rewrite-handler

docker: rewrite handler

Author: rchicoli

Makefile

@@ -18,7 +18,7 @@ DOCKER_COMPOSE      := $(shell which docker-compose)
 
 all: clean docker_build plugin_create plugin_enable clean
 
-local: clean build unit_tests plugin_create plugin_enable clean
+local: clean build unit_tests plugin_create plugin_set plugin_enable clean
 
 clean:
 	@echo ""
@@ -60,6 +60,10 @@ plugin_push:
 	@echo ""
 	docker plugin push $(PLUGIN_NAME):$(PLUGIN_TAG)
 
+plugin_set:
+	@echo ""
+	docker plugin set $(PLUGIN_NAME):$(PLUGIN_TAG) LOG_LEVEL=debug
+
 push: plugin_push
 
 client_version:

main.go

@@ -1,10 +1,14 @@
 package main
 
 import (
+	"encoding/json"
+	"errors"
 	"fmt"
+	"net/http"
 	"os"
 
 	log "github.com/Sirupsen/logrus"
+	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/go-plugins-helpers/sdk"
 
 	"github.com/rchicoli/docker-log-elasticsearch/pkg/docker"
@@ -17,6 +21,23 @@ var logLevels = map[string]log.Level{
 	"error": log.ErrorLevel,
 }
 
+// StartLoggingRequest payload
+type StartLoggingRequest struct {
+	File string      `json:"file,omitempty"`
+	Info logger.Info `json:"info,omitempty"`
+}
+
+// StopLoggingRequest payload
+type StopLoggingRequest struct {
+	File string `json:"file,omitempty"`
+}
+
+// CapabilitiesResponse payload
+type CapabilitiesResponse struct {
+	Cap logger.Capability `json:"capabilities,omitempty"`
+	Err string            `json:"err,omitempty"`
+}
+
 func main() {
 
 	levelVal := os.Getenv("LOG_LEVEL")
@@ -35,8 +56,79 @@ func main() {
 
 	h := sdk.NewHandler(`{"Implements": ["LoggingDriver"]}`)
 	d := docker.NewDriver()
-	docker.Handlers(&h, d)
+
+	h.HandleFunc("/LogDriver.StartLogging", func(w http.ResponseWriter, r *http.Request) {
+		var req StartLoggingRequest
+
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			respond(fmt.Errorf("error: could not decode payload: %v", err), w)
+			return
+		}
+
+		if req.Info.ContainerID == "" {
+			respond(errors.New("error: could not find containerID in request payload"), w)
+			return
+		}
+
+		err := d.StartLogging(req.File, req.Info)
+
+		// reader returns EOF if NoBody is sent
+		respond(err, w)
+	})
+
+	h.HandleFunc("/LogDriver.StopLogging", func(w http.ResponseWriter, r *http.Request) {
+		var req StopLoggingRequest
+
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		err := d.StopLogging(req.File)
+
+		// reader returns EOF if NoBody is sent
+		respond(err, w)
+	})
+
+	h.HandleFunc("/LogDriver.Capabilities", func(w http.ResponseWriter, r *http.Request) {
+		json.NewEncoder(w).Encode(&CapabilitiesResponse{
+			Cap: logger.Capability{ReadLogs: false},
+		})
+	})
+
+	// TODO: implement read logs from elasticsearch
+	// h.HandleFunc("/LogDriver.ReadLogs", func(w http.ResponseWriter, r *http.Request) {
+	// 	var req ReadLogsRequest
+	// 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+	// 		http.Error(w, err.Error(), http.StatusBadRequest)
+	// 		return
+	// 	}
+
+	// 	stream, err := d.ReadLogs(req.Info, req.Config)
+	// 	if err != nil {
+	// 		http.Error(w, err.Error(), http.StatusInternalServerError)
+	// 		return
+	// 	}
+	// 	defer stream.Close()
+
+	// 	w.Header().Set("Content-Type", "application/x-json-stream")
+	// 	wf := ioutils.NewWriteFlusher(w)
+	// 	io.Copy(wf, stream)
+
+	// })
+
 	if err := h.ServeUnix(d.Name(), 0); err != nil {
-		panic(err)
+		log.WithError(err).Fatal("error: serving unix")
+	}
+}
+
+type response struct {
+	Err string `json:"err,omitempty"`
+}
+
+func respond(err error, w http.ResponseWriter) {
+	var res response
+	if err != nil {
+		res.Err = err.Error()
 	}
+	json.NewEncoder(w).Encode(&res)
 }

pkg/docker/container.go

@@ -1,26 +1,28 @@
 package docker
 
 import (
+	"bytes"
 	"context"
-	"fmt"
+	"encoding/binary"
 	"io"
-	"path"
-	"syscall"
+	"os"
+	"strings"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/docker/api/types/plugins/logdriver"
 	"github.com/docker/docker/daemon/logger"
+	protoio "github.com/gogo/protobuf/io"
 	"github.com/rchicoli/docker-log-elasticsearch/pkg/elasticsearch"
+	"github.com/rchicoli/docker-log-elasticsearch/pkg/extension/grok"
 	"github.com/robfig/cron"
-	"github.com/tonistiigi/fifo"
 	"golang.org/x/sync/errgroup"
 )
 
 type container struct {
 	cron      *cron.Cron
 	esClient  elasticsearch.Client
 	indexName string
-	info      logger.Info
 	logger    *log.Entry
 	pipeline  pipeline
 	stream    io.ReadCloser
@@ -32,44 +34,179 @@ type pipeline struct {
 	outputCh chan LogMessage
 }
 
-// newContainer stores the container's configuration in memory
-// and returns a pointer to the container
-func (d *Driver) newContainer(ctx context.Context, file string) (*container, error) {
+// Read reads messages from proto buffer
+func (c *container) Read(ctx context.Context) error {
 
-	filename := path.Base(file)
-	log.WithField("fifo", file).Debug("created fifo file")
+	c.logger.Debug("starting pipeline: Read")
 
-	d.mu.Lock()
-	if _, exists := d.logs[filename]; exists {
-		return nil, fmt.Errorf("error: a logger for this container already exists: %s", filename)
-	}
-	d.mu.Unlock()
+	c.pipeline.group.Go(func() error {
 
-	f, err := fifo.OpenFifo(ctx, file, syscall.O_RDONLY, 0700)
-	if err != nil {
-		return nil, fmt.Errorf("could not open fifo: %q", err)
-	}
+		dec := protoio.NewUint32DelimitedReader(c.stream, binary.BigEndian, 1e6)
+		defer func() {
+			c.logger.Info("closing docker stream")
+			dec.Close()
+			close(c.pipeline.inputCh)
+		}()
 
-	d.mu.Lock()
-	c := &container{stream: f}
-	d.logs[filename] = c
-	d.mu.Unlock()
+		var buf logdriver.LogEntry
+		var err error
 
-	return c, nil
+		for {
+			if err = dec.ReadMsg(&buf); err != nil {
+				if err == io.EOF {
+					c.logger.Debug("shutting down reader eof")
+					break
+				}
+				// the connection has been closed
+				// stop looping and close the input channel
+				// read /proc/self/fd/6: file already closed
+				if strings.Contains(err.Error(), os.ErrClosed.Error()) {
+					c.logger.WithError(err).Debug("shutting down fifo: closed by the writer")
+					break
+				}
+				if err != nil {
+					// the connection has been closed
+					// stop looping and closing the input channel
+					// read /proc/self/fd/6: file already closed
+					c.logger.WithError(err).Debug("shutting down fifo")
+					break
+					// do not return, otherwise group.Go closes the pipeline
+					// return err
+				}
+
+				dec = protoio.NewUint32DelimitedReader(c.stream, binary.BigEndian, 1e6)
+			}
+
+			// in case docker run command throws lots empty line messages
+			if len(bytes.TrimSpace(buf.Line)) == 0 {
+				c.logger.WithField("line", string(buf.Line)).Debug("trim space")
+				continue
+			}
+
+			select {
+			case c.pipeline.inputCh <- buf:
+			case <-ctx.Done():
+				c.logger.WithError(ctx.Err()).Error("closing read pipeline")
+				return ctx.Err()
+			}
+			buf.Reset()
+		}
+
+		return nil
+	})
+
+	return nil
 }
 
-// getContainer retrieves the container's configuration from memory
-func (d *Driver) getContainer(file string) (*container, error) {
+// Parse filters line messages
+func (c *container) Parse(ctx context.Context, info logger.Info, fields, grokMatch, grokPattern, grokPatternFrom, grokPatternSplitter string, grokNamedCapture bool) error {
+
+	c.logger.Debug("starting pipeline: Parse")
+
+	c.pipeline.group.Go(func() error {
+		defer close(c.pipeline.outputCh)
 
-	filename := path.Base(file)
+		groker, err := grok.NewGrok(grokMatch, grokPattern, grokPatternFrom, grokPatternSplitter, grokNamedCapture)
+		if err != nil {
+			return err
+		}
 
-	d.mu.Lock()
-	defer d.mu.Unlock()
+		var logMessage string
+		// custom log message fields
+		msg := getLogMessageFields(fields, info)
 
-	c, exists := d.logs[filename]
-	if !exists {
-		return nil, fmt.Errorf("error: logger not found for socket ID: %v", file)
-	}
+		for m := range c.pipeline.inputCh {
 
-	return c, nil
+			logMessage = string(m.Line)
+
+			// create message
+			msg.Source = m.Source
+			msg.Partial = m.Partial
+			msg.TimeNano = m.TimeNano
+
+			// 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(grokMatch, logMessage, m.Line)
+			if err != nil {
+				c.logger.WithError(err).Error("could not parse line with grok")
+			}
+
+			select {
+			case c.pipeline.outputCh <- msg:
+			case <-ctx.Done():
+				c.logger.WithError(ctx.Err()).Error("closing parse pipeline")
+				return ctx.Err()
+			}
+
+		}
+
+		return nil
+	})
+
+	return nil
 }
+
+// Log sends messages to Elasticsearch Bulk Service
+func (c *container) Log(ctx context.Context, workers, actions, size int, flushInterval time.Duration, stats bool, indexName, tzpe string) error {
+
+	c.logger.Debug("starting pipeline: Log")
+
+	c.pipeline.group.Go(func() error {
+
+		err := c.esClient.NewBulkProcessorService(ctx, workers, actions, size, flushInterval, stats)
+		if err != nil {
+			c.logger.WithError(err).Error("could not create bulk processor")
+		}
+
+		defer func() {
+			if err := c.esClient.Flush(); err != nil {
+				c.logger.WithError(err).Error("could not flush queue")
+			}
+
+			if err := c.esClient.Close(); err != nil {
+				c.logger.WithError(err).Error("could not close client connection")
+			}
+			c.esClient.Stop()
+		}()
+
+		for doc := range c.pipeline.outputCh {
+
+			c.esClient.Add(indexName, tzpe, doc)
+
+			select {
+			case <-ctx.Done():
+				c.logger.WithError(ctx.Err()).Error("closing log pipeline")
+				return ctx.Err()
+			default:
+			}
+		}
+		return nil
+	})
+
+	return nil
+}
+
+// Stats shows metrics related to the bulk service
+// func (d *Driver) Stats(filename string, config Configuration) error {
+// TODO: create metrics from stats
+// d.pipeline.group.Go(func() error {
+// 	stats := d.esClient.Stats()
+
+// 	fields := log.Fields{
+// 		"flushed":   stats.Flushed,
+// 		"committed": stats.Committed,
+// 		"indexed":   stats.Indexed,
+// 		"created":   stats.Created,
+// 		"updated":   stats.Updated,
+// 		"succeeded": stats.Succeeded,
+// 		"failed":    stats.Failed,
+// 	}
+
+// 	for i, w := range stats.Workers {
+// 		fmt.Printf("Worker %d: Number of requests queued: %d\n", i, w.Queued)
+// 		fmt.Printf("           Last response time       : %v\n", w.LastDuration)
+// 		fields[fmt.Sprintf("w%d.queued", i)] = w.Queued
+// 		fields[fmt.Sprintf("w%d.lastduration", i)] = w.LastDuration
+// 	}
+// })
+// }