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

Open
wants to merge 11 commits into
base: main
Choose a base branch
from
Open

Address goroutine leak with dynamically determined log destinations #1848

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 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
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()

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

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

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