Skip to content

fix: fix concurrent map iteration and map write race condition #162

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

Closed
mayankpande88 wants to merge 1 commit into from
Closed

fix: fix concurrent map iteration and map write race condition #162

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
15 changes: 15 additions & 0 deletions containers/container.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -421,10 +421,12 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
ch <- counter(metrics.NetBytesSent, float64(stats.BytesSent), d.DestinationLabelValue(), d.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind, workload_src.Region, workload_src.Zone, workload_dest.Region, workload_dest.Zone, actualDestWorkload.Region, actualDestWorkload.Zone, actualDestWorkload.Instance)
ch <- counter(metrics.NetBytesReceived, float64(stats.BytesReceived), d.DestinationLabelValue(), d.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind, workload_src.Region, workload_src.Zone, workload_dest.Region, workload_dest.Zone, actualDestWorkload.Region, actualDestWorkload.Zone, actualDestWorkload.Instance)
}
c.lock.RLock()
for dst, count := range c.failedConnectionAttempts {
workload := c.ip_resolver.ResolveIP(dst.IP().String())
ch <- counter(metrics.NetConnectionsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind, c.srcWorkload.Region, c.srcWorkload.Zone, workload.Region, workload.Zone, workload.Region, workload.Zone, workload.Instance)
}
c.lock.RUnlock()
Comment on lines +424 to +429

@gemini-code-assist gemini-code-assist Bot Jan 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

medium

While this change correctly adds a lock to prevent a race condition, holding a read lock for the duration of the loop can lead to lock contention, especially since the ch <- counter(...) operation can block.

A better approach is to create a snapshot of the map while holding the lock, and then iterate over the snapshot. This minimizes the time the lock is held. You've already used this "snapshot pattern" for c.logParsers in this same function.

	c.lock.RLock()
	failedAttemptsSnapshot := make(map[common.HostPort]int64, len(c.failedConnectionAttempts))
	for dst, count := range c.failedConnectionAttempts {
		failedAttemptsSnapshot[dst] = count
	}
	c.lock.RUnlock()
	for dst, count := range failedAttemptsSnapshot {
		workload := c.ip_resolver.ResolveIP(dst.IP().String())
		ch <- counter(metrics.NetConnectionsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind, c.srcWorkload.Region, c.srcWorkload.Zone, workload.Region, workload.Zone, workload.Region, workload.Zone, workload.Instance)
	}


// Log connection stats phase timing
connStatsTime := time.Since(phaseStart)
Expand All @@ -436,25 +438,38 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
phaseStart = time.Now()

connections := map[common.DestinationKey]int{}
c.lock.RLock()
for _, conn := range c.activeConnections {
if !conn.Closed.IsZero() {
continue
}
connections[conn.DestinationKey]++
}
c.lock.RUnlock()
for d, count := range connections {
actualDestWorkload := d.GetActualDestinationWorkload()
destWorkload := d.GetDestinationWorkload()
ch <- gauge(metrics.NetConnectionsActive, float64(count), d.DestinationLabelValue(), d.ActualDestinationLabelValue(), c.srcWorkload.Name, c.srcWorkload.Namespace, c.srcWorkload.Kind, destWorkload.Name, destWorkload.Namespace, destWorkload.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind, c.srcWorkload.Region, c.srcWorkload.Zone, destWorkload.Region, destWorkload.Zone, actualDestWorkload.Region, actualDestWorkload.Zone, actualDestWorkload.Instance)
}

c.lock.RLock()
logParsersSnapshot := make(map[string]*LogParser)
for source, p := range c.logParsers {
logParsersSnapshot[source] = p
}
c.lock.RUnlock()

for source, p := range logParsersSnapshot {
for _, ctr := range p.parser.GetCounters() {
if ctr.Level == logparser.LevelCritical || ctr.Level == logparser.LevelError {
c.lock.RLock()
sample, ok := c.logSamples[ctr.Hash]
c.lock.RUnlock()
if !ok {
sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength)
c.lock.Lock()
c.logSamples[ctr.Hash] = sample
c.lock.Unlock()
}
Comment on lines 468 to 473

@gemini-code-assist gemini-code-assist Bot Jan 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

medium

This change correctly adds locking around access to c.logSamples. However, the current implementation has a logical race condition (a form of check-then-act). If two goroutines check for a sample at the same time and find it missing, both will proceed to create and write the sample to the map. This leads to redundant work and one write overwriting the other.

A more robust approach is to use a double-checked locking pattern. After acquiring the write lock, you should check again if the sample has been added by another goroutine before writing.

				if !ok {
					newSample := common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength)
					c.lock.Lock()
					// Re-check if another goroutine has created the sample in the meantime.
					if s, ok2 := c.logSamples[ctr.Hash]; ok2 {
						sample = s // Use the existing sample.
					} else {
						c.logSamples[ctr.Hash] = newSample // Create the sample.
						sample = newSample
					}
					c.lock.Unlock()
				}

ch <- counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample)
}
Expand Down