Add support for copytruncate method when rotating input logs with an …

…external tool in `filestream` input (#23457)

## What does this PR do?

The PR makes the `filestream` log rotation aware to make sure Filebeat can cooperate better with external log rotation tools. The first supported strategy is `copytruncate`.

When `logrotate` rotates e.g. `boot.log` with `copytruncate` the following things happen:
1. all archived files are renamed e.g. `boot.log.2` is renamed `boot.log.3` until `boot.log.1` no longer exists
2. `boot.log` is copied to `boot.log.1`
3. `boot.log` is truncated

You can see my tests on my machine:

Before rotation:
```
root@sleipnir:/home/n# ls -lisaht /var/log/boot.log*
130476 30K -rw------- 1 root root 28K Jan 29 08:59 /var/log/boot.log
130577 36K -rw------- 1 root root 34K Jan 29 08:59 /var/log/boot.log.1
130657 60K -rw------- 1 root root 57K Jan  7 09:51 /var/log/boot.log.2
```
After rotation:
```
root@sleipnir:/home/n# ls -lisaht /var/log/boot.log*
130476   0 -rw------- 1 root root   0 May 25 12:41 /var/log/boot.log
130430 30K -rw------- 1 root root 28K May 25 12:41 /var/log/boot.log.1
130577 36K -rw------- 1 root root 34K Jan 29 08:59 /var/log/boot.log.2
130657 60K -rw------- 1 root root 57K Jan  7 09:51 /var/log/boot.log.3
```

On rotation, the active file is continued and archived files are kept open until EOF is reached.

### Configuration

```yaml
rotation.external.strategy.copytruncate:
  suffix_regex: \.\d$
  count: 10
```

Note: when Filebeat will be able to rotate input logs, its configuration will be under `rotation.internal.*`.

## Why is it important?

Previously, Filebeat was not able to cooperate with external log rotation tools that used `copytruncate` method.

CHANGELOG.next.asciidoc

@@ -828,6 +828,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Update grok patterns for HA Proxy module {issue}25827[25827] {pull}25835[25835]
 - Added dataset `anomalithreatstream` to the `threatintel` module to ingest indicators from Anomali ThreatStream {pull}26350[26350]
 
+- Add support for `copytruncate` method when rotating input logs with an external tool in `filestream` input. {pull}23457[23457]
+
 *Heartbeat*
 
 - Add mime type detection for http responses. {pull}22976[22976]

filebeat/_meta/config/filebeat.inputs.reference.yml.tmpl

@@ -292,6 +292,16 @@ filebeat.inputs:
   # original for harvesting but will report the symlink name as source.
   #prospector.scanner.symlinks: false
 
+  ### Log rotation
+
+  # When an external tool rotates the input files with copytruncate strategy
+  # use this section to help the input find the rotated files.
+  #rotation.external.strategy.copytruncate:
+  # Regex that matches the rotated files.
+  #  suffix_regex: \.\d$
+  # If the rotated filename suffix is a datetime, set it here. 
+  #  dateformat: -20060102
+
   ### State options
 
   # Files for the modification data is older then clean_inactive the state from the registry is removed

filebeat/docs/inputs/input-filestream-file-options.asciidoc

@@ -482,3 +482,56 @@ Set the location of the marker file the following way:
 ----
 file_identity.inode_marker.path: /logs/.filebeat-marker
 ----
+
+=== Log rotation
+
+As log files are constantly written, they must be rotated and purged to prevent
+the logger application from filling up the disk. Rotation is done by an external
+application, thus, {beatname_uc} needs information how to cooperate with it.
+
+When reading from rotating files make sure the paths configuration includes
+both the active file and all rotated files.
+
+By default, {beatname_uc} is able to track files correctly in the following strategies:
+* create: new active file with a unique name is created on rotation
+* rename: rotated files are renamed
+
+However, in case of copytruncate strategy, you should provide additional configuration
+to {beatname_uc}.
+
+[float]
+==== rotation.external.strategy.copytruncate
+
+experimental[]
+
+If the log rotating application copies the contents of the active file and then
+truncates the original file, use these options to help {beatname_uc} to read files
+correctly.
+
+Set the option `suffix_regex` so {beatname_uc} can tell active and rotated files apart. There are
+two supported suffix types in the input: numberic and date.
+
+==== Numeric suffix
+
+If your rotated files have an incrementing index appended to the end of the filename, e.g.
+active file `apache.log` and the rotated files are named `apache.log.1`, `apache.log.2`, etc,
+use the following configuration.
+
+[source,yaml]
+---
+rotation.external.strategy.copytruncate:
+  suffix_regex: \.\d$
+---
+
+==== Date suffix
+
+If the rotation date is appended to the end of the filename, e.g. active file `apache.log` and the
+rotated files are named `apache.log-20210526`, `apache.log-20210527`, etc. use the following configuration:
+
+[source,yaml]
+---
+rotation.external.strategy.copytruncate:
+  suffix_regex: \-\d{6}$
+  dateformat: -20060102
+---
+

filebeat/filebeat.reference.yml

@@ -699,6 +699,16 @@ filebeat.inputs:
   # original for harvesting but will report the symlink name as source.
   #prospector.scanner.symlinks: false
 
+  ### Log rotation
+
+  # When an external tool rotates the input files with copytruncate strategy
+  # use this section to help the input find the rotated files.
+  #rotation.external.strategy.copytruncate:
+  # Regex that matches the rotated files.
+  #  suffix_regex: \.\d$
+  # If the rotated filename suffix is a datetime, set it here. 
+  #  dateformat: -20060102
+
   ### State options
 
   # Files for the modification data is older then clean_inactive the state from the registry is removed

filebeat/input/filestream/config.go

@@ -41,6 +41,7 @@ type config struct {
 	HarvesterLimit uint32                  `config:"harvester_limit" validate:"min=0"`
 	IgnoreOlder    time.Duration           `config:"ignore_older"`
 	IgnoreInactive ignoreInactiveType      `config:"ignore_inactive"`
+	Rotation       *common.ConfigNamespace `config:"rotation"`
 }
 
 type closerConfig struct {
@@ -78,6 +79,17 @@ type backoffConfig struct {
 	Max  time.Duration `config:"max" validate:"nonzero"`
 }
 
+type rotationConfig struct {
+	Strategy *common.ConfigNamespace `config:"strategy" validate:"required"`
+}
+
+type commonRotationConfig struct {
+	SuffixRegex string `config:"suffix_regex" validate:"required"`
+	DateFormat  string `config:"dateformat"`
+}
+
+type copyTruncateConfig commonRotationConfig
+
 func defaultConfig() config {
 	return config{
 		Reader:         defaultReaderConfig(),