Add container_v2 input type as Beta

CHANGELOG.next.asciidoc

@@ -231,6 +231,7 @@ is collected by it.
 - Add CEL partial value debug function. {pull}36652[36652]
 - Added support for new features and removed partial save mechanism in the GCS input. {issue}35847[35847] {pull}36713[36713]
 - Re-use buffers to optimise memory allocation in fingerprint mode of filestream {pull}36736[36736]
+- Add new `container-v2` input type which is based on filestream {issue}34393[34393] {pull}36661[36661]
 
 *Auditbeat*
 

filebeat/docs/filebeat-options.asciidoc

@@ -74,6 +74,7 @@ You can configure {beatname_uc} to use the following inputs:
 * <<{beatname_lc}-input-cloudfoundry>>
 * <<{beatname_lc}-input-cometd>>
 * <<{beatname_lc}-input-container>>
+* <<{beatname_lc}-input-container-v2>>
 * <<{beatname_lc}-input-entity-analytics>>
 * <<{beatname_lc}-input-filestream>>
 * <<{beatname_lc}-input-gcp-pubsub>>
@@ -110,6 +111,8 @@ include::../../x-pack/filebeat/docs/inputs/input-cometd.asciidoc[]
 
 include::inputs/input-container.asciidoc[]
 
+include::inputs/input-container-v2.asciidoc[]
+
 include::../../x-pack/filebeat/docs/inputs/input-entity-analytics.asciidoc[]
 
 include::inputs/input-filestream.asciidoc[]

filebeat/docs/howto/migrate-to-filestream.asciidoc

@@ -1,5 +1,5 @@
 [[migrate-to-filestream]]
-== Migrate `log` input configurations to `filestream`
+== Migrate `log` or `container` input configurations to `filestream`
 
 The `filestream` input has been generally available since 7.14 and it is highly recommended
 you migrate your existing `log` input configurations. The `filestream` input comes with many
@@ -8,11 +8,15 @@ improvements over the old `log` input, such as configurable order for parsers an
 The `log` input is deprecated and will eventually be removed from Filebeat. We are not fixing
 new issues or adding any enhancements to the `log` input. Our focus is on `filestream`.
 
+NOTE: The `container` input is based on the `log` input, so everything in this guide applies to the `container` input too. Additionally, you can migrate a `container` input to a `container-v2` input since it's based on `filestream` and supports the same options.
+
 This manual migration is required only if you've defined `log` inputs manually
 in your stand-alone Filebeat configuration. All the integrations or modules that are still using
 `log` inputs under the hood will be eventually migrated automatically without any additional actions
 required from the user.
 
+WARNING: This migration would not work if you dynamically add inputs during the runtime (e.g. running under Elastic Agent or using the `autodiscover` feature. Migration happens only on Filebeat's startup because it requires exclusive access to the registry.
+
 In this guide, you'll learn how to migrate an existing `log` input configuration.
 
 IMPORTANT: You must replace `log` inputs with `filestream` inputs, make sure you have removed

filebeat/docs/inputs/input-container-v2.asciidoc

@@ -0,0 +1,83 @@
+:type: container-v2
+
+[id="{beatname_lc}-input-{type}"]
+=== Container V2 input
+
+++++
+<titleabbrev>Container V2</titleabbrev>
+++++
+
+WARNING: This input is still in Beta and the configuration options/defaults might change in the future.
+
+Use the `container-v2` input to read container log files.
+
+Unlike its predecessor <<{beatname_lc}-input-container,`container` input>>, this input is based on the <<{beatname_lc}-input-filestream,`filestream` input>> and it uses the <<{beatname_lc}-input-filestream-scan-fingerprint,fingerprint mode>> to identify files, which proved itself to be more reliable, especially in containerized environments.
+
+This input is practically a configuration preset for the <<{beatname_lc}-input-filestream,`filestream` input>>, ready to be used for container logs without much of configuration.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: container-v2
+  id: 'unique-identifier' <1>
+  paths: <2>
+    - '/var/log/containers/*.log'
+----
+
+<1> `id` is not required but highly recommended. If you decide to use the <<{beatname_lc}-input-filestream-take-over, take-over mode>> it requires a unique ID.
+
+<2> `paths` are required
+
+
+NOTE: The <<{beatname_lc}-input-filestream-take-over, take-over mode>> is able to migrate states of the old <<{beatname_lc}-input-container,`container` input>> since it's based on the <<{beatname_lc}-input-log, log input>>. See <<migrate-to-filestream>> for more details about the migration process.
+
+NOTE: '/var/log/containers/*.log' is normally a symlink to '/var/log/pods/*/*/.log',
+so above path can be edited accordingly
+
+==== Configuration options
+
+The `container-v2` input supports the following configuration options plus all the
+<<{beatname_lc}-input-filestream,`filestream` input>> options, however this input has different defaults for some of the `filestream` configuration options:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: container-v2
+  id: 'unique-identifier'
+  paths:
+    - '/var/log/containers/*.log'
+  # container V2 input options mapped to `parsers.container.*` options in `filestream`
+  stream: all
+  format: auto
+  # different defaults for the underlying `filestream` input
+  prospector.scanner.symlinks: true
+  prospector.scanner.fingerprint.enabled: true
+  file_identity.fingerprint: ~
+----
+
+===== `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-v2
+  id: 'my-container-logs'
+  stream: stdout
+  paths:
+    - "/var/log/containers/*.log"
+----
+
+:type!:

filebeat/input/default-inputs/inputs.go

@@ -39,6 +39,7 @@ func Init(info beat.Info, log *logp.Logger, components beater.StateStore) []v2.P
 func genericInputs(log *logp.Logger, components beater.StateStore) []v2.Plugin {
 	return []v2.Plugin{
 		filestream.Plugin(log, components),
+		filestream.ContainerPlugin(log, components),
 		kafka.Plugin(),
 		tcp.Plugin(),
 		udp.Plugin(),

filebeat/input/filestream/container.go

@@ -0,0 +1,123 @@
+// 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 filestream
+
+import (
+	"fmt"
+	"strings"
+
+	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
+	input "github.com/elastic/beats/v7/filebeat/input/v2"
+	"github.com/elastic/beats/v7/libbeat/feature"
+	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+)
+
+const containerPluginName = "container-v2"
+
+func defaultContainerConfig() containerConfig {
+	return containerConfig{
+		Stream: "all",
+		Format: "auto",
+	}
+}
+
+type containerConfig struct {
+	// Stream can be all, stdout or stderr
+	Stream string `config:"stream"`
+
+	// Format can be auto, cri, json-file
+	Format string `config:"format"`
+}
+
+// Validate validates the config.
+func (c *containerConfig) Validate() error {
+	if !stringInSlice(c.Stream, []string{"all", "stdout", "stderr"}) {
+		return fmt.Errorf("invalid value for stream: %s, supported values are: all, stdout, stderr", c.Stream)
+	}
+
+	if !stringInSlice(strings.ToLower(c.Format), []string{"auto", "docker", "cri"}) {
+		return fmt.Errorf("invalid value for format: %s, supported values are: auto, docker, cri", c.Format)
+	}
+
+	return nil
+}
+
+// Plugin creates a new container V2 input plugin for creating a stateful input.
+func ContainerPlugin(log *logp.Logger, store loginp.StateStore) input.Plugin {
+	return input.Plugin{
+		Name:       containerPluginName,
+		Stability:  feature.Beta,
+		Deprecated: false,
+		Info:       "filestream-based container input",
+		Doc:        "The container input collects logs from a running container using filestream",
+		Manager: &loginp.InputManager{
+			Logger:     log,
+			StateStore: store,
+			Type:       containerPluginName,
+			Configure:  configureContainer,
+		},
+	}
+}
+
+type container struct {
+	loginp.Harvester
+}
+
+func (container) Name() string { return containerPluginName }
+
+func configureContainer(cfg *conf.C) (loginp.Prospector, loginp.Harvester, error) {
+	containerConfig := defaultContainerConfig()
+	if err := cfg.Unpack(&containerConfig); err != nil {
+		return nil, nil, fmt.Errorf("failed to read container input config: %w", err)
+	}
+
+	err := cfg.Merge(mapstr.M{
+		"parsers": []mapstr.M{
+			{
+				"container.stream": containerConfig.Stream,
+				"container.format": containerConfig.Format,
+			},
+		},
+		// Set symlinks to true as CRI-O paths could point to symlinks instead of the actual path.
+		"prospector.scanner.symlinks": true,
+		// Most of the time container logs are ingested from file systems without stable inode values
+		"prospector.scanner.fingerprint.enabled": true,
+		"file_identity.fingerprint":              nil,
+	})
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to update container input config: %w", err)
+	}
+
+	prospector, harvester, err := configure(cfg)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create filestream for container input: %w", err)
+	}
+
+	return prospector, container{harvester}, nil
+}
+
+func stringInSlice(str string, list []string) bool {
+	for _, v := range list {
+		if v == str {
+			return true
+		}
+	}
+	return false
+}