Merge pull request #87 from rchicoli/organize-code

Organize code

Author: rchicoli

pkg/docker/config.go

@@ -11,8 +11,8 @@ import (
 	"github.com/docker/docker/daemon/logger"
 )
 
-// LogOpt ...
-type LogOpt struct {
+// Configuration is a type to all log-opt provided
+type Configuration struct {
 	index    string
 	tzpe     string
 	url      string
@@ -38,7 +38,7 @@ type Bulk struct {
 	stats         bool
 }
 
-// Grok ...
+// Grok filter
 type Grok struct {
 	grokPattern         string
 	grokPatternFrom     string
@@ -47,8 +47,8 @@ type Grok struct {
 	grokNamedCapture    bool
 }
 
-func defaultLogOpt() LogOpt {
-	return LogOpt{
+func newConfiguration() Configuration {
+	return Configuration{
 		index:    "docker-%Y.%m.%d",
 		tzpe:     "log",
 		timeout:  1,
@@ -95,7 +95,7 @@ func parseAddress(address string) error {
 }
 
 // ValidateLogOpt looks for es specific log option es-address.
-func (c *LogOpt) validateLogOpt(cfg map[string]string) error {
+func (c *Configuration) validateLogOpt(cfg map[string]string) error {
 	for key, v := range cfg {
 		switch key {
 		case "elasticsearch-url":
@@ -205,9 +205,6 @@ func (c *LogOpt) validateLogOpt(cfg map[string]string) error {
 			}
 			c.grokNamedCapture = s
 
-			// case "tag":
-		// case "labels":
-		// case "env":
 		default:
 			return fmt.Errorf("error: unknown log-opt: %q", v)
 		}
@@ -216,7 +213,7 @@ func (c *LogOpt) validateLogOpt(cfg map[string]string) error {
 	return nil
 }
 
-func getLogOptFields(fields string, info logger.Info) LogMessage {
+func getLogMessageFields(fields string, info logger.Info) LogMessage {
 	var l LogMessage
 	for _, v := range strings.Split(fields, ",") {
 		switch v {

pkg/docker/container.go

@@ -0,0 +1,75 @@
+package docker
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"path"
+	"syscall"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/docker/docker/api/types/plugins/logdriver"
+	"github.com/docker/docker/daemon/logger"
+	"github.com/rchicoli/docker-log-elasticsearch/pkg/elasticsearch"
+	"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
+}
+
+type pipeline struct {
+	group    *errgroup.Group
+	inputCh  chan logdriver.LogEntry
+	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) {
+
+	filename := path.Base(file)
+	log.WithField("fifo", file).Debug("created fifo file")
+
+	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()
+
+	f, err := fifo.OpenFifo(ctx, file, syscall.O_RDONLY, 0700)
+	if err != nil {
+		return nil, fmt.Errorf("could not open fifo: %q", err)
+	}
+
+	d.mu.Lock()
+	c := &container{stream: f}
+	d.logs[filename] = c
+	d.mu.Unlock()
+
+	return c, nil
+}
+
+// getContainer retrieves the container's configuration from memory
+func (d *Driver) getContainer(file string) (*container, error) {
+
+	filename := path.Base(file)
+
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	c, exists := d.logs[filename]
+	if !exists {
+		return nil, fmt.Errorf("error: logger not found for socket ID: %v", file)
+	}
+
+	return c, nil
+}