Compare commits

...
Author SHA1 Message Date
Chris LuandGitHub f6df7126b6 feat(admin): add profiling options for debugging high memory/CPU usage (#8923)
* feat(admin): add profiling options for debugging high memory/CPU usage

Add -debug, -debug.port, -cpuprofile, and -memprofile flags to the admin
command, matching the profiling support already available in master, volume,
and other server commands. This enables investigation of resource usage
issues like #8919.

* refactor(admin): move profiling flags into AdminOptions struct

Move cpuprofile and memprofile flags from global variables into the
AdminOptions struct and init() function for consistency with other flags.

* fix(debug): bind pprof server to localhost only and document profiling flags

StartDebugServer was binding to all interfaces (0.0.0.0), exposing
runtime profiling data to the network. Restrict to 127.0.0.1 since
this is a development/debugging tool.

Also add a "Debugging and Profiling" section to the admin command's
help text documenting the new flags.
2026-04-04 10:05:19 -07:00
Chris LuandGitHub 9add18e169 fix(volume-rust): fix volume balance between Go and Rust servers (#8915)
Two bugs prevented reliable volume balancing when a Rust volume server
is the copy target:

1. find_last_append_at_ns returned None for delete tombstones (Size==0
   in dat header), falling back to file mtime truncated to seconds.
   This caused the tail step to re-send needles from the last sub-second
   window. Fix: change `needle_size <= 0` to `< 0` since Size==0 delete
   needles still have a valid timestamp in their tail.

2. VolumeTailReceiver called read_body_v2 on delete needles, which have
   no DataSize/Data/flags — only checksum+timestamp+padding after the
   header. Fix: skip read_body_v2 when size == 0, reject negative sizes.

Also:
- Unify gRPC server bind: use TcpListener::bind before spawn for both
  TLS and non-TLS paths, propagating bind errors at startup.
- Add mixed Go+Rust cluster test harness and integration tests covering
  VolumeCopy in both directions, copy with deletes, and full balance
  move with tail tombstone propagation and source deletion.
- Make FindOrBuildRustBinary configurable for default vs no-default
  features (4-byte vs 5-byte offsets).
2026-04-04 09:13:23 -07:00
Chris LuandGitHub d1823d3784 fix(s3): include static identities in listing operations (#8903)
* fix(s3): include static identities in listing operations

Static identities loaded from -s3.config file were only stored in the
S3 API server's in-memory state. Listing operations (s3.configure shell
command, aws iam list-users) queried the credential manager which only
returned dynamic identities from the backend store.

Register static identities with the credential manager after loading
so they are included in LoadConfiguration and ListUsers results, and
filtered out before SaveConfiguration to avoid persisting them to the
dynamic store.

Fixes https://github.com/seaweedfs/seaweedfs/discussions/8896

* fix: avoid mutating caller's config and defensive copies

- SaveConfiguration: use shallow struct copy instead of mutating the
  caller's config.Identities field
- SetStaticIdentities: skip nil entries to avoid panics
- GetStaticIdentities: defensively copy PolicyNames slice to avoid
  aliasing the original

* fix: filter nil static identities and sync on config reload

- SetStaticIdentities: filter nil entries from the stored slice (not
  just from staticNames) to prevent panics in LoadConfiguration/ListUsers
- Extract updateCredentialManagerStaticIdentities helper and call it
  from both startup and the grace.OnReload handler so the credential
  manager's static snapshot stays current after config file reloads

* fix: add mutex for static identity fields and fix ListUsers for store callers

- Add sync.RWMutex to protect staticIdentities/staticNames against
  concurrent reads during config reload
- Revert CredentialManager.ListUsers to return only store users, since
  internal callers (e.g. DeletePolicy) look up each user in the store
  and fail on non-existent static entries
- Merge static usernames in the filer gRPC ListUsers handler instead,
  via the new GetStaticUsernames method
- Fix CI: TestIAMPolicyManagement/managed_policy_crud_lifecycle was
  failing because DeletePolicy iterated static users that don't exist
  in the store

* fix: show static identities in admin UI and weed shell

The admin UI and weed shell s3.configure command query the filer's
credential manager via gRPC, which is a separate instance from the S3
server's credential manager. Static identities were only registered
on the S3 server's credential manager, so they never appeared in the
filer's responses.

- Add CredentialManager.LoadS3ConfigFile to parse a static S3 config
  file and register its identities
- Add FilerOptions.s3ConfigFile so the filer can load the same static
  config that the S3 server uses
- Wire s3ConfigFile through in weed mini and weed server modes
- Merge static usernames in filer gRPC ListUsers handler
- Add CredentialManager.GetStaticUsernames helper
- Add sync.RWMutex to protect concurrent access to static identity
  fields
- Avoid importing weed/filer from weed/credential (which pulled in
  filer store init() registrations and broke test isolation)
- Add docker/compose/s3_static_users_example.json

* fix(admin): make static users read-only in admin UI

Static users loaded from the -s3.config file should not be editable
or deletable through the admin UI since they are managed via the
config file.

- Add IsStatic field to ObjectStoreUser, set from credential manager
- Hide edit, delete, and access key buttons for static users in the
  users table template
- Show a "static" badge next to static user names
- Return 403 Forbidden from UpdateUser and DeleteUser API handlers
  when the target user is a static identity

* fix(admin): show details for static users

GetObjectStoreUserDetails called credentialManager.GetUser which only
queries the dynamic store. For static users this returned
ErrUserNotFound. Fall back to GetStaticIdentity when the store lookup
fails.

* fix(admin): load static S3 identities in admin server

The admin server has its own credential manager (gRPC store) which is
a separate instance from the S3 server's and filer's. It had no static
identity data, so IsStaticIdentity returned false (edit/delete buttons
shown) and GetStaticIdentity returned nil (details page failed).

Pass the -s3.config file path through to the admin server and call
LoadS3ConfigFile on its credential manager, matching the approach
used for the filer.

* fix: use protobuf is_static field instead of passing config file path

The previous approach passed -s3.config file path to every component
(filer, admin). This is wrong because the admin server should not need
to know about S3 config files.

Instead, add an is_static field to the Identity protobuf message.
The field is set when static identities are serialized (in
GetStaticIdentities and LoadS3ConfigFile). Any gRPC client that loads
configuration via GetConfiguration automatically sees which identities
are static, without needing the config file.

- Add is_static field (tag 8) to iam_pb.Identity proto message
- Set IsStatic=true in GetStaticIdentities and LoadS3ConfigFile
- Admin GetObjectStoreUsers reads identity.IsStatic from proto
- Admin IsStaticUser helper loads config via gRPC to check the flag
- Filer GetUser gRPC handler falls back to GetStaticIdentity
- Remove s3ConfigFile from AdminOptions and NewAdminServer signature
2026-04-03 20:01:28 -07:00
Chris LuandGitHub 0798b274dd feat(s3): add concurrent chunk prefetch for large file downloads (#8917)
* feat(s3): add concurrent chunk prefetch for large file downloads

Add a pipe-based prefetch pipeline that overlaps chunk fetching with
response writing during S3 GetObject, SSE downloads, and filer proxy.

While chunk N streams to the HTTP response, fetch goroutines for the
next K chunks establish HTTP connections to volume servers ahead of
time, eliminating the RTT gap between sequential chunk fetches.

Uses io.Pipe for minimal memory overhead (~1MB per download regardless
of chunk size, vs buffering entire chunks). Also increases the
streaming read buffer from 64KB to 256KB to reduce syscall overhead.

Benchmark results (64KB chunks, prefetch=4):
- 0ms latency:  1058 → 2362 MB/s (2.2× faster)
- 5ms latency:  11.0 → 41.7 MB/s (3.8× faster)
- 10ms latency: 5.9  → 23.3 MB/s (4.0× faster)
- 20ms latency: 3.1  → 12.1 MB/s (3.9× faster)

* fix: address review feedback for prefetch pipeline

- Fix data race: use *chunkPipeResult (pointer) on channel to avoid
  copying struct while fetch goroutines write to it. Confirmed clean
  with -race detector.
- Remove concurrent map write: retryWithCacheInvalidation no longer
  updates fileId2Url map. Producer only reads it; consumer never writes.
- Use mem.Allocate/mem.Free for copy buffer to reduce GC pressure.
- Add local cancellable context so consumer errors (client disconnect)
  immediately stop the producer and all in-flight fetch goroutines.

* fix(test): remove dead code and add Range header support in test server

- Remove unused allData variable in makeChunksAndServer
- Add Range header handling to createTestServer for partial chunk
  read coverage (206 Partial Content, 416 Range Not Satisfiable)

* fix: correct retry condition and goroutine leak in prefetch pipeline

- Fix retry condition: use result.fetchErr/result.written instead of
  copied to decide cache-invalidation retry. The old condition wrongly
  triggered retry when the fetch succeeded but the response writer
  failed on the first write (copied==0 despite fetcher having data).
  Now matches the sequential path (stream.go:197) which checks whether
  the fetcher itself wrote zero bytes.

- Fix goroutine leak: when the producer's send to the results channel
  is interrupted by context cancellation, the fetch goroutine was
  already launched but the result was never sent to the channel. The
  drain loop couldn't handle it. Now waits on result.done before
  returning so every fetch goroutine is properly awaited.
2026-04-03 19:57:30 -07:00
Chris LuandGitHub 3efe88c718 feat(s3): store and return checksum headers for additional checksum algorithms (#8914)
* feat(s3): store and return checksum headers for additional checksum algorithms

When clients upload with --checksum-algorithm (SHA256, CRC32, etc.),
SeaweedFS validated the checksum but discarded it. The checksum was
never stored in metadata or returned in PUT/HEAD/GET responses.

Now the checksum is computed alongside MD5 during upload, stored in
entry extended attributes, and returned as the appropriate
x-amz-checksum-* header in all responses.

Fixes #8911

* fix(s3): address review feedback and CI failures for checksum support

- Gate GET/HEAD checksum response headers on x-amz-checksum-mode: ENABLED
  per AWS S3 spec, fixing FlexibleChecksumError on ranged GETs and
  multipart copies
- Verify computed checksum against client-provided header value for
  non-chunked uploads, returning BadDigest on mismatch
- Add nil check for getCheckSumWriter to prevent panic
- Handle comma-separated values in X-Amz-Trailer header
- Use ordered slice instead of map for deterministic checksum header
  selection; extract shared mappings into package-level vars

* fix(s3): skip checksum header for ranged GET responses

The stored checksum covers the full object. Returning it for ranged
(partial) responses causes SDK checksum validation failures because the
SDK validates the header value against the partial content received.

Skip emitting x-amz-checksum-* headers when a Range request header is
present, fixing PyArrow large file read failures.

* fix(s3): reject unsupported checksum algorithm with 400

detectRequestedChecksumAlgorithm now returns an error code when
x-amz-sdk-checksum-algorithm or x-amz-checksum-algorithm contains
an unsupported value, instead of silently ignoring it.

* feat(s3): compute composite checksum for multipart uploads

Store the checksum algorithm during CreateMultipartUpload, then during
CompleteMultipartUpload compute a composite checksum from per-part
checksums following the AWS S3 spec: concatenate raw per-part checksums,
hash with the same algorithm, format as "base64-N" where N is part count.

The composite checksum is persisted on the final object entry and
returned in HEAD/GET responses (gated on x-amz-checksum-mode: ENABLED).

Reuses existing per-part checksum storage from putToFiler and the
getCheckSumWriter/checksumHeaders infrastructure.

* fix(s3): validate checksum algorithm in CreateMultipartUpload, error on missing part checksums

- Move detectRequestedChecksumAlgorithm call before mkdir callback so
  an unsupported algorithm returns 400 before the upload is created
- Change computeCompositeChecksum to return an error when a part is
  missing its checksum (the upload was initiated with a checksum
  algorithm, so all parts must have checksums)
- Propagate the error as ErrInvalidPart in CompleteMultipartUpload

* fix(s3): return checksum header in CompleteMultipartUpload response, validate per-part algorithm

- Add ChecksumHeaderName/ChecksumValue fields to CompleteMultipartUploadResult
  and set the x-amz-checksum-* HTTP response header in the handler, matching
  the AWS S3 CompleteMultipartUpload response spec
- Validate that each part's stored checksum algorithm matches the upload's
  expected algorithm before assembling the composite checksum; return an
  error if a part was uploaded with a different algorithm
2026-04-03 18:37:54 -07:00
Chris LuandGitHub 36f37b9b6a fix(filer): remove cancellation guard from RollbackTransaction and clean up #8909 (#8916)
* fix(filer): remove cancellation guard from RollbackTransaction and clean up #8909

RollbackTransaction is a cleanup operation that must succeed even when
the context is cancelled — guarding it causes the exact orphaned state
that #8909 was trying to prevent.

Also:
- Use single-evaluation `if err := ctx.Err(); err != nil` pattern
  instead of double-calling ctx.Err()
- Remove spurious blank lines before guards
- Add context.DeadlineExceeded test coverage
- Simplify tests from ~230 lines to ~130 lines

* fix(filer): call cancel() in expiredCtx and test rollback with expired context

- Call cancel() instead of suppressing it to avoid leaking timer resources
- Test RollbackTransaction with both cancelled and expired contexts
2026-04-03 17:55:27 -07:00
os-pradipbabarandGitHub d5128f00f1 fix: Prevent orphaned metadata from cancelled S3 operations (Issue #8908) (#8909)
fix(filer): check if context was already cancelled before ignoring cancellation
2026-04-03 16:22:46 -07:00
40 changed files with 2979 additions and 149 deletions
@@ -93,7 +93,7 @@ jobs:
- name: Build Go weed binary
run: |
cd weed
go build -o weed .
go build -tags 5BytesOffset -o weed .
chmod +x weed
./weed version
@@ -169,7 +169,7 @@ jobs:
- name: Build Go weed binary
run: |
cd weed
go build -o weed .
go build -tags 5BytesOffset -o weed .
chmod +x weed
./weed version
@@ -0,0 +1,53 @@
{
"identities": [
{
"name": "admin",
"credentials": [
{
"accessKey": "AKIAIOSFODNN7EXAMPLE",
"secretKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
}
],
"actions": [
"Admin",
"Read",
"List",
"Tagging",
"Write"
]
},
{
"name": "steward",
"credentials": [
{
"accessKey": "steward-key",
"secretKey": "steward-secret"
}
],
"actions": [
"Read",
"List",
"Write"
]
},
{
"name": "le001",
"credentials": [
{
"accessKey": "le001-key",
"secretKey": "le001-secret"
}
],
"actions": [
"Read",
"List"
]
},
{
"name": "anonymous",
"actions": [
"Read"
]
}
]
}
+1 -1
View File
@@ -20,7 +20,7 @@ default = ["5bytes"]
[dependencies]
# Async runtime
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"
tokio-stream = { version = "0.1", features = ["net"] }
tokio-io-timeout = "1"
# gRPC + protobuf
+68 -40
View File
@@ -600,35 +600,35 @@ async fn run(
})
};
// Bind the gRPC listener before spawning to propagate bind errors at startup.
let grpc_listener = tokio::net::TcpListener::bind(&grpc_addr)
.await
.unwrap_or_else(|e| panic!("Failed to bind gRPC to {}: {}", grpc_addr, e));
let grpc_local_addr = grpc_listener
.local_addr()
.unwrap_or_else(|e| panic!("Failed to get gRPC local addr: {}", e));
let grpc_handle = {
let grpc_state = state.clone();
let grpc_addr = grpc_addr.clone();
let grpc_tls_acceptor = grpc_tls_acceptor.clone();
let mut shutdown_rx = shutdown_tx.subscribe();
let shutdown_tx_grpc = shutdown_tx.clone();
tokio::spawn(async move {
let addr = tokio::net::lookup_host(&grpc_addr)
.await
.expect("Failed to resolve gRPC address")
.next()
.expect("No addresses found for gRPC bind address");
let grpc_service = VolumeGrpcService {
state: grpc_state.clone(),
};
if let Some(tls_acceptor) = grpc_tls_acceptor {
let listener = tokio::net::TcpListener::bind(&grpc_addr)
.await
.unwrap_or_else(|e| panic!("Failed to bind gRPC to {}: {}", grpc_addr, e));
let incoming = grpc_tls_incoming(listener, tls_acceptor);
let reflection_v1 = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1()
.expect("Failed to build gRPC reflection v1 service");
let reflection_v1alpha = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1alpha()
.expect("Failed to build gRPC reflection v1alpha service");
info!("gRPC server listening on {} (TLS enabled)", addr);
if let Err(e) = build_grpc_server_builder()
let reflection_v1 = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1()
.expect("Failed to build gRPC reflection v1 service");
let reflection_v1alpha = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1alpha()
.expect("Failed to build gRPC reflection v1alpha service");
let result = if let Some(tls_acceptor) = grpc_tls_acceptor {
let incoming = grpc_tls_incoming(grpc_listener, tls_acceptor);
info!("gRPC server listening on {} (TLS enabled)", grpc_local_addr);
build_grpc_server_builder()
.layer(GrpcRequestIdLayer)
.add_service(reflection_v1)
.add_service(reflection_v1alpha)
@@ -637,32 +637,25 @@ async fn run(
let _ = shutdown_rx.recv().await;
})
.await
{
error!("gRPC server error: {}", e);
}
} else {
let reflection_v1 = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1()
.expect("Failed to build gRPC reflection v1 service");
let reflection_v1alpha = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
.build_v1alpha()
.expect("Failed to build gRPC reflection v1alpha service");
info!("gRPC server listening on {}", addr);
if let Err(e) = build_grpc_server_builder()
let incoming =
tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
info!("gRPC server listening on {}", grpc_local_addr);
build_grpc_server_builder()
.layer(GrpcRequestIdLayer)
.add_service(reflection_v1)
.add_service(reflection_v1alpha)
.add_service(build_volume_grpc_service(grpc_service))
.serve_with_shutdown(addr, async move {
.serve_with_incoming_shutdown(incoming, async move {
let _ = shutdown_rx.recv().await;
})
.await
{
error!("gRPC server error: {}", e);
}
};
if let Err(ref e) = result {
error!("gRPC server error: {}", e);
let _ = shutdown_tx_grpc.send(());
}
result
})
};
@@ -771,9 +764,40 @@ async fn run(
}))
};
// Wait for all servers
// Wait for servers. Use select! with &mut so the losing handle is not
// dropped, then await it explicitly afterward.
let mut server_err: Option<String> = None;
let mut http_handle = http_handle;
let mut grpc_handle = grpc_handle;
let grpc_finished_first = tokio::select! {
_ = &mut http_handle => false,
_ = &mut grpc_handle => true,
};
// Inspect the gRPC result (already resolved if it finished first,
// otherwise await it now).
let grpc_result = if grpc_finished_first {
grpc_handle.await
} else {
// HTTP finished first; gRPC is still running. Await it.
grpc_handle.await
};
match grpc_result {
Ok(Ok(())) => {}
Ok(Err(e)) => {
let msg = format!("gRPC server exited with error: {}", e);
error!("{}", msg);
server_err = Some(msg);
// serve error already sent shutdown inside the task
}
Err(e) => {
let msg = format!("gRPC task panicked: {}", e);
error!("{}", msg);
server_err = Some(msg);
let _ = shutdown_tx.send(());
}
}
// Ensure the HTTP handle completes too.
let _ = http_handle.await;
let _ = grpc_handle.await;
if let Some(h) = public_handle {
let _ = h.await;
}
@@ -798,6 +822,10 @@ async fn run(
cpu_profile.finish().map_err(std::io::Error::other)?;
}
if let Some(err_msg) = server_err {
return Err(std::io::Error::other(err_msg).into());
}
info!("Volume server stopped.");
Ok(())
}
+33 -3
View File
@@ -1867,6 +1867,7 @@ impl VolumeServer for VolumeGrpcService {
{
let needle_header = resp.needle_header;
let mut needle_body = resp.needle_body;
let resp_version = resp.version;
if needle_header.is_empty() {
continue;
@@ -1891,8 +1892,36 @@ impl VolumeServer for VolumeGrpcService {
// Parse needle from header + body
let mut n = Needle::default();
n.read_header(&needle_header);
n.read_body_v2(&needle_body)
.map_err(|e| Status::internal(format!("parse needle body: {}", e)))?;
if n.size.0 < 0 {
return Err(Status::invalid_argument(format!(
"unexpected negative needle size {} for needle {}",
n.size.0, n.id.0
)));
} else if n.size.0 > 0 {
// Normal needle: parse the body fields (DataSize, Data, flags, etc.)
n.read_body_v2(&needle_body)
.map_err(|e| Status::internal(format!("parse needle body: {}", e)))?;
} else {
// Delete tombstone (size == 0): body is checksum + timestamp
// (V3) or checksum only (V2) + padding. Validate minimum
// footer length for the protocol version.
use crate::storage::types::{
NEEDLE_CHECKSUM_SIZE, TIMESTAMP_SIZE, VERSION_3, Version,
};
let version = Version(resp_version as u8);
let min_footer = if version >= VERSION_3 {
NEEDLE_CHECKSUM_SIZE + TIMESTAMP_SIZE
} else {
NEEDLE_CHECKSUM_SIZE
};
if needle_body.len() < min_footer {
return Err(Status::invalid_argument(format!(
"tombstone needle {} body too short: got {} bytes, need >= {} for version {}",
n.id.0, needle_body.len(), min_footer, resp_version
)));
}
}
// Write needle to local volume
let mut store = state.store.write().unwrap();
@@ -4047,11 +4076,12 @@ fn find_last_append_at_ns(idx_path: &str, dat_path: &str, version: u32) -> Optio
let mut header = [0u8; 16];
dat_file.read_exact(&mut header).ok()?;
let needle_size = i32::from_be_bytes([header[12], header[13], header[14], header[15]]);
if needle_size <= 0 {
if needle_size < 0 {
return None;
}
// Seek to tail: offset + 16 (header) + size -> checksum (4) + timestamp (8)
// For delete needles (size == 0), the tail is right after the header.
let tail_offset = actual_offset as u64 + 16 + needle_size as u64;
dat_file.seek(SeekFrom::Start(tail_offset)).ok()?;
@@ -0,0 +1,368 @@
package framework
import (
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
// MixedVolumeCluster wraps a Go master + a mix of Go and Rust volume servers
// for integration testing. Go servers occupy indices [0, goCount) and Rust
// servers occupy indices [goCount, goCount+rustCount).
type MixedVolumeCluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string // Go weed binary (master + Go volume servers)
rustVolumeBinary string // Rust volume binary
baseDir string
configDir string
logsDir string
keepLogs bool
masterPort int
masterGrpcPort int
volumePorts []int
volumeGrpcPorts []int
volumePubPorts []int
isRust []bool // which servers are Rust
masterCmd *exec.Cmd
volumeCmds []*exec.Cmd
cleanupOnce sync.Once
}
// StartMixedVolumeCluster starts a cluster with 1 Go master, goCount Go volume
// servers, and rustCount Rust volume servers. Go servers come first in the index.
func StartMixedVolumeCluster(t testing.TB, profile matrix.Profile, goCount, rustCount int) *MixedVolumeCluster {
t.Helper()
if goCount < 0 || rustCount < 0 {
t.Fatalf("goCount and rustCount must be non-negative, got go=%d rust=%d", goCount, rustCount)
}
total := goCount + rustCount
if total < 2 {
t.Fatalf("need at least 2 volume servers, got %d", total)
}
weedBinary, err := FindOrBuildWeedBinary()
if err != nil {
t.Fatalf("resolve weed binary: %v", err)
}
// Only build the Rust binary when Rust servers are requested.
var rustBinary string
if rustCount > 0 {
rustBinary, err = FindOrBuildRustBinary()
if err != nil {
t.Skipf("skipping mixed cluster test: rust binary unavailable: %v", err)
}
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
dirs := []string{configDir, logsDir, masterDataDir}
for i := 0; i < total; i++ {
dirs = append(dirs, filepath.Join(baseDir, fmt.Sprintf("volume%d", i)))
}
for _, dir := range dirs {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
masterPort, masterGrpcPort, err := allocateMasterPortPair()
if err != nil {
t.Fatalf("allocate master port pair: %v", err)
}
// 2 ports per server (admin, grpc); add 1 more when public port is split out.
portsPerServer := 2
if profile.SplitPublicPort {
portsPerServer = 3
}
ports, err := allocatePorts(total * portsPerServer)
if err != nil {
t.Fatalf("allocate volume ports: %v", err)
}
isRust := make([]bool, total)
for i := goCount; i < total; i++ {
isRust[i] = true
}
c := &MixedVolumeCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
rustVolumeBinary: rustBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
keepLogs: keepLogs,
masterPort: masterPort,
masterGrpcPort: masterGrpcPort,
volumePorts: make([]int, total),
volumeGrpcPorts: make([]int, total),
volumePubPorts: make([]int, total),
isRust: isRust,
volumeCmds: make([]*exec.Cmd, total),
}
for i := 0; i < total; i++ {
baseIdx := i * portsPerServer
c.volumePorts[i] = ports[baseIdx]
c.volumeGrpcPorts[i] = ports[baseIdx+1]
if profile.SplitPublicPort {
c.volumePubPorts[i] = ports[baseIdx+2]
} else {
c.volumePubPorts[i] = c.volumePorts[i] // reuse admin port
}
}
// Start master
if err = c.startMaster(masterDataDir); err != nil {
c.Stop()
t.Fatalf("start master: %v", err)
}
helper := &Cluster{logsDir: logsDir}
if err = helper.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
masterLog := helper.tailLog("master.log")
c.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
// Start volume servers
for i := 0; i < total; i++ {
volumeDataDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
if isRust[i] {
err = c.startRustVolume(i, volumeDataDir)
} else {
err = c.startGoVolume(i, volumeDataDir)
}
if err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("start volume server %d (rust=%v): %v\nlog tail:\n%s", i, isRust[i], err, logTail)
}
// Rust uses /healthz, Go uses /status
healthURL := c.VolumeAdminURL(i) + "/status"
if isRust[i] {
healthURL = c.VolumeAdminURL(i) + "/healthz"
}
if err = helper.waitForHTTP(healthURL); err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("wait for volume server %d readiness: %v\nlog tail:\n%s", i, err, logTail)
}
if err = helper.waitForTCP(c.VolumeGRPCAddress(i)); err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("wait for volume server %d grpc readiness: %v\nlog tail:\n%s", i, err, logTail)
}
}
t.Cleanup(func() {
c.Stop()
})
return c
}
func (c *MixedVolumeCluster) Stop() {
if c == nil {
return
}
c.cleanupOnce.Do(func() {
for i := len(c.volumeCmds) - 1; i >= 0; i-- {
stopProcess(c.volumeCmds[i])
}
stopProcess(c.masterCmd)
if !c.keepLogs && !c.testingTB.Failed() {
_ = os.RemoveAll(c.baseDir)
} else if c.baseDir != "" {
c.testingTB.Logf("mixed volume server integration logs kept at %s", c.baseDir)
}
})
}
func (c *MixedVolumeCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.masterPort),
"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
c.masterCmd = exec.Command(c.weedBinary, args...)
c.masterCmd.Dir = c.baseDir
c.masterCmd.Stdout = logFile
c.masterCmd.Stderr = logFile
if err = c.masterCmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
return nil
}
func (c *MixedVolumeCluster) startGoVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"volume",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.volumePorts[index]),
"-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]),
"-port.public=" + strconv.Itoa(c.volumePubPorts[index]),
"-dir=" + dataDir,
"-max=16",
"-master=127.0.0.1:" + strconv.Itoa(c.masterPort),
"-readMode=" + c.profile.ReadMode,
"-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB),
"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
}
if c.profile.InflightUploadTimeout > 0 {
args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
}
if c.profile.InflightDownloadTimeout > 0 {
args = append(args, "-inflightDownloadDataTimeout="+c.profile.InflightDownloadTimeout.String())
}
cmd := exec.Command(c.weedBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
c.volumeCmds[index] = cmd
return nil
}
func (c *MixedVolumeCluster) startRustVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
args := rustVolumeArgs(
c.profile,
c.configDir,
c.masterPort,
c.volumePorts[index],
c.volumeGrpcPorts[index],
c.volumePubPorts[index],
dataDir,
)
cmd := exec.Command(c.rustVolumeBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
c.volumeCmds[index] = cmd
return nil
}
// --- accessor methods (mirror MultiVolumeCluster) ---
func (c *MixedVolumeCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
}
func (c *MixedVolumeCluster) MasterURL() string {
return "http://" + c.MasterAddress()
}
func (c *MixedVolumeCluster) VolumeAdminAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePorts[index]))
}
func (c *MixedVolumeCluster) VolumePublicAddress(index int) string {
if index < 0 || index >= len(c.volumePubPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPorts[index]))
}
func (c *MixedVolumeCluster) VolumeGRPCAddress(index int) string {
if index < 0 || index >= len(c.volumeGrpcPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPorts[index]))
}
func (c *MixedVolumeCluster) VolumeAdminURL(index int) string {
return "http://" + c.VolumeAdminAddress(index)
}
func (c *MixedVolumeCluster) VolumePublicURL(index int) string {
return "http://" + c.VolumePublicAddress(index)
}
func (c *MixedVolumeCluster) BaseDir() string {
return c.baseDir
}
// VolumeServerAddress returns SeaweedFS server address format: ip:httpPort.grpcPort
func (c *MixedVolumeCluster) VolumeServerAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return fmt.Sprintf("%s.%d", c.VolumeAdminAddress(index), c.volumeGrpcPorts[index])
}
func (c *MixedVolumeCluster) IsRust(index int) bool {
if index < 0 || index >= len(c.isRust) {
return false
}
return c.isRust[index]
}
@@ -0,0 +1,567 @@
package volume_server_grpc_test
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
)
// deleteAndWaitForTombstone issues an HTTP DELETE for the given fid on
// volumeURL, asserts a successful response, then polls with GET until
// the file returns 404 (tombstone visible) or the timeout elapses.
func deleteAndWaitForTombstone(t *testing.T, httpClient *http.Client, volumeURL, fid string) {
t.Helper()
req, err := http.NewRequest(http.MethodDelete, fmt.Sprintf("%s/%s", volumeURL, fid), nil)
if err != nil {
t.Fatalf("build delete request for %s: %v", fid, err)
}
resp := framework.DoRequest(t, httpClient, req)
framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusOK {
t.Fatalf("delete %s: expected 200 or 202, got %d", fid, resp.StatusCode)
}
// Poll until GET returns 404 (tombstone flushed to disk).
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
getResp := framework.ReadBytes(t, httpClient, volumeURL, fid)
status := getResp.StatusCode
framework.ReadAllAndClose(t, getResp)
if status == http.StatusNotFound {
return
}
time.Sleep(50 * time.Millisecond)
}
t.Fatalf("delete %s: tombstone not visible after 5s", fid)
}
// TestMixedBalanceCopyGoToRust verifies that VolumeCopy works from a Go volume
// server to a Rust volume server. This is the core operation behind volume
// balancing in a mixed Go+Rust cluster.
func TestMixedBalanceCopyGoToRust(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
// server 0 = Go, server 1 = Rust
conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
defer conn0.Close()
conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
defer conn1.Close()
httpClient := framework.NewHTTPClient()
const volumeID = uint32(50)
// Allocate volume on Go server and upload test data
framework.AllocateVolume(t, goClient, volumeID, "")
testFiles := []struct {
key uint64
cookie uint32
data []byte
}{
{1, 0xAABBCCDD, []byte("hello from Go server")},
{2, 0x11223344, []byte("second file for balance test")},
{3, 0xDEADBEEF, make([]byte, 4096)}, // larger file
}
for _, f := range testFiles {
fid := framework.NewFileID(volumeID, f.key, f.cookie)
resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, f.data)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("upload %s: expected 201, got %d: %s", fid, resp.StatusCode, body)
}
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Read source volume status before copy
sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus on Go server: %v", err)
}
t.Logf("Source: dat=%d idx=%d files=%d version=%d",
sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
sourceStatus.GetFileCount(), sourceStatus.GetVersion())
// Copy volume from Go (server 0) to Rust (server 1)
copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
VolumeId: volumeID,
SourceDataNode: cluster.VolumeServerAddress(0),
})
if err != nil {
t.Fatalf("VolumeCopy call failed: %v", err)
}
var lastAppendAtNs uint64
for {
resp, recvErr := copyStream.Recv()
if recvErr != nil {
if recvErr != io.EOF {
t.Fatalf("VolumeCopy recv error: %v", recvErr)
}
break
}
if resp.GetLastAppendAtNs() != 0 {
lastAppendAtNs = resp.GetLastAppendAtNs()
}
}
t.Logf("VolumeCopy completed, lastAppendAtNs=%d", lastAppendAtNs)
// Verify: read volume status from Rust server
targetStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus on Rust server: %v", err)
}
t.Logf("Target: dat=%d idx=%d files=%d version=%d",
targetStatus.GetDatFileSize(), targetStatus.GetIdxFileSize(),
targetStatus.GetFileCount(), targetStatus.GetVersion())
if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
t.Fatalf("dat file size mismatch: source=%d target=%d",
sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
}
if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
t.Fatalf("idx file size mismatch: source=%d target=%d",
sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
}
if sourceStatus.GetFileCount() != targetStatus.GetFileCount() {
t.Fatalf("file count mismatch: source=%d target=%d",
sourceStatus.GetFileCount(), targetStatus.GetFileCount())
}
// Verify data can be read from Rust server
for _, f := range testFiles {
fid := framework.NewFileID(volumeID, f.key, f.cookie)
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusOK {
t.Fatalf("read %s from Rust server: expected 200, got %d", fid, resp.StatusCode)
}
if !bytes.Equal(body, f.data) {
t.Fatalf("read %s from Rust server: content mismatch (got %d bytes, want %d)", fid, len(body), len(f.data))
}
}
}
// TestMixedBalanceCopyRustToGo verifies that VolumeCopy works from a Rust
// volume server to a Go volume server.
func TestMixedBalanceCopyRustToGo(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
defer conn0.Close()
conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
defer conn1.Close()
httpClient := framework.NewHTTPClient()
const volumeID = uint32(51)
// Allocate volume on Rust server and upload test data
framework.AllocateVolume(t, rustClient, volumeID, "")
testFiles := []struct {
key uint64
cookie uint32
data []byte
}{
{1, 0xAABBCCDD, []byte("hello from Rust server")},
{2, 0x11223344, []byte("second file for reverse balance")},
}
for _, f := range testFiles {
fid := framework.NewFileID(volumeID, f.key, f.cookie)
resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid, f.data)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("upload %s: expected 201, got %d: %s", fid, resp.StatusCode, body)
}
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
sourceStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus on Rust server: %v", err)
}
// Copy volume from Rust (server 1) to Go (server 0)
copyStream, err := goClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
VolumeId: volumeID,
SourceDataNode: cluster.VolumeServerAddress(1),
})
if err != nil {
t.Fatalf("VolumeCopy call failed: %v", err)
}
for {
_, recvErr := copyStream.Recv()
if recvErr != nil {
if recvErr != io.EOF {
t.Fatalf("VolumeCopy recv error: %v", recvErr)
}
break
}
}
targetStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus on Go server: %v", err)
}
if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
t.Fatalf("dat file size mismatch: source=%d target=%d",
sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
}
if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
t.Fatalf("idx file size mismatch: source=%d target=%d",
sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
}
if sourceStatus.GetFileCount() != targetStatus.GetFileCount() {
t.Fatalf("file count mismatch: source=%d target=%d",
sourceStatus.GetFileCount(), targetStatus.GetFileCount())
}
// Verify data can be read from Go server
for _, f := range testFiles {
fid := framework.NewFileID(volumeID, f.key, f.cookie)
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusOK {
t.Fatalf("read %s from Go server: expected 200, got %d", fid, resp.StatusCode)
}
if !bytes.Equal(body, f.data) {
t.Fatalf("read %s from Go server: content mismatch (got %d bytes, want %d)", fid, len(body), len(f.data))
}
}
}
// TestMixedBalanceCopyWithDeletes verifies that VolumeCopy correctly handles
// volumes that have both active and deleted needles.
func TestMixedBalanceCopyWithDeletes(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
defer conn0.Close()
conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
defer conn1.Close()
httpClient := framework.NewHTTPClient()
const volumeID = uint32(52)
framework.AllocateVolume(t, goClient, volumeID, "")
// Upload files
for i := uint64(1); i <= 5; i++ {
fid := framework.NewFileID(volumeID, i, 0x12340000+uint32(i))
resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, []byte(fmt.Sprintf("file-%d", i)))
framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("upload file %d: expected 201, got %d", i, resp.StatusCode)
}
}
// Delete some files and wait for tombstones to be visible
for _, key := range []uint64{2, 4} {
fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
deleteAndWaitForTombstone(t, httpClient, cluster.VolumeAdminURL(0), fid)
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus: %v", err)
}
t.Logf("Source after deletes: dat=%d idx=%d files=%d",
sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
sourceStatus.GetFileCount())
// Copy volume from Go to Rust
copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
VolumeId: volumeID,
SourceDataNode: cluster.VolumeServerAddress(0),
})
if err != nil {
t.Fatalf("VolumeCopy: %v", err)
}
var lastAppendAtNs uint64
for {
resp, recvErr := copyStream.Recv()
if recvErr != nil {
if recvErr != io.EOF {
t.Fatalf("VolumeCopy recv error: %v", recvErr)
}
break
}
if resp.GetLastAppendAtNs() != 0 {
lastAppendAtNs = resp.GetLastAppendAtNs()
}
}
if lastAppendAtNs == 0 {
t.Fatalf("VolumeCopy did not return a lastAppendAtNs timestamp")
}
t.Logf("VolumeCopy completed, lastAppendAtNs=%d", lastAppendAtNs)
targetStatusAfterCopy, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus on Rust: %v", err)
}
t.Logf("Target after copy: dat=%d idx=%d files=%d",
targetStatusAfterCopy.GetDatFileSize(), targetStatusAfterCopy.GetIdxFileSize(),
targetStatusAfterCopy.GetFileCount())
if sourceStatus.GetDatFileSize() != targetStatusAfterCopy.GetDatFileSize() {
t.Fatalf("dat file size mismatch: source=%d target=%d",
sourceStatus.GetDatFileSize(), targetStatusAfterCopy.GetDatFileSize())
}
if sourceStatus.GetIdxFileSize() != targetStatusAfterCopy.GetIdxFileSize() {
t.Fatalf("idx file size mismatch: source=%d target=%d",
sourceStatus.GetIdxFileSize(), targetStatusAfterCopy.GetIdxFileSize())
}
// Tail from the copy checkpoint — source is unchanged (deletes happened
// before copy), so tailing should not append any data.
_, err = rustClient.VolumeTailReceiver(ctx, &volume_server_pb.VolumeTailReceiverRequest{
VolumeId: volumeID,
SinceNs: lastAppendAtNs,
IdleTimeoutSeconds: 3,
SourceVolumeServer: cluster.VolumeServerAddress(0),
})
if err != nil {
t.Fatalf("VolumeTailReceiver: %v", err)
}
targetStatusAfterTail, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("ReadVolumeFileStatus after tail: %v", err)
}
if targetStatusAfterTail.GetDatFileSize() != targetStatusAfterCopy.GetDatFileSize() {
t.Fatalf("dat grew after tail: before=%d after=%d",
targetStatusAfterCopy.GetDatFileSize(), targetStatusAfterTail.GetDatFileSize())
}
if targetStatusAfterTail.GetIdxFileSize() != targetStatusAfterCopy.GetIdxFileSize() {
t.Fatalf("idx grew after tail: before=%d after=%d",
targetStatusAfterCopy.GetIdxFileSize(), targetStatusAfterTail.GetIdxFileSize())
}
// Verify surviving files are readable from Rust
for _, key := range []uint64{1, 3, 5} {
fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusOK {
t.Fatalf("read surviving file %d from Rust: expected 200, got %d", key, resp.StatusCode)
}
expected := fmt.Sprintf("file-%d", key)
if string(body) != expected {
t.Fatalf("data mismatch for file %d: got %q want %q", key, body, expected)
}
}
// Verify deleted files return 404
for _, key := range []uint64{2, 4} {
fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusNotFound {
t.Fatalf("read deleted file %d from Rust: expected 404, got %d", key, resp.StatusCode)
}
}
}
// TestMixedBalanceFullMoveGoToRust exercises the complete volume balance move
// flow: mark readonly → copy → tail → verify sizes → delete source.
// This mirrors the steps in balance_task.go Execute().
func TestMixedBalanceFullMoveGoToRust(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
defer conn0.Close()
conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
defer conn1.Close()
httpClient := framework.NewHTTPClient()
const volumeID = uint32(53)
framework.AllocateVolume(t, goClient, volumeID, "")
// Upload test data
for i := uint64(1); i <= 5; i++ {
fid := framework.NewFileID(volumeID, i, 0xABCD0000+uint32(i))
resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, []byte(fmt.Sprintf("balance-move-file-%d", i)))
framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("upload file %d: expected 201, got %d", i, resp.StatusCode)
}
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
// Step 1: Copy volume to Rust server (source stays writable so we can
// delete after copy to exercise the tail tombstone path)
copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
VolumeId: volumeID,
SourceDataNode: cluster.VolumeServerAddress(0),
})
if err != nil {
t.Fatalf("VolumeCopy: %v", err)
}
var lastAppendAtNs uint64
for {
resp, recvErr := copyStream.Recv()
if recvErr != nil {
if recvErr != io.EOF {
t.Fatalf("VolumeCopy recv: %v", recvErr)
}
break
}
if resp.GetLastAppendAtNs() != 0 {
lastAppendAtNs = resp.GetLastAppendAtNs()
}
}
t.Logf("Copy done, lastAppendAtNs=%d", lastAppendAtNs)
// Step 2: Delete file 3 on the source AFTER copy. This creates a
// tombstone needle that the tail step must propagate to the Rust server.
deleteAndWaitForTombstone(t, httpClient, cluster.VolumeAdminURL(0),
framework.NewFileID(volumeID, 3, 0xABCD0003))
// Step 3: Mark source readonly so no further writes arrive during tail
_, err = goClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("mark readonly: %v", err)
}
// Read source status (the reference for post-tail verification)
sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("read source status: %v", err)
}
t.Logf("Source: dat=%d idx=%d files=%d",
sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
sourceStatus.GetFileCount())
// Step 4: Tail for updates — this must pick up the delete tombstone
_, err = rustClient.VolumeTailReceiver(ctx, &volume_server_pb.VolumeTailReceiverRequest{
VolumeId: volumeID,
SinceNs: lastAppendAtNs,
IdleTimeoutSeconds: 5,
SourceVolumeServer: cluster.VolumeServerAddress(0),
})
if err != nil {
t.Fatalf("VolumeTailReceiver: %v", err)
}
// Step 5: Verify dat/idx sizes match after tail.
// We compare file sizes (byte-level correctness) rather than file_count
// because the tail writes the tombstone via the write path (which
// increments file_count) while the source used the delete path (which
// only increments deletion_count). This is consistent with Go behavior.
targetStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("read target status: %v", err)
}
t.Logf("Target: dat=%d idx=%d files=%d",
targetStatus.GetDatFileSize(), targetStatus.GetIdxFileSize(),
targetStatus.GetFileCount())
if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
t.Fatalf("dat size mismatch after tail: source=%d target=%d",
sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
}
if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
t.Fatalf("idx size mismatch after tail: source=%d target=%d",
sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
}
// Step 6: Delete volume from source
_, err = goClient.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{
VolumeId: volumeID,
})
if err != nil {
t.Fatalf("delete source volume: %v", err)
}
// Verify source volume is gone
_, err = goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
VolumeId: volumeID,
})
if err == nil {
t.Fatalf("expected error reading deleted source volume, got nil")
}
t.Logf("Source volume deleted successfully (error as expected: %v)", err)
// Verify all surviving data is readable from Rust (the new home)
for _, key := range []uint64{1, 2, 4, 5} {
fid := framework.NewFileID(volumeID, key, 0xABCD0000+uint32(key))
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
body := framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusOK {
t.Fatalf("read file %d from Rust after move: expected 200, got %d", key, resp.StatusCode)
}
expected := fmt.Sprintf("balance-move-file-%d", key)
if string(body) != expected {
t.Fatalf("data mismatch for file %d after move: got %q want %q", key, body, expected)
}
}
// Verify deleted file is 404 on Rust (tombstone propagated via tail)
{
fid := framework.NewFileID(volumeID, 3, 0xABCD0003)
resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
framework.ReadAllAndClose(t, resp)
if resp.StatusCode != http.StatusNotFound {
t.Fatalf("deleted file 3 on Rust: expected 404, got %d", resp.StatusCode)
}
}
t.Logf("Full balance move completed: volume %d moved from Go to Rust, source purged", volumeID)
}
+1
View File
@@ -45,6 +45,7 @@ type ObjectStoreUser struct {
SecretKey string `json:"secret_key"`
Permissions []string `json:"permissions"`
PolicyNames []string `json:"policy_names"`
IsStatic bool `json:"is_static"` // loaded from static config file, not editable
}
type ObjectStoreUsersData struct {
+19
View File
@@ -867,6 +867,24 @@ func (s *AdminServer) DeleteS3Bucket(bucketName string) error {
})
}
// IsStaticUser checks if a user is a static identity by loading the
// configuration from the credential manager and checking the IsStatic flag.
func (s *AdminServer) IsStaticUser(username string) bool {
if s.credentialManager == nil {
return false
}
s3cfg, err := s.credentialManager.LoadConfiguration(context.Background())
if err != nil {
return false
}
for _, ident := range s3cfg.Identities {
if ident.Name == username {
return ident.IsStatic
}
}
return false
}
// GetObjectStoreUsers retrieves object store users from identity.json
func (s *AdminServer) GetObjectStoreUsers(ctx context.Context) ([]ObjectStoreUser, error) {
if s.credentialManager == nil {
@@ -890,6 +908,7 @@ func (s *AdminServer) GetObjectStoreUsers(ctx context.Context) ([]ObjectStoreUse
user := ObjectStoreUser{
Username: identity.Name,
Permissions: identity.Actions,
IsStatic: identity.IsStatic,
}
// Set email from account if available
+1 -1
View File
@@ -175,7 +175,7 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
ctx := context.Background()
// Get user using credential manager
// Get user using credential manager (resolves static users via filer gRPC)
identity, err := s.credentialManager.GetUser(ctx, username)
if err != nil {
if err == credential.ErrUserNotFound {
+10
View File
@@ -93,6 +93,11 @@ func (h *UserHandlers) UpdateUser(w http.ResponseWriter, r *http.Request) {
return
}
if h.adminServer.IsStaticUser(username) {
writeJSONError(w, http.StatusForbidden, "Cannot modify static user "+username+" (loaded from config file)")
return
}
var req dash.UpdateUserRequest
if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
@@ -120,6 +125,11 @@ func (h *UserHandlers) DeleteUser(w http.ResponseWriter, r *http.Request) {
return
}
if h.adminServer.IsStaticUser(username) {
writeJSONError(w, http.StatusForbidden, "Cannot delete static user "+username+" (loaded from config file)")
return
}
err := h.adminServer.DeleteObjectStoreUser(username)
if err != nil {
glog.Errorf("Failed to delete user %s: %v", username, err)
+17 -10
View File
@@ -125,6 +125,9 @@ templ ObjectStoreUsers(data dash.ObjectStoreUsersData) {
<div class="d-flex align-items-center">
<i class="fas fa-user me-2 text-muted"></i>
<strong>{user.Username}</strong>
if user.IsStatic {
<span class="badge bg-secondary ms-2" title="Loaded from config file (read-only)">static</span>
}
</div>
</td>
<td>{user.Email}</td>
@@ -133,24 +136,28 @@ templ ObjectStoreUsers(data dash.ObjectStoreUsersData) {
</td>
<td>
<div class="btn-group btn-group-sm" role="group">
<button type="button" class="btn btn-outline-info"
<button type="button" class="btn btn-outline-info"
data-action="show-user-details" data-username={ user.Username }>
<i class="fas fa-info-circle"></i>
</button>
<button type="button" class="btn btn-outline-primary"
data-action="edit-user" data-username={ user.Username }>
<i class="fas fa-edit"></i>
</button>
if user.Username != "anonymous" {
if !user.IsStatic {
<button type="button" class="btn btn-outline-primary"
data-action="edit-user" data-username={ user.Username }>
<i class="fas fa-edit"></i>
</button>
}
if user.Username != "anonymous" && !user.IsStatic {
<button type="button" class="btn btn-outline-secondary"
data-action="manage-access-keys" data-username={ user.Username }>
<i class="fas fa-key"></i>
</button>
}
<button type="button" class="btn btn-outline-danger"
data-action="delete-user" data-username={ user.Username }>
<i class="fas fa-trash"></i>
</button>
if !user.IsStatic {
<button type="button" class="btn btn-outline-danger"
data-action="delete-user" data-username={ user.Username }>
<i class="fas fa-trash"></i>
</button>
}
</div>
</td>
</tr>
File diff suppressed because one or more lines are too long
+25
View File
@@ -30,6 +30,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/security"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/grace"
)
var (
@@ -48,6 +49,10 @@ type AdminOptions struct {
dataDir *string
icebergPort *int
urlPrefix *string
debug *bool
debugPort *int
cpuProfile *string
memProfile *string
}
func init() {
@@ -64,6 +69,10 @@ func init() {
a.readOnlyPassword = cmdAdmin.Flag.String("readOnlyPassword", "", "read-only user password (optional, for view-only access; requires adminPassword to be set)")
a.icebergPort = cmdAdmin.Flag.Int("iceberg.port", 8181, "Iceberg REST Catalog port (0 to hide in UI)")
a.urlPrefix = cmdAdmin.Flag.String("urlPrefix", "", "URL path prefix when running behind a reverse proxy under a subdirectory (e.g. /seaweedfs)")
a.debug = cmdAdmin.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
a.debugPort = cmdAdmin.Flag.Int("debug.port", 6060, "http port for debugging")
a.cpuProfile = cmdAdmin.Flag.String("cpuprofile", "", "cpu profile output file")
a.memProfile = cmdAdmin.Flag.String("memprofile", "", "memory profile output file")
}
var cmdAdmin = &Command{
@@ -140,6 +149,16 @@ var cmdAdmin = &Command{
- All static assets, API endpoints, and navigation links will use the prefix
- Session cookies are scoped to the prefix path
Debugging and Profiling:
- Use -debug to start a pprof HTTP server for live profiling (localhost only)
- Set -debug.port to choose the pprof port (default 6060)
- Profiles are accessible at http://127.0.0.1:<debug.port>/debug/pprof/
- Use -cpuprofile and -memprofile to write profiles to files on shutdown
- WARNING: -debug exposes runtime internals; use only in trusted environments
- Examples:
weed admin -debug -debug.port=6060 -master="localhost:9333"
weed admin -cpuprofile=cpu.prof -memprofile=mem.prof -master="localhost:9333"
Configuration File:
- The security.toml file is read from ".", "$HOME/.seaweedfs/",
"/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order
@@ -149,6 +168,12 @@ var cmdAdmin = &Command{
}
func runAdmin(cmd *Command, args []string) bool {
if *a.debug {
grace.StartDebugServer(*a.debugPort)
}
grace.SetupProfiling(*a.cpuProfile, *a.memProfile)
// Load security configuration
util.LoadSecurityConfiguration()
+10
View File
@@ -81,6 +81,7 @@ type FilerOptions struct {
exposeDirectoryData *bool
tusBasePath *string
certProvider certprovider.Provider
s3ConfigFile *string // optional path to static S3 identity config
}
func init() {
@@ -343,6 +344,15 @@ func (fo *FilerOptions) startFiler() {
glog.V(0).Infof("Initialized credential manager: %s", credentialManager.GetStoreName())
}
// Load static S3 identities from config file if specified
if fo.s3ConfigFile != nil && *fo.s3ConfigFile != "" {
if credentialManager != nil {
if err := credentialManager.LoadS3ConfigFile(*fo.s3ConfigFile); err != nil {
glog.Warningf("Failed to load S3 config file for static identities: %v", err)
}
}
}
fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
Masters: fo.masters,
FilerGroup: *fo.filerGroup,
+4
View File
@@ -820,6 +820,10 @@ func runMini(cmd *Command, args []string) bool {
miniFilerOptions.disableHttp = miniDisableHttp
miniMasterOptions.disableHttp = miniDisableHttp
// Share the S3 static identity config file with the filer so its
// credential manager can also serve static users.
miniFilerOptions.s3ConfigFile = miniS3Config
filerAddress := string(pb.NewServerAddress(*miniIp, *miniFilerOptions.port, *miniFilerOptions.portGrpc))
miniS3Options.filer = &filerAddress
miniWebDavOptions.filer = &filerAddress
+2
View File
@@ -372,6 +372,8 @@ func runServer(cmd *Command, args []string) bool {
} else if *serverIamConfig != "" && *s3Options.iamConfig != *serverIamConfig {
glog.V(0).Infof("both -s3.iam.config(%s) and -iam.config(%s) provided; using -s3.iam.config", *s3Options.iamConfig, *serverIamConfig)
}
// Share the S3 static identity config file with the filer
filerOptions.s3ConfigFile = s3Options.config
go func() {
time.Sleep(2 * time.Second)
s3Options.localFilerSocket = filerOptions.localSocket
+127 -5
View File
@@ -3,14 +3,18 @@ package credential
import (
"context"
"fmt"
"os"
"strings"
"sync"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"google.golang.org/grpc"
"google.golang.org/protobuf/encoding/protojson"
)
// FilerAddressSetter is an interface for credential stores that need a dynamic filer address
@@ -21,6 +25,15 @@ type FilerAddressSetter interface {
// CredentialManager manages user credentials using a configurable store
type CredentialManager struct {
Store CredentialStore
// staticMu protects staticIdentities and staticNames, which are written
// by SetStaticIdentities (startup + config reload) and read concurrently
// by LoadConfiguration, SaveConfiguration, GetStaticUsernames, and IsStaticIdentity.
staticMu sync.RWMutex
// staticIdentities holds identities loaded from a static config file (-s3.config).
// These are included in LoadConfiguration so that listing operations
// return all configured identities, not just dynamic ones from the store.
staticIdentities []*iam_pb.Identity
staticNames map[string]bool
}
// NewCredentialManager creates a new credential manager with the specified store
@@ -74,13 +87,97 @@ func (cm *CredentialManager) GetStoreName() string {
return ""
}
// LoadConfiguration loads the S3 API configuration
func (cm *CredentialManager) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
return cm.Store.LoadConfiguration(ctx)
// SetStaticIdentities registers identities loaded from a static config file.
// These identities are included in LoadConfiguration and ListUsers results
// but are never persisted to the dynamic store.
func (cm *CredentialManager) SetStaticIdentities(identities []*iam_pb.Identity) {
filtered := make([]*iam_pb.Identity, 0, len(identities))
names := make(map[string]bool, len(identities))
for _, ident := range identities {
if ident != nil {
filtered = append(filtered, ident)
names[ident.Name] = true
}
}
cm.staticMu.Lock()
cm.staticIdentities = filtered
cm.staticNames = names
cm.staticMu.Unlock()
}
// SaveConfiguration saves the S3 API configuration
// IsStaticIdentity returns true if the named identity was loaded from static config.
func (cm *CredentialManager) IsStaticIdentity(name string) bool {
cm.staticMu.RLock()
defer cm.staticMu.RUnlock()
return cm.staticNames[name]
}
// GetStaticIdentity returns the protobuf identity for a static user, or nil.
func (cm *CredentialManager) GetStaticIdentity(name string) *iam_pb.Identity {
cm.staticMu.RLock()
defer cm.staticMu.RUnlock()
for _, ident := range cm.staticIdentities {
if ident.Name == name {
return ident
}
}
return nil
}
// GetStaticUsernames returns the names of all static identities.
func (cm *CredentialManager) GetStaticUsernames() []string {
cm.staticMu.RLock()
defer cm.staticMu.RUnlock()
names := make([]string, 0, len(cm.staticIdentities))
for _, ident := range cm.staticIdentities {
names = append(names, ident.Name)
}
return names
}
// LoadConfiguration loads the S3 API configuration from the store and merges
// in any static identities so that listing operations show all users.
func (cm *CredentialManager) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
config, err := cm.Store.LoadConfiguration(ctx)
if err != nil {
return config, err
}
// Merge static identities that are not already in the dynamic config
cm.staticMu.RLock()
staticIdents := cm.staticIdentities
cm.staticMu.RUnlock()
if len(staticIdents) > 0 {
dynamicNames := make(map[string]bool, len(config.Identities))
for _, ident := range config.Identities {
dynamicNames[ident.Name] = true
}
for _, si := range staticIdents {
if !dynamicNames[si.Name] {
config.Identities = append(config.Identities, si)
}
}
}
return config, nil
}
// SaveConfiguration saves the S3 API configuration.
// Static identities are filtered out before saving to the store.
// The caller's config is not mutated.
func (cm *CredentialManager) SaveConfiguration(ctx context.Context, config *iam_pb.S3ApiConfiguration) error {
cm.staticMu.RLock()
staticNames := cm.staticNames
cm.staticMu.RUnlock()
if len(staticNames) > 0 {
var dynamicOnly []*iam_pb.Identity
for _, ident := range config.Identities {
if !staticNames[ident.Name] {
dynamicOnly = append(dynamicOnly, ident)
}
}
configCopy := *config
configCopy.Identities = dynamicOnly
return cm.Store.SaveConfiguration(ctx, &configCopy)
}
return cm.Store.SaveConfiguration(ctx, config)
}
@@ -104,7 +201,12 @@ func (cm *CredentialManager) DeleteUser(ctx context.Context, username string) er
return cm.Store.DeleteUser(ctx, username)
}
// ListUsers returns all usernames
// ListUsers returns usernames from the dynamic store via cm.Store.ListUsers.
// On store error the error is returned directly without merging static entries.
// Static identities (cm.staticIdentities) are NOT included here because
// internal callers (e.g. DeletePolicy) look up each user in the store and
// would fail on non-existent static entries. External callers that need the
// full list should merge GetStaticUsernames separately.
func (cm *CredentialManager) ListUsers(ctx context.Context) ([]string, error) {
return cm.Store.ListUsers(ctx)
}
@@ -169,6 +271,26 @@ func (cm *CredentialManager) UpdatePolicy(ctx context.Context, name string, docu
return cm.Store.PutPolicy(ctx, name, document)
}
// LoadS3ConfigFile reads a static S3 identity config file and registers
// the identities so they appear in LoadConfiguration and listing results.
func (cm *CredentialManager) LoadS3ConfigFile(path string) error {
content, err := os.ReadFile(path)
if err != nil {
return fmt.Errorf("read %s: %w", path, err)
}
config := &iam_pb.S3ApiConfiguration{}
opts := protojson.UnmarshalOptions{DiscardUnknown: true, AllowPartial: true}
if err := opts.Unmarshal(content, config); err != nil {
return fmt.Errorf("parse %s: %w", path, err)
}
for _, ident := range config.Identities {
ident.IsStatic = true
}
cm.SetStaticIdentities(config.Identities)
glog.V(1).Infof("Loaded %d static identities from %s", len(config.Identities), path)
return nil
}
// Shutdown performs cleanup
func (cm *CredentialManager) Shutdown() {
if cm.Store != nil {
+30
View File
@@ -128,6 +128,9 @@ func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefi
}
func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(entry.FullPath)
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
@@ -155,6 +158,9 @@ func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) err
// InsertEntryKnownAbsent skips the pre-insert FindEntry path when the caller has
// already established that the target path does not exist.
func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry *Entry) error {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(entry.FullPath)
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
@@ -178,6 +184,9 @@ func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry
}
func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(entry.FullPath)
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc()
@@ -236,6 +245,9 @@ func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (
}
func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(fp)
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
@@ -264,6 +276,9 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath)
}
func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(existingEntry.FullPath)
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
@@ -288,6 +303,9 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry
}
func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
actualStore := fsw.getActualStore(fp + "/")
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
@@ -394,11 +412,17 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u
}
func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
ctx = context.WithoutCancel(ctx)
return fsw.getDefaultStore().BeginTransaction(ctx)
}
func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
return fsw.getDefaultStore().CommitTransaction(ctx)
}
@@ -413,6 +437,9 @@ func (fsw *FilerStoreWrapper) Shutdown() {
}
func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
return fsw.getDefaultStore().KvPut(ctx, key, value)
}
@@ -421,6 +448,9 @@ func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []by
return fsw.getDefaultStore().KvGet(ctx, key)
}
func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
if err := ctx.Err(); err != nil {
return err
}
ctx = context.WithoutCancel(ctx)
return fsw.getDefaultStore().KvDelete(ctx, key)
}
+136
View File
@@ -2,8 +2,10 @@ package filer
import (
"context"
"errors"
"os"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/stretchr/testify/assert"
@@ -69,3 +71,137 @@ func TestFilerStoreWrapperMimeNormalization(t *testing.T) {
}
}
}
// cancelledCtx returns a context that is already cancelled.
func cancelledCtx() context.Context {
ctx, cancel := context.WithCancel(context.Background())
cancel()
return ctx
}
// expiredCtx returns a context whose deadline has already passed.
func expiredCtx() context.Context {
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
cancel() // release resources immediately as it's already expired
return ctx
}
func TestFilerStoreWrapperWriteOpsRejectCancelledContext(t *testing.T) {
newEntry := func(path string) *Entry {
return &Entry{
FullPath: util.FullPath(path),
Attr: Attr{Mode: 0o660, Mime: "application/octet-stream"},
}
}
// Each write operation that should be guarded.
writeOps := []struct {
name string
run func(*FilerStoreWrapper, context.Context) error
}{
{"InsertEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error {
return fsw.InsertEntry(ctx, newEntry("/test/a"))
}},
{"InsertEntryKnownAbsent", func(fsw *FilerStoreWrapper, ctx context.Context) error {
return fsw.InsertEntryKnownAbsent(ctx, newEntry("/test/b"))
}},
{"UpdateEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error {
_ = fsw.InsertEntry(context.Background(), newEntry("/test/c"))
return fsw.UpdateEntry(ctx, newEntry("/test/c"))
}},
{"DeleteEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error {
_ = fsw.InsertEntry(context.Background(), newEntry("/test/d"))
return fsw.DeleteEntry(ctx, "/test/d")
}},
{"DeleteOneEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error {
e := newEntry("/test/e")
_ = fsw.InsertEntry(context.Background(), e)
return fsw.DeleteOneEntry(ctx, e)
}},
{"DeleteFolderChildren", func(fsw *FilerStoreWrapper, ctx context.Context) error {
_ = fsw.InsertEntry(context.Background(), newEntry("/test/folder/child"))
return fsw.DeleteFolderChildren(ctx, "/test/folder")
}},
{"BeginTransaction", func(fsw *FilerStoreWrapper, ctx context.Context) error {
_, err := fsw.BeginTransaction(ctx)
return err
}},
{"CommitTransaction", func(fsw *FilerStoreWrapper, ctx context.Context) error {
return fsw.CommitTransaction(ctx)
}},
{"KvPut", func(fsw *FilerStoreWrapper, ctx context.Context) error {
return fsw.KvPut(ctx, []byte("k"), []byte("v"))
}},
{"KvDelete", func(fsw *FilerStoreWrapper, ctx context.Context) error {
_ = fsw.KvPut(context.Background(), []byte("k"), []byte("v"))
return fsw.KvDelete(ctx, []byte("k"))
}},
}
badContexts := []struct {
name string
ctx context.Context
wantError error
}{
{"cancelled", cancelledCtx(), context.Canceled},
{"deadline exceeded", expiredCtx(), context.DeadlineExceeded},
}
for _, op := range writeOps {
for _, bc := range badContexts {
t.Run(op.name+"/"+bc.name, func(t *testing.T) {
wrapper := NewFilerStoreWrapper(newStubFilerStore())
err := op.run(wrapper, bc.ctx)
require.Error(t, err)
assert.True(t, errors.Is(err, bc.wantError), "got %v, want %v", err, bc.wantError)
})
}
}
}
func TestFilerStoreWrapperWriteOpsSucceedWithActiveContext(t *testing.T) {
wrapper := NewFilerStoreWrapper(newStubFilerStore())
ctx := context.Background()
entry := &Entry{
FullPath: util.FullPath("/test/obj"),
Attr: Attr{Mode: 0o660},
}
require.NoError(t, wrapper.InsertEntry(ctx, entry))
require.NoError(t, wrapper.UpdateEntry(ctx, entry))
require.NoError(t, wrapper.DeleteOneEntry(ctx, entry))
require.NoError(t, wrapper.InsertEntryKnownAbsent(ctx, entry))
require.NoError(t, wrapper.DeleteEntry(ctx, entry.FullPath))
require.NoError(t, wrapper.KvPut(ctx, []byte("k"), []byte("v")))
require.NoError(t, wrapper.KvDelete(ctx, []byte("k")))
txCtx, err := wrapper.BeginTransaction(ctx)
require.NoError(t, err)
require.NoError(t, wrapper.CommitTransaction(txCtx))
}
func TestFilerStoreWrapperReadOpsSucceedWithCancelledContext(t *testing.T) {
wrapper := NewFilerStoreWrapper(newStubFilerStore())
entry := &Entry{
FullPath: util.FullPath("/test/readable"),
Attr: Attr{Mode: 0o660},
}
require.NoError(t, wrapper.InsertEntry(context.Background(), entry))
require.NoError(t, wrapper.KvPut(context.Background(), []byte("rk"), []byte("rv")))
ctx := cancelledCtx()
_, err := wrapper.FindEntry(ctx, entry.FullPath)
assert.NoError(t, err)
_, err = wrapper.KvGet(ctx, []byte("rk"))
assert.NoError(t, err)
}
// RollbackTransaction must succeed even when the context is cancelled or
// expired, because it is a cleanup operation called after failures.
func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) {
wrapper := NewFilerStoreWrapper(newStubFilerStore())
assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx()))
assert.NoError(t, wrapper.RollbackTransaction(expiredCtx()))
}
+60
View File
@@ -245,6 +245,66 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien
}, nil
}
// PrepareStreamContentWithPrefetch is like PrepareStreamContentWithThrottler but uses
// concurrent chunk prefetching to overlap network I/O. When prefetchAhead > 1, fetch
// goroutines establish HTTP connections to volume servers ahead of time, streaming data
// through io.Pipe with minimal memory overhead.
//
// prefetchAhead controls the number of chunks fetched concurrently:
// - 0 or 1: falls back to sequential fetching (same as PrepareStreamContentWithThrottler)
// - 2+: uses pipe-based prefetch pipeline with that many concurrent fetches
func PrepareStreamContentWithPrefetch(ctx context.Context, masterClient wdclient.HasLookupFileIdFunction, jwtFunc VolumeServerJwtFunction, chunks []*filer_pb.FileChunk, offset int64, size int64, downloadMaxBytesPs int64, prefetchAhead int) (DoStreamContent, error) {
if prefetchAhead <= 1 {
return PrepareStreamContentWithThrottler(ctx, masterClient, jwtFunc, chunks, offset, size, downloadMaxBytesPs)
}
glog.V(4).InfofCtx(ctx, "prepare to stream content with prefetch=%d for chunks: %d", prefetchAhead, len(chunks))
chunkViews := ViewFromChunks(ctx, masterClient.GetLookupFileIdFunction(), chunks, offset, size)
fileId2Url := make(map[string][]string)
for x := chunkViews.Front(); x != nil; x = x.Next {
chunkView := x.Value
var urlStrings []string
var err error
for _, backoff := range getLookupFileIdBackoffSchedule {
if err := ctx.Err(); err != nil {
return nil, err
}
urlStrings, err = masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
if err == nil && len(urlStrings) > 0 {
break
}
if err := ctx.Err(); err != nil {
return nil, err
}
glog.V(4).InfofCtx(ctx, "waiting for chunk: %s", chunkView.FileId)
timer := time.NewTimer(backoff)
select {
case <-ctx.Done():
if !timer.Stop() {
<-timer.C
}
return nil, ctx.Err()
case <-timer.C:
}
}
if err != nil {
glog.V(1).InfofCtx(ctx, "operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
return nil, err
} else if len(urlStrings) == 0 {
errUrlNotFound := fmt.Errorf("operation LookupFileId %s failed, err: urls not found", chunkView.FileId)
glog.ErrorCtx(ctx, errUrlNotFound)
return nil, errUrlNotFound
}
fileId2Url[chunkView.FileId] = urlStrings
}
return func(writer io.Writer) error {
return streamChunksPrefetched(ctx, writer, chunkViews, fileId2Url, jwtFunc, masterClient, offset, size, downloadMaxBytesPs, prefetchAhead)
}, nil
}
func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writer, chunks []*filer_pb.FileChunk, offset int64, size int64) error {
streamFn, err := PrepareStreamContent(masterClient, JwtForVolumeServer, chunks, offset, size)
if err != nil {
+317
View File
@@ -0,0 +1,317 @@
package filer
import (
"bytes"
"context"
"fmt"
"io"
"math/rand"
"net/http"
"net/http/httptest"
"os"
"strings"
"sync"
"testing"
"time"
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
)
func TestMain(m *testing.M) {
// Initialize the global HTTP client required by ReadUrlAsStream
util_http.InitGlobalHttpClient()
os.Exit(m.Run())
}
// mockMasterClientForBenchmark implements HasLookupFileIdFunction and CacheInvalidator
type mockMasterClientForBenchmark struct {
urls map[string][]string
}
func (m *mockMasterClientForBenchmark) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType {
return func(ctx context.Context, fileId string) ([]string, error) {
if urls, ok := m.urls[fileId]; ok {
return urls, nil
}
return nil, fmt.Errorf("fileId %s not found", fileId)
}
}
func (m *mockMasterClientForBenchmark) InvalidateCache(fileId string) {}
// noopJwtFunc returns empty JWT for testing
func noopJwtFunc(fileId string) string {
return ""
}
// createMockVolumeServer creates an httptest server that serves chunk data
// with configurable per-request latency to simulate network conditions.
// The latency is applied once per request (simulating RTT), not per byte.
func createMockVolumeServer(chunkData map[string][]byte, latency time.Duration) *httptest.Server {
var mu sync.RWMutex
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Simulate network latency (connection setup + RTT)
if latency > 0 {
time.Sleep(latency)
}
// Extract fileId from path (e.g., "/1,abc123")
path := r.URL.Path
if strings.HasPrefix(path, "/") {
path = path[1:]
}
mu.RLock()
data, ok := chunkData[path]
mu.RUnlock()
if !ok {
http.Error(w, "not found", http.StatusNotFound)
return
}
// Handle Range header
rangeHeader := r.Header.Get("Range")
if rangeHeader != "" {
var start, end int64
fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end)
if start >= 0 && end < int64(len(data)) && start <= end {
w.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1))
w.WriteHeader(http.StatusPartialContent)
w.Write(data[start : end+1])
return
}
}
w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data)))
w.WriteHeader(http.StatusOK)
w.Write(data)
}))
}
// benchmarkConfig holds parameters for a single benchmark scenario
type benchmarkConfig struct {
numChunks int
chunkSize int
latency time.Duration
prefetch int // 0 = sequential
}
func (c benchmarkConfig) name() string {
name := fmt.Sprintf("chunks=%d/size=%dKB/latency=%dms",
c.numChunks, c.chunkSize/1024, c.latency.Milliseconds())
if c.prefetch > 0 {
name += fmt.Sprintf("/prefetch=%d", c.prefetch)
}
return name
}
// setupBenchmark creates mock infrastructure and returns chunks, master client, and cleanup func
func setupBenchmark(b *testing.B, cfg benchmarkConfig) ([]*filer_pb.FileChunk, *mockMasterClientForBenchmark, func()) {
b.Helper()
// Generate random chunk data
chunkData := make(map[string][]byte, cfg.numChunks)
chunks := make([]*filer_pb.FileChunk, cfg.numChunks)
for i := 0; i < cfg.numChunks; i++ {
fileId := fmt.Sprintf("1,%x", i)
data := make([]byte, cfg.chunkSize)
rand.Read(data)
chunkData[fileId] = data
chunks[i] = &filer_pb.FileChunk{
FileId: fileId,
Offset: int64(i * cfg.chunkSize),
Size: uint64(cfg.chunkSize),
ModifiedTsNs: int64(i),
Fid: &filer_pb.FileId{FileKey: uint64(i)},
}
}
// Start mock volume server
server := createMockVolumeServer(chunkData, cfg.latency)
// Build URL map
urls := make(map[string][]string, cfg.numChunks)
for i := 0; i < cfg.numChunks; i++ {
fileId := fmt.Sprintf("1,%x", i)
urls[fileId] = []string{server.URL + "/" + fileId}
}
masterClient := &mockMasterClientForBenchmark{urls: urls}
cleanup := func() { server.Close() }
return chunks, masterClient, cleanup
}
// runSequentialBenchmark runs the current sequential streaming path
func runSequentialBenchmark(b *testing.B, cfg benchmarkConfig) {
chunks, masterClient, cleanup := setupBenchmark(b, cfg)
defer cleanup()
totalSize := int64(cfg.numChunks * cfg.chunkSize)
b.ResetTimer()
b.SetBytes(totalSize)
for i := 0; i < b.N; i++ {
streamFn, err := PrepareStreamContentWithThrottler(
context.Background(),
masterClient,
noopJwtFunc,
chunks,
0,
totalSize,
0, // no throttle
)
if err != nil {
b.Fatal(err)
}
if err := streamFn(io.Discard); err != nil {
b.Fatal(err)
}
}
}
// BenchmarkStreamSequential benchmarks the current sequential streaming path.
// This provides the BEFORE baseline for comparison.
func BenchmarkStreamSequential(b *testing.B) {
configs := []benchmarkConfig{
// Pure throughput (no latency)
{numChunks: 16, chunkSize: 64 * 1024, latency: 0},
{numChunks: 64, chunkSize: 64 * 1024, latency: 0},
// Moderate latency — shows RTT gap overhead
{numChunks: 16, chunkSize: 64 * 1024, latency: 5 * time.Millisecond},
{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond},
// High latency — significant RTT overhead
{numChunks: 16, chunkSize: 64 * 1024, latency: 20 * time.Millisecond},
{numChunks: 64, chunkSize: 64 * 1024, latency: 10 * time.Millisecond},
}
for _, cfg := range configs {
b.Run(cfg.name(), func(b *testing.B) {
runSequentialBenchmark(b, cfg)
})
}
}
// BenchmarkStreamSequentialVerify is a quick functional test that the benchmark
// infrastructure works correctly — ensures data integrity through the pipeline.
func BenchmarkStreamSequentialVerify(b *testing.B) {
cfg := benchmarkConfig{numChunks: 4, chunkSize: 1024, latency: 0}
chunks, masterClient, cleanup := setupBenchmark(b, cfg)
defer cleanup()
totalSize := int64(cfg.numChunks * cfg.chunkSize)
streamFn, err := PrepareStreamContentWithThrottler(
context.Background(),
masterClient,
noopJwtFunc,
chunks,
0,
totalSize,
0,
)
if err != nil {
b.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
b.Fatal(err)
}
if buf.Len() != int(totalSize) {
b.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
}
}
// runPrefetchBenchmark runs the new prefetch streaming path
func runPrefetchBenchmark(b *testing.B, cfg benchmarkConfig) {
chunks, masterClient, cleanup := setupBenchmark(b, cfg)
defer cleanup()
totalSize := int64(cfg.numChunks * cfg.chunkSize)
b.ResetTimer()
b.SetBytes(totalSize)
for i := 0; i < b.N; i++ {
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(),
masterClient,
noopJwtFunc,
chunks,
0,
totalSize,
0, // no throttle
cfg.prefetch,
)
if err != nil {
b.Fatal(err)
}
if err := streamFn(io.Discard); err != nil {
b.Fatal(err)
}
}
}
// BenchmarkStreamPrefetch benchmarks the new prefetch streaming path.
// Compare against BenchmarkStreamSequential for the AFTER measurement.
func BenchmarkStreamPrefetch(b *testing.B) {
configs := []benchmarkConfig{
// Pure throughput (no latency) — should be similar to sequential
{numChunks: 16, chunkSize: 64 * 1024, latency: 0, prefetch: 4},
{numChunks: 64, chunkSize: 64 * 1024, latency: 0, prefetch: 4},
// Moderate latency — prefetch should eliminate most RTT overhead
{numChunks: 16, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 4},
{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 4},
// High latency — most benefit from prefetch
{numChunks: 16, chunkSize: 64 * 1024, latency: 20 * time.Millisecond, prefetch: 4},
{numChunks: 64, chunkSize: 64 * 1024, latency: 10 * time.Millisecond, prefetch: 4},
// Vary prefetch count with moderate latency
{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 2},
{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 8},
}
for _, cfg := range configs {
b.Run(cfg.name(), func(b *testing.B) {
runPrefetchBenchmark(b, cfg)
})
}
}
// BenchmarkStreamPrefetchVerify verifies data integrity through the prefetch pipeline.
func BenchmarkStreamPrefetchVerify(b *testing.B) {
cfg := benchmarkConfig{numChunks: 4, chunkSize: 1024, latency: 0, prefetch: 4}
chunks, masterClient, cleanup := setupBenchmark(b, cfg)
defer cleanup()
totalSize := int64(cfg.numChunks * cfg.chunkSize)
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(),
masterClient,
noopJwtFunc,
chunks,
0,
totalSize,
0,
cfg.prefetch,
)
if err != nil {
b.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
b.Fatal(err)
}
if buf.Len() != int(totalSize) {
b.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
}
}
+274
View File
@@ -0,0 +1,274 @@
package filer
import (
"context"
"fmt"
"io"
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/mem"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
)
// chunkPipeResult represents a prefetched chunk streaming through a pipe.
// The fetch goroutine writes data into the pipeWriter; the consumer reads from pipeReader.
type chunkPipeResult struct {
chunkView *ChunkView
reader *io.PipeReader
fetchErr error // final error from fetch goroutine
written int64 // bytes written by fetch goroutine
done chan struct{} // closed when fetch goroutine finishes
urlStrings []string // snapshot of URLs at dispatch time (for retry logic)
}
// streamChunksPrefetched streams chunks with concurrent prefetch using io.Pipe.
//
// For each chunk in file order, a goroutine is launched (bounded by a semaphore)
// that establishes an HTTP connection to the volume server and streams data through
// an io.Pipe. The consumer reads from pipes in order, writing to the response.
//
// Memory usage is minimal: pipes are synchronous (no buffering), and only one
// reusable copy buffer is allocated for the consumer.
func streamChunksPrefetched(
ctx context.Context,
writer io.Writer,
chunkViews *IntervalList[*ChunkView],
fileId2Url map[string][]string,
jwtFunc VolumeServerJwtFunction,
masterClient wdclient.HasLookupFileIdFunction,
offset int64,
size int64,
downloadMaxBytesPs int64,
prefetchAhead int,
) error {
downloadThrottler := util.NewWriteThrottler(downloadMaxBytesPs)
// Create a local cancellable context so the consumer can stop the producer
// and all in-flight fetch goroutines on error (e.g., client disconnect).
localCtx, localCancel := context.WithCancel(ctx)
defer localCancel()
// Ordered channel: one entry per chunk, in file order.
// Capacity = prefetchAhead so the producer can run ahead.
// Uses pointer to avoid copying the struct while fetch goroutines write to it.
results := make(chan *chunkPipeResult, prefetchAhead)
// Semaphore to limit concurrent fetch goroutines (and thus HTTP connections).
sem := make(chan struct{}, prefetchAhead)
// Producer: walks chunk list, launches fetch goroutines, sends results in order.
// The producer only reads from fileId2Url (populated before streaming starts),
// so there is no concurrent map access — the consumer never writes to it.
var producerWg sync.WaitGroup
producerWg.Add(1)
go func() {
defer producerWg.Done()
defer close(results)
for x := chunkViews.Front(); x != nil; x = x.Next {
chunkView := x.Value
// Check context before starting new fetch
select {
case <-localCtx.Done():
return
default:
}
// Acquire semaphore slot (bounds concurrent HTTP connections)
select {
case sem <- struct{}{}:
case <-localCtx.Done():
return
}
pr, pw := io.Pipe()
urlStrings := fileId2Url[chunkView.FileId]
jwt := jwtFunc(chunkView.FileId)
result := &chunkPipeResult{
chunkView: chunkView,
reader: pr,
done: make(chan struct{}),
urlStrings: urlStrings,
}
// Launch fetch goroutine
go func(cv *ChunkView, urls []string, jwt string, pw *io.PipeWriter, res *chunkPipeResult) {
defer func() { <-sem }() // release semaphore
defer close(res.done)
written, err := retriedStreamFetchChunkData(
localCtx, pw, urls, jwt,
cv.CipherKey, cv.IsGzipped, cv.IsFullChunk(),
cv.OffsetInChunk, int(cv.ViewSize),
)
res.written = written
res.fetchErr = err
if err != nil {
pw.CloseWithError(err)
} else {
pw.Close()
}
}(chunkView, urlStrings, jwt, pw, result)
// Send result to consumer (blocks if channel full, back-pressuring producer)
select {
case results <- result:
case <-localCtx.Done():
// Consumer gone; close the pipe and wait for the fetch goroutine
// to finish so we don't leak it (this result was never sent to
// the channel, so the drain loop won't handle it).
pr.Close()
<-result.done
return
}
}
}()
// Consumer: reads from results channel in order, writes to response writer.
// Use the SeaweedFS memory pool for the copy buffer to reduce GC pressure.
copyBuf := mem.Allocate(256 * 1024)
defer mem.Free(copyBuf)
remaining := size
var consumeErr error
for result := range results {
chunkView := result.chunkView
// Handle gap before this chunk (zero-fill)
if offset < chunkView.ViewOffset {
gap := chunkView.ViewOffset - offset
remaining -= gap
glog.V(4).InfofCtx(ctx, "prefetch zero [%d,%d)", offset, chunkView.ViewOffset)
if err := writeZero(writer, gap); err != nil {
consumeErr = fmt.Errorf("write zero [%d,%d): %w", offset, chunkView.ViewOffset, err)
result.reader.Close()
break
}
offset = chunkView.ViewOffset
}
// Stream chunk data from pipe to response
start := time.Now()
_, copyErr := io.CopyBuffer(writer, result.reader, copyBuf)
result.reader.Close()
// Wait for fetch goroutine to finish to get final error
<-result.done
// Determine the effective error
err := copyErr
if err == nil && result.fetchErr != nil && result.written == 0 {
err = result.fetchErr
}
// If the fetcher itself failed before writing any data, try cache invalidation
// + re-fetch (same as sequential path stream.go:197). We check result.fetchErr
// and result.written (not copied) to avoid wrongly retrying when the fetch
// succeeded but the response writer failed on the first write.
if result.fetchErr != nil && result.written == 0 {
if err := localCtx.Err(); err != nil {
consumeErr = err
break
}
retryErr := retryWithCacheInvalidation(localCtx, writer, chunkView, result.urlStrings, jwtFunc, masterClient)
if retryErr != nil {
stats.FilerHandlerCounter.WithLabelValues("chunkDownloadError").Inc()
consumeErr = fmt.Errorf("read chunk: %w", retryErr)
break
}
// Retry succeeded
err = nil
} else if err != nil {
if localCtx.Err() != nil {
consumeErr = localCtx.Err()
} else {
stats.FilerHandlerCounter.WithLabelValues("chunkDownloadError").Inc()
consumeErr = fmt.Errorf("read chunk: %w", err)
}
break
}
offset += int64(chunkView.ViewSize)
remaining -= int64(chunkView.ViewSize)
stats.FilerRequestHistogram.WithLabelValues("chunkDownload").Observe(time.Since(start).Seconds())
stats.FilerHandlerCounter.WithLabelValues("chunkDownload").Inc()
downloadThrottler.MaybeSlowdown(int64(chunkView.ViewSize))
}
// Cancel the local context to stop the producer and any in-flight fetchers early.
// This ensures goroutines don't linger after the consumer exits (e.g., on write error).
localCancel()
// Drain remaining results to close pipes and unblock fetch goroutines
for result := range results {
result.reader.Close()
<-result.done
}
// Wait for producer to finish
producerWg.Wait()
if consumeErr != nil {
return consumeErr
}
// Handle trailing zero-fill
if remaining > 0 {
glog.V(4).InfofCtx(ctx, "prefetch zero [%d,%d)", offset, offset+remaining)
if err := writeZero(writer, remaining); err != nil {
return fmt.Errorf("write zero [%d,%d): %w", offset, offset+remaining, err)
}
}
return nil
}
// retryWithCacheInvalidation attempts to re-fetch a chunk after invalidating the URL cache.
// This mirrors the retry logic in PrepareStreamContentWithThrottler's sequential path.
func retryWithCacheInvalidation(
ctx context.Context,
writer io.Writer,
chunkView *ChunkView,
oldUrlStrings []string,
jwtFunc VolumeServerJwtFunction,
masterClient wdclient.HasLookupFileIdFunction,
) error {
invalidator, ok := masterClient.(CacheInvalidator)
if !ok {
return fmt.Errorf("read chunk %s failed and no cache invalidator available", chunkView.FileId)
}
glog.V(0).InfofCtx(ctx, "prefetch read chunk %s failed, invalidating cache and retrying", chunkView.FileId)
invalidator.InvalidateCache(chunkView.FileId)
newUrlStrings, lookupErr := masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
if lookupErr != nil {
glog.WarningfCtx(ctx, "failed to re-lookup chunk %s after cache invalidation: %v", chunkView.FileId, lookupErr)
return fmt.Errorf("re-lookup chunk %s: %w", chunkView.FileId, lookupErr)
}
if len(newUrlStrings) == 0 {
glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", chunkView.FileId)
return fmt.Errorf("re-lookup chunk %s: no locations", chunkView.FileId)
}
if urlSlicesEqual(oldUrlStrings, newUrlStrings) {
glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", chunkView.FileId)
return fmt.Errorf("read chunk %s failed, same locations after cache invalidation", chunkView.FileId)
}
glog.V(0).InfofCtx(ctx, "retrying read chunk %s with new locations: %v", chunkView.FileId, newUrlStrings)
jwt := jwtFunc(chunkView.FileId)
_, err := retriedStreamFetchChunkData(
ctx, writer, newUrlStrings, jwt,
chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(),
chunkView.OffsetInChunk, int(chunkView.ViewSize),
)
return err
}
+365
View File
@@ -0,0 +1,365 @@
package filer
import (
"bytes"
"context"
"fmt"
"io"
"math/rand"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
)
// testMasterClient implements HasLookupFileIdFunction and CacheInvalidator for tests
type testMasterClient struct {
urls map[string][]string
invalidatedCount int32
}
func (m *testMasterClient) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType {
return func(ctx context.Context, fileId string) ([]string, error) {
if urls, ok := m.urls[fileId]; ok {
return urls, nil
}
return nil, fmt.Errorf("fileId %s not found", fileId)
}
}
func (m *testMasterClient) InvalidateCache(fileId string) {
atomic.AddInt32(&m.invalidatedCount, 1)
}
func noopJwt(fileId string) string { return "" }
// createTestServer creates a mock volume server that serves chunk data.
// Supports Range header for partial chunk reads (exercising OffsetInChunk paths).
func createTestServer(chunkData map[string][]byte) *httptest.Server {
var mu sync.RWMutex
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
path := r.URL.Path
if strings.HasPrefix(path, "/") {
path = path[1:]
}
mu.RLock()
data, ok := chunkData[path]
mu.RUnlock()
if !ok {
http.Error(w, "not found", http.StatusNotFound)
return
}
// Handle Range header for partial chunk reads
rangeHeader := r.Header.Get("Range")
if rangeHeader != "" {
var start, end int64
if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err == nil {
if start < 0 || end >= int64(len(data)) || start > end {
w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", len(data)))
w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return
}
rangeData := data[start : end+1]
w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(data)))
w.Header().Set("Content-Length", fmt.Sprintf("%d", len(rangeData)))
w.WriteHeader(http.StatusPartialContent)
w.Write(rangeData)
return
}
}
w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data)))
w.WriteHeader(http.StatusOK)
w.Write(data)
}))
}
// makeChunksAndServer creates N chunks of given size, a mock server, and a master client
func makeChunksAndServer(t *testing.T, numChunks, chunkSize int) ([]*filer_pb.FileChunk, *testMasterClient, map[string][]byte, func()) {
t.Helper()
chunkData := make(map[string][]byte, numChunks)
chunks := make([]*filer_pb.FileChunk, numChunks)
for i := 0; i < numChunks; i++ {
fileId := fmt.Sprintf("1,%x", i)
data := make([]byte, chunkSize)
rand.Read(data)
chunkData[fileId] = data
chunks[i] = &filer_pb.FileChunk{
FileId: fileId,
Offset: int64(i * chunkSize),
Size: uint64(chunkSize),
ModifiedTsNs: int64(i),
Fid: &filer_pb.FileId{FileKey: uint64(i)},
}
}
server := createTestServer(chunkData)
urls := make(map[string][]string, numChunks)
for i := 0; i < numChunks; i++ {
fileId := fmt.Sprintf("1,%x", i)
urls[fileId] = []string{server.URL + "/" + fileId}
}
masterClient := &testMasterClient{urls: urls}
return chunks, masterClient, chunkData, func() { server.Close() }
}
// TestPrefetchInOrderDelivery verifies chunks are written to the output in correct file order
func TestPrefetchInOrderDelivery(t *testing.T) {
chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 8, 4096)
defer cleanup()
totalSize := int64(8 * 4096)
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, 0, totalSize, 0, 4,
)
if err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
t.Fatal(err)
}
// Verify total size
if buf.Len() != int(totalSize) {
t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
}
// Verify data matches chunk-by-chunk in order
result := buf.Bytes()
for i := 0; i < 8; i++ {
fileId := fmt.Sprintf("1,%x", i)
expected := chunkData[fileId]
got := result[i*4096 : (i+1)*4096]
if !bytes.Equal(expected, got) {
t.Fatalf("chunk %d (%s) data mismatch at offset %d", i, fileId, i*4096)
}
}
}
// TestPrefetchSingleChunk verifies the pipeline works with just one chunk
func TestPrefetchSingleChunk(t *testing.T) {
chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 1, 8192)
defer cleanup()
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, 0, 8192, 0, 4,
)
if err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
t.Fatal(err)
}
expected := chunkData["1,0"]
if !bytes.Equal(expected, buf.Bytes()) {
t.Fatal("single chunk data mismatch")
}
}
// TestPrefetchFallbackToSequential verifies prefetch=1 falls back to sequential path
func TestPrefetchFallbackToSequential(t *testing.T) {
chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 4, 1024)
defer cleanup()
totalSize := int64(4 * 1024)
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, 0, totalSize, 0, 1, // prefetch=1 -> sequential
)
if err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
t.Fatal(err)
}
if buf.Len() != int(totalSize) {
t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
}
// Verify data order
result := buf.Bytes()
for i := 0; i < 4; i++ {
fileId := fmt.Sprintf("1,%x", i)
expected := chunkData[fileId]
got := result[i*1024 : (i+1)*1024]
if !bytes.Equal(expected, got) {
t.Fatalf("chunk %d data mismatch", i)
}
}
}
// TestPrefetchContextCancellation verifies all goroutines clean up on cancellation
func TestPrefetchContextCancellation(t *testing.T) {
// Use a slow server so cancellation happens mid-stream
var requestCount int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&requestCount, 1)
// Slow response
time.Sleep(100 * time.Millisecond)
w.Header().Set("Content-Length", "1024")
w.WriteHeader(http.StatusOK)
w.Write(make([]byte, 1024))
}))
defer server.Close()
numChunks := 16
chunks := make([]*filer_pb.FileChunk, numChunks)
urls := make(map[string][]string, numChunks)
for i := 0; i < numChunks; i++ {
fileId := fmt.Sprintf("1,%x", i)
chunks[i] = &filer_pb.FileChunk{
FileId: fileId, Offset: int64(i * 1024), Size: 1024,
ModifiedTsNs: int64(i), Fid: &filer_pb.FileId{FileKey: uint64(i)},
}
urls[fileId] = []string{server.URL + "/" + fileId}
}
masterClient := &testMasterClient{urls: urls}
// Cancel after a short time
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
streamFn, err := PrepareStreamContentWithPrefetch(
ctx, masterClient, noopJwt,
chunks, 0, int64(numChunks*1024), 0, 4,
)
if err != nil {
// URL resolution may fail due to cancellation — that's expected
return
}
err = streamFn(io.Discard)
if err == nil {
t.Fatal("expected error from cancelled context")
}
// Verify not all chunks were requested (cancellation stopped early)
reqs := atomic.LoadInt32(&requestCount)
if reqs >= int32(numChunks) {
t.Logf("warning: all %d chunks were requested despite cancellation (got %d)", numChunks, reqs)
}
}
// TestPrefetchRangeRequest verifies prefetch works with offset/size subset
func TestPrefetchRangeRequest(t *testing.T) {
chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 8, 4096)
defer cleanup()
// Request only chunks 2-5 (offset=8192, size=16384)
offset := int64(2 * 4096)
size := int64(4 * 4096)
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, offset, size, 0, 4,
)
if err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
t.Fatal(err)
}
if buf.Len() != int(size) {
t.Fatalf("expected %d bytes, got %d", size, buf.Len())
}
// Verify data matches chunks 2-5
result := buf.Bytes()
for i := 2; i < 6; i++ {
fileId := fmt.Sprintf("1,%x", i)
expected := chunkData[fileId]
start := (i - 2) * 4096
got := result[start : start+4096]
if !bytes.Equal(expected, got) {
t.Fatalf("chunk %d data mismatch in range request", i)
}
}
}
// TestPrefetchLargePrefetchCount verifies prefetch > numChunks is handled gracefully
func TestPrefetchLargePrefetchCount(t *testing.T) {
chunks, masterClient, _, cleanup := makeChunksAndServer(t, 3, 1024)
defer cleanup()
totalSize := int64(3 * 1024)
// prefetch=10 but only 3 chunks — should work fine
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, 0, totalSize, 0, 10,
)
if err != nil {
t.Fatal(err)
}
var buf bytes.Buffer
if err := streamFn(&buf); err != nil {
t.Fatal(err)
}
if buf.Len() != int(totalSize) {
t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
}
}
// TestPrefetchConcurrentDownloads verifies multiple concurrent prefetch streams
func TestPrefetchConcurrentDownloads(t *testing.T) {
chunks, masterClient, _, cleanup := makeChunksAndServer(t, 8, 2048)
defer cleanup()
totalSize := int64(8 * 2048)
var wg sync.WaitGroup
errors := make(chan error, 4)
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
streamFn, err := PrepareStreamContentWithPrefetch(
context.Background(), masterClient, noopJwt,
chunks, 0, totalSize, 0, 4,
)
if err != nil {
errors <- err
return
}
if err := streamFn(io.Discard); err != nil {
errors <- err
}
}()
}
wg.Wait()
close(errors)
for err := range errors {
t.Fatalf("concurrent download error: %v", err)
}
}
+1
View File
@@ -186,6 +186,7 @@ message Identity {
bool disabled = 5; // User status: false = enabled (default), true = disabled
repeated string service_account_ids = 6; // IDs of service accounts owned by this user
repeated string policy_names = 7;
bool is_static = 8; // Loaded from static config file (read-only, not editable via API)
}
message Credential {
+11 -2
View File
@@ -1399,6 +1399,7 @@ type Identity struct {
Disabled bool `protobuf:"varint,5,opt,name=disabled,proto3" json:"disabled,omitempty"` // User status: false = enabled (default), true = disabled
ServiceAccountIds []string `protobuf:"bytes,6,rep,name=service_account_ids,json=serviceAccountIds,proto3" json:"service_account_ids,omitempty"` // IDs of service accounts owned by this user
PolicyNames []string `protobuf:"bytes,7,rep,name=policy_names,json=policyNames,proto3" json:"policy_names,omitempty"`
IsStatic bool `protobuf:"varint,8,opt,name=is_static,json=isStatic,proto3" json:"is_static,omitempty"` // Loaded from static config file (read-only, not editable via API)
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -1482,6 +1483,13 @@ func (x *Identity) GetPolicyNames() []string {
return nil
}
func (x *Identity) GetIsStatic() bool {
if x != nil {
return x.IsStatic
}
return false
}
type Credential struct {
state protoimpl.MessageState `protogen:"open.v1"`
AccessKey string `protobuf:"bytes,1,opt,name=access_key,json=accessKey,proto3" json:"access_key,omitempty"`
@@ -3013,7 +3021,7 @@ const file_iam_proto_rawDesc = "" +
"\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" +
"\amembers\x18\x02 \x03(\tR\amembers\x12!\n" +
"\fpolicy_names\x18\x03 \x03(\tR\vpolicyNames\x12\x1a\n" +
"\bdisabled\x18\x04 \x01(\bR\bdisabled\"\x88\x02\n" +
"\bdisabled\x18\x04 \x01(\bR\bdisabled\"\xa5\x02\n" +
"\bIdentity\x12\x12\n" +
"\x04name\x18\x01 \x01(\tR\x04name\x124\n" +
"\vcredentials\x18\x02 \x03(\v2\x12.iam_pb.CredentialR\vcredentials\x12\x18\n" +
@@ -3021,7 +3029,8 @@ const file_iam_proto_rawDesc = "" +
"\aaccount\x18\x04 \x01(\v2\x0f.iam_pb.AccountR\aaccount\x12\x1a\n" +
"\bdisabled\x18\x05 \x01(\bR\bdisabled\x12.\n" +
"\x13service_account_ids\x18\x06 \x03(\tR\x11serviceAccountIds\x12!\n" +
"\fpolicy_names\x18\a \x03(\tR\vpolicyNames\"b\n" +
"\fpolicy_names\x18\a \x03(\tR\vpolicyNames\x12\x1b\n" +
"\tis_static\x18\b \x01(\bR\bisStatic\"b\n" +
"\n" +
"Credential\x12\x1d\n" +
"\n" +
+57
View File
@@ -295,6 +295,10 @@ func NewIdentityAccessManagementWithStore(option *S3ApiServerOption, filerClient
// This serves as an in-memory "static" configuration
iam.loadEnvironmentVariableCredentials()
// Update credential manager with all static identities (file + env vars)
// so that listing operations via filer gRPC also include them.
iam.updateCredentialManagerStaticIdentities()
// Determine whether to enable S3 authentication based on configuration
// For "weed mini" without any S3 config, default to allowing all access (isAuthEnabled = false)
// If any credentials are configured (via file, filer, or env vars), enable authentication
@@ -1069,6 +1073,59 @@ func (iam *IdentityAccessManagement) IsStaticIdentity(identityName string) bool
return iam.staticIdentityNames[identityName]
}
// updateCredentialManagerStaticIdentities syncs the current set of static
// identities to the credential manager. Call this after any operation that
// changes static identities (startup, config file reload, etc.).
func (iam *IdentityAccessManagement) updateCredentialManagerStaticIdentities() {
if iam.credentialManager != nil {
iam.credentialManager.SetStaticIdentities(iam.GetStaticIdentities())
}
}
// GetStaticIdentities returns protobuf representations of all static identities.
// This is used to include static identities in listing operations (ListUsers, etc.)
func (iam *IdentityAccessManagement) GetStaticIdentities() []*iam_pb.Identity {
iam.m.RLock()
defer iam.m.RUnlock()
var result []*iam_pb.Identity
for _, ident := range iam.identities {
if !ident.IsStatic {
continue
}
var policyNames []string
if len(ident.PolicyNames) > 0 {
policyNames = make([]string, len(ident.PolicyNames))
copy(policyNames, ident.PolicyNames)
}
pbIdent := &iam_pb.Identity{
Name: ident.Name,
Disabled: ident.Disabled,
PolicyNames: policyNames,
IsStatic: true,
}
for _, action := range ident.Actions {
pbIdent.Actions = append(pbIdent.Actions, string(action))
}
for _, cred := range ident.Credentials {
pbIdent.Credentials = append(pbIdent.Credentials, &iam_pb.Credential{
AccessKey: cred.AccessKey,
SecretKey: cred.SecretKey,
Status: cred.Status,
})
}
if ident.Account != nil {
pbIdent.Account = &iam_pb.Account{
Id: ident.Account.Id,
DisplayName: ident.Account.DisplayName,
EmailAddress: ident.Account.EmailAddress,
}
}
result = append(result, pbIdent)
}
return result
}
func (iam *IdentityAccessManagement) lookupByAccessKey(accessKey string) (identity *Identity, cred *Credential, found bool) {
iam.m.RLock()
defer iam.m.RUnlock()
+157 -31
View File
@@ -65,6 +65,12 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM
uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "")
// Validate checksum algorithm before creating the upload directory
_, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r)
if checksumErrCode != s3err.ErrNone {
return nil, checksumErrCode
}
// Prepare error handling outside callback scope
var encryptionError error
@@ -95,6 +101,12 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM
}
s3a.applyMultipartEncryptionConfig(entry, encryptionConfig)
// Store the requested checksum algorithm so CompleteMultipartUpload can compute
// a composite checksum from per-part checksums
if checksumHeaderName != "" {
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName)
}
// Extract and store object lock metadata from request headers
// This ensures object lock settings from create_multipart_upload are preserved
if err := s3a.extractObjectLockMetadataFromRequest(r, entry); err != nil {
@@ -129,6 +141,11 @@ type CompleteMultipartUploadResult struct {
Bucket *string `xml:"Bucket,omitempty"`
Key *string `xml:"Key,omitempty"`
ETag *string `xml:"ETag,omitempty"`
// Checksum fields — returned as HTTP response headers, not in the XML body
ChecksumHeaderName string `xml:"-"`
ChecksumValue string `xml:"-"`
// VersionId is NOT included in XML body - it should only be in x-amz-version-id HTTP header
// Store the VersionId internally for setting HTTP header, but don't marshal to XML
@@ -173,15 +190,17 @@ type multipartPartBoundary struct {
}
type multipartCompletionState struct {
deleteEntries []*filer_pb.Entry
partEntries map[int][]*filer_pb.Entry
pentry *filer_pb.Entry
mime string
finalParts []*filer_pb.FileChunk
offset int64
partBoundaries []multipartPartBoundary
multipartETag string
entityWithTtl bool
deleteEntries []*filer_pb.Entry
partEntries map[int][]*filer_pb.Entry
pentry *filer_pb.Entry
mime string
finalParts []*filer_pb.FileChunk
offset int64
partBoundaries []multipartPartBoundary
multipartETag string
entityWithTtl bool
checksumHeaderName string // e.g. "X-Amz-Checksum-Crc32", empty if no checksum
checksumValue string // composite base64 checksum with "-N" suffix
}
func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult {
@@ -346,16 +365,36 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
}
}
// Compute composite checksum from per-part checksums if the upload
// was initiated with a checksum algorithm (stored in upload dir entry)
checksumHeaderName := ""
checksumValue := ""
if pentry.Extended != nil {
if algoName, ok := pentry.Extended[s3_constants.ExtChecksumAlgorithm]; ok {
checksumHeaderName = string(algoName)
}
}
if checksumHeaderName != "" {
var checksumErr error
checksumValue, checksumErr = computeCompositeChecksum(checksumHeaderName, partEntries, completedPartNumbers)
if checksumErr != nil {
glog.Errorf("completeMultipartUpload: composite checksum computation failed: %v", checksumErr)
return nil, nil, s3err.ErrInvalidPart
}
}
return &multipartCompletionState{
deleteEntries: deleteEntries,
partEntries: partEntries,
pentry: pentry,
mime: mime,
finalParts: finalParts,
offset: offset,
partBoundaries: partBoundaries,
multipartETag: calculateMultipartETag(partEntries, completedPartNumbers),
entityWithTtl: entityWithTtl,
deleteEntries: deleteEntries,
partEntries: partEntries,
pentry: pentry,
mime: mime,
finalParts: finalParts,
offset: offset,
partBoundaries: partBoundaries,
multipartETag: calculateMultipartETag(partEntries, completedPartNumbers),
entityWithTtl: entityWithTtl,
checksumHeaderName: checksumHeaderName,
checksumValue: checksumValue,
}, nil, s3err.ErrNone
}
@@ -442,6 +481,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
// Persist ETag to ensure subsequent HEAD/GET uses the same value
versionEntry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
// Store composite checksum if computed from per-part checksums
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
versionEntry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
versionEntry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
}
// Preserve ALL SSE metadata from the first part (if any)
// SSE metadata is stored in individual parts, not the upload directory
@@ -490,11 +534,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
// For versioned buckets, all content is stored in .versions directory
// The latest version information is tracked in the .versions directory metadata
output = &CompleteMultipartUploadResult{
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
VersionId: aws.String(versionId),
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
VersionId: aws.String(versionId),
ChecksumHeaderName: completionState.checksumHeaderName,
ChecksumValue: completionState.checksumValue,
}
return s3err.ErrNone
}
@@ -534,6 +580,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
}
// Persist ETag to ensure subsequent HEAD/GET uses the same value
entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
// Store composite checksum if computed from per-part checksums
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
}
if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
entry.Attributes.Mime = completionState.pentry.Attributes.Mime
} else if completionState.mime != "" {
@@ -547,10 +598,12 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
// Note: Suspended versioning should NOT return VersionId field according to AWS S3 spec
output = &CompleteMultipartUploadResult{
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
ChecksumHeaderName: completionState.checksumHeaderName,
ChecksumValue: completionState.checksumValue,
// VersionId field intentionally omitted for suspended versioning
}
return s3err.ErrNone
@@ -589,6 +642,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
}
// Persist ETag to ensure subsequent HEAD/GET uses the same value
entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
// Store composite checksum if computed from per-part checksums
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
}
if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
entry.Attributes.Mime = completionState.pentry.Attributes.Mime
} else if completionState.mime != "" {
@@ -606,10 +664,12 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
// For non-versioned buckets, return response without VersionId
output = &CompleteMultipartUploadResult{
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
Bucket: input.Bucket,
ETag: aws.String(etagQuote),
Key: objectKey(input.Key),
ChecksumHeaderName: completionState.checksumHeaderName,
ChecksumValue: completionState.checksumValue,
}
return s3err.ErrNone
})
@@ -1030,6 +1090,72 @@ func calculateMultipartETag(partEntries map[int][]*filer_pb.Entry, completedPart
return fmt.Sprintf("%x-%d", md5.Sum(etags), len(completedPartNumbers))
}
// computeCompositeChecksum computes a composite checksum from per-part checksums.
// It concatenates the raw (decoded) per-part checksums, hashes the result with the
// same algorithm, and returns the value as "base64-N" where N is the part count.
// This follows the AWS S3 multipart checksum specification.
// Returns an error if a part is missing its checksum (the upload was initiated with
// a checksum algorithm, so all parts must have been uploaded with checksums).
func computeCompositeChecksum(checksumHeaderName string, partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) (string, error) {
// Determine the algorithm from the header name
algo := checksumAlgorithmFromHeaderName(checksumHeaderName)
if algo == ChecksumAlgorithmNone {
return "", fmt.Errorf("unknown checksum algorithm for header %q", checksumHeaderName)
}
// Collect raw per-part checksums
var combined []byte
for _, partNumber := range completedPartNumbers {
entries, ok := partEntries[partNumber]
if !ok || len(entries) == 0 {
return "", fmt.Errorf("part %d not found", partNumber)
}
if len(entries) > 1 {
sortEntriesByLatestChunk(entries)
}
entry := entries[0]
if entry.Extended == nil {
return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
}
// Validate the part's checksum algorithm matches the upload's expected algorithm
partAlgo, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]
if !ok || len(partAlgo) == 0 {
return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
}
if string(partAlgo) != checksumHeaderName {
return "", fmt.Errorf("part %d checksum algorithm mismatch: upload expects %s but part has %s", partNumber, checksumHeaderName, string(partAlgo))
}
partChecksumB64, ok := entry.Extended[s3_constants.ExtChecksumValue]
if !ok || len(partChecksumB64) == 0 {
return "", fmt.Errorf("part %d missing checksum value: upload initiated with %s but part has no checksum value", partNumber, checksumHeaderName)
}
raw, err := base64.StdEncoding.DecodeString(string(partChecksumB64))
if err != nil {
return "", fmt.Errorf("part %d has invalid checksum encoding: %w", partNumber, err)
}
combined = append(combined, raw...)
}
// Hash the concatenated raw checksums
h := getCheckSumWriter(algo)
if h == nil {
return "", fmt.Errorf("failed to create hash writer for %s", checksumHeaderName)
}
h.Write(combined)
compositeRaw := h.Sum(nil)
return fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(compositeRaw), len(completedPartNumbers)), nil
}
// checksumAlgorithmFromHeaderName maps a canonical header name back to its algorithm.
func checksumAlgorithmFromHeaderName(headerName string) ChecksumAlgorithm {
for _, entry := range checksumHeaders {
if entry.name == headerName {
return entry.alg
}
}
return ChecksumAlgorithmNone
}
func getEtagFromEntry(entry *filer_pb.Entry) string {
if entry.Extended != nil {
if etagBytes, ok := entry.Extended[s3_constants.ExtETagKey]; ok {
+4
View File
@@ -20,6 +20,10 @@ const (
ExtLatestVersionIsDeleteMarker = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
ExtMultipartObjectKey = "key"
// S3 checksum storage keys (use x-seaweedfs- prefix to avoid leaking in generic header loop)
ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm"
ExtChecksumValue = "x-seaweedfs-checksum-value"
// Bucket Policy
ExtBucketPolicyKey = "Seaweed-X-Amz-Bucket-Policy"
+10
View File
@@ -76,6 +76,16 @@ const (
AmzObjectLockRetainUntilDate = "X-Amz-Object-Lock-Retain-Until-Date"
AmzObjectLockLegalHold = "X-Amz-Object-Lock-Legal-Hold"
// S3 checksum headers
AmzChecksumAlgorithm = "X-Amz-Checksum-Algorithm"
AmzChecksumCRC32 = "X-Amz-Checksum-Crc32"
AmzChecksumCRC32C = "X-Amz-Checksum-Crc32c"
AmzChecksumCRC64NVME = "X-Amz-Checksum-Crc64nvme"
AmzChecksumSHA1 = "X-Amz-Checksum-Sha1"
AmzChecksumSHA256 = "X-Amz-Checksum-Sha256"
AmzTrailer = "X-Amz-Trailer"
AmzSdkChecksumAlgorithm = "X-Amz-Sdk-Checksum-Algorithm"
// S3 conditional headers
IfMatch = "If-Match"
IfNoneMatch = "If-None-Match"
+2
View File
@@ -215,6 +215,7 @@ func (e *EmbeddedIamApi) writeIamErrorResponse(w http.ResponseWriter, r *http.Re
}
// GetS3ApiConfiguration loads the S3 API configuration from the credential manager.
// The credential manager automatically includes static identities in the result.
func (e *EmbeddedIamApi) GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) error {
if e.getS3ApiConfigurationFunc != nil {
return e.getS3ApiConfigurationFunc(s3cfg)
@@ -228,6 +229,7 @@ func (e *EmbeddedIamApi) GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration)
}
// PutS3ApiConfiguration saves the S3 API configuration to the credential manager.
// The credential manager automatically filters out static identities before saving.
func (e *EmbeddedIamApi) PutS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) error {
if e.putS3ApiConfigurationFunc != nil {
return e.putS3ApiConfigurationFunc(s3cfg)
+19 -2
View File
@@ -1053,7 +1053,7 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
// Prepare streaming function with simple master client wrapper
tStreamPrep := time.Now()
// Use filerClient directly (not wrapped) so it can support cache invalidation
streamFn, err := filer.PrepareStreamContentWithThrottler(
streamFn, err := filer.PrepareStreamContentWithPrefetch(
ctx,
s3a.filerClient,
filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
@@ -1061,6 +1061,7 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
offset,
size,
0, // no throttling
4, // prefetch 4 chunks ahead for overlapped fetching
)
streamPrepTime = time.Since(tStreamPrep)
if err != nil {
@@ -1928,7 +1929,7 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry
}
// Create streaming reader - use filerClient directly for cache invalidation support
streamFn, err := filer.PrepareStreamContentWithThrottler(
streamFn, err := filer.PrepareStreamContentWithPrefetch(
ctx,
s3a.filerClient,
filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
@@ -1936,6 +1937,7 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry
0,
totalSize,
0,
4, // prefetch 4 chunks ahead for overlapped fetching
)
if err != nil {
return nil, err
@@ -2068,6 +2070,21 @@ func (s3a *S3ApiServer) setResponseHeaders(w http.ResponseWriter, r *http.Reques
}
}
// Set checksum header if stored in metadata, but only when:
// 1. The request contains "x-amz-checksum-mode: ENABLED" (per AWS S3 spec)
// 2. The request is NOT a ranged GET (Range header absent)
// The stored checksum covers the full object; returning it for partial
// responses causes SDK checksum validation failures.
if r != nil && r.Header.Get("X-Amz-Checksum-Mode") == "ENABLED" && r.Header.Get("Range") == "" {
if entry.Extended != nil {
if algoName, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]; ok {
if checksumVal, ok := entry.Extended[s3_constants.ExtChecksumValue]; ok {
w.Header().Set(string(algoName), string(checksumVal))
}
}
}
}
// Apply S3 passthrough headers from query parameters
// AWS S3 supports overriding response headers via query parameters like:
// ?response-cache-control=no-cache&response-content-type=application/json
@@ -180,6 +180,11 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
w.Header().Set("x-amz-version-id", *response.VersionId)
}
// Set composite checksum header if present
if response.ChecksumHeaderName != "" && response.ChecksumValue != "" {
w.Header().Set(response.ChecksumHeaderName, response.ChecksumValue)
}
stats_collect.RecordBucketActiveTime(bucket)
stats_collect.S3UploadedObjectsCounter.WithLabelValues(bucket).Inc()
+130 -1
View File
@@ -8,6 +8,7 @@ import (
"encoding/json"
"errors"
"fmt"
"hash"
"io"
"net/http"
"net/url"
@@ -71,6 +72,9 @@ type SSEResponseMetadata struct {
SSEType string
KMSKeyID string
BucketKeyEnabled bool
// Checksum fields for S3 additional checksum support
ChecksumHeaderName string // e.g. "X-Amz-Checksum-Sha256"
ChecksumValue string // base64-encoded checksum value
}
func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
@@ -358,6 +362,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
plaintextHash := md5.New()
dataReader = io.TeeReader(dataReader, plaintextHash)
// Detect and set up additional checksum computation (S3 checksum algorithm support)
checksumAlgo, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r)
if checksumErrCode != s3err.ErrNone {
return "", checksumErrCode, SSEResponseMetadata{}
}
var checksumHash hash.Hash
if checksumAlgo != ChecksumAlgorithmNone {
checksumHash = getCheckSumWriter(checksumAlgo)
if checksumHash != nil {
dataReader = io.TeeReader(dataReader, checksumHash)
}
}
// Handle all SSE encryption types in a unified manner
sseResult, sseErrorCode := s3a.handleAllSSEEncryption(r, dataReader, partOffset)
if sseErrorCode != s3err.ErrNone {
@@ -634,6 +651,24 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
// Store ETag in Extended attribute for future retrieval (e.g. multipart parts)
entry.Extended[s3_constants.ExtETagKey] = []byte(etag)
// Store additional checksum if one was computed
checksumBase64 := ""
if checksumHash != nil && checksumHeaderName != "" {
checksumBase64 = base64.StdEncoding.EncodeToString(checksumHash.Sum(nil))
// Verify against client-provided checksum if present in request headers
// (non-chunked uploads send the value directly; chunked uploads validate in the reader)
if expectedChecksum := r.Header.Get(checksumHeaderName); expectedChecksum != "" {
if expectedChecksum != checksumBase64 {
glog.Warningf("putToFiler: checksum mismatch for %s: expected %s, got %s", checksumHeaderName, expectedChecksum, checksumBase64)
s3a.deleteOrphanedChunks(chunkResult.FileChunks)
return "", s3err.ErrBadDigest, SSEResponseMetadata{}
}
}
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName)
entry.Extended[s3_constants.ExtChecksumValue] = []byte(checksumBase64)
glog.V(3).Infof("putToFiler: stored checksum %s=%s for %s", checksumHeaderName, checksumBase64, filePath)
}
// Set object owner according to bucket ownership settings.
s3a.setObjectOwnerFromRequest(r, bucket, entry)
@@ -802,7 +837,9 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
// Build SSE response metadata with encryption details
responseMetadata := SSEResponseMetadata{
SSEType: sseType,
SSEType: sseType,
ChecksumHeaderName: checksumHeaderName,
ChecksumValue: checksumBase64,
}
// For SSE-KMS, include key ID and bucket-key-enabled flag from stored metadata
@@ -816,6 +853,93 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
return etag, s3err.ErrNone, responseMetadata
}
// checksumAlgorithmMapping maps algorithm name strings to their enum and header name.
var checksumAlgorithmMapping = map[string]struct {
alg ChecksumAlgorithm
name string
}{
"CRC32": {ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
"CRC32C": {ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
"CRC64NVME": {ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
"SHA1": {ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
"SHA256": {ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
}
// trailerToChecksumAlgorithm maps trailer header names to their algorithm and canonical header name.
var trailerToChecksumAlgorithm = map[string]struct {
alg ChecksumAlgorithm
name string
}{
"x-amz-checksum-crc32": {ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
"x-amz-checksum-crc32c": {ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
"x-amz-checksum-crc64nvme": {ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
"x-amz-checksum-sha1": {ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
"x-amz-checksum-sha256": {ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
}
// checksumHeaders is the ordered list of individual checksum headers to check.
// Using a slice ensures deterministic selection order.
var checksumHeaders = []struct {
header string
alg ChecksumAlgorithm
name string
}{
{s3_constants.AmzChecksumCRC32, ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
{s3_constants.AmzChecksumCRC32C, ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
{s3_constants.AmzChecksumCRC64NVME, ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
{s3_constants.AmzChecksumSHA1, ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
{s3_constants.AmzChecksumSHA256, ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
}
// detectRequestedChecksumAlgorithm detects the checksum algorithm requested by the client.
// It checks the x-amz-sdk-checksum-algorithm header, x-amz-checksum-algorithm header,
// x-amz-trailer header (including comma-separated values), and individual x-amz-checksum-*
// headers. Returns the algorithm enum, the canonical HTTP header name, and an error code
// if an unsupported algorithm is specified.
func detectRequestedChecksumAlgorithm(r *http.Request) (ChecksumAlgorithm, string, s3err.ErrorCode) {
// Check x-amz-sdk-checksum-algorithm (set by AWS SDKs)
if algo := r.Header.Get(s3_constants.AmzSdkChecksumAlgorithm); algo != "" {
if m, ok := checksumAlgorithmMapping[strings.ToUpper(algo)]; ok {
return m.alg, m.name, s3err.ErrNone
}
glog.Warningf("unsupported checksum algorithm in %s: %q", s3_constants.AmzSdkChecksumAlgorithm, algo)
return ChecksumAlgorithmNone, "", s3err.ErrInvalidRequest
}
// Check x-amz-checksum-algorithm header
if algo := r.Header.Get(s3_constants.AmzChecksumAlgorithm); algo != "" {
if m, ok := checksumAlgorithmMapping[strings.ToUpper(algo)]; ok {
return m.alg, m.name, s3err.ErrNone
}
glog.Warningf("unsupported checksum algorithm in %s: %q", s3_constants.AmzChecksumAlgorithm, algo)
return ChecksumAlgorithmNone, "", s3err.ErrInvalidRequest
}
// Check x-amz-trailer header (used by chunked uploads, may be comma-separated)
if trailer := r.Header.Get(s3_constants.AmzTrailer); trailer != "" {
for _, part := range strings.Split(trailer, ",") {
part = strings.TrimSpace(strings.ToLower(part))
if part == "" {
continue
}
if m, ok := trailerToChecksumAlgorithm[part]; ok {
return m.alg, m.name, s3err.ErrNone
}
// Non-checksum trailers (e.g. x-amz-server-side-encryption) are fine — skip them
}
}
// Check individual checksum headers (non-chunked uploads send the value directly)
// Uses ordered slice for deterministic selection
for _, entry := range checksumHeaders {
if r.Header.Get(entry.header) != "" {
return entry.alg, entry.name, s3err.ErrNone
}
}
return ChecksumAlgorithmNone, "", s3err.ErrNone
}
const defaultFileMode = uint32(0660)
// resolveFileMode determines the file permission mode for an S3 upload.
@@ -883,6 +1007,11 @@ func (s3a *S3ApiServer) setSSEResponseHeaders(w http.ResponseWriter, r *http.Req
w.Header().Set(s3_constants.AmzServerSideEncryptionBucketKeyEnabled, "true")
}
}
// Set checksum response header if a checksum was computed
if sseMetadata.ChecksumHeaderName != "" && sseMetadata.ChecksumValue != "" {
w.Header().Set(sseMetadata.ChecksumHeaderName, sseMetadata.ChecksumValue)
}
}
func filerErrorToS3Error(err error) s3err.ErrorCode {
+1
View File
@@ -276,6 +276,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
glog.Errorf("fail to load config file %s: %v", option.Config, err)
} else {
glog.V(1).Infof("Loaded %d identities from config file %s", len(s3ApiServer.iam.identities), option.Config)
s3ApiServer.iam.updateCredentialManagerStaticIdentities()
}
})
}
+1
View File
@@ -82,6 +82,7 @@ type FilerOption struct {
AllowedOrigins []string
ExposeDirectoryData bool
TusBasePath string
S3ConfigFile string // optional path to static S3 identity config file
CredentialManager *credential.CredentialManager
}
@@ -101,6 +101,10 @@ func (s *IamGrpcServer) GetUser(ctx context.Context, req *iam_pb.GetUserRequest)
identity, err := s.credentialManager.GetUser(ctx, req.Username)
if err != nil {
if err == credential.ErrUserNotFound {
// Fall back to static identities (loaded from -s3.config file)
if si := s.credentialManager.GetStaticIdentity(req.Username); si != nil {
return &iam_pb.GetUserResponse{Identity: si}, nil
}
return nil, status.Errorf(codes.NotFound, "user %s not found", req.Username)
}
glog.Errorf("Failed to get user %s: %v", req.Username, err)
@@ -166,6 +170,20 @@ func (s *IamGrpcServer) ListUsers(ctx context.Context, req *iam_pb.ListUsersRequ
return nil, err
}
// Merge static identities (from -s3.config file) into the result
staticNames := s.credentialManager.GetStaticUsernames()
if len(staticNames) > 0 {
dynamicSet := make(map[string]bool, len(usernames))
for _, name := range usernames {
dynamicSet[name] = true
}
for _, name := range staticNames {
if !dynamicSet[name] {
usernames = append(usernames, name)
}
}
}
return &iam_pb.ListUsersResponse{
Usernames: usernames,
}, nil
+1 -1
View File
@@ -223,7 +223,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
// Matches S3 API behavior. Request context (ctx) is used for metadata operations above.
streamCtx, streamCancel := context.WithCancel(context.WithoutCancel(ctx))
streamFn, err := filer.PrepareStreamContentWithThrottler(streamCtx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
streamFn, err := filer.PrepareStreamContentWithPrefetch(streamCtx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs, 4)
if err != nil {
streamCancel()
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
+2 -2
View File
@@ -15,8 +15,8 @@ import (
// The server runs in a goroutine and serves pprof endpoints at /debug/pprof/*.
func StartDebugServer(debugPort int) {
go func() {
addr := fmt.Sprintf(":%d", debugPort)
glog.V(0).Infof("Starting debug server for pprof at http://localhost%s/debug/pprof/", addr)
addr := fmt.Sprintf("127.0.0.1:%d", debugPort)
glog.V(0).Infof("Starting debug server for pprof at http://%s/debug/pprof/", addr)
if err := http.ListenAndServe(addr, nil); err != nil && err != http.ErrServerClosed {
glog.Errorf("Failed to start debug server on %s: %v", addr, err)
}
+1 -1
View File
@@ -372,7 +372,7 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte,
var (
m int
)
buf := mem.Allocate(64 * 1024)
buf := mem.Allocate(256 * 1024)
defer mem.Free(buf)
for {