[Logs]: Added a support for journald (#1618)

.circleci/images/builder/Dockerfile

@@ -5,7 +5,7 @@ RUN sed -i 's/^#\s*\(deb.*universe\)$/\1/g' /etc/apt/sources.list \
     && sed -i 's/main/main contrib non-free/' /etc/apt/sources.list
 
 RUN apt-get update && apt-get install -y python2.7-dev autoconf autogen intltool libssl1.0-dev
-RUN apt-get install -y libsnmp-base libsnmp-dev libpq-dev snmp-mibs-downloader
+RUN apt-get install -y libsnmp-base libsnmp-dev libpq-dev snmp-mibs-downloader libsystemd-dev
 
 # Ruby,,,
 RUN mkdir -p /usr/local/etc \

Gopkg.toml

@@ -98,6 +98,10 @@
   name = "github.com/spf13/viper"
   version = "~v1.0.0"
 
+[[constraint]]
+  name = "github.com/coreos/go-systemd"
+  version = "~v16"
+
 [[constraint]]
   name = "github.com/stretchr/testify"
   version = "~v1.2.1"

LICENSE-3rdparty.csv

@@ -4,6 +4,7 @@ core,github.com/beevik/ntp,BSD-2-Clause
 core,github.com/cihub/seelog,BSD-3-Clause
 core,github.com/codemirror/CodeMirror,MIT
 core,github.com/coreos/etcd,Apache-2.0
+core,github.com/coreos/go-systemd,Apache-2.0
 core,github.com/DataDog/agent-payload,BSD-3-Clause
 core,github.com/DataDog/gohai,MIT
 core,github.com/DataDog/mmh3,MIT

README.md

@@ -33,7 +33,7 @@ To start working on the Agent, you can build the `master` branch:
 2. cd into the project folder: `cd $GOPATH/src/github.com/DataDog/datadog-agent`.
 3. install project's dependencies: `invoke deps`.
    Make sure that `$GOPATH/bin` is in your `$PATH` otherwise this step might fail.
-4. build the whole project with `invoke agent.build --build-exclude=snmp`
+4. build the whole project with `invoke agent.build --build-exclude=snmp,systemd`
 
 Please refer to the [Agent Developer Guide](docs/dev/README.md) for more details.
 

docs/dev/agent_build.md

@@ -16,7 +16,7 @@ invoke agent.build --build-include=zstd,etcd,cpython
 Conversely, if you want to exclude something:
 
 ```
-invoke agent.build --build-exclude=snmp,cpython
+invoke agent.build --build-exclude=snmp,systemd,cpython
 ```
 
 This is the complete list of the available components:
@@ -35,6 +35,7 @@ This is the complete list of the available components:
 * `snmp`: build the SNMP check.
 * `zk`: enable Zookeeper as a configuration store.
 * `zstd`: use Zstandard instead of Zlib.
+* `systemd`: enable systemd journal log collection
 
 Please note you might need to provide some extra dependencies in your dev
 environment to build certain bits (see [development environment][dev-env]).

docs/dev/agent_dev_env.md

@@ -115,6 +115,15 @@ need to work/debug on the SNMP integration, you could just build the agent witho
 it (see [Building the Agent][building] for how to do it) and avoid the dependencies
 setup efforts altogether.
 
+### Systemd
+
+The agent is able to collect systemd journal logs using a wrapper on the systemd utility library.
+
+On Linux:
+```
+sudo apt-get install libsystemd-dev
+```
+
 ## Docker
 
 If you want to build a Docker image containing the Agent, or if you wan to run

pkg/logs/agent.go

@@ -9,6 +9,7 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/logs/auditor"
 	"github.com/DataDog/datadog-agent/pkg/logs/config"
 	"github.com/DataDog/datadog-agent/pkg/logs/input/container"
+	"github.com/DataDog/datadog-agent/pkg/logs/input/journald"
 	"github.com/DataDog/datadog-agent/pkg/logs/input/listener"
 	"github.com/DataDog/datadog-agent/pkg/logs/input/tailer"
 	"github.com/DataDog/datadog-agent/pkg/logs/message"
@@ -29,6 +30,7 @@ type Agent struct {
 	containersScanner *container.Scanner
 	filesScanner      *tailer.Scanner
 	networkListener   *listener.Listener
+	journaldLauncher  *journald.Launcher
 	pipelineProvider  pipeline.Provider
 }
 
@@ -50,11 +52,13 @@ func NewAgent(sources *config.LogSources) *Agent {
 	containersScanner := container.New(sources.GetValidSources(), pipelineProvider, auditor)
 	networkListeners := listener.New(sources.GetValidSources(), pipelineProvider)
 	filesScanner := tailer.New(sources.GetValidSources(), config.LogsAgent.GetInt("logs_config.open_files_limit"), pipelineProvider, auditor, tailer.DefaultSleepDuration)
+	journaldLauncher := journald.New(sources.GetValidSources(), pipelineProvider, auditor)
 
 	return &Agent{
 		auditor:           auditor,
 		containersScanner: containersScanner,
 		filesScanner:      filesScanner,
+		journaldLauncher:  journaldLauncher,
 		networkListener:   networkListeners,
 		pipelineProvider:  pipelineProvider,
 	}
@@ -69,6 +73,7 @@ func (a *Agent) Start() {
 		a.filesScanner,
 		a.networkListener,
 		a.containersScanner,
+		a.journaldLauncher,
 	)
 }
 
@@ -80,6 +85,7 @@ func (a *Agent) Stop() {
 			a.filesScanner,
 			a.networkListener,
 			a.containersScanner,
+			a.journaldLauncher,
 		),
 		a.pipelineProvider,
 		a.auditor,

pkg/logs/config/integration_config.go

@@ -19,10 +19,11 @@ import (
 
 // Logs source types
 const (
-	TCPType    = "tcp"
-	UDPType    = "udp"
-	FileType   = "file"
-	DockerType = "docker"
+	TCPType      = "tcp"
+	UDPType      = "udp"
+	FileType     = "file"
+	DockerType   = "docker"
+	JournaldType = "journald"
 )
 
 // Logs rule types
@@ -58,7 +59,11 @@ type LogsConfig struct {
 	Type string
 
 	Port int    // Network
-	Path string // File
+	Path string // File, Journald
+
+	IncludeUnits         []string `mapstructure:"include_units"`         // Journald
+	ExcludeUnits         []string `mapstructure:"exclude_units"`         // Journald
+	DisableNormalization bool     `mapstructure:"disable_normalization"` // Journald
 
 	Image string // Docker
 	Label string // Docker
@@ -185,29 +190,22 @@ func integrationConfigsFromDirectory(dir string, prefix string) []string {
 }
 
 func validateConfig(config LogsConfig) error {
-
 	switch config.Type {
-	case FileType,
-		DockerType,
-		TCPType,
-		UDPType:
+	case FileType, DockerType, TCPType, UDPType, JournaldType:
 	default:
 		return fmt.Errorf("A source must have a valid type (got %s)", config.Type)
 	}
 
-	if config.Type == FileType && config.Path == "" {
+	switch {
+	case config.Type == FileType && config.Path == "":
 		return fmt.Errorf("A file source must have a path")
-	}
-
-	if config.Type == TCPType && config.Port == 0 {
+	case config.Type == TCPType && config.Port == 0:
 		return fmt.Errorf("A tcp source must have a port")
-	}
-
-	if config.Type == UDPType && config.Port == 0 {
+	case config.Type == UDPType && config.Port == 0:
 		return fmt.Errorf("A udp source must have a port")
+	default:
+		return nil
 	}
-
-	return nil
 }
 
 // validateProcessingRules checks the rules and raises errors if one is misconfigured

pkg/logs/config/integration_config_test.go

@@ -18,16 +18,16 @@ const testsPath = "tests"
 
 func TestAvailableIntegrationConfigs(t *testing.T) {
 	ddconfdPath := filepath.Join(testsPath, "complete", "conf.d")
-	assert.Equal(t, []string{"integration.yaml", "integration2.yml", "misconfigured_integration.yaml", "integration.d/integration3.yaml", "integration.d/integration4.yaml"}, availableIntegrationConfigs(ddconfdPath))
+	assert.Equal(t, []string{"integration.yaml", "integration2.yml", "integration5.yml", "misconfigured_integration.yaml", "integration.d/integration3.yaml", "integration.d/integration4.yaml"}, availableIntegrationConfigs(ddconfdPath))
 }
 
 func TestBuildLogsAgentIntegrationsConfigs(t *testing.T) {
 	ddconfdPath := filepath.Join(testsPath, "complete", "conf.d")
 	allSources, err := buildLogSources(ddconfdPath, false)
 
 	assert.Nil(t, err)
-	assert.Equal(t, 5, len(allSources.GetValidSources()))
-	assert.Equal(t, 6, len(allSources.GetSources()))
+	assert.Equal(t, 6, len(allSources.GetValidSources()))
+	assert.Equal(t, 7, len(allSources.GetSources()))
 
 	sources := allSources.GetValidSources()
 
@@ -44,11 +44,13 @@ func TestBuildLogsAgentIntegrationsConfigs(t *testing.T) {
 	assert.Equal(t, "", sources[1].Config.Source)
 	assert.Equal(t, 0, len(sources[1].Config.Tags))
 
-	assert.Equal(t, "docker", sources[2].Config.Type)
-	assert.Equal(t, "test", sources[2].Config.Image)
+	assert.Equal(t, "journald", sources[2].Config.Type)
+
+	assert.Equal(t, "docker", sources[3].Config.Type)
+	assert.Equal(t, "test", sources[3].Config.Image)
 
-	assert.Equal(t, []string{"env:prod", "foo:bar"}, sources[3].Config.Tags)
 	assert.Equal(t, []string{"env:prod", "foo:bar"}, sources[4].Config.Tags)
+	assert.Equal(t, []string{"env:prod", "foo:bar"}, sources[5].Config.Tags)
 
 	// processing
 	assert.Equal(t, 0, len(sources[0].Config.ProcessingRules))

pkg/logs/config/tests/complete/conf.d/integration5.yml

@@ -0,0 +1,2 @@
+logs:
+  - type: journald    

pkg/logs/input/journald/launcher.go

@@ -0,0 +1,77 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2018 Datadog, Inc.
+
+package journald
+
+import (
+	log "github.com/cihub/seelog"
+
+	"github.com/DataDog/datadog-agent/pkg/logs/auditor"
+	"github.com/DataDog/datadog-agent/pkg/logs/config"
+	"github.com/DataDog/datadog-agent/pkg/logs/pipeline"
+	"github.com/DataDog/datadog-agent/pkg/logs/restart"
+)
+
+// Launcher is in charge of starting and stopping new journald tailers
+type Launcher struct {
+	sources          []*config.LogSource
+	pipelineProvider pipeline.Provider
+	auditor          *auditor.Auditor
+	tailers          map[string]*Tailer
+}
+
+// New returns a new Launcher.
+func New(sources []*config.LogSource, pipelineProvider pipeline.Provider, auditor *auditor.Auditor) *Launcher {
+	journaldSources := []*config.LogSource{}
+	for _, source := range sources {
+		if source.Config.Type == config.JournaldType {
+			journaldSources = append(journaldSources, source)
+		}
+	}
+	return &Launcher{
+		sources:          journaldSources,
+		pipelineProvider: pipelineProvider,
+		auditor:          auditor,
+		tailers:          make(map[string]*Tailer),
+	}
+}
+
+// Start starts new tailers.
+func (l *Launcher) Start() {
+	for _, source := range l.sources {
+		identifier := source.Config.Path
+		if _, exists := l.tailers[identifier]; exists {
+			// set up only one tailer per journal
+			continue
+		}
+		tailer, err := l.setupTailer(source)
+		if err != nil {
+			log.Warn("Could not set up journald tailer: ", err)
+		} else {
+			l.tailers[identifier] = tailer
+		}
+	}
+}
+
+// Stop stops all active tailers
+func (l *Launcher) Stop() {
+	stopper := restart.NewParallelStopper()
+	for identifier, tailer := range l.tailers {
+		stopper.Add(tailer)
+		delete(l.tailers, identifier)
+	}
+	stopper.Stop()
+}
+
+// setupTailer configures and starts a new tailer,
+// returns the tailer or an error.
+func (l *Launcher) setupTailer(source *config.LogSource) (*Tailer, error) {
+	tailer := NewTailer(source, l.pipelineProvider.NextPipelineChan())
+	err := tailer.Start(l.auditor.GetLastCommittedOffset(tailer.Identifier()))
+	if err != nil {
+		return nil, err
+	}
+	return tailer, nil
+}

pkg/logs/input/journald/launcher_test.go

@@ -0,0 +1,34 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2018 Datadog, Inc.
+
+// +build systemd
+
+package journald
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/DataDog/datadog-agent/pkg/logs/auditor"
+	"github.com/DataDog/datadog-agent/pkg/logs/config"
+	pipeline "github.com/DataDog/datadog-agent/pkg/logs/pipeline/mock"
+)
+
+func TestShouldStartOnlyOneTailerPerJournal(t *testing.T) {
+	sources := []*config.LogSource{
+		config.NewLogSource("", &config.LogsConfig{Type: config.JournaldType}),
+		config.NewLogSource("", &config.LogsConfig{Type: config.JournaldType}),
+	}
+	launcher := New(sources, pipeline.NewMockProvider(), auditor.New(nil, ""))
+
+	// expect two new tailers
+	launcher.Start()
+	assert.Equal(t, 1, len(launcher.tailers))
+
+	// expect all tailers to be released
+	launcher.Stop()
+	assert.Equal(t, 0, len(launcher.tailers))
+}