Skip to content

Address goroutine leak with dynamically determined log destinations #1848

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 14 commits into from
Oct 17, 2025
Merged

Address goroutine leak with dynamically determined log destinations #1848

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
a312b0f
adding some debug info
dricross Aug 27, 2025
128847f
Stop the queue
dricross Aug 29, 2025
a7510db
Create one configurer for the lifetime of CloudWatchLogs
dricross Sep 3, 2025
8778ad9
Fixup unit tests
dricross Sep 3, 2025
a5424dc
Add reference counter to cwDest
dricross Sep 4, 2025
90e380c
Remove unneeded changes
dricross Sep 4, 2025
2b15fa5
Remove debug printing
dricross Sep 5, 2025
4233347
Add some refreshing comments
dricross Sep 5, 2025
4082795
Locking cwDest on Publish. Remove vestigial functionality
dricross Sep 8, 2025
50d569c
Add error log when negative refCount is detected
dricross Sep 10, 2025
17ca8f5
defer dest notification. stop underlying sender. assert sender is
dricross Oct 7, 2025
012ea3f
Merge branch 'main' into dricross/logfiletailerdebug
dricross Oct 16, 2025
45e6bd1
Merge branch 'main' into dricross/logfiletailerdebug
dricross Oct 16, 2025
4e7e9dd
Merge branch 'main' into dricross/logfiletailerdebug
dricross Oct 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions logs/logs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,13 +65,13 @@ type LogBackend interface {
// e.g. a particular log stream in cloudwatchlogs.
type LogDest interface {
Publish(events []LogEvent) error
NotifySourceStopped()
}

// LogAgent is the agent handles pure log pipelines
type LogAgent struct {
Config *config.Config
backends map[string]LogBackend
destNames map[LogDest]string
collections []LogCollection
retentionAlreadyAttempted map[string]bool
}
Expand All @@ -80,7 +80,6 @@ func NewLogAgent(c *config.Config) *LogAgent {
return &LogAgent{
Config: c,
backends: make(map[string]LogBackend),
destNames: make(map[LogDest]string),
retentionAlreadyAttempted: make(map[string]bool),
}
}
Expand Down Expand Up @@ -136,7 +135,6 @@ func (l *LogAgent) Run(ctx context.Context) {
}
retention = l.checkRetentionAlreadyAttempted(retention, logGroup)
dest := backend.CreateDest(logGroup, logStream, retention, logGroupClass, src)
l.destNames[dest] = dname
log.Printf("I! [logagent] piping log from %s/%s(%s) to %s with retention %d", logGroup, logStream, description, dname, retention)
go l.runSrcToDest(src, dest)
}
Expand All @@ -148,8 +146,10 @@ func (l *LogAgent) Run(ctx context.Context) {
}

func (l *LogAgent) runSrcToDest(src LogSrc, dest LogDest) {

eventsCh := make(chan LogEvent)
defer src.Stop()
defer dest.NotifySourceStopped()
Copy link
Contributor

@jefchien jefchien Oct 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it matter which order this is in? If the source is stopped after the destination reports it, would that be an issue?

Copy link
Contributor Author

@dricross dricross Oct 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The order doesn't matter. The two objects are only linked through this function. Maybe the naming isn't optimal here which might cause some confusion. The NotifySourceStopped method on dest doesn't notify the source, it notifies dest that the source is stopped.


closed := false
src.SetOutput(func(e LogEvent) {
Expand All @@ -168,11 +168,11 @@ func (l *LogAgent) runSrcToDest(src LogSrc, dest LogDest) {
for e := range eventsCh {
err := dest.Publish([]LogEvent{e})
if err == ErrOutputStopped {
log.Printf("I! [logagent] Log destination %v has stopped, finalizing %v/%v", l.destNames[dest], src.Group(), src.Stream())
Copy link
Contributor Author

@dricross dricross Sep 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There was actually a concurrent map write condition here. This non-sync map is written to in the Run routine and is potentially read from the runSrcToDest goroutine.

log.Printf("I! [logagent] Log destination %v has stopped, finalizing %v/%v", src.Destination(), src.Group(), src.Stream())
return
}
if err != nil {
log.Printf("E! [logagent] Failed to publish log to %v, error: %v", l.destNames[dest], err)
log.Printf("E! [logagent] Failed to publish log to %v, error: %v", src.Destination(), err)
return
}
}
Expand Down
8 changes: 2 additions & 6 deletions plugins/inputs/logfile/logfile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,9 @@ type LogFile struct {
started bool
}

func NewLogFile() *LogFile {
var _ logs.LogCollection = (*LogFile)(nil)

func NewLogFile() *LogFile {
return &LogFile{
configs: make(map[*FileConfig]map[string]*tailerSrc),
done: make(chan struct{}),
Expand Down Expand Up @@ -251,11 +252,6 @@ func (t *LogFile) FindLogSrc() []logs.LogSrc {
}
}

destination := fileconfig.Destination
if destination == "" {
destination = t.Destination
}

Comment on lines -254 to -258
Copy link
Contributor Author

@dricross dricross Sep 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This variable is unused

src := NewTailerSrc(
groupName, streamName,
t.Destination,
Expand Down
Loading
Loading