Add container input, deprecate docker in favor of it

filebeat/autodiscover/builder/hints/config.go

@@ -26,7 +26,7 @@ type config struct {
 
 func defaultConfig() config {
  rawCfg := map[string]interface{}{
- "type": "docker",
+ "type": "container",
  "containers": map[string]interface{}{
  "paths": []string{
  // To be able to use this builder with CRI-O replace paths with:

filebeat/docs/filebeat-options.asciidoc

@@ -44,6 +44,7 @@ You can configure {beatname_uc} to use the following inputs:
 
 * <<{beatname_lc}-input-log>>
 * <<{beatname_lc}-input-stdin>>
+* <<{beatname_lc}-input-container>>
 * <<{beatname_lc}-input-redis>>
 * <<{beatname_lc}-input-udp>>
 * <<{beatname_lc}-input-docker>>
@@ -57,6 +58,8 @@ include::inputs/input-log.asciidoc[]
 
 include::inputs/input-stdin.asciidoc[]
 
+include::inputs/input-container.asciidoc[]
+
 include::inputs/input-redis.asciidoc[]
 
 include::inputs/input-udp.asciidoc[]

filebeat/docs/inputs/input-container.asciidoc

@@ -0,0 +1,64 @@
+:type: container
+
+[id="{beatname_lc}-input-{type}"]
+=== Container input
+
+++++
+<titleabbrev>Container</titleabbrev>
+++++
+
+Use the `container` input to read containers log files.
+
+This input searches for container logs under the given path, and parse them into
+common message lines, extracting timestamps too. Everything happens before line
+filtering, multiline, and JSON decoding, so this input can be used in
+combination with those settings.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: container
+ paths: <1>
+ - '/var/lib/docker/containers/*/*.log'
+----
+
+<1> `path` is required. All other settings are optional.
+
+==== Configuration options
+
+The `container` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+===== `stream`
+
+Reads from the specified streams only: `all`, `stdout` or `stderr`. The default
+is `all`.
+
+===== `format`
+
+Use the given format when reading the log file: `auto`, `docker` or `cri`. The
+default is `auto`, it will automatically detect the format. To disable
+autodetection set any of the other options.
+
+
+The following input configures {beatname_uc} to read the `stdout` stream from
+all containers under the default Kubernetes logs path:
+
+[source,yaml]
+----
+- type: container
+ stream: stdout
+ paths:
+ - "/var/log/containers/*.log"
+----
+
+include::../inputs/input-common-harvester-options.asciidoc[]
+
+include::../inputs/input-common-file-options.asciidoc[]
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../inputs/input-common-options.asciidoc[]
+
+:type!:

filebeat/docs/inputs/input-docker.asciidoc

@@ -7,6 +7,8 @@
 <titleabbrev>Docker</titleabbrev>
 ++++
 
+deprecated[7.2.0, Use `container` input instead.]
+
 Use the `docker` input to read logs from Docker containers.
 
 This input searches for container logs under its path, and parse them into
@@ -103,4 +105,4 @@ include::../inputs/input-common-file-options.asciidoc[]
 [id="{beatname_lc}-input-{type}-common-options"]
 include::../inputs/input-common-options.asciidoc[]
 
-:type!:
+:type!:

filebeat/harvester/util.go

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

filebeat/input/container/config.go

@@ -0,0 +1,31 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package container
+
+var defaultConfig = config{
+ Stream: "all",
+ Format: "auto",
+}
+
+type config struct {
+ // Stream can be all, stdout or stderr
+ Stream string `config:"stream"`
+
+ // Format can be auto, cri, json-file
+ Format string `config:"format"`
+}

filebeat/input/container/input.go

@@ -0,0 +1,117 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package container
+
+import (
+ "fmt"
+ "strings"
+
+ "github.com/elastic/beats/filebeat/channel"
+ "github.com/elastic/beats/filebeat/input"
+ "github.com/elastic/beats/filebeat/input/log"
+ "github.com/elastic/beats/libbeat/common"
+
+ "github.com/pkg/errors"
+)
+
+func init() {
+ err := input.Register("container", NewInput)
+ if err != nil {
+ panic(err)
+ }
+}
+
+// NewInput creates a new container input
+func NewInput(
+ cfg *common.Config,
+ outletFactory channel.Connector,
+ context input.Context,
+) (input.Input, error) {
+ // Wrap log input with custom docker settings
+ config := defaultConfig
+ if err := cfg.Unpack(&config); err != nil {
+ return nil, errors.Wrap(err, "reading container input config")
+ }
+
+ if err := validateStream(config.Stream); err != nil {
+ return nil, err
+ }
+
+ if err := validateFormat(config.Format); err != nil {
+ return nil, err
+ }
+
+ // Set partial line joining to true (both json-file and CRI)
+ if err := cfg.SetBool("docker-json.partial", -1, true); err != nil {
+ return nil, errors.Wrap(err, "update input config")
+ }
+
+ if err := cfg.SetBool("docker-json.cri_flags", -1, true); err != nil {
+ return nil, errors.Wrap(err, "update input config")
+ }
+
+ // Allow stream selection (stdout/stderr/all)
+ if err := cfg.SetString("docker-json.stream", -1, config.Stream); err != nil {
+ return nil, errors.Wrap(err, "update input config")
+ }
+
+ if err := cfg.SetString("docker-json.format", -1, config.Format); err != nil {
+ return nil, errors.Wrap(err, "update input config")
+ }
+
+ // Set symlinks to true as CRI-O paths could point to symlinks instead of the actual path.
+ if err := cfg.SetBool("symlinks", -1, true); err != nil {
+ return nil, errors.Wrap(err, "update input config")
+ }
+
+ // Add stream to meta to ensure different state per stream
+ if config.Stream != "all" {
+ if context.Meta == nil {
+ context.Meta = map[string]string{}
+ }
+ context.Meta["stream"] = config.Stream
+ }
+
+ return log.NewInput(cfg, outletFactory, context)
+}
+
+func validateStream(val string) error {
+ if stringInSlice(val, []string{"all", "stdout", "stderr"}) {
+ return nil
+ }
+
+ return fmt.Errorf("Invalid value for stream: %s, supported values are: all, stdout, stderr", val)
+}
+
+func validateFormat(val string) error {
+ val = strings.ToLower(val)
+ if stringInSlice(val, []string{"auto", "docker", "cri"}) {
+ return nil
+ }
+
+ return fmt.Errorf("Invalid value for format: %s, supported values are: auto, docker, cri", val)
+}
+
+func stringInSlice(str string, list []string) bool {
+ for _, v := range list {
+ if v == str {
+ return true
+ }
+ }
+ return false
+}

filebeat/input/docker/config.go

@@ -44,8 +44,6 @@ type containers struct {
  IDs []string `config:"ids"`
  Path string `config:"path"`
 
- Paths []string `config:"paths"`
-
  // Stream can be all, stdout or stderr
  Stream string `config:"stream"`
 }

filebeat/input/docker/input.go

@@ -25,6 +25,7 @@ import (
  "github.com/elastic/beats/filebeat/input"
  "github.com/elastic/beats/filebeat/input/log"
  "github.com/elastic/beats/libbeat/common"
+ "github.com/elastic/beats/libbeat/common/cfgwarn"
  "github.com/elastic/beats/libbeat/logp"
 
  "github.com/pkg/errors"
@@ -45,25 +46,18 @@ func NewInput(
 ) (input.Input, error) {
  logger := logp.NewLogger("docker")
 
+ cfgwarn.Deprecate("8.0.0", "'docker' input deprecated. Use 'container' input instead.")
+
  // Wrap log input with custom docker settings
  config := defaultConfig
  if err := cfg.Unpack(&config); err != nil {
  return nil, errors.Wrap(err, "reading docker input config")
  }
 
- // Docker input should make sure that no callers should ever pass empty strings as container IDs or paths
+ // Docker input should make sure that no callers should ever pass empty strings as container IDs
  // Hence we explicitly make sure that we catch such things and print stack traces in the event of
  // an invocation so that it can be fixed.
- var ids, paths []string
- for _, p := range config.Containers.Paths {
- if p != "" {
- paths = append(paths, p)
- } else {
- logger.Error("Docker paths can't be empty for Docker input config")
- logger.Debugw("Empty path for Docker logfile was received", logp.Stack("stacktrace"))
- }
- }
-
+ var ids []string
  for _, containerID := range config.Containers.IDs {
  if containerID != "" {
  ids = append(ids, containerID)
@@ -73,23 +67,12 @@ func NewInput(
  }
  }
 
- if len(ids) == 0 && len(paths) == 0 {
- return nil, errors.New("Docker input requires at least one entry under 'containers.ids' or 'containers.paths'")
- }
-
- // IDs + Path and Paths are mutually exclusive. Ensure that only one of them are set in a given configuration
- if len(ids) != 0 && len(paths) != 0 {
- return nil, errors.New("can not provide both 'containers.ids' and 'containers.paths' in the same input config")
+ if len(ids) == 0 {
+ return nil, errors.New("Docker input requires at least one entry under 'containers.ids''")
  }
 
- if len(ids) != 0 {
- for idx, containerID := range ids {
- cfg.SetString("paths", idx, path.Join(config.Containers.Path, containerID, "*.log"))
- }
- } else {
- for idx, p := range paths {
- cfg.SetString("paths", idx, p)
- }
+ for idx, containerID := range ids {
+ cfg.SetString("paths", idx, path.Join(config.Containers.Path, containerID, "*.log"))
  }
 
  if err := checkStream(config.Containers.Stream); err != nil {
@@ -108,13 +91,8 @@ func NewInput(
  return nil, errors.Wrap(err, "update input config")
  }
 
- if err := cfg.SetBool("docker-json.force_cri_logs", -1, config.CRIForce); err != nil {
- return nil, errors.Wrap(err, "update input config")
- }
-
- if len(paths) != 0 {
- // Set symlinks to true as CRI-O paths could point to symlinks instead of the actual path.
- if err := cfg.SetBool("symlinks", -1, true); err != nil {
+ if config.CRIForce {
+ if err := cfg.SetString("docker-json.format", -1, "cri"); err != nil {
  return nil, errors.Wrap(err, "update input config")
  }
  }

filebeat/input/log/config.go

@@ -109,7 +109,7 @@ type config struct {
  DockerJSON *struct {
  Stream string `config:"stream"`
  Partial bool `config:"partial"`
- ForceCRI bool  `config:"force_cri_logs"`
+ Format string `config:"format"`
  CRIFlags bool `config:"cri_flags"`
  } `config:"docker-json"`
 }