Add experimental Docker json-file prospector (#5402)

CHANGELOG.asciidoc

@@ -107,6 +107,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Remove error log from runnerfactory as error is returned by API. {pull}5085[5085]
 - Remove error log from runnerfactory as error is returned by API. {pull}5085[5085]
 - Add Kubernetes manifests to deploy Filebeat. {pull}5349[5349]
+- Add experimental Docker `json-file` prospector . {pull}5402[5402]
 
 *Heartbeat*
 

filebeat/_meta/fields.common.yml

@@ -23,6 +23,12 @@
       description: >
         The content of the line read from the log file.
 
+    - name: stream
+      type: keyword
+      required: false
+      description: >
+        Log stream when reading container logs, can be 'stdout' or 'stderr'
+
     - name: prospector.type
       required: true
       description: >

filebeat/docs/fields.asciidoc

@@ -934,6 +934,16 @@ required: True
 The content of the line read from the log file.
 
 
+[float]
+=== `stream`
+
+type: keyword
+
+required: False
+
+Log stream when reading container logs, can be 'stdout' or 'stderr'
+
+
 [float]
 === `prospector.type`
 

filebeat/harvester/reader/docker_json.go

@@ -0,0 +1,54 @@
+package reader
+
+import (
+	"bytes"
+	"encoding/json"
+	"time"
+
+	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/pkg/errors"
+)
+
+// DockerJSON processor renames a given field
+type DockerJSON struct {
+	reader Reader
+}
+
+type dockerLog struct {
+	Timestamp string `json:"time"`
+	Log       string `json:"log"`
+	Stream    string `json:"stream"`
+}
+
+// NewDockerJSON creates a new reader renaming a field
+func NewDockerJSON(r Reader) *DockerJSON {
+	return &DockerJSON{reader: r}
+}
+
+// Next returns the next line.
+func (p *DockerJSON) Next() (Message, error) {
+	message, err := p.reader.Next()
+	if err != nil {
+		return message, err
+	}
+
+	var line dockerLog
+	dec := json.NewDecoder(bytes.NewReader(message.Content))
+	if err = dec.Decode(&line); err != nil {
+		return message, errors.Wrap(err, "decoding docker JSON")
+	}
+
+	// Parse timestamp
+	ts, err := time.Parse(time.RFC3339, line.Timestamp)
+	if err != nil {
+		return message, errors.Wrap(err, "parsing docker timestamp")
+	}
+
+	message.AddFields(common.MapStr{
+		"stream": line.Stream,
+	})
+	message.Content = []byte(line.Log)
+	message.Ts = ts
+	return message, nil
+}

filebeat/harvester/reader/docker_json_test.go

@@ -0,0 +1,60 @@
+package reader
+
+import (
+	"testing"
+	"time"
+
+	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDockerJSON(t *testing.T) {
+	tests := []struct {
+		input           []byte
+		expectedError   bool
+		expectedMessage Message
+	}{
+		// Common log message
+		{
+			input: []byte(`{"log":"1:M 09 Nov 13:27:36.276 # User requested shutdown...\n","stream":"stdout","time":"2017-11-09T13:27:36.277747246Z"}`),
+			expectedMessage: Message{
+				Content: []byte("1:M 09 Nov 13:27:36.276 # User requested shutdown...\n"),
+				Fields:  common.MapStr{"stream": "stdout"},
+				Ts:      time.Date(2017, 11, 9, 13, 27, 36, 277747246, time.UTC),
+			},
+		},
+		// Wrong JSON
+		{
+			input:         []byte(`this is not JSON`),
+			expectedError: true,
+		},
+		// Missing time
+		{
+			input:         []byte(`{"log":"1:M 09 Nov 13:27:36.276 # User requested shutdown...\n","stream":"stdout"}`),
+			expectedError: true,
+		},
+	}
+
+	for _, test := range tests {
+		r := mockReader{message: test.input}
+		json := NewDockerJSON(r)
+		message, err := json.Next()
+
+		assert.Equal(t, test.expectedError, err != nil)
+
+		if !test.expectedError {
+			assert.EqualValues(t, test.expectedMessage, message)
+		}
+	}
+}
+
+type mockReader struct {
+	message []byte
+}
+
+func (m mockReader) Next() (Message, error) {
+	return Message{
+		Content: m.message,
+	}, nil
+}

filebeat/harvester/util.go

@@ -4,10 +4,11 @@ import "github.com/elastic/beats/libbeat/common/match"
 
 // Contains available prospector types
 const (
-	LogType   = "log"
-	StdinType = "stdin"
-	RedisType = "redis"
-	UdpType   = "udp"
+	LogType    = "log"
+	StdinType  = "stdin"
+	RedisType  = "redis"
+	UdpType    = "udp"
+	DockerType = "docker"
 )
 
 // MatchAny checks if the text matches any of the regular expressions

filebeat/include/list.go

@@ -8,6 +8,7 @@ package include
 
 import (
 	// This list is automatically generated by `make imports`
+	_ "github.com/elastic/beats/filebeat/prospector/docker"
 	_ "github.com/elastic/beats/filebeat/prospector/log"
 	_ "github.com/elastic/beats/filebeat/prospector/redis"
 	_ "github.com/elastic/beats/filebeat/prospector/stdin"

filebeat/prospector/docker/config.go

@@ -0,0 +1,17 @@
+package docker
+
+var defaultConfig = config{
+	Containers: containers{
+		IDs:  []string{},
+		Path: "/var/lib/docker/containers",
+	},
+}
+
+type config struct {
+	Containers containers `config:"containers"`
+}
+
+type containers struct {
+	IDs  []string `config:"ids"`
+	Path string   `config:"path"`
+}

filebeat/prospector/docker/prospector.go

@@ -0,0 +1,42 @@
+package docker
+
+import (
+	"path"
+
+	"github.com/elastic/beats/filebeat/channel"
+	"github.com/elastic/beats/filebeat/prospector"
+	"github.com/elastic/beats/filebeat/prospector/log"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+
+	"github.com/pkg/errors"
+)
+
+func init() {
+	err := prospector.Register("docker", NewProspector)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// NewProspector creates a new docker prospector
+func NewProspector(cfg *common.Config, outletFactory channel.Factory, context prospector.Context) (prospector.Prospectorer, error) {
+	cfgwarn.Experimental("Docker prospector is enabled.")
+
+	config := defaultConfig
+	if err := cfg.Unpack(&config); err != nil {
+		return nil, errors.Wrap(err, "reading docker prospector config")
+	}
+
+	// Wrap log prospector with custom docker settings
+	if len(config.Containers.IDs) > 0 {
+		for idx, containerID := range config.Containers.IDs {
+			cfg.SetString("paths", idx, path.Join(config.Containers.Path, containerID, "*.log"))
+		}
+	}
+
+	if err := cfg.SetBool("docker-json", -1, true); err != nil {
+		return nil, errors.Wrap(err, "update prospector config")
+	}
+	return log.NewProspector(cfg, outletFactory, context)
+}

filebeat/prospector/log/config.go

@@ -83,6 +83,9 @@ type config struct {
 	MaxBytes     int                     `config:"max_bytes" validate:"min=0,nonzero"`
 	Multiline    *reader.MultilineConfig `config:"multiline"`
 	JSON         *reader.JSONConfig      `config:"json"`
+
+	// Hidden on purpose, used by the docker prospector:
+	DockerJSON bool `config:"docker-json"`
 }
 
 type LogConfig struct {

filebeat/prospector/log/harvester.go

@@ -126,6 +126,8 @@ func (h *Harvester) open() error {
 		return h.openStdin()
 	case harvester.LogType:
 		return h.openFile()
+	case harvester.DockerType:
+		return h.openFile()
 	default:
 		return fmt.Errorf("Invalid harvester type: %+v", h.config)
 	}
@@ -515,6 +517,11 @@ func (h *Harvester) newLogFileReader() (reader.Reader, error) {
 		return nil, err
 	}
 
+	if h.config.DockerJSON {
+		// Docker json-file format, add custom parsing to the pipeline
+		r = reader.NewDockerJSON(r)
+	}
+
 	if h.config.JSON != nil {
 		r = reader.NewJSON(r, h.config.JSON)
 	}