mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
e48bf9a79192a5dcf999fdc0b3171a4df589fbca
99
Commits
| Author | SHA1 | Message | Date | |
|---|---|---|---|---|
 chrislu
|
e48bf9a791 |
security: upgrade Jetty from 9.4.53 to 12.0.16
- Upgrade from 9.4.53.v20231009 to 12.0.16 (meets requirement >12.0.9) - Addresses security vulnerabilities in older Jetty versions - Externalized version to jetty.version property for easier maintenance - Added jetty-util, jetty-io, jetty-security to dependencyManagement - Ensures all Jetty transitive dependencies use secure version |
||
|
chrislu
|
a1a14259c3 |
fix: add -max=0 to volume server for unlimited volumes
- Add -max=0 flag to volume server command - Allows volume server to create unlimited 50MB volumes - Fixes 'No writable volumes' error during Spark tests - Volume server will create new volumes as needed for writes - Consistent with other integration test configurations |
||
|
chrislu
|
c49abc0c2f |
security: upgrade Apache ZooKeeper to 3.9.4
- Upgrade from 3.9.3 to 3.9.4 (latest stable) - Ensures all known security vulnerabilities are patched - Fixes GHSA-g93m-8x6h-g5gv, GHSA-r978-9m6m-6gm6, GHSA-2hmj-97jw-28jh |
||
|
chrislu
|
a051452fe6 |
security: upgrade Apache ZooKeeper to 3.9.3
- Upgrade from 3.9.1 to 3.9.3 - Fixes GHSA-g93m-8x6h-g5gv: Authentication bypass in Admin Server - Fixes GHSA-r978-9m6m-6gm6: Information disclosure in persistent watchers - Fixes GHSA-2hmj-97jw-28jh: Insufficient permission check in snapshot/restore - Addresses high and moderate severity vulnerabilities |
||
|
chrislu
|
150deefdc0 |
fix: aggressively suppress Parquet DEBUG logging
- Set Parquet I/O loggers to OFF (completely disabled) - Add log4j.configuration system property to ensure config is used - Override Spark's default log4j configuration - Prevents thousands of record-level DEBUG messages in CI logs |
||
|
chrislu
|
b018588c14 |
security: upgrade Netty to 4.1.124.Final (patched version)
- Upgrade from 4.1.118.Final to 4.1.124.Final - Fixes GHSA-prj3-ccx8-p6x4: MadeYouReset HTTP/2 DDoS vulnerability - 4.1.124.Final is the confirmed patched version per GitHub advisory - All versions <= 4.1.123.Final are vulnerable |
||
|
chrislu
|
4dd55783da |
security: upgrade Netty to 4.1.118.Final
- Upgrade from 4.1.115.Final to 4.1.118.Final - Fixes CVE-2025-24970: improper validation in SslHandler - Fixes CVE-2024-47535: unsafe environment file reading on Windows - Fixes CVE-2024-29025: HttpPostRequestDecoder resource exhaustion - Addresses GHSA-prj3-ccx8-p6x4 and related vulnerabilities |
||
|
chrislu
|
fab383dc10 |
fix: use 127.0.0.1 for volume server IP registration
- Change volume -ip from seaweedfs-volume to 127.0.0.1 - Change -publicUrl from localhost:8080 to 127.0.0.1:8080 - Volume server now registers with master using 127.0.0.1 - Filer will return 127.0.0.1:8080 URL that's resolvable from host - Fixes UnknownHostException for seaweedfs-volume hostname |
||
|
chrislu
|
707e7732a7 |
fix: suppress verbose Parquet DEBUG logging
- Set org.apache.parquet to WARN level - Set org.apache.parquet.io to ERROR level - Suppress RecordConsumerLoggingWrapper and MessageColumnIO DEBUG logs - Reduces CI log noise from thousands of record-level messages - Keeps important error messages visible |
||
|
chrislu
|
8c13794a49 |
security: upgrade Netty to 4.1.115.Final to fix CVE
- Upgrade netty.version from 4.1.100.Final to 4.1.115.Final - Fixes GHSA-prj3-ccx8-p6x4: MadeYouReset HTTP/2 DDoS vulnerability - Netty 4.1.115.Final includes patches for high severity DoS attack - Addresses GitHub dependency review security alert |
||
|
chrislu
|
abaf933971 |
fix: add publicUrl to volume server for host network access
- Add -publicUrl=localhost:8080 to volume server command - Ensures filer returns localhost URL instead of Docker service name - Fixes UnknownHostException when tests run on host network - Volume server is accessible via localhost from CI runner |
||
|
chrislu
|
01e20a350c |
refactor: extract surefire JVM args to property
- Move multi-line argLine to surefire.jvm.args property - Reference property in argLine for cleaner configuration - Improves maintainability and readability - Follows Maven best practices for JVM argument management - Avoids potential whitespace parsing issues |
||
|
chrislu
|
3074b1ee2f |
refactor: externalize seaweedfs-hadoop3-client version to property
- Add seaweedfs.hadoop3.client.version property set to 3.80
- Replace hardcoded version with ${seaweedfs.hadoop3.client.version}
- Enables easier version management from single location
- Follows Maven best practices for dependency versioning
|
||
|
chrislu
|
7b548be48e |
security: add dependencyManagement to fix vulnerable transitives
- Pin Jackson to 2.15.3 (fixes multiple CVEs in older versions) - Pin Netty to 4.1.100.Final (fixes CVEs in transport/codec) - Pin Apache Avro to 1.11.4 (fixes deserialization CVEs) - Pin Apache ZooKeeper to 3.9.1 (fixes authentication bypass) - Pin commons-compress to 1.26.0 (fixes zip slip vulnerabilities) - Pin commons-io to 2.15.1 (fixes path traversal) - Pin Guava to 32.1.3-jre (fixes temp directory vulnerabilities) - Pin SnakeYAML to 2.2 (fixes arbitrary code execution) - Pin Jetty to 9.4.53 (fixes multiple HTTP vulnerabilities) - Overrides vulnerable versions from Spark/Hadoop transitives |
||
|
chrislu
|
ec08fadf85 |
fix: align maven-compiler-plugin with compiler properties
- Change compiler plugin source/target from hardcoded 1.8 to use properties - Ensures consistency with maven.compiler.source/target set to 11 - Prevents version mismatch between properties and plugin configuration - Aligns with surefire Java 9+ module arguments |
||
|
chrislu
|
e8e9df2680 |
test: improve docker-compose config for Spark tests
- Add -volumeSizeLimitMB=50 to master (consistent with other integration tests) - Add -defaultReplication=000 to master for explicit single-copy storage - Add explicit -port and -port.grpc flags to all services - Add -preStopSeconds=1 to volume for faster shutdown - Add healthchecks to master and volume services - Use service_healthy conditions for proper startup ordering - Improve healthcheck intervals and timeouts for faster startup - Use -ip flag instead of -ip.bind for service identity |
||
|
chrislu
|
dd9c1c6190 |
fix: add -peers=none to master command for standalone mode
- Ensures master runs in standalone single-node mode - Prevents master from trying to form a cluster - Required for proper initialization in test environment |
||
|
chrislu
|
e63feb48fc |
ci: fix Maven artifact restoration in workflow
- Add step to restore Maven artifacts from download to ~/.m2/repository - Restructure artifact upload to use consistent directory layout - Remove obsolete 'version' field from docker-compose.yml to eliminate warnings - Ensures SeaweedFS Java dependencies are available during test execution |
||
|
chrislu
|
786f5de7bb |
ci: refactor Spark workflow for DRY and robustness
1. Add explicit permissions (least privilege): - contents: read - checks: write (for test reports) - pull-requests: write (for PR comments) 2. Extract duplicate build steps into shared 'build-deps' job: - Eliminates duplication between spark-tests and spark-example - Build artifacts are uploaded and reused by dependent jobs - Reduces CI time and ensures consistency 3. Fix spark-example service startup verification: - Match robust approach from spark-tests job - Add explicit timeout and failure handling - Verify all services (master, volume, filer) - Include diagnostic logging on failure - Prevents silent failures and obscure errors These changes improve maintainability, security, and reliability of the Spark integration test workflow. |
||
|
chrislu
|
b35463c8b4 |
spark: fix flaky test by sorting DataFrame before first()
- In testLargeDataset(), add orderBy("value") before calling first()
- Parquet files don't guarantee row order, so first() on unordered
DataFrame can return any row, making assertions flaky
- Sorting by 'value' ensures the first row is always the one with
value=0, making the test deterministic and reliable
|
||
|
chrislu
|
6f905acfd2 |
spark: update compiler source/target from Java 8 to Java 11
- Fix inconsistency between maven.compiler.source/target (1.8) and surefire JVM args (Java 9+ module flags like --add-opens) - Update to Java 11 to match CI environment (GitHub Actions uses Java 11) - Docker environment uses Java 17 which is also compatible - Java 11+ is required for the --add-opens/--add-exports flags used in the surefire configuration |
||
|
chrislu
|
fba36f8d1c |
java: fix NPE in SeaweedWrite and Makefile env var scope
- Add null check for HttpEntity in SeaweedWrite.multipartUpload() to prevent NPE when response.getEntity() returns null - Fix Makefile test target to properly export SEAWEEDFS_TEST_ENABLED by setting it on the same command line as mvn test - Update docker-compose commands to use V2 syntax (docker compose) for consistency with GitHub Actions workflow |
||
|
chrislu
|
e15e2e23c4 | fix building | ||
|
chrislu
|
3841fac4fe | go mod tidy | ||
|
chrislu
|
89a6d42cee | Complete Spark integration test suite | ||
|
Chris Lu
|
64dcbbb25b |
test read write by s3fs and PyArrow native file system for s3 (#7520)
* test read write by s3fs and PyArrow native file system for s3 * address comments * add github action |
||
|
Chris LuandCopilot
|
c1b8d4bf0d |
S3: adds FilerClient to use cached volume id (#7518)
* adds FilerClient to use cached volume id
* refactor: MasterClient embeds vidMapClient to eliminate ~150 lines of duplication
- Create masterVolumeProvider that implements VolumeLocationProvider
- MasterClient now embeds vidMapClient instead of maintaining duplicate cache logic
- Removed duplicate methods: LookupVolumeIdsWithFallback, getStableVidMap, etc.
- MasterClient still receives real-time updates via KeepConnected streaming
- Updates call inherited addLocation/deleteLocation from vidMapClient
- Benefits: DRY principle, shared singleflight, cache chain logic reused
- Zero behavioral changes - only architectural improvement
* refactor: mount uses FilerClient for efficient volume location caching
- Add configurable vidMap cache size (default: 5 historical snapshots)
- Add FilerClientOption struct for clean configuration
* GrpcTimeout: default 5 seconds (prevents hanging requests)
* UrlPreference: PreferUrl or PreferPublicUrl
* CacheSize: number of historical vidMap snapshots (for volume moves)
- NewFilerClient uses option struct for better API extensibility
- Improved error handling in filerVolumeProvider.LookupVolumeIds:
* Distinguish genuine 'not found' from communication failures
* Log volumes missing from filer response
* Return proper error context with volume count
* Document that filer Locations lacks Error field (unlike master)
- FilerClient.GetLookupFileIdFunction() handles URL preference automatically
- Mount (WFS) creates FilerClient with appropriate options
- Benefits for weed mount:
* Singleflight: Deduplicates concurrent volume lookups
* Cache history: Old volume locations available briefly when volumes move
* Configurable cache depth: Tune for different deployment environments
* Battle-tested vidMap cache with cache chain
* Better concurrency handling with timeout protection
* Improved error visibility and debugging
- Old filer.LookupFn() kept for backward compatibility
- Performance improvement for mount operations with high concurrency
* fix: prevent vidMap swap race condition in LookupFileIdWithFallback
- Hold vidMapLock.RLock() during entire vm.LookupFileId() call
- Prevents resetVidMap() from swapping vidMap mid-operation
- Ensures atomic access to the current vidMap instance
- Added documentation warnings to getStableVidMap() about swap risks
- Enhanced withCurrentVidMap() documentation for clarity
This fixes a subtle race condition where:
1. Thread A: acquires lock, gets vm pointer, releases lock
2. Thread B: calls resetVidMap(), swaps vc.vidMap
3. Thread A: calls vm.LookupFileId() on old/stale vidMap
While the old vidMap remains valid (in cache chain), holding the lock
ensures we consistently use the current vidMap for the entire operation.
* fix: FilerClient supports multiple filer addresses for high availability
Critical fix: FilerClient now accepts []ServerAddress instead of single address
- Prevents mount failure when first filer is down (regression fix)
- Implements automatic failover to remaining filers
- Uses round-robin with atomic index tracking (same pattern as WFS.WithFilerClient)
- Retries all configured filers before giving up
- Updates successful filer index for future requests
Changes:
- NewFilerClient([]pb.ServerAddress, ...) instead of (pb.ServerAddress, ...)
- filerVolumeProvider references FilerClient for failover access
- LookupVolumeIds tries all filers with util.Retry pattern
- Mount passes all option.FilerAddresses for HA
- S3 wraps single filer in slice for API consistency
This restores the high availability that existed in the old implementation
where mount would automatically failover between configured filers.
* fix: restore leader change detection in KeepConnected stream loop
Critical fix: Leader change detection was accidentally removed from the streaming loop
- Master can announce leader changes during an active KeepConnected stream
- Without this check, client continues talking to non-leader until connection breaks
- This can lead to stale data or operational errors
The check needs to be in TWO places:
1. Initial response (lines 178-187): Detect redirect on first connect
2. Stream loop (lines 203-209): Detect leader changes during active stream
Restored the loop check that was accidentally removed during refactoring.
This ensures the client immediately reconnects to new leader when announced.
* improve: address code review findings on error handling and documentation
1. Master provider now preserves per-volume errors
- Surface detailed errors from master (e.g., misconfiguration, deletion)
- Return partial results with aggregated errors using errors.Join
- Callers can now distinguish specific volume failures from general errors
- Addresses issue of losing vidLoc.Error details
2. Document GetMaster initialization contract
- Add comprehensive documentation explaining blocking behavior
- Clarify that KeepConnectedToMaster must be started first
- Provide typical initialization pattern example
- Prevent confusing timeouts during warm-up
3. Document partial results API contract
- LookupVolumeIdsWithFallback explicitly documents partial results
- Clear examples of how to handle result + error combinations
- Helps prevent callers from discarding valid partial results
4. Add safeguards to legacy filer.LookupFn
- Add deprecation warning with migration guidance
- Implement simple 10,000 entry cache limit
- Log warning when limit reached
- Recommend wdclient.FilerClient for new code
- Prevents unbounded memory growth in long-running processes
These changes improve API clarity and operational safety while maintaining
backward compatibility.
* fix: handle partial results correctly in LookupVolumeIdsWithFallback callers
Two callers were discarding partial results by checking err before processing
the result map. While these are currently single-volume lookups (so partial
results aren't possible), the code was fragile and would break if we ever
batched multiple volumes together.
Changes:
- Check result map FIRST, then conditionally check error
- If volume is found in result, use it (ignore errors about other volumes)
- If volume is NOT found and err != nil, include error context with %w
- Add defensive comments explaining the pattern for future maintainers
This makes the code:
1. Correct for future batched lookups
2. More informative (preserves underlying error details)
3. Consistent with filer_grpc_server.go which already handles this correctly
Example: If looking up ["1", "2", "999"] and only 999 fails, callers
looking for volumes 1 or 2 will succeed instead of failing unnecessarily.
* improve: address remaining code review findings
1. Lazy initialize FilerClient in mount for proxy-only setups
- Only create FilerClient when VolumeServerAccess != "filerProxy"
- Avoids wasted work when all reads proxy through filer
- filerClient is nil for proxy mode, initialized for direct access
2. Fix inaccurate deprecation comment in filer.LookupFn
- Updated comment to reflect current behavior (10k bounded cache)
- Removed claim of "unbounded growth" after adding size limit
- Still directs new code to wdclient.FilerClient for better features
3. Audit all MasterClient usages for KeepConnectedToMaster
- Verified all production callers start KeepConnectedToMaster early
- Filer, Shell, Master, Broker, Benchmark, Admin all correct
- IAM creates MasterClient but never uses it (harmless)
- Test code doesn't need KeepConnectedToMaster (mocks)
All callers properly follow the initialization pattern documented in
GetMaster(), preventing unexpected blocking or timeouts.
* fix: restore observability instrumentation in MasterClient
During the refactoring, several important stats counters and logging
statements were accidentally removed from tryConnectToMaster. These are
critical for monitoring and debugging the health of master client connections.
Restored instrumentation:
1. stats.MasterClientConnectCounter("total") - tracks all connection attempts
2. stats.MasterClientConnectCounter(FailedToKeepConnected) - when KeepConnected stream fails
3. stats.MasterClientConnectCounter(FailedToReceive) - when Recv() fails in loop
4. stats.MasterClientConnectCounter(Failed) - when overall gprcErr occurs
5. stats.MasterClientConnectCounter(OnPeerUpdate) - when peer updates detected
Additionally restored peer update logging:
- "+ filer@host noticed group.type address" for node additions
- "- filer@host noticed group.type address" for node removals
- Only logs updates matching the client's FilerGroup for noise reduction
This information is valuable for:
- Monitoring cluster health and connection stability
- Debugging cluster membership changes
- Tracking master failover and reconnection patterns
- Identifying network issues between clients and masters
No functional changes - purely observability restoration.
* improve: implement gRPC-aware retry for FilerClient volume lookups
The previous implementation used util.Retry which only retries errors
containing the string "transport". This is insufficient for handling
the full range of transient gRPC errors.
Changes:
1. Added isRetryableGrpcError() to properly inspect gRPC status codes
- Retries: Unavailable, DeadlineExceeded, ResourceExhausted, Aborted
- Falls back to string matching for non-gRPC network errors
2. Replaced util.Retry with custom retry loop
- 3 attempts with exponential backoff (1s, 1.5s, 2.25s)
- Tries all N filers on each attempt (N*3 total attempts max)
- Fast-fails on non-retryable errors (NotFound, PermissionDenied, etc.)
3. Improved logging
- Shows both filer attempt (x/N) and retry attempt (y/3)
- Logs retry reason and wait time for debugging
Benefits:
- Better handling of transient gRPC failures (server restarts, load spikes)
- Faster failure for permanent errors (no wasted retries)
- More informative logs for troubleshooting
- Maintains existing HA failover across multiple filers
Example: If all 3 filers return Unavailable (server overload):
- Attempt 1: try all 3 filers, wait 1s
- Attempt 2: try all 3 filers, wait 1.5s
- Attempt 3: try all 3 filers, fail
Example: If filer returns NotFound (volume doesn't exist):
- Attempt 1: try all 3 filers, fast-fail (no retry)
* fmt
* improve: add circuit breaker to skip known-unhealthy filers
The previous implementation tried all filers on every failure, including
known-unhealthy ones. This wasted time retrying permanently down filers.
Problem scenario (3 filers, filer0 is down):
- Last successful: filer1 (saved as filerIndex=1)
- Next lookup when filer1 fails:
Retry 1: filer1(fail) → filer2(fail) → filer0(fail, wastes 5s timeout)
Retry 2: filer1(fail) → filer2(fail) → filer0(fail, wastes 5s timeout)
Retry 3: filer1(fail) → filer2(fail) → filer0(fail, wastes 5s timeout)
Total wasted: 15 seconds on known-bad filer!
Solution: Circuit breaker pattern
- Track consecutive failures per filer (atomic int32)
- Skip filers with 3+ consecutive failures
- Re-check unhealthy filers every 30 seconds
- Reset failure count on success
New behavior:
- filer0 fails 3 times → marked unhealthy
- Future lookups skip filer0 for 30 seconds
- After 30s, re-check filer0 (allows recovery)
- If filer0 succeeds, reset failure count to 0
Benefits:
1. Avoids wasting time on known-down filers
2. Still sticks to last healthy filer (via filerIndex)
3. Allows recovery (30s re-check window)
4. No configuration needed (automatic)
Implementation details:
- filerHealth struct tracks failureCount (atomic) + lastFailureTime
- shouldSkipUnhealthyFiler(): checks if we should skip this filer
- recordFilerSuccess(): resets failure count to 0
- recordFilerFailure(): increments count, updates timestamp
- Logs when skipping unhealthy filers (V(2) level)
Example with circuit breaker:
- filer0 down, saved filerIndex=1 (filer1 healthy)
- Lookup 1: filer1(ok) → Done (0.01s)
- Lookup 2: filer1(fail) → filer2(ok) → Done, save filerIndex=2 (0.01s)
- Lookup 3: filer2(fail) → skip filer0 (unhealthy) → filer1(ok) → Done (0.01s)
Much better than wasting 15s trying filer0 repeatedly!
* fix: OnPeerUpdate should only process updates for matching FilerGroup
Critical bug: The OnPeerUpdate callback was incorrectly moved outside the
FilerGroup check when restoring observability instrumentation. This caused
clients to process peer updates for ALL filer groups, not just their own.
Problem:
Before: mc.OnPeerUpdate only called for update.FilerGroup == mc.FilerGroup
Bug: mc.OnPeerUpdate called for ALL updates regardless of FilerGroup
Impact:
- Multi-tenant deployments with separate filer groups would see cross-group
updates (e.g., group A clients processing group B updates)
- Could cause incorrect cluster membership tracking
- OnPeerUpdate handlers (like Filer's DLM ring updates) would receive
irrelevant updates from other groups
Example scenario:
Cluster has two filer groups: "production" and "staging"
Production filer connects with FilerGroup="production"
Incorrect behavior (bug):
- Receives "staging" group updates
- Incorrectly adds staging filers to production DLM ring
- Cross-tenant data access issues
Correct behavior (fixed):
- Only receives "production" group updates
- Only adds production filers to production DLM ring
- Proper isolation between groups
Fix:
Moved mc.OnPeerUpdate(update, time.Now()) back INSIDE the FilerGroup check
where it belongs, matching the original implementation.
The logging and stats counter were already correctly scoped to matching
FilerGroup, so they remain inside the if block as intended.
* improve: clarify Aborted error handling in volume lookups
Added documentation and logging to address the concern that codes.Aborted
might not always be retryable in all contexts.
Context-specific justification for treating Aborted as retryable:
Volume location lookups (LookupVolume RPC) are simple, read-only operations:
- No transactions
- No write conflicts
- No application-level state changes
- Idempotent (safe to retry)
In this context, Aborted is most likely caused by:
- Filer restarting/recovering (transient)
- Connection interrupted mid-request (transient)
- Server-side resource cleanup (transient)
NOT caused by:
- Application-level conflicts (no writes)
- Transaction failures (no transactions)
- Logical errors (read-only lookup)
Changes:
1. Added detailed comment explaining the context-specific reasoning
2. Added V(1) logging when treating Aborted as retryable
- Helps detect misclassification if it occurs
- Visible in verbose logs for troubleshooting
3. Split switch statement for clarity (one case per line)
If future analysis shows Aborted should not be retried, operators will
now have visibility via logs to make that determination. The logging
provides evidence for future tuning decisions.
Alternative approaches considered but not implemented:
- Removing Aborted entirely (too conservative for read-only ops)
- Message content inspection (adds complexity, no known patterns yet)
- Different handling per RPC type (premature optimization)
* fix: IAM server must start KeepConnectedToMaster for masterClient usage
The IAM server creates and uses a MasterClient but never started
KeepConnectedToMaster, which could cause blocking if IAM config files
have chunks requiring volume lookups.
Problem flow:
NewIamApiServerWithStore()
→ creates masterClient
→ ❌ NEVER starts KeepConnectedToMaster
GetS3ApiConfigurationFromFiler()
→ filer.ReadEntry(iama.masterClient, ...)
→ StreamContent(masterClient, ...) if file has chunks
→ masterClient.GetLookupFileIdFunction()
→ GetMaster(ctx) ← BLOCKS indefinitely waiting for connection!
While IAM config files (identity & policies) are typically small and
stored inline without chunks, the code path exists and would block
if the files ever had chunks.
Fix:
Start KeepConnectedToMaster in background goroutine right after
creating masterClient, following the documented pattern:
mc := wdclient.NewMasterClient(...)
go mc.KeepConnectedToMaster(ctx)
This ensures masterClient is usable if ReadEntry ever needs to
stream chunked content from volume servers.
Note: This bug was dormant because IAM config files are small (<256 bytes)
and SeaweedFS stores small files inline in Entry.Content, not as chunks.
The bug would only manifest if:
- IAM config grew > 256 bytes (inline threshold)
- Config was stored as chunks on volume servers
- ReadEntry called StreamContent
- GetMaster blocked indefinitely
Now all 9 production MasterClient instances correctly follow the pattern.
* fix: data race on filerHealth.lastFailureTime in circuit breaker
The circuit breaker tracked lastFailureTime as time.Time, which was
written in recordFilerFailure and read in shouldSkipUnhealthyFiler
without synchronization, causing a data race.
Data race scenario:
Goroutine 1: recordFilerFailure(0)
health.lastFailureTime = time.Now() // ❌ unsynchronized write
Goroutine 2: shouldSkipUnhealthyFiler(0)
time.Since(health.lastFailureTime) // ❌ unsynchronized read
→ RACE DETECTED by -race detector
Fix:
Changed lastFailureTime from time.Time to int64 (lastFailureTimeNs)
storing Unix nanoseconds for atomic access:
Write side (recordFilerFailure):
atomic.StoreInt64(&health.lastFailureTimeNs, time.Now().UnixNano())
Read side (shouldSkipUnhealthyFiler):
lastFailureNs := atomic.LoadInt64(&health.lastFailureTimeNs)
if lastFailureNs == 0 { return false } // Never failed
lastFailureTime := time.Unix(0, lastFailureNs)
time.Since(lastFailureTime) > 30*time.Second
Benefits:
- Atomic reads/writes (no data race)
- Efficient (int64 is 8 bytes, always atomic on 64-bit systems)
- Zero value (0) naturally means "never failed"
- No mutex needed (lock-free circuit breaker)
Note: sync/atomic was already imported for failureCount, so no new
import needed.
* fix: create fresh timeout context for each filer retry attempt
The timeout context was created once at function start and reused across
all retry attempts, causing subsequent retries to run with progressively
shorter (or expired) deadlines.
Problem flow:
Line 244: timeoutCtx, cancel := context.WithTimeout(ctx, 5s)
defer cancel()
Retry 1, filer 0: client.LookupVolume(timeoutCtx, ...) ← 5s available ✅
Retry 1, filer 1: client.LookupVolume(timeoutCtx, ...) ← 3s left
Retry 1, filer 2: client.LookupVolume(timeoutCtx, ...) ← 0.5s left
Retry 2, filer 0: client.LookupVolume(timeoutCtx, ...) ← EXPIRED! ❌
Result: Retries always fail with DeadlineExceeded, defeating the purpose
of retries.
Fix:
Moved context.WithTimeout inside the per-filer loop, creating a fresh
timeout context for each attempt:
for x := 0; x < n; x++ {
timeoutCtx, cancel := context.WithTimeout(ctx, fc.grpcTimeout)
err := pb.WithGrpcFilerClient(..., func(client) {
resp, err := client.LookupVolume(timeoutCtx, ...)
...
})
cancel() // Clean up immediately after call
}
Benefits:
- Each filer attempt gets full fc.grpcTimeout (default 5s)
- Retries actually have time to complete
- No context leaks (cancel called after each attempt)
- More predictable timeout behavior
Example with fix:
Retry 1, filer 0: fresh 5s timeout ✅
Retry 1, filer 1: fresh 5s timeout ✅
Retry 2, filer 0: fresh 5s timeout ✅
Total max time: 3 retries × 3 filers × 5s = 45s (plus backoff)
Note: The outer ctx (from caller) still provides overall cancellation if
the caller cancels or times out the entire operation.
* fix: always reset vidMap cache on master reconnection
The previous refactoring removed the else block that resets vidMap when
the first message from a newly connected master is not a VolumeLocation.
Problem scenario:
1. Client connects to master-1 and builds vidMap cache
2. Master-1 fails, client connects to master-2
3. First message from master-2 is a ClusterNodeUpdate (not VolumeLocation)
4. Old code: vidMap is reset and updated ✅
5. New code: vidMap is NOT reset ❌
6. Result: Client uses stale cache from master-1 → data access errors
Example flow with bug:
Connect to master-2
First message: ClusterNodeUpdate {filer.x added}
→ No resetVidMap() call
→ vidMap still has master-1's stale volume locations
→ Client reads from wrong volume servers → 404 errors
Fix:
Restored the else block that resets vidMap when first message is not
a VolumeLocation:
if resp.VolumeLocation != nil {
// ... check leader, reset, and update ...
} else {
// First message is ClusterNodeUpdate or other type
// Must still reset to avoid stale data
mc.resetVidMap()
}
This ensures the cache is always cleared when establishing a new master
connection, regardless of what the first message type is.
Root cause:
During the vidMapClient refactoring, this else block was accidentally
dropped, making failover behavior fragile and non-deterministic (depends
on which message type arrives first from the new master).
Impact:
- High severity for master failover scenarios
- Could cause read failures, 404s, or wrong data access
- Only manifests when first message is not VolumeLocation
* fix: goroutine and connection leak in IAM server shutdown
The IAM server's KeepConnectedToMaster goroutine used context.Background(),
which is non-cancellable, causing the goroutine and its gRPC connections
to leak on server shutdown.
Problem:
go masterClient.KeepConnectedToMaster(context.Background())
- context.Background() never cancels
- KeepConnectedToMaster goroutine runs forever
- gRPC connection to master stays open
- No way to stop cleanly on server shutdown
Result: Resource leaks when IAM server is stopped
Fix:
1. Added shutdownContext and shutdownCancel to IamApiServer struct
2. Created cancellable context in NewIamApiServerWithStore:
shutdownCtx, shutdownCancel := context.WithCancel(context.Background())
3. Pass shutdownCtx to KeepConnectedToMaster:
go masterClient.KeepConnectedToMaster(shutdownCtx)
4. Added Shutdown() method to invoke cancel:
func (iama *IamApiServer) Shutdown() {
if iama.shutdownCancel != nil {
iama.shutdownCancel()
}
}
5. Stored masterClient reference on IamApiServer for future use
Benefits:
- Goroutine stops cleanly when Shutdown() is called
- gRPC connections are closed properly
- No resource leaks on server restart/stop
- Shutdown() is idempotent (safe to call multiple times)
Usage (for future graceful shutdown):
iamServer, _ := iamapi.NewIamApiServer(...)
defer iamServer.Shutdown()
// or in signal handler:
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)
go func() {
<-sigChan
iamServer.Shutdown()
os.Exit(0)
}()
Note: Current command implementations (weed/command/iam.go) don't have
shutdown paths yet, but this makes IAM server ready for proper lifecycle
management when that infrastructure is added.
* refactor: remove unnecessary KeepMasterClientConnected wrapper in filer
The Filer.KeepMasterClientConnected() method was an unnecessary wrapper that
just forwarded to MasterClient.KeepConnectedToMaster(). This wrapper added
no value and created inconsistency with other components that call
KeepConnectedToMaster directly.
Removed:
filer.go:178-180
func (fs *Filer) KeepMasterClientConnected(ctx context.Context) {
fs.MasterClient.KeepConnectedToMaster(ctx)
}
Updated caller:
filer_server.go:181
- go fs.filer.KeepMasterClientConnected(context.Background())
+ go fs.filer.MasterClient.KeepConnectedToMaster(context.Background())
Benefits:
- Consistent with other components (S3, IAM, Shell, Mount)
- Removes unnecessary indirection
- Clearer that KeepConnectedToMaster runs in background goroutine
- Follows the documented pattern from MasterClient.GetMaster()
Note: shell/commands.go was verified and already correctly starts
KeepConnectedToMaster in a background goroutine (shell_liner.go:51):
go commandEnv.MasterClient.KeepConnectedToMaster(ctx)
* fix: use client ID instead of timeout for gRPC signature parameter
The pb.WithGrpcFilerClient signature parameter is meant to be a client
identifier for logging and tracking (added as 'sw-client-id' gRPC metadata
in streaming mode), not a timeout value.
Problem:
timeoutMs := int32(fc.grpcTimeout.Milliseconds()) // 5000 (5 seconds)
err := pb.WithGrpcFilerClient(false, timeoutMs, filerAddress, ...)
- Passing timeout (5000ms) as signature/client ID
- Misuse of API: signature should be a unique client identifier
- Timeout is already handled by timeoutCtx passed to gRPC call
- Inconsistent with other callers (all use 0 or proper client ID)
How WithGrpcFilerClient uses signature parameter:
func WithGrpcClient(..., signature int32, ...) {
if streamingMode && signature != 0 {
md := metadata.New(map[string]string{"sw-client-id": fmt.Sprintf("%d", signature)})
ctx = metadata.NewOutgoingContext(ctx, md)
}
...
}
It's for client identification, not timeout control!
Fix:
1. Added clientId int32 field to FilerClient struct
2. Initialize with rand.Int31() in NewFilerClient for unique ID
3. Removed timeoutMs variable (and misleading comment)
4. Use fc.clientId in pb.WithGrpcFilerClient call
Before:
err := pb.WithGrpcFilerClient(false, timeoutMs, ...)
^^^^^^^^^ Wrong! (5000)
After:
err := pb.WithGrpcFilerClient(false, fc.clientId, ...)
^^^^^^^^^^^^ Correct! (random int31)
Benefits:
- Correct API usage (signature = client ID, not timeout)
- Timeout still works via timeoutCtx (unchanged)
- Consistent with other pb.WithGrpcFilerClient callers
- Enables proper client tracking on filer side via gRPC metadata
- Each FilerClient instance has unique ID for debugging
Examples of correct usage elsewhere:
weed/iamapi/iamapi_server.go:145 pb.WithGrpcFilerClient(false, 0, ...)
weed/command/s3.go:215 pb.WithGrpcFilerClient(false, 0, ...)
weed/shell/commands.go:110 pb.WithGrpcFilerClient(streamingMode, 0, ...)
All use 0 (or a proper signature), not a timeout value.
* fix: add timeout to master volume lookup to prevent indefinite blocking
The masterVolumeProvider.LookupVolumeIds method was using the context
directly without a timeout, which could cause it to block indefinitely
if the master is slow to respond or unreachable.
Problem:
err := pb.WithMasterClient(false, p.masterClient.GetMaster(ctx), ...)
resp, err := client.LookupVolume(ctx, &master_pb.LookupVolumeRequest{...})
- No timeout on gRPC call to master
- Could block indefinitely if master is unresponsive
- Inconsistent with FilerClient which uses 5s timeout
- This is a fallback path (cache miss) but still needs protection
Scenarios where this could hang:
1. Master server under heavy load (slow response)
2. Network issues between client and master
3. Master server hung or deadlocked
4. Master in process of shutting down
Fix:
timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
err := pb.WithMasterClient(false, p.masterClient.GetMaster(timeoutCtx), ...)
resp, err := client.LookupVolume(timeoutCtx, &master_pb.LookupVolumeRequest{...})
Benefits:
- Prevents indefinite blocking on master lookup
- Consistent with FilerClient timeout pattern (5 seconds)
- Faster failure detection when master is unresponsive
- Caller's context still honored (timeout is in addition, not replacement)
- Improves overall system resilience
Note: 5 seconds is a reasonable default for volume lookups:
- Long enough for normal master response (~10-50ms)
- Short enough to fail fast on issues
- Matches FilerClient's grpcTimeout default
* purge
* refactor: address code review feedback on comments and style
Fixed several code quality issues identified during review:
1. Corrected backoff algorithm description in filer_client.go:
- Changed "Exponential backoff" to "Multiplicative backoff with 1.5x factor"
- The formula waitTime * 3/2 produces 1s, 1.5s, 2.25s, not exponential 2^n
- More accurate terminology prevents confusion
2. Removed redundant nil check in vidmap_client.go:
- After the for loop, node is guaranteed to be non-nil
- Loop either returns early or assigns non-nil value to node
- Simplified: if node != nil { node.cache.Store(nil) } → node.cache.Store(nil)
3. Added startup logging to IAM server for consistency:
- Log when master client connection starts
- Matches pattern in S3ApiServer (line 100 in s3api_server.go)
- Improves operational visibility during startup
- Added missing glog import
4. Fixed indentation in filer/reader_at.go:
- Lines 76-91 had incorrect indentation (extra tab level)
- Line 93 also misaligned
- Now properly aligned with surrounding code
5. Updated deprecation comment to follow Go convention:
- Changed "DEPRECATED:" to "Deprecated:" (standard Go format)
- Tools like staticcheck and IDEs recognize the standard format
- Enables automated deprecation warnings in tooling
- Better developer experience
All changes are cosmetic and do not affect functionality.
* fmt
* refactor: make circuit breaker parameters configurable in FilerClient
The circuit breaker failure threshold (3) and reset timeout (30s) were
hardcoded, making it difficult to tune the client's behavior in different
deployment environments without modifying the code.
Problem:
func shouldSkipUnhealthyFiler(index int32) bool {
if failureCount < 3 { // Hardcoded threshold
return false
}
if time.Since(lastFailureTime) > 30*time.Second { // Hardcoded timeout
return false
}
}
Different environments have different needs:
- High-traffic production: may want lower threshold (2) for faster failover
- Development/testing: may want higher threshold (5) to tolerate flaky networks
- Low-latency services: may want shorter reset timeout (10s)
- Batch processing: may want longer reset timeout (60s)
Solution:
1. Added fields to FilerClientOption:
- FailureThreshold int32 (default: 3)
- ResetTimeout time.Duration (default: 30s)
2. Added fields to FilerClient:
- failureThreshold int32
- resetTimeout time.Duration
3. Applied defaults in NewFilerClient with option override:
failureThreshold := int32(3)
resetTimeout := 30 * time.Second
if opt.FailureThreshold > 0 {
failureThreshold = opt.FailureThreshold
}
if opt.ResetTimeout > 0 {
resetTimeout = opt.ResetTimeout
}
4. Updated shouldSkipUnhealthyFiler to use configurable values:
if failureCount < fc.failureThreshold { ... }
if time.Since(lastFailureTime) > fc.resetTimeout { ... }
Benefits:
✓ Tunable for different deployment environments
✓ Backward compatible (defaults match previous hardcoded values)
✓ No breaking changes to existing code
✓ Better maintainability and flexibility
Example usage:
// Aggressive failover for low-latency production
fc := wdclient.NewFilerClient(filers, dialOpt, dc, &wdclient.FilerClientOption{
FailureThreshold: 2,
ResetTimeout: 10 * time.Second,
})
// Tolerant of flaky networks in development
fc := wdclient.NewFilerClient(filers, dialOpt, dc, &wdclient.FilerClientOption{
FailureThreshold: 5,
ResetTimeout: 60 * time.Second,
})
* retry parameters
* refactor: make retry and timeout parameters configurable
Made retry logic and gRPC timeouts configurable across FilerClient and
MasterClient to support different deployment environments and network
conditions.
Problem 1: Hardcoded retry parameters in FilerClient
waitTime := time.Second // Fixed at 1s
maxRetries := 3 // Fixed at 3 attempts
waitTime = waitTime * 3 / 2 // Fixed 1.5x multiplier
Different environments have different needs:
- Unstable networks: may want more retries (5) with longer waits (2s)
- Low-latency production: may want fewer retries (2) with shorter waits (500ms)
- Batch processing: may want exponential backoff (2x) instead of 1.5x
Problem 2: Hardcoded gRPC timeout in MasterClient
timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
Master lookups may need different timeouts:
- High-latency cross-region: may need 10s timeout
- Local network: may use 2s timeout for faster failure detection
Solution for FilerClient:
1. Added fields to FilerClientOption:
- MaxRetries int (default: 3)
- InitialRetryWait time.Duration (default: 1s)
- RetryBackoffFactor float64 (default: 1.5)
2. Added fields to FilerClient:
- maxRetries int
- initialRetryWait time.Duration
- retryBackoffFactor float64
3. Updated LookupVolumeIds to use configurable values:
waitTime := fc.initialRetryWait
maxRetries := fc.maxRetries
for retry := 0; retry < maxRetries; retry++ {
...
waitTime = time.Duration(float64(waitTime) * fc.retryBackoffFactor)
}
Solution for MasterClient:
1. Added grpcTimeout field to MasterClient (default: 5s)
2. Initialize in NewMasterClient with 5 * time.Second default
3. Updated masterVolumeProvider to use p.masterClient.grpcTimeout
Benefits:
✓ Tunable for different network conditions and deployment scenarios
✓ Backward compatible (defaults match previous hardcoded values)
✓ No breaking changes to existing code
✓ Consistent configuration pattern across FilerClient and MasterClient
Example usage:
// Fast-fail for low-latency production with stable network
fc := wdclient.NewFilerClient(filers, dialOpt, dc, &wdclient.FilerClientOption{
MaxRetries: 2,
InitialRetryWait: 500 * time.Millisecond,
RetryBackoffFactor: 2.0, // Exponential backoff
GrpcTimeout: 2 * time.Second,
})
// Patient retries for unstable network or batch processing
fc := wdclient.NewFilerClient(filers, dialOpt, dc, &wdclient.FilerClientOption{
MaxRetries: 5,
InitialRetryWait: 2 * time.Second,
RetryBackoffFactor: 1.5,
GrpcTimeout: 10 * time.Second,
})
Note: MasterClient timeout is currently set at construction time and not
user-configurable via NewMasterClient parameters. Future enhancement could
add a MasterClientOption struct similar to FilerClientOption.
* fix: rename vicCacheLock to vidCacheLock for consistency
Fixed typo in variable name for better code consistency and readability.
Problem:
vidCache := make(map[string]*filer_pb.Locations)
var vicCacheLock sync.RWMutex // Typo: vic instead of vid
vicCacheLock.RLock()
locations, found := vidCache[vid]
vicCacheLock.RUnlock()
The variable name 'vicCacheLock' is inconsistent with 'vidCache'.
Both should use 'vid' prefix (volume ID) not 'vic'.
Fix:
Renamed all 5 occurrences:
- var vicCacheLock → var vidCacheLock (line 56)
- vicCacheLock.RLock() → vidCacheLock.RLock() (line 62)
- vicCacheLock.RUnlock() → vidCacheLock.RUnlock() (line 64)
- vicCacheLock.Lock() → vidCacheLock.Lock() (line 81)
- vicCacheLock.Unlock() → vidCacheLock.Unlock() (line 91)
Benefits:
✓ Consistent variable naming convention
✓ Clearer intent (volume ID cache lock)
✓ Better code readability
✓ Easier code navigation
* fix: use defer cancel() with anonymous function for proper context cleanup
Fixed context cancellation to use defer pattern correctly in loop iteration.
Problem:
for x := 0; x < n; x++ {
timeoutCtx, cancel := context.WithTimeout(ctx, fc.grpcTimeout)
err := pb.WithGrpcFilerClient(...)
cancel() // Only called on normal return, not on panic
}
Issues with original approach:
1. If pb.WithGrpcFilerClient panics, cancel() is never called → context leak
2. If callback returns early (though unlikely here), cleanup might be missed
3. Not following Go best practices for context.WithTimeout usage
Problem with naive defer in loop:
for x := 0; x < n; x++ {
timeoutCtx, cancel := context.WithTimeout(ctx, fc.grpcTimeout)
defer cancel() // ❌ WRONG: All defers accumulate until function returns
}
In Go, defer executes when the surrounding *function* returns, not when
the loop iteration ends. This would accumulate n deferred cancel() calls
and leak contexts until LookupVolumeIds returns.
Solution: Wrap in anonymous function
for x := 0; x < n; x++ {
err := func() error {
timeoutCtx, cancel := context.WithTimeout(ctx, fc.grpcTimeout)
defer cancel() // ✅ Executes when anonymous function returns (per iteration)
return pb.WithGrpcFilerClient(...)
}()
}
Benefits:
✓ Context always cancelled, even on panic
✓ defer executes after each iteration (not accumulated)
✓ Follows Go best practices for context.WithTimeout
✓ No resource leaks during retry loop execution
✓ Cleaner error handling
Reference:
Go documentation for context.WithTimeout explicitly shows:
ctx, cancel := context.WithTimeout(...)
defer cancel()
This is the idiomatic pattern that should always be followed.
* Can't use defer directly in loop
* improve: add data center preference and URL shuffling for consistent performance
Added missing data center preference and load distribution (URL shuffling)
to ensure consistent performance and behavior across all code paths.
Problem 1: PreferPublicUrl path missing DC preference and shuffling
Location: weed/wdclient/filer_client.go lines 184-192
The custom PreferPublicUrl implementation was simply iterating through
locations and building URLs without considering:
1. Data center proximity (latency optimization)
2. Load distribution across volume servers
Before:
for _, loc := range locations {
url := loc.PublicUrl
if url == "" { url = loc.Url }
fullUrls = append(fullUrls, "http://"+url+"/"+fileId)
}
return fullUrls, nil
After:
var sameDcUrls, otherDcUrls []string
dataCenter := fc.GetDataCenter()
for _, loc := range locations {
url := loc.PublicUrl
if url == "" { url = loc.Url }
httpUrl := "http://" + url + "/" + fileId
if dataCenter != "" && dataCenter == loc.DataCenter {
sameDcUrls = append(sameDcUrls, httpUrl)
} else {
otherDcUrls = append(otherDcUrls, httpUrl)
}
}
rand.Shuffle(len(sameDcUrls), ...)
rand.Shuffle(len(otherDcUrls), ...)
fullUrls = append(sameDcUrls, otherDcUrls...)
Problem 2: Cache miss path missing URL shuffling
Location: weed/wdclient/vidmap_client.go lines 95-108
The cache miss path (fallback lookup) was missing URL shuffling, while
the cache hit path (vm.LookupFileId) already shuffles URLs. This
inconsistency meant:
- Cache hit: URLs shuffled → load distributed
- Cache miss: URLs not shuffled → first server always hit
Before:
var sameDcUrls, otherDcUrls []string
// ... build URLs ...
fullUrls = append(sameDcUrls, otherDcUrls...)
return fullUrls, nil
After:
var sameDcUrls, otherDcUrls []string
// ... build URLs ...
rand.Shuffle(len(sameDcUrls), ...)
rand.Shuffle(len(otherDcUrls), ...)
fullUrls = append(sameDcUrls, otherDcUrls...)
return fullUrls, nil
Benefits:
✓ Reduced latency by preferring same-DC volume servers
✓ Even load distribution across all volume servers
✓ Consistent behavior between cache hit/miss paths
✓ Consistent behavior between PreferUrl and PreferPublicUrl
✓ Matches behavior of existing vidMap.LookupFileId implementation
Impact on performance:
- Lower read latency (same-DC preference)
- Better volume server utilization (load spreading)
- No single volume server becomes a hotspot
Note: Added math/rand import to vidmap_client.go for shuffle support.
* Update weed/wdclient/masterclient.go
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
* improve: call IAM server Shutdown() for best-effort cleanup
Added call to iamApiServer.Shutdown() to ensure cleanup happens when possible,
and documented the limitations of the current approach.
Problem:
The Shutdown() method was defined in IamApiServer but never called anywhere,
meaning the KeepConnectedToMaster goroutine would continue running even when
the IAM server stopped, causing resource leaks.
Changes:
1. Store iamApiServer instance in weed/command/iam.go
- Changed: _, iamApiServer_err := iamapi.NewIamApiServer(...)
- To: iamApiServer, iamApiServer_err := iamapi.NewIamApiServer(...)
2. Added defer call for best-effort cleanup
- defer iamApiServer.Shutdown()
- This will execute if startIamServer() returns normally
3. Added logging in Shutdown() method
- Log when shutdown is triggered for visibility
4. Documented limitations and future improvements
- Added note that defer only works for normal function returns
- SeaweedFS commands don't currently have signal handling
- Suggested future enhancement: add SIGTERM/SIGINT handling
Current behavior:
- ✓ Cleanup happens if HTTP server fails to start (glog.Fatalf path)
- ✓ Cleanup happens if Serve() returns with error (unlikely)
- ✗ Cleanup does NOT happen on SIGTERM/SIGINT (process killed)
The last case is a limitation of the current command architecture - all
SeaweedFS commands (s3, filer, volume, master, iam) lack signal handling
for graceful shutdown. This is a systemic issue that affects all services.
Future enhancement:
To properly handle SIGTERM/SIGINT, the command layer would need:
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)
go func() {
httpServer.Serve(listener) // Non-blocking
}()
<-sigChan
glog.V(0).Infof("Received shutdown signal")
iamApiServer.Shutdown()
httpServer.Shutdown(context.Background())
This would require refactoring the command structure for all services,
which is out of scope for this change.
Benefits of current approach:
✓ Best-effort cleanup (better than nothing)
✓ Proper cleanup in error paths
✓ Documented for future improvement
✓ Consistent with how other SeaweedFS services handle lifecycle
* data racing in test
---------
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
|
||
|
dependabot[bot]andchrislu
|
7dce429e61 |
chore(deps): bump golang.org/x/crypto from 0.43.0 to 0.45.0 (#7511)
* chore(deps): bump golang.org/x/crypto from 0.43.0 to 0.45.0 Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.43.0 to 0.45.0. - [Commits](https://github.com/golang/crypto/compare/v0.43.0...v0.45.0) --- updated-dependencies: - dependency-name: golang.org/x/crypto dependency-version: 0.45.0 dependency-type: direct:production ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
dependabot[bot]
|
d2daea1552 |
chore(deps): bump golang.org/x/crypto from 0.43.0 to 0.45.0 in /test/kafka/kafka-client-loadtest (#7510)
chore(deps): bump golang.org/x/crypto Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.43.0 to 0.45.0. - [Commits](https://github.com/golang/crypto/compare/v0.43.0...v0.45.0) --- updated-dependencies: - dependency-name: golang.org/x/crypto dependency-version: 0.45.0 dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> |
||
|
Chris Lu
|
c6b6ea40e6 |
filer store: add foundationdb (#7178)
* add foundationdb * Update foundationdb_store.go * fix * apply the patch * avoid panic on error * address comments * remove extra data * address comments * adds more debug messages * fix range listing * delete with prefix range; list with right start key * fix docker files * use the more idiomatic FoundationDB KeySelectors * address comments * proper errors * fix API versions * more efficient * recursive deletion * clean up * clean up * pagination, one transaction for deletion * error checking * Use fdb.Strinc() to compute the lexicographically next string and create a proper range * fix docker * Update README.md * delete in batches * delete in batches * fix build * add foundationdb build * Updated FoundationDB Version * Fixed glibc/musl Incompatibility (Alpine → Debian) * Update container_foundationdb_version.yml * build SeaweedFS * build tag * address comments * separate transaction * address comments * fix build * empty vs no data * fixes * add go test * Install FoundationDB client libraries * nil compare |
||
|
Chris Lu
|
8be9e258fc |
S3: Add tests for PyArrow with native S3 filesystem (#7508)
* PyArrow native S3 filesystem * add sse-s3 tests * update * minor * ENABLE_SSE_S3 * Update test_pyarrow_native_s3.py * clean up * refactoring * Update test_pyarrow_native_s3.py |
||
|
Chris Luandgemini-code-assist[bot]
|
ca84a8a713 |
S3: Directly read write volume servers (#7481)
* Lazy Versioning Check, Conditional SSE Entry Fetch, HEAD Request Optimization * revert Reverted the conditional versioning check to always check versioning status Reverted the conditional SSE entry fetch to always fetch entry metadata Reverted the conditional versioning check to always check versioning status Reverted the conditional SSE entry fetch to always fetch entry metadata * Lazy Entry Fetch for SSE, Skip Conditional Header Check * SSE-KMS headers are present, this is not an SSE-C request (mutually exclusive) * SSE-C is mutually exclusive with SSE-S3 and SSE-KMS * refactor * Removed Premature Mutual Exclusivity Check * check for the presence of the X-Amz-Server-Side-Encryption header * not used * fmt * directly read write volume servers * HTTP Range Request Support * set header * md5 * copy object * fix sse * fmt * implement sse * sse continue * fixed the suffix range bug (bytes=-N for "last N bytes") * debug logs * Missing PartsCount Header * profiling * url encoding * test_multipart_get_part * headers * debug * adjust log level * handle part number * Update s3api_object_handlers.go * nil safety * set ModifiedTsNs * remove * nil check * fix sse header * same logic as filer * decode values * decode ivBase64 * s3: Fix SSE decryption JWT authentication and streaming errors Critical fix for SSE (Server-Side Encryption) test failures: 1. **JWT Authentication Bug** (Root Cause): - Changed from GenJwtForFilerServer to GenJwtForVolumeServer - S3 API now uses correct JWT when directly reading from volume servers - Matches filer's authentication pattern for direct volume access - Fixes 'unexpected EOF' and 500 errors in SSE tests 2. **Streaming Error Handling**: - Added error propagation in getEncryptedStreamFromVolumes goroutine - Use CloseWithError() to properly communicate stream failures - Added debug logging for streaming errors 3. **Response Header Timing**: - Removed premature WriteHeader(http.StatusOK) call - Let Go's http package write status automatically on first write - Prevents header lock when errors occur during streaming 4. **Enhanced SSE Decryption Debugging**: - Added IV/Key validation and logging for SSE-C, SSE-KMS, SSE-S3 - Better error messages for missing or invalid encryption metadata - Added glog.V(2) debugging for decryption setup This fixes SSE integration test failures where encrypted objects could not be retrieved due to volume server authentication failures. The JWT bug was causing volume servers to reject requests, resulting in truncated/empty streams (EOF) or internal errors. * s3: Fix SSE multipart upload metadata preservation Critical fix for SSE multipart upload test failures (SSE-C and SSE-KMS): **Root Cause - Incomplete SSE Metadata Copying**: The old code only tried to copy 'SeaweedFSSSEKMSKey' from the first part to the completed object. This had TWO bugs: 1. **Wrong Constant Name** (Key Mismatch Bug): - Storage uses: SeaweedFSSSEKMSKeyHeader = 'X-SeaweedFS-SSE-KMS-Key' - Old code read: SeaweedFSSSEKMSKey = 'x-seaweedfs-sse-kms-key' - Result: SSE-KMS metadata was NEVER copied → 500 errors 2. **Missing SSE-C and SSE-S3 Headers**: - SSE-C requires: IV, Algorithm, KeyMD5 - SSE-S3 requires: encrypted key data + standard headers - Old code: copied nothing for SSE-C/SSE-S3 → decryption failures **Fix - Complete SSE Header Preservation**: Now copies ALL SSE headers from first part to completed object: - SSE-C: SeaweedFSSSEIV, CustomerAlgorithm, CustomerKeyMD5 - SSE-KMS: SeaweedFSSSEKMSKeyHeader, AwsKmsKeyId, ServerSideEncryption - SSE-S3: SeaweedFSSSES3Key, ServerSideEncryption Applied consistently to all 3 code paths: 1. Versioned buckets (creates version file) 2. Suspended versioning (creates main object with null versionId) 3. Non-versioned buckets (creates main object) **Why This Is Correct**: The headers copied EXACTLY match what putToFiler stores during part upload (lines 496-521 in s3api_object_handlers_put.go). This ensures detectPrimarySSEType() can correctly identify encrypted multipart objects and trigger inline decryption with proper metadata. Fixes: TestSSEMultipartUploadIntegration (SSE-C and SSE-KMS subtests) * s3: Add debug logging for versioning state diagnosis Temporary debug logging to diagnose test_versioning_obj_plain_null_version_overwrite_suspended failure. Added glog.V(0) logging to show: 1. setBucketVersioningStatus: when versioning status is changed 2. PutObjectHandler: what versioning state is detected (Enabled/Suspended/none) 3. PutObjectHandler: which code path is taken (putVersionedObject vs putSuspendedVersioningObject) This will help identify if: - The versioning status is being set correctly in bucket config - The cache is returning stale/incorrect versioning state - The switch statement is correctly routing to suspended vs enabled handlers * s3: Enhanced versioning state tracing for suspended versioning diagnosis Added comprehensive logging across the entire versioning state flow: PutBucketVersioningHandler: - Log requested status (Enabled/Suspended) - Log when calling setBucketVersioningStatus - Log success/failure of status change setBucketVersioningStatus: - Log bucket and status being set - Log when config is updated - Log completion with error code updateBucketConfig: - Log versioning state being written to cache - Immediate cache verification after Set - Log if cache verification fails getVersioningState: - Log bucket name and state being returned - Log if object lock forces VersioningEnabled - Log errors This will reveal: 1. If PutBucketVersioning(Suspended) is reaching the handler 2. If the cache update succeeds 3. What state getVersioningState returns during PUT 4. Any cache consistency issues Expected to show why bucket still reports 'Enabled' after 'Suspended' call. * s3: Add SSE chunk detection debugging for multipart uploads Added comprehensive logging to diagnose why TestSSEMultipartUploadIntegration fails: detectPrimarySSEType now logs: 1. Total chunk count and extended header count 2. All extended headers with 'sse'/'SSE'/'encryption' in the name 3. For each chunk: index, SseType, and whether it has metadata 4. Final SSE type counts (SSE-C, SSE-KMS, SSE-S3) This will reveal if: - Chunks are missing SSE metadata after multipart completion - Extended headers are copied correctly from first part - The SSE detection logic is working correctly Expected to show if chunks have SseType=0 (none) or proper SSE types set. * s3: Trace SSE chunk metadata through multipart completion and retrieval Added end-to-end logging to track SSE chunk metadata lifecycle: **During Multipart Completion (filer_multipart.go)**: 1. Log finalParts chunks BEFORE mkFile - shows SseType and metadata 2. Log versionEntry.Chunks INSIDE mkFile callback - shows if mkFile preserves SSE info 3. Log success after mkFile completes **During GET Retrieval (s3api_object_handlers.go)**: 1. Log retrieved entry chunks - shows SseType and metadata after retrieval 2. Log detected SSE type result This will reveal at which point SSE chunk metadata is lost: - If finalParts have SSE metadata but versionEntry.Chunks don't → mkFile bug - If versionEntry.Chunks have SSE metadata but retrieved chunks don't → storage/retrieval bug - If chunks never have SSE metadata → multipart completion SSE processing bug Expected to show chunks with SseType=NONE during retrieval even though they were created with proper SseType during multipart completion. * s3: Fix SSE-C multipart IV base64 decoding bug **Critical Bug Found**: SSE-C multipart uploads were failing because: Root Cause: - entry.Extended[SeaweedFSSSEIV] stores base64-encoded IV (24 bytes for 16-byte IV) - SerializeSSECMetadata expects raw IV bytes (16 bytes) - During multipart completion, we were passing base64 IV directly → serialization error Error Message: "Failed to serialize SSE-C metadata for chunk in part X: invalid IV length: expected 16 bytes, got 24" Fix: - Base64-decode IV before passing to SerializeSSECMetadata - Added error handling for decode failures Impact: - SSE-C multipart uploads will now correctly serialize chunk metadata - Chunks will have proper SSE metadata for decryption during GET This fixes the SSE-C subtest of TestSSEMultipartUploadIntegration. SSE-KMS still has a separate issue (error code 23) being investigated. * fixes * kms sse * handle retry if not found in .versions folder and should read the normal object * quick check (no retries) to see if the .versions/ directory exists * skip retry if object is not found * explicit update to avoid sync delay * fix map update lock * Remove fmt.Printf debug statements * Fix SSE-KMS multipart base IV fallback to fail instead of regenerating * fmt * Fix ACL grants storage logic * header handling * nil handling * range read for sse content * test range requests for sse objects * fmt * unused code * upload in chunks * header case * fix url * bucket policy error vs bucket not found * jwt handling * fmt * jwt in request header * Optimize Case-Insensitive Prefix Check * dead code * Eliminated Unnecessary Stream Prefetch for Multipart SSE * range sse * sse * refactor * context * fmt * fix type * fix SSE-C IV Mismatch * Fix Headers Being Set After WriteHeader * fix url parsing * propergate sse headers * multipart sse-s3 * aws sig v4 authen * sse kms * set content range * better errors * Update s3api_object_handlers_copy.go * Update s3api_object_handlers.go * Update s3api_object_handlers.go * avoid magic number * clean up * Update s3api_bucket_policy_handlers.go * fix url parsing * context * data and metadata both use background context * adjust the offset * SSE Range Request IV Calculation * adjust logs * IV relative to offset in each part, not the whole file * collect logs * offset * fix offset * fix url * logs * variable * jwt * Multipart ETag semantics: conditionally set object-level Md5 for single-chunk uploads only. * sse * adjust IV and offset * multipart boundaries * ensures PUT and GET operations return consistent ETags * Metadata Header Case * CommonPrefixes Sorting with URL Encoding * always sort * remove the extra PathUnescape call * fix the multipart get part ETag * the FileChunk is created without setting ModifiedTsNs * Sort CommonPrefixes lexicographically to match AWS S3 behavior * set md5 for multipart uploads * prevents any potential data loss or corruption in the small-file inline storage path * compiles correctly * decryptedReader will now be properly closed after use * Fixed URL encoding and sort order for CommonPrefixes * Update s3api_object_handlers_list.go * SSE-x Chunk View Decryption * Different IV offset calculations for single-part vs multipart objects * still too verbose in logs * less logs * ensure correct conversion * fix listing * nil check * minor fixes * nil check * single character delimiter * optimize * range on empty object or zero-length * correct IV based on its position within that part, not its position in the entire object * adjust offset * offset Fetch FULL encrypted chunk (not just the range) Adjust IV by PartOffset/ChunkOffset only Decrypt full chunk Skip in the DECRYPTED stream to reach OffsetInChunk * look breaking * refactor * error on no content * handle intra-block byte skipping * Incomplete HTTP Response Error Handling * multipart SSE * Update s3api_object_handlers.go * address comments * less logs * handling directory * Optimized rejectDirectoryObjectWithoutSlash() to avoid unnecessary lookups * Revert "handling directory" This reverts commit 3a335f0ac33c63f51975abc63c40e5328857a74b. * constant * Consolidate nil entry checks in GetObjectHandler * add range tests * Consolidate redundant nil entry checks in HeadObjectHandler * adjust logs * SSE type * large files * large files Reverted the plain-object range test * ErrNoEncryptionConfig * Fixed SSERangeReader Infinite Loop Vulnerability * Fixed SSE-KMS Multipart ChunkReader HTTP Body Leak * handle empty directory in S3, added PyArrow tests * purge unused code * Update s3_parquet_test.py * Update requirements.txt * According to S3 specifications, when both partNumber and Range are present, the Range should apply within the selected part's boundaries, not to the full object. * handle errors * errors after writing header * https * fix: Wait for volume assignment readiness before running Parquet tests The test-implicit-dir-with-server test was failing with an Internal Error because volume assignment was not ready when tests started. This fix adds a check that attempts a volume assignment and waits for it to succeed before proceeding with tests. This ensures that: 1. Volume servers are registered with the master 2. Volume growth is triggered if needed 3. The system can successfully assign volumes for writes Fixes the timeout issue where boto3 would retry 4 times and fail with 'We encountered an internal error, please try again.' * sse tests * store derived IV * fix: Clean up gRPC ports between tests to prevent port conflicts The second test (test-implicit-dir-with-server) was failing because the volume server's gRPC port (18080 = VOLUME_PORT + 10000) was still in use from the first test. The cleanup code only killed HTTP port processes, not gRPC port processes. Added cleanup for gRPC ports in all stop targets: - Master gRPC: MASTER_PORT + 10000 (19333) - Volume gRPC: VOLUME_PORT + 10000 (18080) - Filer gRPC: FILER_PORT + 10000 (18888) This ensures clean state between test runs in CI. * add import * address comments * docs: Add placeholder documentation files for Parquet test suite Added three missing documentation files referenced in test/s3/parquet/README.md: 1. TEST_COVERAGE.md - Documents 43 total test cases (17 Go unit tests, 6 Python integration tests, 20 Python end-to-end tests) 2. FINAL_ROOT_CAUSE_ANALYSIS.md - Explains the s3fs compatibility issue with PyArrow, the implicit directory problem, and how the fix works 3. MINIO_DIRECTORY_HANDLING.md - Compares MinIO's directory handling approach with SeaweedFS's implementation Each file contains: - Title and overview - Key technical details relevant to the topic - TODO sections for future expansion These placeholder files resolve the broken README links and provide structure for future detailed documentation. * clean up if metadata operation failed * Update s3_parquet_test.py * clean up * Update Makefile * Update s3_parquet_test.py * Update Makefile * Handle ivSkip for non-block-aligned offsets * Update README.md * stop volume server faster * stop volume server in 1 second * different IV for each chunk in SSE-S3 and SSE-KMS * clean up if fails * testing upload * error propagation * fmt * simplify * fix copying * less logs * endian * Added marshaling error handling * handling invalid ranges * error handling for adding to log buffer * fix logging * avoid returning too quickly and ensure proper cleaning up * Activity Tracking for Disk Reads * Cleanup Unused Parameters * Activity Tracking for Kafka Publishers * Proper Test Error Reporting * refactoring * less logs * less logs * go fmt * guard it with if entry.Attributes.TtlSec > 0 to match the pattern used elsewhere. * Handle bucket-default encryption config errors explicitly for multipart * consistent activity tracking * obsolete code for s3 on filer read/write handlers * Update weed/s3api/s3api_object_handlers_list.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> |
||
|
chrislu
|
aef5121c36 | faster master startup | ||
|
dependabot[bot]andchrislu
|
cae62db31b |
chore(deps): bump golang.org/x/image from 0.32.0 to 0.33.0 (#7497)
* chore(deps): bump golang.org/x/image from 0.32.0 to 0.33.0 Bumps [golang.org/x/image](https://github.com/golang/image) from 0.32.0 to 0.33.0. - [Commits](https://github.com/golang/image/compare/v0.32.0...v0.33.0) --- updated-dependencies: - dependency-name: golang.org/x/image dependency-version: 0.33.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
Chris Lu
|
508d06d9a5 |
S3: Enforce bucket policy (#7471)
* evaluate policies during authorization * cache bucket policy * refactor * matching with regex special characters * Case Sensitivity, pattern cache, Dead Code Removal * Fixed Typo, Restored []string Case, Added Cache Size Limit * hook up with policy engine * remove old implementation * action mapping * validate * if not specified, fall through to IAM checks * fmt * Fail-close on policy evaluation errors * Explicit `Allow` bypasses IAM checks * fix error message * arn:seaweed => arn:aws * remove legacy support * fix tests * Clean up bucket policy after this test * fix for tests * address comments * security fixes * fix tests * temp comment out |
||
|
dependabot[bot]andchrislu
|
d8dd3e4c53 |
chore(deps): bump golang.org/x/sys from 0.37.0 to 0.38.0 (#7459)
* chore(deps): bump golang.org/x/sys from 0.37.0 to 0.38.0 Bumps [golang.org/x/sys](https://github.com/golang/sys) from 0.37.0 to 0.38.0. - [Commits](https://github.com/golang/sys/compare/v0.37.0...v0.38.0) --- updated-dependencies: - dependency-name: golang.org/x/sys dependency-version: 0.38.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
dependabot[bot]andchrislu
|
735516cf0a |
chore(deps): bump github.com/shirou/gopsutil/v4 from 4.25.9 to 4.25.10 (#7457)
* chore(deps): bump github.com/shirou/gopsutil/v4 from 4.25.9 to 4.25.10 Bumps [github.com/shirou/gopsutil/v4](https://github.com/shirou/gopsutil) from 4.25.9 to 4.25.10. - [Release notes](https://github.com/shirou/gopsutil/releases) - [Commits](https://github.com/shirou/gopsutil/compare/v4.25.9...v4.25.10) --- updated-dependencies: - dependency-name: github.com/shirou/gopsutil/v4 dependency-version: 4.25.10 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
Konstantin Lebedevandchrislu
|
084b377f87 |
do delete expired entries on s3 list request (#7426)
* do delete expired entries on s3 list request
https://github.com/seaweedfs/seaweedfs/issues/6837
* disable delete expires s3 entry in filer
* pass opt allowDeleteObjectsByTTL to all servers
* delete on get and head
* add lifecycle expiration s3 tests
* fix opt allowDeleteObjectsByTTL for server
* fix test lifecycle expiration
* fix IsExpired
* fix locationPrefix for updateEntriesTTL
* fix s3tests
* resolv coderabbitai
* GetS3ExpireTime on filer
* go mod
* clear TtlSeconds for volume
* move s3 delete expired entry to filer
* filer delete meta and data
* del unusing func removeExpiredObject
* test s3 put
* test s3 put multipart
* allowDeleteObjectsByTTL by default
* fix pipline tests
* rm dublicate SeaweedFSExpiresS3
* revert expiration tests
* fix updateTTL
* rm log
* resolv comment
* fix delete version object
* fix S3Versioning
* fix delete on FindEntry
* fix delete chunks
* fix sqlite not support concurrent writes/reads
* move deletion out of listing transaction; delete entries and empty folders
* Revert "fix sqlite not support concurrent writes/reads"
This reverts commit
|
||
|
chrislu
|
f4f2718ba0 | adjust test | ||
|
Chris Lu
|
498ac8903f |
S3: prevent deleting buckets with object locking (#7434)
* prevent deleting buckets with object locking * addressing comments * Update s3api_bucket_handlers.go * address comments * early return * refactor * simplify * constant * go fmt |
||
|
Chris Lu
|
eaecd8328b |
S3: add fallback for CORS (#7404)
* add fallback for cors * refactor * expose aws headers * add fallback to test * refactor * Only falls back to global config when there's explicitly no bucket-level config. * fmt * Update s3_cors_http_test.go * refactoring |
||
|
|
8667aa8e17 |
chore(deps): bump github.com/prometheus/procfs from 0.17.0 to 0.19.1 (#7388)
* chore(deps): bump github.com/prometheus/procfs from 0.17.0 to 0.19.1 Bumps [github.com/prometheus/procfs](https://github.com/prometheus/procfs) from 0.17.0 to 0.19.1. - [Release notes](https://github.com/prometheus/procfs/releases) - [Commits](https://github.com/prometheus/procfs/compare/v0.17.0...v0.19.1) --- updated-dependencies: - dependency-name: github.com/prometheus/procfs dependency-version: 0.19.1 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
|
fe532592cc |
chore(deps): bump golang.org/x/net from 0.45.0 to 0.46.0 (#7386)
* chore(deps): bump golang.org/x/net from 0.45.0 to 0.46.0 Bumps [golang.org/x/net](https://github.com/golang/net) from 0.45.0 to 0.46.0. - [Commits](https://github.com/golang/net/compare/v0.45.0...v0.46.0) --- updated-dependencies: - dependency-name: golang.org/x/net dependency-version: 0.46.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
chrislu
|
290c245095 | go mod tidy | ||
|
|
82fbf15b71 |
chore(deps): bump golang.org/x/image from 0.30.0 to 0.32.0 (#7343)
* chore(deps): bump golang.org/x/image from 0.30.0 to 0.32.0 Bumps [golang.org/x/image](https://github.com/golang/image) from 0.30.0 to 0.32.0. - [Commits](https://github.com/golang/image/compare/v0.30.0...v0.32.0) --- updated-dependencies: - dependency-name: golang.org/x/image dependency-version: 0.32.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> |
||
|
|
da18e3dc82 |
chore(deps): bump golang.org/x/crypto from 0.42.0 to 0.43.0 (#7347)
* chore(deps): bump golang.org/x/crypto from 0.42.0 to 0.43.0 Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.42.0 to 0.43.0. - [Commits](https://github.com/golang/crypto/compare/v0.42.0...v0.43.0) --- updated-dependencies: - dependency-name: golang.org/x/crypto dependency-version: 0.43.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * go mod * go mod 2 * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> |
||
|
dependabot[bot]andchrislu
|
5bec9e4114 |
chore(deps): bump github.com/klauspost/compress from 1.18.0 to 1.18.1 (#7344)
* chore(deps): bump github.com/klauspost/compress from 1.18.0 to 1.18.1 Bumps [github.com/klauspost/compress](https://github.com/klauspost/compress) from 1.18.0 to 1.18.1. - [Release notes](https://github.com/klauspost/compress/releases) - [Changelog](https://github.com/klauspost/compress/blob/master/.goreleaser.yml) - [Commits](https://github.com/klauspost/compress/compare/v1.18.0...v1.18.1) --- updated-dependencies: - dependency-name: github.com/klauspost/compress dependency-version: 1.18.1 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> * go mod * go mod tidy --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: chrislu <chris.lu@gmail.com> |
||
|
Chris Lu
|
97f3028782 |
Clean up logs and deprecated functions (#7339)
* less logs * fix deprecated grpc.Dial |
||
|
Chris Lu
|
8d63a9cf5f |
Fixes for kafka gateway (#7329)
* fix race condition
* save checkpoint every 2 seconds
* Inlined the session creation logic to hold the lock continuously
* comment
* more logs on offset resume
* only recreate if we need to seek backward (requested offset < current offset), not on any mismatch
* Simplified GetOrCreateSubscriber to always reuse existing sessions
* atomic currentStartOffset
* fmt
* avoid deadlock
* fix locking
* unlock
* debug
* avoid race condition
* refactor dedup
* consumer group that does not join group
* increase deadline
* use client timeout wait
* less logs
* add some delays
* adjust deadline
* Update fetch.go
* more time
* less logs, remove unused code
* purge unused
* adjust return values on failures
* clean up consumer protocols
* avoid goroutine leak
* seekable subscribe messages
* ack messages to broker
* reuse cached records
* pin s3 test version
* adjust s3 tests
* verify produced messages are consumed
* track messages with testStartTime
* removing the unnecessary restart logic and relying on the seek mechanism we already implemented
* log read stateless
* debug fetch offset APIs
* fix tests
* fix go mod
* less logs
* test: increase timeouts for consumer group operations in E2E tests
Consumer group operations (coordinator discovery, offset fetch/commit) are
slower in CI environments with limited resources. This increases timeouts to:
- ProduceMessages: 10s -> 30s (for when consumer groups are active)
- ConsumeWithGroup: 30s -> 60s (for offset fetch/commit operations)
Fixes the TestOffsetManagement timeout failures in GitHub Actions CI.
* feat: add context timeout propagation to produce path
This commit adds proper context propagation throughout the produce path,
enabling client-side timeouts to be honored on the broker side. Previously,
only fetch operations respected client timeouts - produce operations continued
indefinitely even if the client gave up.
Changes:
- Add ctx parameter to ProduceRecord and ProduceRecordValue signatures
- Add ctx parameter to PublishRecord and PublishRecordValue in BrokerClient
- Add ctx parameter to handleProduce and related internal functions
- Update all callers (protocol handlers, mocks, tests) to pass context
- Add context cancellation checks in PublishRecord before operations
Benefits:
- Faster failure detection when client times out
- No orphaned publish operations consuming broker resources
- Resource efficiency improvements (no goroutine/stream/lock leaks)
- Consistent timeout behavior between produce and fetch paths
- Better error handling with proper cancellation signals
This fixes the root cause of CI test timeouts where produce operations
continued indefinitely after clients gave up, leading to cascading delays.
* feat: add disk I/O fallback for historical offset reads
This commit implements async disk I/O fallback to handle cases where:
1. Data is flushed from memory before consumers can read it (CI issue)
2. Consumers request historical offsets not in memory
3. Small LogBuffer retention in resource-constrained environments
Changes:
- Add readHistoricalDataFromDisk() helper function
- Update ReadMessagesAtOffset() to call ReadFromDiskFn when offset < bufferStartOffset
- Properly handle maxMessages and maxBytes limits during disk reads
- Return appropriate nextOffset after disk reads
- Log disk read operations at V(2) and V(3) levels
Benefits:
- Fixes CI test failures where data is flushed before consumption
- Enables consumers to catch up even if they fall behind memory retention
- No blocking on hot path (disk read only for historical data)
- Respects existing ReadFromDiskFn timeout handling
How it works:
1. Try in-memory read first (fast path)
2. If offset too old and ReadFromDiskFn configured, read from disk
3. Return disk data with proper nextOffset
4. Consumer continues reading seamlessly
This fixes the 'offset 0 too old (earliest in-memory: 5)' error in
TestOffsetManagement where messages were flushed before consumer started.
* fmt
* feat: add in-memory cache for disk chunk reads
This commit adds an LRU cache for disk chunks to optimize repeated reads
of historical data. When multiple consumers read the same historical offsets,
or a single consumer refetches the same data, the cache eliminates redundant
disk I/O.
Cache Design:
- Chunk size: 1000 messages per chunk
- Max chunks: 16 (configurable, ~16K messages cached)
- Eviction policy: LRU (Least Recently Used)
- Thread-safe with RWMutex
- Chunk-aligned offsets for efficient lookups
New Components:
1. DiskChunkCache struct - manages cached chunks
2. CachedDiskChunk struct - stores chunk data with metadata
3. getCachedDiskChunk() - checks cache before disk read
4. cacheDiskChunk() - stores chunks with LRU eviction
5. extractMessagesFromCache() - extracts subset from cached chunk
How It Works:
1. Read request for offset N (e.g., 2500)
2. Calculate chunk start: (2500 / 1000) * 1000 = 2000
3. Check cache for chunk starting at 2000
4. If HIT: Extract messages 2500-2999 from cached chunk
5. If MISS: Read chunk 2000-2999 from disk, cache it, extract 2500-2999
6. If cache full: Evict LRU chunk before caching new one
Benefits:
- Eliminates redundant disk I/O for popular historical data
- Reduces latency for repeated reads (cache hit ~1ms vs disk ~100ms)
- Supports multiple consumers reading same historical offsets
- Automatically evicts old chunks when cache is full
- Zero impact on hot path (in-memory reads unchanged)
Performance Impact:
- Cache HIT: ~99% faster than disk read
- Cache MISS: Same as disk read (with caching overhead ~1%)
- Memory: ~16MB for 16 chunks (16K messages x 1KB avg)
Example Scenario (CI tests):
- Producer writes offsets 0-4
- Data flushes to disk
- Consumer 1 reads 0-4 (cache MISS, reads from disk, caches chunk 0-999)
- Consumer 2 reads 0-4 (cache HIT, served from memory)
- Consumer 1 rebalances, re-reads 0-4 (cache HIT, no disk I/O)
This optimization is especially valuable in CI environments where:
- Small memory buffers cause frequent flushing
- Multiple consumers read the same historical data
- Disk I/O is relatively slow compared to memory access
* fix: commit offsets in Cleanup() before rebalancing
This commit adds explicit offset commit in the ConsumerGroupHandler.Cleanup()
method, which is called during consumer group rebalancing. This ensures all
marked offsets are committed BEFORE partitions are reassigned to other consumers,
significantly reducing duplicate message consumption during rebalancing.
Problem:
- Cleanup() was not committing offsets before rebalancing
- When partition reassigned to another consumer, it started from last committed offset
- Uncommitted messages (processed but not yet committed) were read again by new consumer
- This caused ~100-200% duplicate messages during rebalancing in tests
Solution:
- Add session.Commit() in Cleanup() method
- This runs after all ConsumeClaim goroutines have exited
- Ensures all MarkMessage() calls are committed before partition release
- New consumer starts from the last processed offset, not an older committed offset
Benefits:
- Dramatically reduces duplicate messages during rebalancing
- Improves at-least-once semantics (closer to exactly-once for normal cases)
- Better performance (less redundant processing)
- Cleaner test results (expected duplicates only from actual failures)
Kafka Rebalancing Lifecycle:
1. Rebalance triggered (consumer join/leave, timeout, etc.)
2. All ConsumeClaim goroutines cancelled
3. Cleanup() called ← WE COMMIT HERE NOW
4. Partitions reassigned to other consumers
5. New consumer starts from last committed offset ← NOW MORE UP-TO-DATE
Expected Results:
- Before: ~100-200% duplicates during rebalancing (2-3x reads)
- After: <10% duplicates (only from uncommitted in-flight messages)
This is a critical fix for production deployments where consumer churn
(scaling, restarts, failures) causes frequent rebalancing.
* fmt
* feat: automatic idle partition cleanup to prevent memory bloat
Implements automatic cleanup of topic partitions with no active publishers
or subscribers to prevent memory accumulation from short-lived topics.
**Key Features:**
1. Activity Tracking (local_partition.go)
- Added lastActivityTime field to LocalPartition
- UpdateActivity() called on publish, subscribe, and message reads
- IsIdle() checks if partition has no publishers/subscribers
- GetIdleDuration() returns time since last activity
- ShouldCleanup() determines if partition eligible for cleanup
2. Cleanup Task (local_manager.go)
- Background goroutine runs every 1 minute (configurable)
- Removes partitions idle for > 5 minutes (configurable)
- Automatically removes empty topics after all partitions cleaned
- Proper shutdown handling with WaitForCleanupShutdown()
3. Broker Integration (broker_server.go)
- StartIdlePartitionCleanup() called on broker startup
- Default: check every 1 minute, cleanup after 5 minutes idle
- Transparent operation with sensible defaults
**Cleanup Process:**
- Checks: partition.Publishers.Size() == 0 && partition.Subscribers.Size() == 0
- Calls partition.Shutdown() to:
- Flush all data to disk (no data loss)
- Stop 3 goroutines (loopFlush, loopInterval, cleanupLoop)
- Free in-memory buffers (~100KB-10MB per partition)
- Close LogBuffer resources
- Removes partition from LocalTopic.Partitions
- Removes topic if no partitions remain
**Benefits:**
- Prevents memory bloat from short-lived topics
- Reduces goroutine count (3 per partition cleaned)
- Zero configuration required
- Data remains on disk, can be recreated on demand
- No impact on active partitions
**Example Logs:**
I Started idle partition cleanup task (check: 1m, timeout: 5m)
I Cleaning up idle partition topic-0 (idle for 5m12s, publishers=0, subscribers=0)
I Cleaned up 2 idle partition(s)
**Memory Freed per Partition:**
- In-memory message buffer: ~100KB-10MB
- Disk buffer cache
- 3 goroutines
- Publisher/subscriber tracking maps
- Condition variables and mutexes
**Related Issue:**
Prevents memory accumulation in systems with high topic churn or
many short-lived consumer groups, improving long-term stability
and resource efficiency.
**Testing:**
- Compiles cleanly
- No linting errors
- Ready for integration testing
fmt
* refactor: reduce verbosity of debug log messages
Changed debug log messages with bracket prefixes from V(1)/V(2) to V(3)/V(4)
to reduce log noise in production. These messages were added during development
for detailed debugging and are still available with higher verbosity levels.
Changes:
- glog.V(2).Infof("[") -> glog.V(4).Infof("[") (~104 messages)
- glog.V(1).Infof("[") -> glog.V(3).Infof("[") (~30 messages)
Affected files:
- weed/mq/broker/broker_grpc_fetch.go
- weed/mq/broker/broker_grpc_sub_offset.go
- weed/mq/kafka/integration/broker_client_fetch.go
- weed/mq/kafka/integration/broker_client_subscribe.go
- weed/mq/kafka/integration/seaweedmq_handler.go
- weed/mq/kafka/protocol/fetch.go
- weed/mq/kafka/protocol/fetch_partition_reader.go
- weed/mq/kafka/protocol/handler.go
- weed/mq/kafka/protocol/offset_management.go
Benefits:
- Cleaner logs in production (default -v=0)
- Still available for deep debugging with -v=3 or -v=4
- No code behavior changes, only log verbosity
- Safer than deletion - messages preserved for debugging
Usage:
- Default (-v=0): Only errors and important events
- -v=1: Standard info messages
- -v=2: Detailed info messages
- -v=3: Debug messages (previously V(1) with brackets)
- -v=4: Verbose debug (previously V(2) with brackets)
* refactor: change remaining glog.Infof debug messages to V(3)
Changed remaining debug log messages with bracket prefixes from
glog.Infof() to glog.V(3).Infof() to prevent them from showing
in production logs by default.
Changes (8 messages across 3 files):
- glog.Infof("[") -> glog.V(3).Infof("[")
Files updated:
- weed/mq/broker/broker_grpc_fetch.go (4 messages)
- [FetchMessage] CALLED! debug marker
- [FetchMessage] request details
- [FetchMessage] LogBuffer read start
- [FetchMessage] LogBuffer read completion
- weed/mq/kafka/integration/broker_client_fetch.go (3 messages)
- [FETCH-STATELESS-CLIENT] received messages
- [FETCH-STATELESS-CLIENT] converted records (with data)
- [FETCH-STATELESS-CLIENT] converted records (empty)
- weed/mq/kafka/integration/broker_client_publish.go (1 message)
- [GATEWAY RECV] _schemas topic debug
Now ALL debug messages with bracket prefixes require -v=3 or higher:
- Default (-v=0): Clean production logs ✅
- -v=3: All debug messages visible
- -v=4: All verbose debug messages visible
Result: Production logs are now clean with default settings!
* remove _schemas debug
* less logs
* fix: critical bug causing 51% message loss in stateless reads
CRITICAL BUG FIX: ReadMessagesAtOffset was returning error instead of
attempting disk I/O when data was flushed from memory, causing massive
message loss (6254 out of 12192 messages = 51% loss).
Problem:
In log_read_stateless.go lines 120-131, when data was flushed to disk
(empty previous buffer), the code returned an 'offset out of range' error
instead of attempting disk I/O. This caused consumers to skip over flushed
data entirely, leading to catastrophic message loss.
The bug occurred when:
1. Data was written to LogBuffer
2. Data was flushed to disk due to buffer rotation
3. Consumer requested that offset range
4. Code found offset in expected range but not in memory
5. ❌ Returned error instead of reading from disk
Root Cause:
Lines 126-131 had early return with error when previous buffer was empty:
// Data not in memory - for stateless fetch, we don't do disk I/O
return messages, startOffset, highWaterMark, false,
fmt.Errorf("offset %d out of range...")
This comment was incorrect - we DO need disk I/O for flushed data!
Fix:
1. Lines 120-132: Changed to fall through to disk read logic instead of
returning error when previous buffer is empty
2. Lines 137-177: Enhanced disk read logic to handle TWO cases:
- Historical data (offset < bufferStartOffset)
- Flushed data (offset >= bufferStartOffset but not in memory)
Changes:
- Line 121: Log "attempting disk read" instead of breaking
- Line 130-132: Fall through to disk read instead of returning error
- Line 141: Changed condition from 'if startOffset < bufferStartOffset'
to 'if startOffset < currentBufferEnd' to handle both cases
- Lines 143-149: Add context-aware logging for both historical and flushed data
- Lines 154-159: Add context-aware error messages
Expected Results:
- Before: 51% message loss (6254/12192 missing)
- After: <1% message loss (only from rebalancing, which we already fixed)
- Duplicates: Should remain ~47% (from rebalancing, expected until offsets committed)
Testing:
- ✅ Compiles successfully
- Ready for integration testing with standard-test
Related Issues:
- This explains the massive data loss in recent load tests
- Disk I/O fallback was implemented but not reachable due to early return
- Disk chunk cache is working but was never being used for flushed data
Priority: CRITICAL - Fixes production-breaking data loss bug
* perf: add topic configuration cache to fix 60% CPU overhead
CRITICAL PERFORMANCE FIX: Added topic configuration caching to eliminate
massive CPU overhead from repeated filer reads and JSON unmarshaling on
EVERY fetch request.
Problem (from CPU profile):
- ReadTopicConfFromFiler: 42.45% CPU (5.76s out of 13.57s)
- protojson.Unmarshal: 25.64% CPU (3.48s)
- GetOrGenerateLocalPartition called on EVERY FetchMessage request
- No caching - reading from filer and unmarshaling JSON every time
- This caused filer, gateway, and broker to be extremely busy
Root Cause:
GetOrGenerateLocalPartition() is called on every FetchMessage request and
was calling ReadTopicConfFromFiler() without any caching. Each call:
1. Makes gRPC call to filer (expensive)
2. Reads JSON from disk (expensive)
3. Unmarshals protobuf JSON (25% of CPU!)
The disk I/O fix (previous commit) made this worse by enabling more reads,
exposing this performance bottleneck.
Solution:
Added topicConfCache similar to existing topicExistsCache:
Changes to broker_server.go:
- Added topicConfCacheEntry struct
- Added topicConfCache map to MessageQueueBroker
- Added topicConfCacheMu RWMutex for thread safety
- Added topicConfCacheTTL (30 seconds)
- Initialize cache in NewMessageBroker()
Changes to broker_topic_conf_read_write.go:
- Modified GetOrGenerateLocalPartition() to check cache first
- Cache HIT: Return cached config immediately (V(4) log)
- Cache MISS: Read from filer, cache result, proceed
- Added invalidateTopicConfCache() for cache invalidation
- Added import "time" for cache TTL
Cache Strategy:
- TTL: 30 seconds (matches topicExistsCache)
- Thread-safe with RWMutex
- Cache key: topic.String() (e.g., "kafka.loadtest-topic-0")
- Invalidation: Call invalidateTopicConfCache() when config changes
Expected Results:
- Before: 60% CPU on filer reads + JSON unmarshaling
- After: <1% CPU (only on cache miss every 30s)
- Filer load: Reduced by ~99% (from every fetch to once per 30s)
- Gateway CPU: Dramatically reduced
- Broker CPU: Dramatically reduced
- Throughput: Should increase significantly
Performance Impact:
With 50 msgs/sec per topic × 5 topics = 250 fetches/sec:
- Before: 250 filer reads/sec (25000% overhead!)
- After: 0.17 filer reads/sec (5 topics / 30s TTL)
- Reduction: 99.93% fewer filer calls
Testing:
- ✅ Compiles successfully
- Ready for load test to verify CPU reduction
Priority: CRITICAL - Fixes production-breaking performance issue
Related: Works with previous commit (disk I/O fix) to enable correct and fast reads
* fmt
* refactor: merge topicExistsCache and topicConfCache into unified topicCache
Merged two separate caches into one unified cache to simplify code and
reduce memory usage. The unified cache stores both topic existence and
configuration in a single structure.
Design:
- Single topicCacheEntry with optional *ConfigureTopicResponse
- If conf != nil: topic exists with full configuration
- If conf == nil: topic doesn't exist (negative cache)
- Same 30-second TTL for both existence and config caching
Changes to broker_server.go:
- Removed topicExistsCacheEntry struct
- Removed topicConfCacheEntry struct
- Added unified topicCacheEntry struct (conf can be nil)
- Removed topicExistsCache, topicExistsCacheMu, topicExistsCacheTTL
- Removed topicConfCache, topicConfCacheMu, topicConfCacheTTL
- Added unified topicCache, topicCacheMu, topicCacheTTL
- Updated NewMessageBroker() to initialize single cache
Changes to broker_topic_conf_read_write.go:
- Modified GetOrGenerateLocalPartition() to use unified cache
- Added negative caching (conf=nil) when topic not found
- Renamed invalidateTopicConfCache() to invalidateTopicCache()
- Single cache lookup instead of two separate checks
Changes to broker_grpc_lookup.go:
- Modified TopicExists() to use unified cache
- Check: exists = (entry.conf != nil)
- Only cache negative results (conf=nil) in TopicExists
- Positive results cached by GetOrGenerateLocalPartition
- Removed old invalidateTopicExistsCache() function
Changes to broker_grpc_configure.go:
- Updated invalidateTopicExistsCache() calls to invalidateTopicCache()
- Two call sites updated
Benefits:
1. Code Simplification: One cache instead of two
2. Memory Reduction: Single map, single mutex, single TTL
3. Consistency: No risk of cache desync between existence and config
4. Less Lock Contention: One lock instead of two
5. Easier Maintenance: Single invalidation function
6. Same Performance: Still eliminates 60% CPU overhead
Cache Behavior:
- TopicExists: Lightweight check, only caches negative (conf=nil)
- GetOrGenerateLocalPartition: Full config read, caches positive (conf != nil)
- Both share same 30s TTL
- Both use same invalidation on topic create/update/delete
Testing:
- ✅ Compiles successfully
- Ready for integration testing
This refactor maintains all performance benefits while simplifying
the codebase and reducing memory footprint.
* fix: add cache to LookupTopicBrokers to eliminate 26% CPU overhead
CRITICAL: LookupTopicBrokers was bypassing cache, causing 26% CPU overhead!
Problem (from CPU profile):
- LookupTopicBrokers: 35.74% CPU (9s out of 25.18s)
- ReadTopicConfFromFiler: 26.41% CPU (6.65s)
- protojson.Unmarshal: 16.64% CPU (4.19s)
- LookupTopicBrokers called b.fca.ReadTopicConfFromFiler() directly on line 35
- Completely bypassed our unified topicCache!
Root Cause:
LookupTopicBrokers is called VERY frequently by clients (every fetch request
needs to know partition assignments). It was calling ReadTopicConfFromFiler
directly instead of using the cache, causing:
1. Expensive gRPC calls to filer on every lookup
2. Expensive JSON unmarshaling on every lookup
3. 26%+ CPU overhead on hot path
4. Our cache optimization was useless for this critical path
Solution:
Created getTopicConfFromCache() helper and updated all callers:
Changes to broker_topic_conf_read_write.go:
- Added getTopicConfFromCache() - public API for cached topic config reads
- Implements same caching logic: check cache -> read filer -> cache result
- Handles both positive (conf != nil) and negative (conf == nil) caching
- Refactored GetOrGenerateLocalPartition() to use new helper (code dedup)
- Now only 14 lines instead of 60 lines (removed duplication)
Changes to broker_grpc_lookup.go:
- Modified LookupTopicBrokers() to call getTopicConfFromCache()
- Changed from: b.fca.ReadTopicConfFromFiler(t) (no cache)
- Changed to: b.getTopicConfFromCache(t) (with cache)
- Added comment explaining this fixes 26% CPU overhead
Cache Strategy:
- First call: Cache MISS -> read filer + unmarshal JSON -> cache for 30s
- Next 1000+ calls in 30s: Cache HIT -> return cached config immediately
- No filer gRPC, no JSON unmarshaling, near-zero CPU
- Cache invalidated on topic create/update/delete
Expected CPU Reduction:
- Before: 26.41% on ReadTopicConfFromFiler + 16.64% on JSON unmarshal = 43% CPU
- After: <0.1% (only on cache miss every 30s)
- Expected total broker CPU: 25.18s -> ~8s (67% reduction!)
Performance Impact (with 250 lookups/sec):
- Before: 250 filer reads/sec + 250 JSON unmarshals/sec
- After: 0.17 filer reads/sec (5 topics / 30s TTL)
- Reduction: 99.93% fewer expensive operations
Code Quality:
- Eliminated code duplication (60 lines -> 14 lines in GetOrGenerateLocalPartition)
- Single source of truth for cached reads (getTopicConfFromCache)
- Clear API: "Always use getTopicConfFromCache, never ReadTopicConfFromFiler directly"
Testing:
- ✅ Compiles successfully
- Ready to deploy and measure CPU improvement
Priority: CRITICAL - Completes the cache optimization to achieve full performance fix
* perf: optimize broker assignment validation to eliminate 14% CPU overhead
CRITICAL: Assignment validation was running on EVERY LookupTopicBrokers call!
Problem (from CPU profile):
- ensureTopicActiveAssignments: 14.18% CPU (2.56s out of 18.05s)
- EnsureAssignmentsToActiveBrokers: 14.18% CPU (2.56s)
- ConcurrentMap.IterBuffered: 12.85% CPU (2.32s) - iterating all brokers
- Called on EVERY LookupTopicBrokers request, even with cached config!
Root Cause:
LookupTopicBrokers flow was:
1. getTopicConfFromCache() - returns cached config (fast ✅)
2. ensureTopicActiveAssignments() - validates assignments (slow ❌)
Even though config was cached, we still validated assignments every time,
iterating through ALL active brokers on every single request. With 250
requests/sec, this meant 250 full broker iterations per second!
Solution:
Move assignment validation inside getTopicConfFromCache() and only run it
on cache misses:
Changes to broker_topic_conf_read_write.go:
- Modified getTopicConfFromCache() to validate assignments after filer read
- Validation only runs on cache miss (not on cache hit)
- If hasChanges: Save to filer immediately, invalidate cache, return
- If no changes: Cache config with validated assignments
- Added ensureTopicActiveAssignmentsUnsafe() helper (returns bool)
- Kept ensureTopicActiveAssignments() for other callers (saves to filer)
Changes to broker_grpc_lookup.go:
- Removed ensureTopicActiveAssignments() call from LookupTopicBrokers
- Assignment validation now implicit in getTopicConfFromCache()
- Added comments explaining the optimization
Cache Behavior:
- Cache HIT: Return config immediately, skip validation (saves 14% CPU!)
- Cache MISS: Read filer -> validate assignments -> cache result
- If broker changes detected: Save to filer, invalidate cache, return
- Next request will re-read and re-validate (ensures consistency)
Performance Impact:
With 30-second cache TTL and 250 lookups/sec:
- Before: 250 validations/sec × 10ms each = 2.5s CPU/sec (14% overhead)
- After: 0.17 validations/sec (only on cache miss)
- Reduction: 99.93% fewer validations
Expected CPU Reduction:
- Before (with cache): 18.05s total, 2.56s validation (14%)
- After (with optimization): ~15.5s total (-14% = ~2.5s saved)
- Combined with previous cache fix: 25.18s -> ~15.5s (38% total reduction)
Cache Consistency:
- Assignments validated when config first cached
- If broker membership changes, assignments updated and saved
- Cache invalidated to force fresh read
- All brokers eventually converge on correct assignments
Testing:
- ✅ Compiles successfully
- Ready to deploy and measure CPU improvement
Priority: CRITICAL - Completes optimization of LookupTopicBrokers hot path
* fmt
* perf: add partition assignment cache in gateway to eliminate 13.5% CPU overhead
CRITICAL: Gateway calling LookupTopicBrokers on EVERY fetch to translate
Kafka partition IDs to SeaweedFS partition ranges!
Problem (from CPU profile):
- getActualPartitionAssignment: 13.52% CPU (1.71s out of 12.65s)
- Called bc.client.LookupTopicBrokers on line 228 for EVERY fetch
- With 250 fetches/sec, this means 250 LookupTopicBrokers calls/sec!
- No caching at all - same overhead as broker had before optimization
Root Cause:
Gateway needs to translate Kafka partition IDs (0, 1, 2...) to SeaweedFS
partition ranges (0-341, 342-682, etc.) for every fetch request. This
translation requires calling LookupTopicBrokers to get partition assignments.
Without caching, every fetch request triggered:
1. gRPC call to broker (LookupTopicBrokers)
2. Broker reads from its cache (fast now after broker optimization)
3. gRPC response back to gateway
4. Gateway computes partition range mapping
The gRPC round-trip overhead was consuming 13.5% CPU even though broker
cache was fast!
Solution:
Added partitionAssignmentCache to BrokerClient:
Changes to types.go:
- Added partitionAssignmentCacheEntry struct (assignments + expiresAt)
- Added cache fields to BrokerClient:
* partitionAssignmentCache map[string]*partitionAssignmentCacheEntry
* partitionAssignmentCacheMu sync.RWMutex
* partitionAssignmentCacheTTL time.Duration
Changes to broker_client.go:
- Initialize partitionAssignmentCache in NewBrokerClientWithFilerAccessor
- Set partitionAssignmentCacheTTL to 30 seconds (same as broker)
Changes to broker_client_publish.go:
- Added "time" import
- Modified getActualPartitionAssignment() to check cache first:
* Cache HIT: Use cached assignments (fast ✅)
* Cache MISS: Call LookupTopicBrokers, cache result for 30s
- Extracted findPartitionInAssignments() helper function
* Contains range calculation and partition matching logic
* Reused for both cached and fresh lookups
Cache Behavior:
- First fetch: Cache MISS -> LookupTopicBrokers (~2ms) -> cache for 30s
- Next 7500 fetches in 30s: Cache HIT -> immediate return (~0.01ms)
- Cache automatically expires after 30s, re-validates on next fetch
Performance Impact:
With 250 fetches/sec and 5 topics:
- Before: 250 LookupTopicBrokers/sec = 500ms CPU overhead
- After: 0.17 LookupTopicBrokers/sec (5 topics / 30s TTL)
- Reduction: 99.93% fewer gRPC calls
Expected CPU Reduction:
- Before: 12.65s total, 1.71s in getActualPartitionAssignment (13.5%)
- After: ~11s total (-13.5% = 1.65s saved)
- Benefit: 13% lower CPU, more capacity for actual message processing
Cache Consistency:
- Same 30-second TTL as broker's topic config cache
- Partition assignments rarely change (only on topic reconfiguration)
- 30-second staleness is acceptable for partition mapping
- Gateway will eventually converge with broker's view
Testing:
- ✅ Compiles successfully
- Ready to deploy and measure CPU improvement
Priority: CRITICAL - Eliminates major performance bottleneck in gateway fetch path
* perf: add RecordType inference cache to eliminate 37% gateway CPU overhead
CRITICAL: Gateway was creating Avro codecs and inferring RecordTypes on
EVERY fetch request for schematized topics!
Problem (from CPU profile):
- NewCodec (Avro): 17.39% CPU (2.35s out of 13.51s)
- inferRecordTypeFromAvroSchema: 20.13% CPU (2.72s)
- Total schema overhead: 37.52% CPU
- Called during EVERY fetch to check if topic is schematized
- No caching - recreating expensive goavro.Codec objects repeatedly
Root Cause:
In the fetch path, isSchematizedTopic() -> matchesSchemaRegistryConvention()
-> ensureTopicSchemaFromRegistryCache() -> inferRecordTypeFromCachedSchema()
-> inferRecordTypeFromAvroSchema() was being called.
The inferRecordTypeFromAvroSchema() function created a NEW Avro decoder
(which internally calls goavro.NewCodec()) on every call, even though:
1. The schema.Manager already has a decoder cache by schema ID
2. The same schemas are used repeatedly for the same topics
3. goavro.NewCodec() is expensive (parses JSON, builds schema tree)
This was wasteful because:
- Same schema string processed repeatedly
- No reuse of inferred RecordType structures
- Creating codecs just to infer types, then discarding them
Solution:
Added inferredRecordTypes cache to Handler:
Changes to handler.go:
- Added inferredRecordTypes map[string]*schema_pb.RecordType to Handler
- Added inferredRecordTypesMu sync.RWMutex for thread safety
- Initialize cache in NewTestHandlerWithMock() and NewSeaweedMQBrokerHandlerWithDefaults()
Changes to produce.go:
- Added glog import
- Modified inferRecordTypeFromAvroSchema():
* Check cache first (key: schema string)
* Cache HIT: Return immediately (V(4) log)
* Cache MISS: Create decoder, infer type, cache result
- Modified inferRecordTypeFromProtobufSchema():
* Same caching strategy (key: "protobuf:" + schema)
- Modified inferRecordTypeFromJSONSchema():
* Same caching strategy (key: "json:" + schema)
Cache Strategy:
- Key: Full schema string (unique per schema content)
- Value: Inferred *schema_pb.RecordType
- Thread-safe with RWMutex (optimized for reads)
- No TTL - schemas don't change for a topic
- Memory efficient - RecordType is small compared to codec
Performance Impact:
With 250 fetches/sec across 5 topics (1-3 schemas per topic):
- Before: 250 codec creations/sec + 250 inferences/sec = ~5s CPU
- After: 3-5 codec creations total (one per schema) = ~0.05s CPU
- Reduction: 99% fewer expensive operations
Expected CPU Reduction:
- Before: 13.51s total, 5.07s schema operations (37.5%)
- After: ~8.5s total (-37.5% = 5s saved)
- Benefit: 37% lower gateway CPU, more capacity for message processing
Cache Consistency:
- Schemas are immutable once registered in Schema Registry
- If schema changes, schema ID changes, so safe to cache indefinitely
- New schemas automatically cached on first use
- No need for invalidation or TTL
Additional Optimizations:
- Protobuf and JSON Schema also cached (same pattern)
- Prevents future bottlenecks as more schema formats are used
- Consistent caching approach across all schema types
Testing:
- ✅ Compiles successfully
- Ready to deploy and measure CPU improvement under load
Priority: HIGH - Eliminates major performance bottleneck in gateway schema path
* fmt
* fix Node ID Mismatch, and clean up log messages
* clean up
* Apply client-specified timeout to context
* Add comprehensive debug logging for Noop record processing
- Track Produce v2+ request reception with API version and request body size
- Log acks setting, timeout, and topic/partition information
- Log record count from parseRecordSet and any parse errors
- **CRITICAL**: Log when recordCount=0 fallback extraction attempts
- Log record extraction with NULL value detection (Noop records)
- Log record key in hex for Noop key identification
- Track each record being published to broker
- Log offset assigned by broker for each record
- Log final response with offset and error code
This enables root cause analysis of Schema Registry Noop record timeout issue.
* fix: Remove context timeout propagation from produce that breaks consumer init
Commit
|
||
|
Chris Lu
|
3d25f206c8 |
S3: Signature verification should not check permissions (#7335)
* Signature verification should not check permissions - that's done later in authRequest * test permissions during signature verfication * fix s3 test path * s3tests_boto3 => s3tests * remove extra lines |