fix: harden disk exporter concurrency#4422
Open
JaydipGabani wants to merge 21 commits into
Open
Conversation
Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR addresses a data race in the disk export driver by adding synchronization around accesses to Writer.openConnections.
Changes:
- Add
Writer.mulocking aroundopenConnectionswrites inCreateConnection. - Add locking around
openConnectionsreads/writes inUpdateConnection,CloseConnection, andPublish. - Reorder
UpdateConnectionto validate config before locking, then safely read/update the connection.
Comments suppressed due to low confidence (3)
pkg/export/disk/disk.go:151
CloseConnectionnow holdsr.muacrosscloseAndRemoveFilesWithRetry(conn), which includes retries and filesystem operations. This can stall all concurrent publishes/connection updates while a close is retrying, and can also delay startup ofbackgroundCleanup(the goroutine will block on the same mutex). A safer pattern is to remove the connection fromopenConnectionsunder lock, release the lock, perform close/remove, then re-lock only to updateclosedConnections/cleanup state on failure.
r.mu.Lock()
defer r.mu.Unlock()
conn, ok := r.openConnections[connectionName]
if !ok {
return fmt.Errorf("connection %s not found for disk driver", connectionName)
}
defer delete(r.openConnections, connectionName)
err := r.closeAndRemoveFilesWithRetry(conn)
if err != nil {
pkg/export/disk/disk.go:205
Publishtakesr.mu.Lock()for the entire method, including JSON marshaling and all filesystem operations (handleAuditStart,WriteString,handleAuditEnd, directory walks). This serializes all publishes across all connections and can become a bottleneck under load. Consider using the mutex only to safely fetch/store the connection from the map, and protect per-connection mutable state (File/currentAuditRun) with a per-connection lock (or store pointers in the map and use finer-grained locking).
r.mu.Lock()
defer r.mu.Unlock()
conn, ok := r.openConnections[connectionName]
if !ok {
return fmt.Errorf("invalid connection: %s not found for disk driver", connectionName)
}
var violation util.ExportMsg
if violation, ok = data.(util.ExportMsg); !ok {
return fmt.Errorf("invalid data type: cannot convert data to exportMsg")
}
if violation.Message == util.AuditStartedMsg {
err := conn.handleAuditStart(violation.ID, topic)
if err != nil {
return fmt.Errorf("error handling audit start: %w", err)
}
r.openConnections[connectionName] = conn
}
jsonData, err := json.Marshal(data)
if err != nil {
return fmt.Errorf("error marshaling data: %w", err)
}
if conn.File == nil {
return fmt.Errorf("failed to write violation: no file provided")
}
_, err = conn.File.WriteString(string(jsonData) + "\n")
if err != nil {
return fmt.Errorf("error writing message to disk: %w", err)
pkg/export/disk/disk.go:129
r.muis held while callingcloseAndRemoveFilesWithRetry(conn), which can sleep/retry and do filesystem I/O. This blocks all Publish/Create/Close operations (even for other connections) for the duration of the backoff and can significantly reduce export throughput. Consider limiting the critical section to map access (e.g., read/updateopenConnectionsunder lock, but perform close/remove outside the lock, potentially using a per-connection lock or temporarily removing/marking the connection to prevent concurrent Publish).
r.mu.Lock()
defer r.mu.Unlock()
conn, exists := r.openConnections[connectionName]
if !exists {
return fmt.Errorf("connection %s for disk driver not found", connectionName)
}
if conn.Path != path {
if err := r.closeAndRemoveFilesWithRetry(conn); err != nil {
return fmt.Errorf("error updating connection %s, %w", connectionName, err)
}
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #4422 +/- ##
==========================================
- Coverage 54.49% 45.32% -9.17%
==========================================
Files 134 288 +154
Lines 12329 21298 +8969
==========================================
+ Hits 6719 9654 +2935
- Misses 5116 10824 +5708
- Partials 494 820 +326
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
aramase
reviewed
Mar 9, 2026
aramase
left a comment
aramase
left a comment
Contributor
There was a problem hiding this comment.
Could you add a concurrency test exercising concurrent Publish/UpdateConnection/CloseConnection to validate the fix and prevent regressions.
Contributor
Author
Some of these tests are being added here - #4409. I will add what is not covered by this PR. |
Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
aramase
reviewed
Mar 24, 2026
…cleanupDone Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
Comments suppressed due to low confidence (2)
pkg/export/disk/disk.go:109
- CreateConnection overwrites any existing entry in openConnections unconditionally. If CreateConnection is called again for an already-in-use connection (e.g., while an audit run has an open File), this will drop the prior Connection struct (leaking/losing the file handle and lock state) and can cause subsequent Publish calls to fail with "no file provided". Consider either returning an error when the connection already exists, or preserving existing runtime state (File/currentAuditRun) when the config is effectively the same, and explicitly closing/cleaning up when replacing a live connection.
func (r *Writer) CreateConnection(_ context.Context, connectionName string, config interface{}) error {
path, maxResults, ttl, err := unmarshalConfig(config)
if err != nil {
return fmt.Errorf("error creating connection %s: %w", connectionName, err)
}
r.mu.Lock()
defer r.mu.Unlock()
r.openConnections[connectionName] = Connection{
Path: path,
MaxAuditResults: int(maxResults),
ClosedConnectionTTL: ttl,
}
return nil
pkg/export/disk/disk.go:223
- Publish now holds the single Writer mutex across filesystem work (MkdirAll/Chmod/OpenFile/Flock in handleAuditStart, WriteString, Rename/cleanup in handleAuditEnd). This serializes all publishes across all connections and can become a throughput bottleneck under concurrent publishing. Consider using an RWMutex (RLock for normal writes, Lock for audit start/end/connection mutations) or a per-connection lock so different connections can publish concurrently without racing on connection state.
// Publish holds the global lock for its entire body because Connection values
// share a *os.File pointer; releasing the lock mid-method would let concurrent
// publishers race on the same file descriptor.
func (r *Writer) Publish(_ context.Context, connectionName string, data interface{}, topic string) error {
r.mu.Lock()
defer r.mu.Unlock()
conn, ok := r.openConnections[connectionName]
if !ok {
return fmt.Errorf("invalid connection: %s not found for disk driver", connectionName)
}
var violation util.ExportMsg
if violation, ok = data.(util.ExportMsg); !ok {
return fmt.Errorf("invalid data type: cannot convert data to exportMsg")
}
if violation.Message == util.AuditStartedMsg {
err := conn.handleAuditStart(violation.ID, topic)
if err != nil {
return fmt.Errorf("error handling audit start: %w", err)
}
r.openConnections[connectionName] = conn
}
jsonData, err := json.Marshal(data)
if err != nil {
return fmt.Errorf("error marshaling data: %w", err)
}
if conn.File == nil {
return fmt.Errorf("failed to write violation: no file provided")
}
_, err = conn.File.WriteString(string(jsonData) + "\n")
if err != nil {
return fmt.Errorf("error writing message to disk: %w", err)
Contributor
|
This PR has been automatically marked as stale because it has not had recent activity. It will be closed in 14 days if no further activity occurs. Thank you for your contributions. |
# Conflicts: # pkg/export/disk/disk.go
Comment on lines
289
to
294
| if violation.Message == util.AuditStartedMsg { | ||
| err := conn.handleAuditStart(violation.ID, topic) | ||
| if err != nil { | ||
| return fmt.Errorf("error handling audit start: %w", err) | ||
| } | ||
| r.openConnections[connectionName] = conn | ||
| } |
Prevent stale failed-close cleanup from deleting an export path that has been recreated by an active disk connection. Retry close and remove failures through the configured backoff, and clear closed files from queued cleanup state so retries do not attempt to close the same file descriptor again. Add focused regression coverage for reused paths, cleanup retries, and path-update visibility. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Allow relative disk export paths to keep existing Connection configs and documented Helm examples working while retaining traversal and filesystem-root validation. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Add regression coverage for CreateConnection and UpdateConnection rejecting paths currently reserved for cleanup while preserving existing connection state. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
JaydipGabani
changed the title
Prevent CreateConnection from overwriting existing connection state and leaking open audit files. Add regression coverage that duplicate creates preserve the existing file handle and connection configuration. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Share the close/remove cleanup fallback between immediate cleanup and background retry so Writer values without an injected cleanup function do not panic during retry. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Reject paths that resolve to "." so cleanup cannot os.RemoveAll the working directory. Create the target directory only after the cleanup-in-progress check, and reserve it during a path change, so a rejected or concurrent operation cannot recreate or delete a directory mid-cleanup. Split validatePath from directory creation to keep config parsing side-effect free, and schedule the next retry from the current time instead of a stale snapshot. Signed-off-by: Jaydip Gabani <gabanijaydip@gmail.com>
Comment on lines
+74
to
+78
| if ttlValue, ok := cfg["closedConnectionTTL"]; ok { | ||
| ttlStr, ok := ttlValue.(string) | ||
| if !ok { | ||
| return "", 0.0, 0, fmt.Errorf("invalid ttl format: expected string") | ||
| } |
Comment on lines
+95
to
+100
| if err := ensureDirectory(path); err != nil { | ||
| r.mu.Lock() | ||
| if r.connectionLockIsCurrentLocked(connectionName, connLock) { | ||
| delete(r.openConnections, connectionName) | ||
| } | ||
| r.mu.Unlock() |
| r.openConnections[connectionName] = conn | ||
| } | ||
| r.mu.Unlock() | ||
| return fmt.Errorf("error updating connection %s, %w", connectionName, err) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What this PR does / why we need it:
Hardens the disk export driver concurrency and cleanup paths. The change adds per-connection synchronization around
CreateConnection,UpdateConnection,CloseConnection, andPublish, keeps cleanup I/O out of the global writer mutex, and prevents stale per-connection locks from mutating newer state.It also tightens disk file handling for audit exports: failed cleanup paths are retried without deleting paths still used by active connections, audit files remain tracked across
Publisherror paths, failed lock acquisition closes opened files, repeated audit starts are rejected, and audittopic/auditIDvalues are validated before being used as filesystem path components.Additional disk hardening includes deterministic
.logretention cleanup, non-blocking file locks, narrower explicit file/directory permissions, validation for fractionalmaxAuditResultsand malformedclosedConnectionTTL, and preservation of existing relative-pathConnectionconfigs.Which issue(s) this PR fixes (optional, using
fixes #<issue number>(, fixes #<issue_number>, ...)format, will close the issue(s) when the PR gets merged):N/A
Special notes for your reviewer: