add fixed size buffer to handle fd in log tailer with feature flag #1577

movence · 2025-03-03T19:31:27Z

Revisions

Rev2

rename only supported backpressure_mode value to fd_release
add time out (5m) to blocked wait for a new file in Reopen
remove unnecessary logging

Rev1

move backpressure_drop feature flag to individual file level config for granular control
rebase

Description of the issue

The current CWA strategy for log collection prioritizes log integrity without consideration for disk space. CWA holds onto open file descriptors until all the log lines have been successfully pushed to CloudWatch Logs. If another process deletes the file, the file descriptor held by the agent will be unaffected. If CWA is unable to publish the logs, it will retry for 14 days. For some customers, this is the expectation and preferred behavior particularly for sensitive files such as security logs.

On Unix, when a file is removed by a process, the file name is unlinked from the underlying file instead of deleted. If there’s an open reference to the file via file descriptor, the file will not be deleted.

Description of changes

Add buffer-based (size of 1) log handling in tailer for file rotation
Add new runSender goroutine. tailer now sends single event at a time to channel that the sender reads from
Enable this new path behind feature flag backpressure_drop
Add feature toggle flag that is read either from agent json config or env var with the value from agent config takes precedence. This is to keep the original behavior with auto_removal which reads all contents from a rotate log file before deleting them.
Feature flag value is then added to env-config.json by the translator so it can be added as env var of systemd process of the agent at start up time

NOTE
The auto_removal setting takes precedence, disabling any tailer-sender modifications for log entries when enabled. With auto_removal, all existing files are expected to be fully processed before they are deleted. However, the backpressure_drop flag contradicts this behavior by stopping the processing of previous files upon rotation. To ensure backward compatibility for customers who depend on the auto_removal functionality, we must retain the current behavior.

License

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Tests

Block logs endpoint by pointing API URI (eg logs.us-west-2.amazonaws.com) to localhost. Then use a bash script to generate 10 lines/sec to /var/log/test/application.log, rotating the log every 3 minutes with a ~30-minute bake period.. Keep a maximum of 3 rotated files

#### Agent version 1.300052.0
> pgrep -f "/opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent"
21390
> ls -l /proc/$(pgrep -f "/opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent")/fd | grep application | wc -l
12
> ls -l /proc/$(pgrep -f "/opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent")/fd | grep application | head -5
lr-x------ 1 root root 64 Mar  6 03:12 10 -> /var/log/test/application.log.20250306_031458 (deleted)
lr-x------ 1 root root 64 Mar  6 03:12 11 -> /var/log/test/application.log.20250306_031530 (deleted)
lr-x------ 1 root root 64 Mar  6 03:12 12 -> /var/log/test/application.log.20250306_031602 (deleted)
lr-x------ 1 root root 64 Mar  6 03:12 13 -> /var/log/test/application.log.20250306_031634 (deleted)
lr-x------ 1 root root 64 Mar  6 03:12 14 -> /var/log/test/application.log.20250306_031706 (deleted)


#### With `backpressure_drop` enabled
> pgrep -f "/opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent"
27016
> ls -l /proc/$(pgrep -f "/opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent")/fd | grep application | wc -l
0


#### agent config used
{
  "agent": {
    "debug": true,
    "run_as_user": "root"
  },
  "logs": {
    "backpressure_drop": true,
    "logs_collected": {
      "files": {
        "collect_list": [
          {
            "file_path": "/var/log/test/application.log*",
            "log_group_name": "application-logs",
            "log_stream_name": "{instance_id}-application",
            #"auto_removal": true
          }
        ]
      }
    }
  }
}

Requirements

Before commit the code, please do the following steps.

Run make fmt and make fmt-sh
Run make lint

plugins/inputs/logfile/tail/tail.go

jefchien · 2025-03-10T14:39:23Z

plugins/inputs/logfile/tail/tail.go

@@ -210,6 +212,15 @@ func (tail *Tail) reopen() error {
 		break
 	}
 	OpenFileCount.Add(1)
+
+	tail.openReader()


This looks like logic that was similar to tailFileSync. Do we need to restart the watcher or any of the other parts of that function?

plugins/inputs/logfile/tailersrc.go

jefchien · 2025-03-10T15:25:47Z

plugins/inputs/logfile/tailersrc.go

@@ -310,6 +326,7 @@ func (ts *tailerSrc) cleanUp() {

 	if ts.outputFn != nil {
 		ts.outputFn(nil) // inform logs agent the tailer src's exit, to stop runSrcToDest
+		ts.outputFn = nil


Why is this needed?

this is to prevent any unseen race condition due to added go routines trying to call outputFn. mutex has been added around this to double protect as well.

I think there has to be a better way to do this without setting the outputFn to nil. The race condition is in the outputFn itself right? Wouldn't it be better to fix it there?

jefchien · 2025-03-10T15:26:57Z

plugins/inputs/logfile/tailersrc.go

+				log.Printf("D! [logfile] runSender buffer was closed for %s", ts.tailer.Filename)
+				return
+			}
+			if e != nil && ts.outputFn != nil {


Can this ever be false?

Yes, with clean up path.

jefchien · 2025-03-10T15:28:53Z

plugins/inputs/logfile/logfile.go

 }

 func NewLogFile() *LogFile {
+	backpressureDrop := envconfig.IsBackpressureDropEnabled()


Alternatively, this could be set in the log configuration. The translator could be the one that reads the environment variable or JSON config and sets the field. Would allow for more granular control this way since you could opt-in/opt-out per log file.

jefchien · 2025-03-10T15:31:57Z

translator/tocwconfig/toenvconfig/toEnvConfig_test.go

+			for k, v := range tt.envVars {
+				os.Setenv(k, v)
+			}
+			defer func() {
+				for k := range tt.envVars {
+					os.Unsetenv(k)
+				}
+			}()


nit: t.Setenv will do the cleanup for us.

Suggested change

for k, v := range tt.envVars {

os.Setenv(k, v)

}

defer func() {

for k := range tt.envVars {

os.Unsetenv(k)

}

}()

for k, v := range tt.envVars {

t.Setenv(k, v)

}

doesn't look like so. dropping Unsetenv will fail subsequent tests after setting the env var.

We might also have leak in our tests by not properly closing files.

plugins/inputs/logfile/logfile.go

plugins/inputs/logfile/tail/tail.go

plugins/inputs/logfile/tailersrc.go

translator/tocwconfig/toenvconfig/toEnvConfig.go

plugins/inputs/logfile/logfile.go

plugins/inputs/logfile/tailersrc.go

jefchien · 2025-03-14T18:24:25Z

plugins/inputs/logfile/tailersrc.go

@@ -310,6 +326,7 @@ func (ts *tailerSrc) cleanUp() {

 	if ts.outputFn != nil {
 		ts.outputFn(nil) // inform logs agent the tailer src's exit, to stop runSrcToDest
+		ts.outputFn = nil


I think there has to be a better way to do this without setting the outputFn to nil. The race condition is in the outputFn itself right? Wouldn't it be better to fix it there?

translator/tocwconfig/toenvconfig/toEnvConfig.go

plugins/inputs/logfile/tailersrc.go

jefchien · 2025-03-18T15:55:16Z

plugins/inputs/logfile/tailersrc.go

+					select {
+					case ts.buffer <- e:
+						// sent event after buffer gets freed up
+						if err := ts.tailer.Reopen(false); err != nil {


What happens if the file isn't closed before the buffer has capacity? It looks like it'll close and then reopen the file, which defeats the purpose of the timer.

What happens if a file is closed and deleted before it tries to reopen the file?

added time out to handle deleted file without getting blocked indifinitely

plugins/inputs/logfile/fileconfig.go

logs/logs.go

internal/logscommon/const.go

jefchien · 2025-03-19T16:43:07Z

plugins/inputs/logfile/tailersrc.go

+					select {
+					case ts.buffer <- e:
+						// sent event after buffer gets freed up
+						if err := ts.tailer.Reopen(false); err != nil {


What happens if a file is closed and deleted before it tries to reopen the file?

lisguo · 2025-03-19T20:13:59Z

plugins/inputs/logfile/tailersrc.go

@@ -163,7 +179,12 @@ func (ts *tailerSrc) Done(offset fileOffset) {
 }

 func (ts *tailerSrc) Stop() {
-	close(ts.done)
+	ts.stopOnce.Do(func() {


do we need this? did you find that in your testing we had multiple Stop() invocations?

plugins/inputs/logfile/tailersrc_test.go

jefchien · 2025-03-20T15:38:35Z

internal/logscommon/const.go

+type BackpressureMode string
+
+const (
+	LogBackpressureModeFDDrop BackpressureMode = "fd_release"


nit: Rename to match the value.

Suggested change

LogBackpressureModeFDDrop BackpressureMode = "fd_release"

BackpressureModeFDRelease BackpressureMode = "fd_release"

jefchien · 2025-03-20T15:39:06Z

logs/logs.go

 	src.SetOutput(func(e LogEvent) {
+		if closed {


nit: Is this needed? Are we continuing to call the function after it's closed?

plugins/inputs/logfile/fileconfig.go

jefchien · 2025-03-20T15:51:28Z

plugins/inputs/logfile/tailersrc.go

@@ -96,27 +101,33 @@ func NewTailerSrc(
 	maxEventSize int,
 	truncateSuffix string,
 	retentionInDays int,
+	backpressureMode string,


nit: Should be the enum type. There's no point in having an enum if it's just converted to a string everywhere.

jefchien · 2025-03-20T16:09:06Z

plugins/inputs/logfile/logfile.go

@@ -287,8 +289,8 @@ func (t *LogFile) getTargetFiles(fileconfig *FileConfig) ([]string, error) {
 	var targetFileName string
 	var targetModTime time.Time
 	for matchedFileName, matchedFileInfo := range g.Match() {
+		t.Log.Debugf("Processing matched file: %s, ModTime: %v", matchedFileName, matchedFileInfo.ModTime())


nit: Remove these debug logs.

jefchien · 2025-03-20T16:12:35Z

plugins/inputs/logfile/tailersrc.go

-	stateFileMode = 0644
-	bufferLimit   = 50
+	stateFileMode     = 0644
+	tailCloseInterval = 3 * time.Second


nit: It's less of an interval and more of a threshold.

translator/tocwconfig/toenvconfig/toEnvConfig.go

jefchien · 2025-03-20T17:12:55Z

plugins/inputs/logfile/tailersrc.go

+						// sent event after buffer gets freed up
+						if ts.tailer.IsFileClosed() { // skip file closing if not already closed
+							if err := ts.tailer.Reopen(false); err != nil {
+								log.Printf("E! [logfile] error reopening file %s: %v", ts.tailer.Filename, err)


nit: We might not want to log this if it just wasn't able to find the file anymore. Might be misleading.

jefchien · 2025-03-20T17:15:09Z

plugins/inputs/logfile/logfile.go

 		fileBaseName := filepath.Base(matchedFileName)
 		if blacklistP != nil && blacklistP.MatchString(fileBaseName) {
 			continue
 		}
 		if !fileconfig.PublishMultiLogs {
+			t.Log.Debugf("Single file mode - current target: %s (ModTime: %v), candidate: %s (ModTime: %v)",


nit: I think this can become excessive logging as well depending on the number of files that match.

movence force-pushed the fix-logs-fd-leak branch 2 times, most recently from 754d84a to 0dc46cd Compare March 4, 2025 21:37

movence marked this pull request as ready for review March 5, 2025 15:17

movence requested a review from a team as a code owner March 5, 2025 15:17

movence force-pushed the fix-logs-fd-leak branch from e3a0758 to 15d84b5 Compare March 6, 2025 03:33

jefchien reviewed Mar 10, 2025

View reviewed changes

movence force-pushed the fix-logs-fd-leak branch 2 times, most recently from 4fb59d8 to 3f2a3f3 Compare March 14, 2025 13:39

lisguo reviewed Mar 14, 2025

View reviewed changes

jefchien reviewed Mar 14, 2025

View reviewed changes

movence force-pushed the fix-logs-fd-leak branch from d97315f to e144b1e Compare March 18, 2025 17:22

jefchien reviewed Mar 18, 2025

View reviewed changes

jefchien reviewed Mar 19, 2025

View reviewed changes

lisguo reviewed Mar 19, 2025

View reviewed changes

movence force-pushed the fix-logs-fd-leak branch from 37f9101 to 9ed151a Compare March 20, 2025 15:46

jefchien reviewed Mar 20, 2025

View reviewed changes

jefchien previously approved these changes Mar 20, 2025

View reviewed changes

lisguo previously approved these changes Mar 20, 2025

View reviewed changes

translator/tocwconfig/toenvconfig/toEnvConfig.go Outdated Show resolved Hide resolved

movence added 12 commits March 20, 2025 13:07

add fixed size buffer to handle fd in tailer

1a59c7b

lint

4d908dd

use util func to handle feature flag

9c0116e

update flag name to be more aligned with functionality

89b056c

skip adding env var when disabled

0291af2

clean up

f48b874

fix panic

36d3cc3

test

9e47b2c

test

50f6481

use timer to close fd

5aecca8

clean up and update test

4cb8139

fix race condition

21994d4

movence added 11 commits March 20, 2025 13:07

fix flaky test

f1a57aa

fix test on windows

85156cd

relax test

4f93b3e

clean up

43d58a9

simplify thread control

3dadff9

move feature flag to individual file level config for granular control

6fbd3ff

bypass/fix lint

a4a7ac5

update config name and use string value instead bool

56fe9a3

add time out in waiting on new file. change mode name and some clean up

5758e61

lint

311a123

nits

35154e1

movence dismissed stale reviews from lisguo and jefchien via 35154e1 March 20, 2025 17:07

movence force-pushed the fix-logs-fd-leak branch from bb19fd4 to 35154e1 Compare March 20, 2025 17:07

jefchien previously approved these changes Mar 20, 2025

View reviewed changes

jefchien reviewed Mar 20, 2025

View reviewed changes

remove more logging

73dbe88

movence dismissed jefchien’s stale review via 73dbe88 March 20, 2025 17:25

jefchien approved these changes Mar 20, 2025

View reviewed changes

Merge branch 'main' into fix-logs-fd-leak

959e1c2

lisguo approved these changes Mar 20, 2025

View reviewed changes

sky333999 merged commit 7e2fbb1 into aws:main Mar 20, 2025
7 checks passed

	LogBackpressureModeFDDrop BackpressureMode = "fd_release"
	BackpressureModeFDRelease BackpressureMode = "fd_release"

add fixed size buffer to handle fd in log tailer with feature flag #1577

add fixed size buffer to handle fd in log tailer with feature flag #1577

Uh oh!

Conversation

movence commented Mar 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Revisions

Description of the issue

Description of changes

License

Tests

Requirements

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

movence Mar 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

movence commented Mar 3, 2025 •

edited

Loading

movence Mar 13, 2025 •

edited

Loading