Compare commits

..
Author SHA1 Message Date
Chris LuandClaude Opus 4.6 b1c862e7e3 fix: prevent unhandled promise rejection for scheduler status fetch
Attach .catch at promise creation to mark it as handled by the runtime,
then check the result type after await. This prevents an
unhandledrejection event if the fetch rejects before the await.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 19:27:17 -08:00
Chris LuandClaude Opus 4.6 05902c7101 fix: use lifecycle context for job-type budget and retry backoff
- Use r.ctx (cancelled on shutdown) as parent for per-job-type budget
  context instead of context.Background(), so in-flight work is
  cancelled promptly on plugin shutdown.
- Replace waitForShutdownOrTimer with waitForShutdownOrCtx in retry
  backoff so retries respect the per-job-type budget context.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 19:27:06 -08:00
Chris LuandClaude Opus 4.6 e0cccae794 fix: add -scheduler.idleSleep flag to prevent EC integration test hang
The new 17-minute default idle sleep caused TestEcEndToEnd to hang
because the scheduler would not re-check for work frequently enough
after the first iteration found nothing.

Add -scheduler.idleSleep CLI flag (in seconds) to configure the idle
sleep duration. The EC integration test now passes -scheduler.idleSleep=2
so detection runs every 2 seconds when idle.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 19:05:56 -08:00
Chris LuandClaude Opus 4.6 6289beb8f5 fix: add ClusterContextProvider to shutdown test and handle status fetch errors
Address PR review nitpicks:
- Add ClusterContextProvider to TestGracefulShutdownDuringIteration so
  the scheduler loop actually starts (New() requires it).
- Wrap schedulerStatusPromise await in try/catch in plugin.templ so a
  failed status fetch does not break refreshJobsAndActivities rendering.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 18:49:37 -08:00
Chris LuandClaude Opus 4.6 35786d72e0 fix: unconditional lock release and context-aware executor reservation
Address PR review findings:
- Make defer releaseLock() unconditional in runSchedulerIteration since
  acquireAdminLock guarantees a non-nil release func on success.
- Add ctx parameter to reserveScheduledExecutor so it honors the
  per-job-type budget context, preventing waits from exceeding the
  30-minute budget.
- Add waitForShutdownOrCtx helper that also selects on ctx.Done().
- Update all reserveScheduledExecutor call sites in tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 18:49:28 -08:00
Chris LuandClaude Opus 4.6 51d715b4e6 plugin: update and add tests for sequential iteration model
Update TestListSchedulerStatesIncludesPolicyAndState to use
currentJobType for DetectionInFlight and assert MaxJobTypeDurationSeconds.

Add new tests:
- TestIterationAcquiresLockOnce: verify single lock acquire per iteration
- TestIterationReturnsfalseWhenLockFails: lock failure returns no-work
- TestSchedulerPhaseTracking: phase transitions and finishIteration
- TestGetSchedulerStatusIncludesIterationFields: status field population
- TestGracefulShutdownDuringIteration: clean shutdown mid-iteration
- TestIdleSleepDurationDefault: default 17m idle sleep
- TestIdleSleepDurationCustom: custom idle sleep from Options

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 16:42:48 -08:00
Chris LuandClaude Opus 4.6 4f97f7d413 plugin: add scheduler iteration status card to UI
Add a Scheduler Iteration card showing the current phase, active job
type, idle sleep duration, and last iteration result.

Update the Scheduler State table to replace In Flight / Next Detection
/ Interval columns with Status (Processing/Idle) and Max Duration
columns to match the new iteration model.

Add renderSchedulerIterationStatus JS function and update
refreshJobsAndActivities to fetch and render iteration status.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 16:42:26 -08:00
Chris LuandClaude Opus 4.6 c8ad7b528d plugin: update scheduler status to expose iteration state
Add Phase, CurrentJobType, IdleSleepSeconds, IterationStartedAt,
LastIterationEndedAt, and LastIterationWorkDetected to SchedulerStatus.

Set SchedulerTickSeconds to match IdleSleepSeconds for backward
compatibility.

Add MaxJobTypeDurationSeconds to SchedulerJobTypeStatus and remove
NextDetectionAt and DetectionIntervalSeconds.

Update GetSchedulerStatus to read iteration tracking fields and
adapt waiting status logic for the new model.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 16:42:10 -08:00
Chris LuandClaude Opus 4.6 41f22f5c92 plugin: rewrite scheduler to sequential lock-based iteration model
Replace the ticker-based parallel goroutine scheduler with a
single-lock sequential-processing iteration model:

- schedulerLoop now calls runSchedulerIteration in a loop, sleeping
  idleSleepDuration (default 17m) when no work is found, or looping
  immediately when work is detected.

- runSchedulerIteration acquires the admin lock once, loads cluster
  context once, then processes all enabled job types sequentially.

- runJobTypeIteration runs detection then execution for one job type
  within a per-job-type budget (default 30m) enforced via context.

- Remove markDetectionDue, finishDetection, pruneSchedulerState,
  runSchedulerTick, and runScheduledDetection which are no longer
  needed.

- dispatchScheduledProposals and executeScheduledJobWithExecutor now
  accept a parent context for budget enforcement.

- Remove DetectionInterval from schedulerPolicy, add MaxJobTypeDuration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 16:41:59 -08:00
Chris LuandClaude Opus 4.6 05138c6031 plugin: add iteration tracking fields to Plugin struct
Replace per-job-type detection maps (nextDetectionAt, detectionInFlight)
with iteration-level tracking fields: schedulerPhase, currentJobType,
iterationStartedAt, lastIterationEndedAt, lastIterationWorkDetected.

Add IdleSleepDuration to Options and idleSleepDuration to Plugin.
Add MaxJobTypeDurationSeconds to SchedulerJobTypeState and remove
NextDetectionAt and DetectionIntervalSeconds which are no longer
needed in the sequential iteration model.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 16:40:50 -08:00
298 changed files with 3151 additions and 23567 deletions
+5 -5
View File
@@ -32,7 +32,7 @@ jobs:
org.opencontainers.image.vendor=Chris Lu
-
name: Set up QEMU
uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v1
uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v1
-
name: Create BuildKit config
run: |
@@ -42,28 +42,28 @@ jobs:
EOF
-
name: Set up Docker Buildx
uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1
uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v1
with:
buildkitd-flags: "--debug"
buildkitd-config: /tmp/buildkitd.toml
-
name: Login to Docker Hub
if: github.event_name != 'pull_request'
uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
-
name: Login to GHCR
if: github.event_name != 'pull_request'
uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
with:
registry: ghcr.io
username: ${{ secrets.GHCR_USERNAME }}
password: ${{ secrets.GHCR_TOKEN }}
-
name: Build
uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v2
uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v2
with:
context: ./docker
push: ${{ github.event_name != 'pull_request' }}
@@ -126,14 +126,14 @@ jobs:
echo "seaweedfs_ref=$seaweed" >> "$GITHUB_OUTPUT"
- name: Set up QEMU
uses: docker/setup-qemu-action@v4
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Login to Docker Hub
if: github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
@@ -150,7 +150,7 @@ jobs:
fi
- name: Build and push image
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
with:
context: ./docker
push: ${{ github.event_name != 'pull_request' }}
+7 -7
View File
@@ -98,7 +98,7 @@ jobs:
org.opencontainers.image.vendor=Chris Lu
- name: Set up QEMU
if: matrix.platform != 'amd64'
uses: docker/setup-qemu-action@v4
uses: docker/setup-qemu-action@v3
- name: Create BuildKit config
run: |
cat > /tmp/buildkitd.toml <<EOF
@@ -106,25 +106,25 @@ jobs:
mirrors = ["https://mirror.gcr.io"]
EOF
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
with:
buildkitd-flags: "--debug"
buildkitd-config: /tmp/buildkitd.toml
- name: Login to Docker Hub
if: github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Login to GHCR
if: github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ secrets.GHCR_USERNAME }}
password: ${{ secrets.GHCR_TOKEN }}
- name: Build ${{ matrix.platform }} ${{ matrix.variant }}
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
env:
DOCKER_BUILDKIT: 1
with:
@@ -180,12 +180,12 @@ jobs:
ghcr.io/chrislusf/seaweedfs
tags: type=raw,value=${{ github.event_name == 'workflow_dispatch' && github.event.inputs.image_tag || 'latest' }},suffix=${{ steps.config.outputs.tag_suffix }}
- name: Login to Docker Hub
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Login to GHCR
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ secrets.GHCR_USERNAME }}
@@ -35,14 +35,14 @@ jobs:
org.opencontainers.image.vendor=Chris Lu
-
name: Set up QEMU
uses: docker/setup-qemu-action@v4
uses: docker/setup-qemu-action@v3
-
name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
-
name: Login to Docker Hub
if: github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
@@ -57,7 +57,7 @@ jobs:
fi
-
name: Build
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
with:
context: ./docker
push: ${{ github.event_name != 'pull_request' }}
+12 -12
View File
@@ -60,16 +60,16 @@ jobs:
build_args: TAGS=5BytesOffset
tag_suffix: _large_disk
# Full tags - multi-arch
# Full tags - amd64 only
- variant: full
platforms: linux/amd64,linux/arm64
platforms: linux/amd64
dockerfile: ./docker/Dockerfile.go_build
build_args: TAGS=elastic,gocdk,rclone,sqlite,tarantool,tikv,ydb
tag_suffix: _full
# Large disk + full tags - multi-arch
# Large disk + full tags - amd64 only
- variant: large_disk_full
platforms: linux/amd64,linux/arm64
platforms: linux/amd64
dockerfile: ./docker/Dockerfile.go_build
build_args: TAGS=5BytesOffset,elastic,gocdk,rclone,sqlite,tarantool,tikv,ydb
tag_suffix: _large_disk_full
@@ -117,7 +117,7 @@ jobs:
- name: Set up QEMU
if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && contains(matrix.platforms, 'arm')
uses: docker/setup-qemu-action@v4
uses: docker/setup-qemu-action@v3
- name: Create BuildKit config
if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
@@ -129,20 +129,20 @@ jobs:
- name: Set up Docker Buildx
if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
with:
buildkitd-config: /tmp/buildkitd.toml
- name: Login to Docker Hub
if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Login to GHCR
if: (github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant) && github.event_name != 'pull_request'
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ secrets.GHCR_USERNAME }}
@@ -150,7 +150,7 @@ jobs:
- name: Build and push ${{ matrix.variant }}
if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
env:
DOCKER_BUILDKIT: 1
with:
@@ -198,14 +198,14 @@ jobs:
steps:
- name: Login to Docker Hub
if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Login to GHCR
if: github.event_name != 'workflow_dispatch' || github.event.inputs.variant == 'all' || github.event.inputs.variant == matrix.variant
uses: docker/login-action@v4
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ secrets.GHCR_USERNAME }}
@@ -82,19 +82,19 @@ jobs:
echo "seaweedfs_ref=$seaweed" >> "$GITHUB_OUTPUT"
- name: Set up QEMU
uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v1
uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v1
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v1
uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v1
- name: Login to Docker Hub
uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v1
uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v1
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Build and push image
uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v2
uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v2
with:
context: ./docker
push: true
+1 -1
View File
@@ -26,7 +26,7 @@ jobs:
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: '1.24'
- name: Build Telemetry Server
if: github.event_name == 'workflow_dispatch' && inputs.deploy
+1 -1
View File
@@ -11,4 +11,4 @@ jobs:
- name: 'Checkout Repository'
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8
- name: 'Dependency Review'
uses: actions/dependency-review-action@2031cfc080254a8a887f58cffee85186f0e49e48
uses: actions/dependency-review-action@05fe4576374b728f0c523d6a13d64c25081e0803
+8 -7
View File
@@ -23,16 +23,17 @@ jobs:
runs-on: ubuntu-22.04
timeout-minutes: 30
steps:
- name: Check out code into the Go module directory
uses: actions/checkout@v6
- name: Set up Go
uses: actions/setup-go@v6
- name: Set up Go 1.x
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
with:
go-version-file: 'go.mod'
go-version: ^1.13
id: go
- name: Check out code into the Go module directory
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Cache Docker layers
uses: actions/cache@v5
+1 -1
View File
@@ -18,7 +18,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
id: go
- name: Check out code into the Go module directory
+1 -1
View File
@@ -28,7 +28,7 @@ jobs:
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: '1.24'
- name: Build weed binary
run: |
+3 -2
View File
@@ -22,6 +22,7 @@ permissions:
contents: read
env:
GO_VERSION: '1.24'
TEST_TIMEOUT: '45m'
jobs:
@@ -34,10 +35,10 @@ jobs:
- name: Checkout code
uses: actions/checkout@v6
- name: Set up Go
- name: Set up Go ${{ env.GO_VERSION }}
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: ${{ env.GO_VERSION }}
- name: Install FUSE and dependencies
run: |
+18 -15
View File
@@ -19,12 +19,13 @@ jobs:
name: Go Vet
runs-on: ubuntu-latest
steps:
- name: Check out code into the Go module directory
uses: actions/checkout@v6
- name: Set up Go
uses: actions/setup-go@v6
- name: Set up Go 1.x
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
with:
go-version-file: 'go.mod'
go-version: ^1.13
id: go
- name: Check out code into the Go module directory
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
- name: Get dependencies
run: |
cd weed; go get -v -t -d ./...
@@ -41,12 +42,13 @@ jobs:
name: Build
runs-on: ubuntu-latest
steps:
- name: Check out code into the Go module directory
uses: actions/checkout@v6
- name: Set up Go
uses: actions/setup-go@v6
- name: Set up Go 1.x
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
with:
go-version-file: 'go.mod'
go-version: ^1.13
id: go
- name: Check out code into the Go module directory
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
- name: Build
run: cd weed; go build -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v .
@@ -54,11 +56,12 @@ jobs:
name: Test
runs-on: ubuntu-latest
steps:
- name: Check out code into the Go module directory
uses: actions/checkout@v6
- name: Set up Go
uses: actions/setup-go@v6
- name: Set up Go 1.x
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
with:
go-version-file: 'go.mod'
go-version: ^1.13
id: go
- name: Check out code into the Go module directory
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
- name: Test
run: cd weed; go test -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v ./...
+2 -2
View File
@@ -26,14 +26,14 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
cache-dependency-path: |
**/go.sum
id: go
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Install dependencies
run: |
+7 -7
View File
@@ -37,7 +37,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
id: go
- name: Check out code
@@ -82,7 +82,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
id: go
- name: Check out code
@@ -132,7 +132,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
cache-dependency-path: |
**/go.sum
@@ -311,7 +311,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
cache-dependency-path: |
**/go.sum
@@ -473,7 +473,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
cache-dependency-path: |
**/go.sum
@@ -631,7 +631,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
cache-dependency-path: |
**/go.sum
@@ -789,7 +789,7 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
id: go
- name: Check out code
@@ -30,6 +30,7 @@ permissions:
contents: read
env:
GO_VERSION: '1.24'
TEST_TIMEOUT: '10m'
jobs:
@@ -42,10 +43,10 @@ jobs:
- name: Checkout code
uses: actions/checkout@v6
- name: Set up Go
- name: Set up Go ${{ env.GO_VERSION }}
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: ${{ env.GO_VERSION }}
- name: Build SeaweedFS
run: |
+2 -2
View File
@@ -25,14 +25,14 @@ jobs:
- name: Set up Go 1.x
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
id: go
- name: Check out code
uses: actions/checkout@v6
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Cache Docker layers
uses: actions/cache@v5
+2 -11
View File
@@ -5,8 +5,6 @@ on:
paths:
- 'weed/iam/**'
- 'weed/s3api/**'
- 'weed/credential/**'
- 'weed/pb/**'
- 'test/s3/iam/**'
- '.github/workflows/s3-iam-tests.yml'
push:
@@ -14,8 +12,6 @@ on:
paths:
- 'weed/iam/**'
- 'weed/s3api/**'
- 'weed/credential/**'
- 'weed/pb/**'
- 'test/s3/iam/**'
- '.github/workflows/s3-iam-tests.yml'
@@ -84,7 +80,7 @@ jobs:
timeout-minutes: 25
strategy:
matrix:
test-type: ["basic", "advanced", "policy-enforcement", "group"]
test-type: ["basic", "advanced", "policy-enforcement"]
steps:
- name: Check out code
@@ -121,7 +117,7 @@ jobs:
"basic")
echo "Running basic IAM functionality tests..."
make clean setup start-services wait-for-services
go test -v -timeout 15m -run "TestS3IAMAuthentication|TestS3IAMBasicWorkflow|TestS3IAMTokenValidation|TestIAMUserManagement|TestIAMAccessKeyManagement|TestIAMPolicyManagement" ./...
go test -v -timeout 15m -run "TestS3IAMAuthentication|TestS3IAMBasicWorkflow|TestS3IAMTokenValidation|TestIAM" ./...
;;
"advanced")
echo "Running advanced IAM feature tests..."
@@ -133,11 +129,6 @@ jobs:
make clean setup start-services wait-for-services
go test -v -timeout 15m -run "TestS3IAMPolicyEnforcement|TestS3IAMBucketPolicy|TestS3IAMContextual" ./...
;;
"group")
echo "Running IAM group management tests..."
make clean setup start-services wait-for-services
go test -v -timeout 15m -run "TestIAMGroup" ./...
;;
*)
echo "Unknown test type: ${{ matrix.test-type }}"
exit 1
+2 -2
View File
@@ -41,7 +41,7 @@ jobs:
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
- name: Set up Python ${{ matrix.python-version }}
@@ -148,7 +148,7 @@ jobs:
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
cache: true
- name: Run Go unit tests
@@ -29,7 +29,7 @@ jobs:
id: go
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Build SeaweedFS binary for Linux
run: |
+1 -1
View File
@@ -33,7 +33,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Install SeaweedFS
run: |
+5 -5
View File
@@ -144,7 +144,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Pre-pull Trino image
run: docker pull trinodb/trino:479
@@ -271,7 +271,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Pre-pull Spark image
run: docker pull apache/spark:3.5.1
@@ -337,7 +337,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Pre-pull RisingWave image
run: |
@@ -405,7 +405,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Pre-pull Python image
run: docker pull python:3
@@ -471,7 +471,7 @@ jobs:
id: go
- name: Set up Docker
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
- name: Pre-pull Python image
run: docker pull python:3
+3 -2
View File
@@ -24,6 +24,7 @@ permissions:
contents: read
env:
GO_VERSION: '1.24'
TEST_TIMEOUT: '15m'
jobs:
@@ -36,10 +37,10 @@ jobs:
- name: Checkout code
uses: actions/checkout@v6
- name: Set up Go
- name: Set up Go ${{ env.GO_VERSION }}
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: ${{ env.GO_VERSION }}
- name: Install dependencies
run: |
@@ -43,7 +43,7 @@ jobs:
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: '1.24'
- name: Build SeaweedFS binary
run: |
@@ -24,7 +24,7 @@ jobs:
- uses: actions/setup-go@v6
with:
go-version: ^1.25
go-version: ^1.24
- name: Build SeaweedFS
run: |
@@ -28,6 +28,7 @@ permissions:
contents: read
env:
GO_VERSION: '1.24'
TEST_TIMEOUT: '30m'
jobs:
@@ -45,10 +46,10 @@ jobs:
- name: Checkout code
uses: actions/checkout@v6
- name: Set up Go
- name: Set up Go ${{ env.GO_VERSION }}
uses: actions/setup-go@v6
with:
go-version-file: 'go.mod'
go-version: ${{ env.GO_VERSION }}
- name: Build SeaweedFS binary
run: |
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25 AS builder
FROM golang:1.24 AS builder
RUN apt-get update && \
apt-get install -y build-essential wget ca-certificates && \
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine as builder
RUN apk add git g++ fuse
RUN mkdir -p /go/src/github.com/seaweedfs/
ARG BRANCH=${BRANCH:-master}
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25 AS builder
FROM golang:1.24 AS builder
RUN apt-get update
RUN apt-get install -y build-essential libsnappy-dev zlib1g-dev libbz2-dev libgflags-dev liblz4-dev libzstd-dev
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25 AS builder
FROM golang:1.24 AS builder
RUN apt-get update
RUN apt-get install -y build-essential libsnappy-dev zlib1g-dev libbz2-dev libgflags-dev liblz4-dev libzstd-dev
-1
View File
@@ -7,7 +7,6 @@
[master.maintenance]
# periodically run these scripts are the same as running them from 'weed shell'
# Scripts are skipped while an admin server is connected.
scripts = """
lock
ec.encode -fullPercent=95 -quietFor=1h
+7 -7
View File
@@ -1,6 +1,6 @@
module github.com/seaweedfs/seaweedfs
go 1.25.0
go 1.24.9
require (
cloud.google.com/go v0.123.0 // indirect
@@ -63,7 +63,7 @@ require (
github.com/prometheus/client_golang v1.23.2
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.67.2 // indirect
github.com/prometheus/procfs v0.20.1
github.com/prometheus/procfs v0.19.2
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/seaweedfs/goexif v1.0.3
@@ -87,7 +87,7 @@ require (
github.com/xdg-go/stringprep v1.0.4 // indirect
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
go.etcd.io/etcd/client/v3 v3.6.7
go.mongodb.org/mongo-driver v1.17.9
go.mongodb.org/mongo-driver v1.17.6
go.opencensus.io v0.24.0 // indirect
gocloud.dev v0.45.0
gocloud.dev/pubsub/natspubsub v0.45.0
@@ -97,7 +97,7 @@ require (
golang.org/x/image v0.36.0
golang.org/x/net v0.49.0
golang.org/x/oauth2 v0.35.0
golang.org/x/sys v0.42.0
golang.org/x/sys v0.41.0
golang.org/x/text v0.34.0 // indirect
golang.org/x/tools v0.41.0 // indirect
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
@@ -124,13 +124,13 @@ require (
github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3
github.com/arangodb/go-driver v1.6.9
github.com/armon/go-metrics v0.4.1
github.com/aws/aws-sdk-go-v2 v1.41.3
github.com/aws/aws-sdk-go-v2 v1.41.1
github.com/aws/aws-sdk-go-v2/config v1.32.7
github.com/aws/aws-sdk-go-v2/credentials v1.19.7
github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0
github.com/cognusion/imaging v1.0.2
github.com/fluent/fluent-logger-golang v1.10.1
github.com/getsentry/sentry-go v0.43.0
github.com/getsentry/sentry-go v0.42.0
github.com/go-ldap/ldap/v3 v3.4.12
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/google/flatbuffers/go v0.0.0-20230108230133-3b8644d32c50
@@ -341,7 +341,7 @@ require (
github.com/aws/aws-sdk-go-v2/service/sso v1.30.9 // indirect
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 // indirect
github.com/aws/aws-sdk-go-v2/service/sts v1.41.6
github.com/aws/smithy-go v1.24.2
github.com/aws/smithy-go v1.24.0
github.com/boltdb/bolt v1.3.1 // indirect
github.com/bradenaw/juniper v0.15.3 // indirect
github.com/bradfitz/iter v0.0.0-20191230175014-e8f45d346db8 // indirect
+12 -12
View File
@@ -716,8 +716,8 @@ github.com/armon/go-metrics v0.4.1/go.mod h1:E6amYzXo6aW1tqzoZGT755KkbgrJsSdpwZ+
github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
github.com/aws/aws-sdk-go v1.55.8 h1:JRmEUbU52aJQZ2AjX4q4Wu7t4uZjOu71uyNmaWlUkJQ=
github.com/aws/aws-sdk-go v1.55.8/go.mod h1:ZkViS9AqA6otK+JBBNH2++sx1sgxrPKcSzPPvQkUtXk=
github.com/aws/aws-sdk-go-v2 v1.41.3 h1:4kQ/fa22KjDt13QCy1+bYADvdgcxpfH18f0zP542kZA=
github.com/aws/aws-sdk-go-v2 v1.41.3/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o=
github.com/aws/aws-sdk-go-v2 v1.41.1 h1:ABlyEARCDLN034NhxlRUSZr4l71mh+T5KAeGh6cerhU=
github.com/aws/aws-sdk-go-v2 v1.41.1/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
github.com/aws/aws-sdk-go-v2/config v1.32.7 h1:vxUyWGUwmkQ2g19n7JY/9YL8MfAIl7bTesIUykECXmY=
@@ -758,8 +758,8 @@ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 h1:gd84Omyu9JLriJVCbGApcLz
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13/go.mod h1:sTGThjphYE4Ohw8vJiRStAcu3rbjtXRsdNB0TvZ5wwo=
github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 h1:5fFjR/ToSOzB2OQ/XqWpZBmNvmP/pJ1jOWYlFDJTjRQ=
github.com/aws/aws-sdk-go-v2/service/sts v1.41.6/go.mod h1:qgFDZQSD/Kys7nJnVqYlWKnh0SSdMjAi0uSwON4wgYQ=
github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng=
github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
github.com/aws/smithy-go v1.24.0 h1:LpilSUItNPFr1eY85RYgTIg5eIEPtvFbskaFcmmIUnk=
github.com/aws/smithy-go v1.24.0/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0=
github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk=
github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg=
github.com/bazelbuild/rules_go v0.46.0 h1:CTefzjN/D3Cdn3rkrM6qMWuQj59OBcuOjyIp3m4hZ7s=
@@ -1053,8 +1053,8 @@ github.com/gabriel-vasile/mimetype v1.4.11 h1:AQvxbp830wPhHTqc1u7nzoLT+ZFxGY7emj
github.com/gabriel-vasile/mimetype v1.4.11/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s=
github.com/geoffgarside/ber v1.2.0 h1:/loowoRcs/MWLYmGX9QtIAbA+V/FrnVLsMMPhwiRm64=
github.com/geoffgarside/ber v1.2.0/go.mod h1:jVPKeCbj6MvQZhwLYsGwaGI52oUorHoHKNecGT85ZCc=
github.com/getsentry/sentry-go v0.43.0 h1:XbXLpFicpo8HmBDaInk7dum18G9KSLcjZiyUKS+hLW4=
github.com/getsentry/sentry-go v0.43.0/go.mod h1:XDotiNZbgf5U8bPDUAfvcFmOnMQQceESxyKaObSssW0=
github.com/getsentry/sentry-go v0.42.0 h1:eeFMACuZTbUQf90RE8dE4tXeSe4CZyfvR1MBL7RLEt8=
github.com/getsentry/sentry-go v0.42.0/go.mod h1:eRXCoh3uvmjQLY6qu63BjUZnaBu5L5WhMV1RwYO8W5s=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w=
github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM=
@@ -1769,8 +1769,8 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc=
github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws=
github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
github.com/pterm/pterm v0.12.27/go.mod h1:PhQ89w4i95rhgE+xedAoqous6K9X+r6aSOI2eFF7DZI=
github.com/pterm/pterm v0.12.29/go.mod h1:WI3qxgvoQFFGKGjGnJR849gU0TsEOvKn5Q8LlY1U7lg=
github.com/pterm/pterm v0.12.30/go.mod h1:MOqLIyMOgmTDz9yorcYbcw+HsgoZo3BQfg2wtl3HEFE=
@@ -2113,8 +2113,8 @@ go.etcd.io/etcd/client/pkg/v3 v3.6.7 h1:vvzgyozz46q+TyeGBuFzVuI53/yd133CHceNb/Ah
go.etcd.io/etcd/client/pkg/v3 v3.6.7/go.mod h1:2IVulJ3FZ/czIGl9T4lMF1uxzrhRahLqe+hSgy+Kh7Q=
go.etcd.io/etcd/client/v3 v3.6.7 h1:9WqA5RpIBtdMxAy1ukXLAdtg2pAxNqW5NUoO2wQrE6U=
go.etcd.io/etcd/client/v3 v3.6.7/go.mod h1:2XfROY56AXnUqGsvl+6k29wrwsSbEh1lAouQB1vHpeE=
go.mongodb.org/mongo-driver v1.17.9 h1:IexDdCuuNJ3BHrELgBlyaH9p60JXAvdzWR128q+U5tU=
go.mongodb.org/mongo-driver v1.17.9/go.mod h1:LlOhpH5NUEfhxcAwG0UEkMqwYcc4JU18gtCdGudk/tQ=
go.mongodb.org/mongo-driver v1.17.6 h1:87JUG1wZfWsr6rIz3ZmpH90rL5tea7O3IHuSwHUpsss=
go.mongodb.org/mongo-driver v1.17.6/go.mod h1:Hy04i7O2kC4RS06ZrhPRqj/u4DTYkFDAAccj+rVKqgQ=
go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
@@ -2509,8 +2509,8 @@ golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
golang.org/x/telemetry v0.0.0-20260109210033-bd525da824e2 h1:O1cMQHRfwNpDfDJerqRoE2oD+AFlyid87D40L/OkkJo=
golang.org/x/telemetry v0.0.0-20260109210033-bd525da824e2/go.mod h1:b7fPSJ0pKZ3ccUh8gnTONJxhn3c/PS6tyzQvyqw4iA8=
+2 -2
View File
@@ -1,6 +1,6 @@
apiVersion: v1
description: SeaweedFS
name: seaweedfs
appVersion: "4.17"
appVersion: "4.13"
# Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
version: 4.17.0
version: 4.0.413
+1 -10
View File
@@ -23,15 +23,8 @@
#
# Adjust storageClass and sizes to match your cluster's available StorageClasses.
# On OpenShift you can discover them with: oc get storageclass
global:
enableReplication: true
# replication type is XYZ:
# X number of replica in other data centers
# Y number of replica in other racks in the same data center
# Z number of replica in other servers in the same rack
replicationPlacement: "000" # no data replica
master:
replicas: 1
data:
type: "persistentVolumeClaim"
size: "10Gi"
@@ -56,7 +49,6 @@ master:
type: RuntimeDefault
volume:
replicas: 1
dataDirs:
- name: data1
type: "persistentVolumeClaim"
@@ -83,7 +75,6 @@ volume:
type: RuntimeDefault
filer:
replicas: 1
data:
type: "persistentVolumeClaim"
size: "25Gi"
@@ -243,7 +243,8 @@ spec:
{{- if $httpsPort }}
-s3.port.https={{ $httpsPort }} \
{{- end }}
{{ include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
-s3.cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-s3.key.file=/usr/local/share/ca-certificates/client/tls.key \
{{- end }}
{{- if or .Values.allInOne.s3.enableAuth .Values.s3.enableAuth .Values.filer.s3.enableAuth }}
-s3.config=/etc/sw/s3/seaweedfs_s3_config \
@@ -345,9 +346,6 @@ spec:
- name: client-cert
mountPath: /usr/local/share/ca-certificates/client/
readOnly: true
{{- if .Values.allInOne.s3.enabled }}
{{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
{{- end }}
{{- end }}
{{ tpl .Values.allInOne.extraVolumeMounts . | nindent 12 }}
ports:
@@ -475,9 +473,6 @@ spec:
- name: client-cert
secret:
secretName: {{ include "seaweedfs.fullname" . }}-client-cert
{{- if .Values.allInOne.s3.enabled }}
{{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
{{- end }}
{{- end }}
{{ tpl .Values.allInOne.extraVolumes . | nindent 8 }}
{{- if .Values.allInOne.nodeSelector }}
@@ -17,9 +17,6 @@ metadata:
spec:
type: {{ .Values.allInOne.service.type | default "ClusterIP" }}
internalTrafficPolicy: {{ .Values.allInOne.service.internalTrafficPolicy | default "Cluster" }}
{{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) .Values.allInOne.s3.trafficDistribution }}
trafficDistribution: {{ include "seaweedfs.trafficDistribution" (dict "value" .Values.allInOne.s3.trafficDistribution "Capabilities" .Capabilities) }}
{{- end }}
ports:
# Master ports
- name: "swfs-master"
@@ -200,7 +200,8 @@ spec:
{{- if .Values.filer.s3.httpsPort }}
-s3.port.https={{ .Values.filer.s3.httpsPort }} \
{{- end }}
{{ include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }}
-s3.cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-s3.key.file=/usr/local/share/ca-certificates/client/tls.key \
{{- end }}
{{- if .Values.filer.s3.enableAuth }}
-s3.config=/etc/sw/seaweedfs_s3_config \
@@ -253,9 +254,6 @@ spec:
- name: client-cert
readOnly: true
mountPath: /usr/local/share/ca-certificates/client
{{- if .Values.filer.s3.enabled }}
{{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
{{- end }}
{{- end }}
{{ tpl .Values.filer.extraVolumeMounts . | nindent 12 | trim }}
ports:
@@ -386,9 +384,6 @@ spec:
- name: client-cert
secret:
secretName: {{ include "seaweedfs.fullname" . }}-client-cert
{{- if .Values.filer.s3.enabled }}
{{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
{{- end }}
{{- end }}
{{ tpl .Values.filer.extraVolumes . | indent 8 | trim }}
{{- if .Values.filer.nodeSelector }}
@@ -127,7 +127,8 @@ spec:
{{- if .Values.s3.httpsPort }}
-port.https={{ .Values.s3.httpsPort }} \
{{- end }}
{{ include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "") | nindent 14 }}
-cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-key.file=/usr/local/share/ca-certificates/client/tls.key \
{{- end }}
{{- if .Values.s3.domainName }}
-domainName={{ .Values.s3.domainName }} \
@@ -175,7 +176,6 @@ spec:
- name: client-cert
readOnly: true
mountPath: /usr/local/share/ca-certificates/client/
{{- include "seaweedfs.s3.tlsVolumeMount" . | nindent 12 }}
{{- end }}
{{ tpl .Values.s3.extraVolumeMounts . | nindent 12 | trim }}
ports:
@@ -267,7 +267,6 @@ spec:
- name: client-cert
secret:
secretName: {{ include "seaweedfs.fullname" . }}-client-cert
{{- include "seaweedfs.s3.tlsVolume" . | nindent 8 }}
{{- end }}
{{ tpl .Values.s3.extraVolumes . | indent 8 | trim }}
{{- if .Values.s3.nodeSelector }}
@@ -16,9 +16,8 @@ metadata:
{{- end }}
spec:
internalTrafficPolicy: {{ .Values.s3.internalTrafficPolicy | default "Cluster" }}
{{- $td := .Values.s3.trafficDistribution | default .Values.filer.s3.trafficDistribution }}
{{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) $td }}
trafficDistribution: {{ include "seaweedfs.trafficDistribution" (dict "value" $td "Capabilities" .Capabilities) }}
{{- if and (semverCompare ">=1.31-0" .Capabilities.KubeVersion.GitVersion) (or .Values.s3.trafficDistribution .Values.filer.s3.trafficDistribution) }}
trafficDistribution: {{ include "seaweedfs.trafficDistribution" . }}
{{- end }}
ports:
- name: "swfs-s3"
@@ -338,41 +338,11 @@ Create the name of the service account to use
{{- .Values.global.serviceAccountName | default "seaweedfs" -}}
{{- end -}}
{{/* S3 TLS cert/key arguments, using custom secret if s3.tlsSecret is set */}}
{{- define "seaweedfs.s3.tlsArgs" -}}
{{- $prefix := .prefix -}}
{{- $root := .root -}}
{{- if $root.Values.s3.tlsSecret -}}
-{{ $prefix }}cert.file=/usr/local/share/ca-certificates/s3/tls.crt \
-{{ $prefix }}key.file=/usr/local/share/ca-certificates/s3/tls.key \
{{- else -}}
-{{ $prefix }}cert.file=/usr/local/share/ca-certificates/client/tls.crt \
-{{ $prefix }}key.file=/usr/local/share/ca-certificates/client/tls.key \
{{- end -}}
{{- end -}}
{{/* S3 custom TLS volume mount */}}
{{- define "seaweedfs.s3.tlsVolumeMount" -}}
{{- if .Values.s3.tlsSecret }}
- name: s3-tls-cert
readOnly: true
mountPath: /usr/local/share/ca-certificates/s3/
{{- end }}
{{- end -}}
{{/* S3 custom TLS volume */}}
{{- define "seaweedfs.s3.tlsVolume" -}}
{{- if .Values.s3.tlsSecret }}
- name: s3-tls-cert
secret:
secretName: {{ .Values.s3.tlsSecret }}
{{- end }}
{{- end -}}
{{/* Generate a compatible trafficDistribution value due to "PreferClose" fast deprecation in k8s v1.35.
Accepts a dict with "value" (the trafficDistribution string) and "Capabilities". */}}
{{/* Generate a compatible trafficDistribution value due to "PreferClose" fast deprecation in k8s v1.35 */}}
{{- define "seaweedfs.trafficDistribution" -}}
{{- if .value -}}
{{- and (eq .value "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .value -}}
{{- if .Values.s3.trafficDistribution -}}
{{- and (eq .Values.s3.trafficDistribution "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .Values.s3.trafficDistribution -}}
{{- else if .Values.filer.s3.trafficDistribution -}}
{{- and (eq .Values.filer.s3.trafficDistribution "PreferClose") (semverCompare ">=1.35-0" .Capabilities.KubeVersion.GitVersion) | ternary "PreferSameZone" .Values.filer.s3.trafficDistribution -}}
{{- end -}}
{{- end -}}
@@ -51,7 +51,7 @@ metadata:
annotations:
"helm.sh/hook": post-install,post-upgrade
"helm.sh/hook-weight": "-5"
"helm.sh/hook-delete-policy": before-hook-creation,hook-succeeded
"helm.sh/hook-delete-policy": hook-succeeded
spec:
template:
metadata:
@@ -64,11 +64,9 @@ spec:
{{- if .Values.filer.podSecurityContext.enabled }}
securityContext: {{- omit .Values.filer.podSecurityContext "enabled" | toYaml | nindent 8 }}
{{- end }}
{{- include "seaweedfs.imagePullSecrets" $ | nindent 6 }}
containers:
- name: post-install-job
image: {{ template "master.image" . }}
imagePullPolicy: {{ $.Values.global.imagePullPolicy | default "IfNotPresent" }}
env:
- name: WEED_CLUSTER_DEFAULT
value: "sw"
@@ -94,7 +92,6 @@ spec:
- "/bin/sh"
- "-ec"
- |
set -o pipefail
wait_for_service() {
local url=$1
local max_attempts=60 # 5 minutes total (5s * 60)
@@ -120,7 +117,8 @@ spec:
wait_for_service "http://$WEED_CLUSTER_SW_MASTER{{ .Values.master.readinessProbe.httpGet.path }}"
wait_for_service "http://$WEED_CLUSTER_SW_FILER{{ .Values.filer.readinessProbe.httpGet.path }}"
{{- end }}
{{- range $createBuckets }}
set -o pipefail
{{- range $createBuckets }}
{{- $bucketName := .name }}
{{- $bucketLock := or .lock .objectLock .withLock }}
bucket_list=$(/bin/echo 's3.bucket.list' | /usr/bin/weed shell) || { echo "Error listing s3 buckets"; exit 1; }
@@ -189,10 +187,6 @@ spec:
{{- end }}
- containerPort: {{ .Values.master.grpcPort }}
#name: swfs-master-grpc
{{- with coalesce .Values.allInOne.s3.createBucketsHook.resources .Values.s3.createBucketsHook.resources .Values.filer.s3.createBucketsHook.resources }}
resources:
{{- toYaml . | nindent 10 }}
{{- end }}
{{- if .Values.filer.containerSecurityContext.enabled }}
securityContext: {{- omit .Values.filer.containerSecurityContext "enabled" | toYaml | nindent 12 }}
{{- end }}
-21
View File
@@ -909,10 +909,6 @@ filer:
# versioning: Enabled
# - name: bucket-b
# anonymousRead: false
# Resource requests/limits for the post-install bucket creation hook job.
# Should map directly to the value of the resources field for a PodSpec.
createBucketsHook:
resources: {}
s3:
enabled: false
@@ -923,13 +919,6 @@ s3:
port: 8333
# add additional https port
httpsPort: 0
# Use a custom TLS certificate secret for the S3 HTTPS endpoint.
# When set, this Kubernetes Secret (must contain tls.crt and tls.key) is used
# instead of the internal self-signed client certificate generated by cert-manager.
# This allows using a publicly trusted certificate (e.g., from Let's Encrypt)
# so that S3 clients don't need to trust the internal CA.
# Requires global.enableSecurity to be true.
tlsSecret: null
metricsPort: 9327
# Iceberg catalog REST port (Apache Iceberg REST Catalog API)
# Set to a port number to enable, or 0/null to disable
@@ -1080,11 +1069,6 @@ s3:
failureThreshold: 100
timeoutSeconds: 10
# Resource requests/limits for the post-install bucket creation hook job.
# Should map directly to the value of the resources field for a PodSpec.
createBucketsHook:
resources: {}
ingress:
enabled: false
className: ""
@@ -1469,7 +1453,6 @@ allInOne:
# The s3-secret.yaml template only reads from .Values.s3.credentials.
# See: s3.credentials.admin.accessKey, s3.credentials.read.accessKey
auditLogConfig: null # S3 audit log configuration (null inherits from s3.auditLogConfig)
trafficDistribution: null # Service traffic distribution (e.g., "PreferClose"); auto-converts to "PreferSameZone" on k8s >=1.35
# You may specify buckets to be created during the install process.
# Buckets may be exposed publicly by setting `anonymousRead` to `true`
# ttl format: [1-255][m|h|d|w|M|y] (e.g., 7d)
@@ -1483,10 +1466,6 @@ allInOne:
# versioning: Enabled
# - name: bucket-b
# anonymousRead: false
# Resource requests/limits for the post-install bucket creation hook job.
# Should map directly to the value of the resources field for a PodSpec.
createBucketsHook:
resources: {}
# SFTP server configuration
# Note: Most parameters below default to null, which means they inherit from
@@ -100,12 +100,10 @@ message ListEntriesRequest {
string startFromFileName = 3;
bool inclusiveStartFrom = 4;
uint32 limit = 5;
int64 snapshot_ts_ns = 6;
}
message ListEntriesResponse {
Entry entry = 1;
int64 snapshot_ts_ns = 2;
}
message RemoteEntry {
@@ -205,7 +203,6 @@ message CreateEntryRequest {
message CreateEntryResponse {
string error = 1;
SubscribeMetadataResponse metadata_event = 2;
}
message UpdateEntryRequest {
@@ -215,7 +212,6 @@ message UpdateEntryRequest {
repeated int32 signatures = 4;
}
message UpdateEntryResponse {
SubscribeMetadataResponse metadata_event = 1;
}
message AppendToEntryRequest {
@@ -240,7 +236,6 @@ message DeleteEntryRequest {
message DeleteEntryResponse {
string error = 1;
SubscribeMetadataResponse metadata_event = 2;
}
message AtomicRenameEntryRequest {
@@ -474,7 +469,6 @@ message CacheRemoteObjectToLocalClusterRequest {
}
message CacheRemoteObjectToLocalClusterResponse {
Entry entry = 1;
SubscribeMetadataResponse metadata_event = 2;
}
/////////////////////////
+1 -1
View File
@@ -1,5 +1,5 @@
# Multi-stage build for Go Sidecar
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Install build dependencies
RUN apk add --no-cache git ca-certificates tzdata
@@ -1,5 +1,5 @@
# Multi-stage build for Test Client
FROM golang:1.25-alpine AS builder
FROM golang:1.23-alpine AS builder
# Install build dependencies
RUN apk add --no-cache git ca-certificates tzdata
@@ -135,7 +135,7 @@ func ensureEnvironment(t *testing.T) {
// 6. Start Admin
os.RemoveAll(filepath.Join("tmp", "admin"))
os.MkdirAll(filepath.Join("tmp", "admin"), 0755)
startWeed(t, "admin", "admin", "-master=localhost:9333", "-port=23646", "-dataDir=./tmp/admin")
startWeed(t, "admin", "admin", "-master=localhost:9333", "-port=23646", "-dataDir=./tmp/admin", "-scheduler.idleSleep=2")
waitForUrl(t, AdminUrl+"/health", 60)
t.Log("Environment started successfully")
@@ -153,38 +153,6 @@ func waitForUrl(t *testing.T, url string, retries int) {
t.Fatalf("Timeout waiting for %s", url)
}
func fetchJSON(url string, out interface{}) error {
resp, err := http.Get(url)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("status %d: %s", resp.StatusCode, string(body))
}
return json.NewDecoder(resp.Body).Decode(out)
}
func mapField(obj map[string]interface{}, key string) (interface{}, bool) {
if obj == nil {
return nil, false
}
if value, ok := obj[key]; ok {
return value, true
}
return nil, false
}
func mapFieldAny(obj map[string]interface{}, keys ...string) (interface{}, bool) {
for _, key := range keys {
if value, ok := mapField(obj, key); ok {
return value, true
}
}
return nil, false
}
func TestEcEndToEnd(t *testing.T) {
defer cleanup()
ensureEnvironment(t)
@@ -307,7 +275,6 @@ func TestEcEndToEnd(t *testing.T) {
startTime := time.Now()
ecVerified := false
var lastBody []byte
debugTick := 0
for time.Since(startTime) < 300*time.Second {
// 3.1 Check Master Topology
@@ -333,104 +300,25 @@ func TestEcEndToEnd(t *testing.T) {
}
}
// 3.2 Debug: Check workers, jobs, and scheduler status
debugTick++
var workers []map[string]interface{}
// 3.2 Debug: Check workers and jobs
wResp, wErr := http.Get(AdminUrl + "/api/plugin/workers")
workerCount := 0
ecDetectorCount := 0
ecExecutorCount := 0
if err := fetchJSON(AdminUrl+"/api/plugin/workers", &workers); err == nil {
if wErr == nil {
var workers []interface{}
json.NewDecoder(wResp.Body).Decode(&workers)
wResp.Body.Close()
workerCount = len(workers)
for _, worker := range workers {
capsValue, ok := mapFieldAny(worker, "capabilities", "Capabilities")
if !ok {
continue
}
caps, ok := capsValue.(map[string]interface{})
if !ok {
continue
}
if capValue, ok := caps["erasure_coding"].(map[string]interface{}); ok {
if capValue["can_detect"] == true {
ecDetectorCount++
}
if capValue["can_execute"] == true {
ecExecutorCount++
}
}
}
}
var tasks []map[string]interface{}
tResp, tErr := http.Get(AdminUrl + "/api/plugin/jobs?limit=1000")
taskCount := 0
ecTaskCount := 0
ecTaskStates := map[string]int{}
if err := fetchJSON(AdminUrl+"/api/plugin/jobs?limit=1000", &tasks); err == nil {
if tErr == nil {
var tasks []interface{}
json.NewDecoder(tResp.Body).Decode(&tasks)
tResp.Body.Close()
taskCount = len(tasks)
for _, task := range tasks {
jobType, _ := task["job_type"].(string)
state, _ := task["state"].(string)
if jobType == "erasure_coding" {
ecTaskCount++
ecTaskStates[state]++
}
}
}
t.Logf("Waiting for EC... (Workers: %d det=%d exec=%d, Tasks: %d ec=%d, EC States: %+v)",
workerCount, ecDetectorCount, ecExecutorCount, taskCount, ecTaskCount, ecTaskStates)
if debugTick%3 == 0 {
var pluginStatus map[string]interface{}
if err := fetchJSON(AdminUrl+"/api/plugin/status", &pluginStatus); err == nil {
t.Logf("Plugin status: enabled=%v worker_count=%v worker_grpc_port=%v configured=%v",
pluginStatus["enabled"], pluginStatus["worker_count"], pluginStatus["worker_grpc_port"], pluginStatus["configured"])
}
var schedulerStatus map[string]interface{}
if err := fetchJSON(AdminUrl+"/api/plugin/scheduler-status", &schedulerStatus); err == nil {
if schedValue, ok := schedulerStatus["scheduler"].(map[string]interface{}); ok {
t.Logf("Scheduler status: current_job_type=%v phase=%v last_iteration_had_jobs=%v idle_sleep_seconds=%v last_iteration_done_at=%v next_detection_at=%v",
schedValue["current_job_type"], schedValue["current_phase"],
schedValue["last_iteration_had_jobs"], schedValue["idle_sleep_seconds"], schedValue["last_iteration_done_at"], schedValue["next_detection_at"])
} else {
t.Logf("Scheduler status: %v", schedulerStatus)
}
}
var schedulerStates []map[string]interface{}
if err := fetchJSON(AdminUrl+"/api/plugin/scheduler-states", &schedulerStates); err == nil {
for _, state := range schedulerStates {
if state["job_type"] == "erasure_coding" {
t.Logf("EC scheduler state: enabled=%v detection_in_flight=%v detector_available=%v executor_workers=%v next_detection_at=%v last_run_status=%v last_run_started_at=%v last_run_completed_at=%v",
state["enabled"], state["detection_in_flight"], state["detector_available"],
state["executor_worker_count"], state["next_detection_at"], state["last_run_status"],
state["last_run_started_at"], state["last_run_completed_at"])
break
}
}
}
var jobTypes []map[string]interface{}
if err := fetchJSON(AdminUrl+"/api/plugin/job-types", &jobTypes); err == nil {
var names []string
for _, jobType := range jobTypes {
if name, ok := jobType["job_type"].(string); ok && name != "" {
names = append(names, name)
}
}
t.Logf("Plugin job types: %v", names)
}
var activities []map[string]interface{}
if err := fetchJSON(AdminUrl+"/api/plugin/activities?job_type=erasure_coding&limit=5", &activities); err == nil {
for i := len(activities) - 1; i >= 0; i-- {
act := activities[i]
t.Logf("EC activity: stage=%v message=%v occurred_at=%v", act["stage"], act["message"], act["occurred_at"])
}
}
}
t.Logf("Waiting for EC... (Workers: %d, Active Tasks: %d)", workerCount, taskCount)
time.Sleep(10 * time.Second)
}
+1 -1
View File
@@ -1,6 +1,6 @@
# Simplified single-stage build for SeaweedFS with FoundationDB support
# Force x86_64 platform to use AMD64 FoundationDB packages
FROM --platform=linux/amd64 golang:1.25-bookworm
FROM --platform=linux/amd64 golang:1.24-bookworm
ARG FOUNDATIONDB_VERSION=7.4.5
ENV FOUNDATIONDB_VERSION=${FOUNDATIONDB_VERSION}
+1 -1
View File
@@ -1,5 +1,5 @@
# Multi-stage Dockerfile to build SeaweedFS with FoundationDB support for ARM64
FROM --platform=linux/arm64 golang:1.25-bookworm AS builder
FROM --platform=linux/arm64 golang:1.24-bookworm AS builder
ARG FOUNDATIONDB_VERSION=7.4.5
ENV FOUNDATIONDB_VERSION=${FOUNDATIONDB_VERSION}
+1 -1
View File
@@ -1,5 +1,5 @@
# Test environment with Go and FoundationDB support
FROM golang:1.25-bookworm
FROM golang:1.24-bookworm
# Install system dependencies
RUN apt-get update && apt-get install -y \
+1 -1
View File
@@ -123,7 +123,7 @@
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.9.5</version>
<version>3.9.4</version>
</dependency>
<!-- Apache Commons - Fix CVEs -->
+1 -1
View File
@@ -1,5 +1,5 @@
# Dockerfile for Kafka Gateway Integration Testing
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Install build dependencies
RUN apk add --no-cache git make gcc musl-dev sqlite-dev
+1 -1
View File
@@ -1,5 +1,5 @@
# Dockerfile for building SeaweedFS components from the current workspace
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
RUN apk add --no-cache git make gcc musl-dev sqlite-dev
+1 -1
View File
@@ -1,5 +1,5 @@
# Dockerfile for Kafka Integration Test Setup
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Install build dependencies
RUN apk add --no-cache git make gcc musl-dev
@@ -2,7 +2,7 @@
# Multi-stage build for cross-platform support
# Stage 1: Builder
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
WORKDIR /app
+13 -48
View File
@@ -27,21 +27,19 @@ type VolumeServer struct {
address string
baseDir string
mu sync.Mutex
receivedFiles map[string]uint64
mountRequests []*volume_server_pb.VolumeEcShardsMountRequest
deleteRequests []*volume_server_pb.VolumeDeleteRequest
markReadonlyCalls int
markWritableCalls int
readFileStatusCalls int
vacuumGarbageRatio float64
vacuumCheckCalls int
vacuumCompactCalls int
vacuumCommitCalls int
vacuumCleanupCalls int
volumeCopyCalls int
volumeMountCalls int
tailReceiverCalls int
mu sync.Mutex
receivedFiles map[string]uint64
mountRequests []*volume_server_pb.VolumeEcShardsMountRequest
deleteRequests []*volume_server_pb.VolumeDeleteRequest
markReadonlyCalls int
vacuumGarbageRatio float64
vacuumCheckCalls int
vacuumCompactCalls int
vacuumCommitCalls int
vacuumCleanupCalls int
volumeCopyCalls int
volumeMountCalls int
tailReceiverCalls int
}
// NewVolumeServer starts a test volume server using the provided base directory.
@@ -153,20 +151,6 @@ func (v *VolumeServer) MarkReadonlyCount() int {
return v.markReadonlyCalls
}
// MarkWritableCount returns the number of writable calls.
func (v *VolumeServer) MarkWritableCount() int {
v.mu.Lock()
defer v.mu.Unlock()
return v.markWritableCalls
}
// ReadFileStatusCount returns the number of ReadVolumeFileStatus calls.
func (v *VolumeServer) ReadFileStatusCount() int {
v.mu.Lock()
defer v.mu.Unlock()
return v.readFileStatusCalls
}
// Shutdown stops the volume server.
func (v *VolumeServer) Shutdown() {
if v.server != nil {
@@ -296,25 +280,6 @@ func (v *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_serve
return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil
}
func (v *VolumeServer) VolumeMarkWritable(ctx context.Context, req *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) {
v.mu.Lock()
v.markWritableCalls++
v.mu.Unlock()
return &volume_server_pb.VolumeMarkWritableResponse{}, nil
}
func (v *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) {
v.mu.Lock()
v.readFileStatusCalls++
v.mu.Unlock()
return &volume_server_pb.ReadVolumeFileStatusResponse{
VolumeId: req.VolumeId,
DatFileSize: 1024,
IdxFileSize: 16,
FileCount: 1,
}, nil
}
func (v *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_server_pb.VacuumVolumeCheckRequest) (*volume_server_pb.VacuumVolumeCheckResponse, error) {
v.mu.Lock()
v.vacuumCheckCalls++
@@ -37,9 +37,7 @@ func TestVolumeBalanceDetectionIntegration(t *testing.T) {
MasterGrpcAddresses: []string{master.Address()},
}, 10)
require.NoError(t, err)
// With default batch_size=20 and 10 overloaded volumes vs 1 underloaded,
// all moves are grouped into a single batch proposal.
require.Len(t, proposals, 1, "expected exactly one batch proposal")
require.Len(t, proposals, 1)
proposal := proposals[0]
require.Equal(t, "volume_balance", proposal.JobType)
@@ -48,15 +46,8 @@ func TestVolumeBalanceDetectionIntegration(t *testing.T) {
params := &worker_pb.TaskParams{}
require.NoError(t, proto.Unmarshal(paramsValue.GetBytesValue(), params))
bp := params.GetBalanceParams()
require.NotNil(t, bp, "expected BalanceParams in batch proposal")
require.Greater(t, len(bp.Moves), 1, "batch proposal should contain multiple moves")
for _, move := range bp.Moves {
require.NotZero(t, move.VolumeId)
require.NotEmpty(t, move.SourceNode)
require.NotEmpty(t, move.TargetNode)
}
require.NotEmpty(t, params.Sources)
require.NotEmpty(t, params.Targets)
}
func buildBalanceVolumeListResponse(t *testing.T) *master_pb.VolumeListResponse {
@@ -8,12 +8,10 @@ import (
pluginworkers "github.com/seaweedfs/seaweedfs/test/plugin_workers"
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/proto"
)
func TestVolumeBalanceExecutionIntegration(t *testing.T) {
@@ -62,92 +60,8 @@ func TestVolumeBalanceExecutionIntegration(t *testing.T) {
require.GreaterOrEqual(t, source.MarkReadonlyCount(), 1)
require.GreaterOrEqual(t, len(source.DeleteRequests()), 1)
copyCalls, _, tailCalls := target.BalanceStats()
copyCalls, mountCalls, tailCalls := target.BalanceStats()
require.GreaterOrEqual(t, copyCalls, 1)
require.GreaterOrEqual(t, mountCalls, 1)
require.GreaterOrEqual(t, tailCalls, 1)
}
func TestVolumeBalanceBatchExecutionIntegration(t *testing.T) {
dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
handler := pluginworker.NewVolumeBalanceHandler(dialOption)
harness := pluginworkers.NewHarness(t, pluginworkers.HarnessConfig{
WorkerOptions: pluginworker.WorkerOptions{
GrpcDialOption: dialOption,
},
Handlers: []pluginworker.JobHandler{handler},
})
harness.WaitForJobType("volume_balance")
// Create one source and one target fake volume server.
source := pluginworkers.NewVolumeServer(t, "")
target := pluginworkers.NewVolumeServer(t, "")
// Build a batch job with 3 volume moves from source → target.
volumeIDs := []uint32{401, 402, 403}
moves := make([]*worker_pb.BalanceMoveSpec, len(volumeIDs))
for i, vid := range volumeIDs {
moves[i] = &worker_pb.BalanceMoveSpec{
VolumeId: vid,
SourceNode: source.Address(),
TargetNode: target.Address(),
Collection: "batch-test",
}
}
params := &worker_pb.TaskParams{
TaskId: "batch-balance-test",
TaskParams: &worker_pb.TaskParams_BalanceParams{
BalanceParams: &worker_pb.BalanceTaskParams{
MaxConcurrentMoves: 2,
Moves: moves,
},
},
}
paramBytes, err := proto.Marshal(params)
require.NoError(t, err)
job := &plugin_pb.JobSpec{
JobId: "batch-balance-test",
JobType: "volume_balance",
Parameters: map[string]*plugin_pb.ConfigValue{
"task_params_pb": {
Kind: &plugin_pb.ConfigValue_BytesValue{BytesValue: paramBytes},
},
},
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
result, err := harness.Plugin().ExecuteJob(ctx, job, nil, 1)
require.NoError(t, err)
require.NotNil(t, result)
require.True(t, result.Success, "batch balance job should succeed; result: %+v", result)
// Each of the 3 moves should have marked the source readonly and deleted.
require.Equal(t, len(volumeIDs), source.MarkReadonlyCount(),
"each move should mark source volume readonly")
require.Equal(t, len(volumeIDs), len(source.DeleteRequests()),
"each move should delete the source volume")
// Verify delete requests reference the expected volume IDs.
deletedVols := make(map[uint32]bool)
for _, req := range source.DeleteRequests() {
deletedVols[req.VolumeId] = true
}
for _, vid := range volumeIDs {
require.True(t, deletedVols[vid], "volume %d should have been deleted from source", vid)
}
// Pre-delete verification should have called ReadVolumeFileStatus on both
// source and target for each volume.
require.Equal(t, len(volumeIDs), source.ReadFileStatusCount(),
"each move should read source volume status before delete")
require.Equal(t, len(volumeIDs), target.ReadFileStatusCount(),
"each move should read target volume status before delete")
// Target should have received copy and tail calls for all 3 volumes.
copyCalls, _, tailCalls := target.BalanceStats()
require.Equal(t, len(volumeIDs), copyCalls, "target should receive one copy per volume")
require.Equal(t, len(volumeIDs), tailCalls, "target should receive one tail per volume")
}
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Set working directory
WORKDIR /app
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Set working directory
WORKDIR /app
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.24-alpine AS builder
# Install git and other build dependencies
RUN apk add --no-cache git make
+4 -4
View File
@@ -140,7 +140,7 @@ func TestCORSConfigurationManagement(t *testing.T) {
Bucket: aws.String(bucketName),
CORSConfiguration: corsConfig,
})
require.NoError(t, err, "Should be able to put CORS configuration")
assert.NoError(t, err, "Should be able to put CORS configuration")
// Wait for metadata subscription to update cache
time.Sleep(50 * time.Millisecond)
@@ -149,9 +149,9 @@ func TestCORSConfigurationManagement(t *testing.T) {
getResp, err := client.GetBucketCors(context.TODO(), &s3.GetBucketCorsInput{
Bucket: aws.String(bucketName),
})
require.NoError(t, err, "Should be able to get CORS configuration")
require.NotNil(t, getResp.CORSRules, "CORS configuration should not be nil")
require.Len(t, getResp.CORSRules, 1, "Should have one CORS rule")
assert.NoError(t, err, "Should be able to get CORS configuration")
assert.NotNil(t, getResp.CORSRules, "CORS configuration should not be nil")
assert.Len(t, getResp.CORSRules, 1, "Should have one CORS rule")
rule := getResp.CORSRules[0]
assert.Equal(t, []string{"*"}, rule.AllowedHeaders, "Allowed headers should match")
+1 -1
View File
@@ -1,5 +1,5 @@
# Multi-stage build for SeaweedFS S3 with IAM
FROM golang:1.25-alpine AS builder
FROM golang:1.23-alpine AS builder
# Install build dependencies
RUN apk add --no-cache git make curl wget
+1 -4
View File
@@ -185,9 +185,6 @@ test-context: ## Test only contextual policy enforcement
test-presigned: ## Test only presigned URL integration
go test -v -run TestS3IAMPresignedURLIntegration ./...
test-group: ## Run IAM group management tests
go test -v -run "TestIAMGroup" ./...
test-sts: ## Run all STS tests
go test -v -run "TestSTS" ./...
@@ -266,7 +263,7 @@ docker-build: ## Build custom SeaweedFS image for Docker tests
# All PHONY targets
.PHONY: test test-quick run-tests setup start-services stop-services wait-for-services clean logs status debug
.PHONY: test-auth test-policy test-expiration test-multipart test-bucket-policy test-context test-presigned test-group test-sts test-sts-assume-role test-sts-ldap
.PHONY: test-auth test-policy test-expiration test-multipart test-bucket-policy test-context test-presigned test-sts test-sts-assume-role test-sts-ldap
.PHONY: benchmark ci watch install-deps docker-test docker-up docker-down docker-logs docker-build
.PHONY: test-distributed test-performance test-stress test-versioning-stress test-keycloak-full test-all-previously-skipped setup-all-tests help-advanced
-792
View File
@@ -1,792 +0,0 @@
package iam
import (
"encoding/xml"
"io"
"net/http"
"net/url"
"strings"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestIAMGroupLifecycle tests the full lifecycle of group management:
// CreateGroup, GetGroup, ListGroups, DeleteGroup
func TestIAMGroupLifecycle(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-group-lifecycle"
t.Run("create_group", func(t *testing.T) {
resp, err := iamClient.CreateGroup(&iam.CreateGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
assert.Equal(t, groupName, *resp.Group.GroupName)
})
t.Run("get_group", func(t *testing.T) {
resp, err := iamClient.GetGroup(&iam.GetGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
assert.Equal(t, groupName, *resp.Group.GroupName)
})
t.Run("list_groups_contains_created", func(t *testing.T) {
resp, err := iamClient.ListGroups(&iam.ListGroupsInput{})
require.NoError(t, err)
found := false
for _, g := range resp.Groups {
if *g.GroupName == groupName {
found = true
break
}
}
assert.True(t, found, "Created group should appear in ListGroups")
})
t.Run("create_duplicate_group_fails", func(t *testing.T) {
_, err := iamClient.CreateGroup(&iam.CreateGroupInput{
GroupName: aws.String(groupName),
})
assert.Error(t, err, "Creating a duplicate group should fail")
})
t.Run("delete_group", func(t *testing.T) {
_, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
// Verify it's gone
resp, err := iamClient.ListGroups(&iam.ListGroupsInput{})
require.NoError(t, err)
for _, g := range resp.Groups {
assert.NotEqual(t, groupName, *g.GroupName,
"Deleted group should not appear in ListGroups")
}
})
t.Run("delete_nonexistent_group_fails", func(t *testing.T) {
_, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{
GroupName: aws.String("nonexistent-group-xyz"),
})
assert.Error(t, err)
})
}
// TestIAMGroupMembership tests adding and removing users from groups
func TestIAMGroupMembership(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-group-members"
userName := "test-user-for-group"
// Setup: create group and user
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
_, err = iamClient.CreateUser(&iam.CreateUserInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
defer iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
t.Run("add_user_to_group", func(t *testing.T) {
_, err := iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
require.NoError(t, err)
})
t.Run("get_group_shows_member", func(t *testing.T) {
resp, err := iamClient.GetGroup(&iam.GetGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
found := false
for _, u := range resp.Users {
if *u.UserName == userName {
found = true
break
}
}
assert.True(t, found, "Added user should appear in GetGroup members")
})
t.Run("list_groups_for_user", func(t *testing.T) {
resp, err := iamClient.ListGroupsForUser(&iam.ListGroupsForUserInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
found := false
for _, g := range resp.Groups {
if *g.GroupName == groupName {
found = true
break
}
}
assert.True(t, found, "Group should appear in ListGroupsForUser")
})
t.Run("add_duplicate_member_is_idempotent", func(t *testing.T) {
_, err := iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
// Should succeed (idempotent) or return a benign error
// AWS IAM allows duplicate add without error
assert.NoError(t, err)
})
t.Run("remove_user_from_group", func(t *testing.T) {
_, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
require.NoError(t, err)
// Verify removal
resp, err := iamClient.GetGroup(&iam.GetGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
for _, u := range resp.Users {
assert.NotEqual(t, userName, *u.UserName,
"Removed user should not appear in group members")
}
})
}
// TestIAMGroupPolicyAttachment tests attaching and detaching policies from groups
func TestIAMGroupPolicyAttachment(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-group-policies"
policyName := "test-group-attach-policy"
policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`
// Setup: create group and policy
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName),
PolicyDocument: aws.String(policyDoc),
})
require.NoError(t, err)
policyArn := createPolicyResp.Policy.Arn
// Cleanup in correct order: detach policy, delete group, delete policy
t.Cleanup(func() {
if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: policyArn,
}); err != nil {
t.Logf("cleanup: failed to detach group policy: %v", err)
}
if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
t.Logf("cleanup: failed to delete group: %v", err)
}
if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: policyArn}); err != nil {
t.Logf("cleanup: failed to delete policy: %v", err)
}
})
t.Run("attach_group_policy", func(t *testing.T) {
_, err := iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: policyArn,
})
require.NoError(t, err)
})
t.Run("list_attached_group_policies", func(t *testing.T) {
resp, err := iamClient.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
found := false
for _, p := range resp.AttachedPolicies {
if *p.PolicyName == policyName {
found = true
break
}
}
assert.True(t, found, "Attached policy should appear in ListAttachedGroupPolicies")
})
t.Run("detach_group_policy", func(t *testing.T) {
_, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: policyArn,
})
require.NoError(t, err)
// Verify detachment
resp, err := iamClient.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
for _, p := range resp.AttachedPolicies {
assert.NotEqual(t, policyName, *p.PolicyName,
"Detached policy should not appear in ListAttachedGroupPolicies")
}
})
}
// TestIAMGroupPolicyEnforcement tests that group policies are enforced during S3 operations.
// Creates a user with no direct policies, adds them to a group with S3 access,
// and verifies they can access S3 through the group policy.
func TestIAMGroupPolicyEnforcement(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-enforcement-group"
userName := "test-enforcement-user"
policyName := "test-enforcement-policy"
bucketName := "test-group-enforce-bucket"
policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::` + bucketName + `","arn:aws:s3:::` + bucketName + `/*"]}]}`
// Create user
_, err = iamClient.CreateUser(&iam.CreateUserInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
// Create access key for the user
keyResp, err := iamClient.CreateAccessKey(&iam.CreateAccessKeyInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
accessKeyId := *keyResp.AccessKey.AccessKeyId
secretKey := *keyResp.AccessKey.SecretAccessKey
// Create an S3 client with the user's credentials
userS3Client := createS3Client(t, accessKeyId, secretKey)
// Create group
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{
GroupName: aws.String(groupName),
})
require.NoError(t, err)
// Create policy
createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName),
PolicyDocument: aws.String(policyDoc),
})
require.NoError(t, err)
policyArn := createPolicyResp.Policy.Arn
// Cleanup in correct order: remove user from group, detach policy,
// delete access key, delete user, delete group, delete policy
t.Cleanup(func() {
if _, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
}); err != nil {
t.Logf("cleanup: failed to remove user from group: %v", err)
}
if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: policyArn,
}); err != nil {
t.Logf("cleanup: failed to detach group policy: %v", err)
}
if _, err := iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
UserName: aws.String(userName),
AccessKeyId: keyResp.AccessKey.AccessKeyId,
}); err != nil {
t.Logf("cleanup: failed to delete access key: %v", err)
}
if _, err := iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)}); err != nil {
t.Logf("cleanup: failed to delete user: %v", err)
}
if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
t.Logf("cleanup: failed to delete group: %v", err)
}
if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: policyArn}); err != nil {
t.Logf("cleanup: failed to delete policy: %v", err)
}
})
// Register bucket cleanup on parent test with admin credentials
// (userS3Client may lack permissions by cleanup time)
adminS3, err := framework.CreateS3ClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
t.Cleanup(func() {
if _, err := adminS3.DeleteObject(&s3.DeleteObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String("test-key"),
}); err != nil {
t.Logf("cleanup: failed to delete object: %v", err)
}
if _, err := adminS3.DeleteBucket(&s3.DeleteBucketInput{Bucket: aws.String(bucketName)}); err != nil {
t.Logf("cleanup: failed to delete bucket: %v", err)
}
})
t.Run("user_without_group_denied", func(t *testing.T) {
// User has no policies and is not in any group — should be denied
_, err := userS3Client.CreateBucket(&s3.CreateBucketInput{
Bucket: aws.String(bucketName),
})
require.Error(t, err, "User without any policies should be denied")
awsErr, ok := err.(awserr.Error)
require.True(t, ok, "Expected awserr.Error")
assert.Equal(t, "AccessDenied", awsErr.Code())
})
t.Run("user_with_group_policy_allowed", func(t *testing.T) {
// Attach policy to group
_, err := iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: policyArn,
})
require.NoError(t, err)
// Add user to group
_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
require.NoError(t, err)
// Wait for policy propagation, then create bucket
require.Eventually(t, func() bool {
_, err = userS3Client.CreateBucket(&s3.CreateBucketInput{
Bucket: aws.String(bucketName),
})
return err == nil
}, 10*time.Second, 500*time.Millisecond, "User with group policy should be allowed")
// Should also be able to put/get objects
_, err = userS3Client.PutObject(&s3.PutObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String("test-key"),
Body: aws.ReadSeekCloser(strings.NewReader("test-data")),
})
require.NoError(t, err, "User should be able to put objects through group policy")
})
t.Run("user_removed_from_group_denied", func(t *testing.T) {
// Remove user from group
_, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
require.NoError(t, err)
// Wait for policy propagation — user should now be denied
var lastErr error
require.Eventually(t, func() bool {
_, lastErr = userS3Client.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(bucketName),
})
return lastErr != nil
}, 10*time.Second, 500*time.Millisecond, "User removed from group should be denied")
awsErr, ok := lastErr.(awserr.Error)
require.True(t, ok, "Expected awserr.Error")
assert.Equal(t, "AccessDenied", awsErr.Code())
})
}
// TestIAMGroupDisabledPolicyEnforcement tests that disabled groups do not contribute policies.
// Uses the raw IAM API (callIAMAPI) since the AWS SDK doesn't support custom group status.
func TestIAMGroupDisabledPolicyEnforcement(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
if !isSeaweedFSRunning(t) {
t.Skip("SeaweedFS is not running at", TestIAMEndpoint)
}
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-disabled-group"
userName := "test-disabled-grp-user"
policyName := "test-disabled-grp-policy"
bucketName := "test-disabled-grp-bucket"
policyDoc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::` + bucketName + `","arn:aws:s3:::` + bucketName + `/*"]}]}`
// Create user, group, policy
_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
keyResp, err := iamClient.CreateAccessKey(&iam.CreateAccessKeyInput{UserName: aws.String(userName)})
require.NoError(t, err)
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(groupName)})
require.NoError(t, err)
createPolicyResp, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName), PolicyDocument: aws.String(policyDoc),
})
require.NoError(t, err)
// Cleanup in correct order: remove user from group, detach policy,
// delete access key, delete user, delete group, delete policy
t.Cleanup(func() {
if _, err := iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(groupName), UserName: aws.String(userName),
}); err != nil {
t.Logf("cleanup: failed to remove user from group: %v", err)
}
if _, err := iamClient.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
GroupName: aws.String(groupName),
PolicyArn: aws.String("arn:aws:iam:::policy/" + policyName),
}); err != nil {
t.Logf("cleanup: failed to detach group policy: %v", err)
}
if _, err := iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
UserName: aws.String(userName), AccessKeyId: keyResp.AccessKey.AccessKeyId,
}); err != nil {
t.Logf("cleanup: failed to delete access key: %v", err)
}
if _, err := iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)}); err != nil {
t.Logf("cleanup: failed to delete user: %v", err)
}
if _, err := iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)}); err != nil {
t.Logf("cleanup: failed to delete group: %v", err)
}
if _, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: createPolicyResp.Policy.Arn}); err != nil {
t.Logf("cleanup: failed to delete policy: %v", err)
}
})
// Setup: attach policy, add user, create bucket with admin
_, err = iamClient.AttachGroupPolicy(&iam.AttachGroupPolicyInput{
GroupName: aws.String(groupName), PolicyArn: createPolicyResp.Policy.Arn,
})
require.NoError(t, err)
_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(groupName), UserName: aws.String(userName),
})
require.NoError(t, err)
userS3Client := createS3Client(t, *keyResp.AccessKey.AccessKeyId, *keyResp.AccessKey.SecretAccessKey)
// Create bucket using admin first so we can test listing
adminS3, err := framework.CreateS3ClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
_, err = adminS3.CreateBucket(&s3.CreateBucketInput{Bucket: aws.String(bucketName)})
require.NoError(t, err)
defer adminS3.DeleteBucket(&s3.DeleteBucketInput{Bucket: aws.String(bucketName)})
t.Run("enabled_group_allows_access", func(t *testing.T) {
require.Eventually(t, func() bool {
_, err := userS3Client.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(bucketName),
})
return err == nil
}, 10*time.Second, 500*time.Millisecond, "User in enabled group should have access")
})
t.Run("disabled_group_denies_access", func(t *testing.T) {
// Disable group via raw IAM API (no SDK support for this extension)
resp, err := callIAMAPIAuthenticated(t, framework, "UpdateGroup", url.Values{
"GroupName": {groupName},
"Disabled": {"true"},
})
require.NoError(t, err)
defer resp.Body.Close()
require.Equal(t, http.StatusOK, resp.StatusCode, "UpdateGroup (disable) should return 200")
// Wait for propagation — user should be denied
var lastErr error
require.Eventually(t, func() bool {
_, lastErr = userS3Client.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(bucketName),
})
return lastErr != nil
}, 10*time.Second, 500*time.Millisecond, "User in disabled group should be denied access")
awsErr, ok := lastErr.(awserr.Error)
require.True(t, ok, "Expected awserr.Error")
assert.Equal(t, "AccessDenied", awsErr.Code())
})
t.Run("re_enabled_group_restores_access", func(t *testing.T) {
// Re-enable the group
resp, err := callIAMAPIAuthenticated(t, framework, "UpdateGroup", url.Values{
"GroupName": {groupName},
"Disabled": {"false"},
})
require.NoError(t, err)
defer resp.Body.Close()
require.Equal(t, http.StatusOK, resp.StatusCode, "UpdateGroup (re-enable) should return 200")
// Wait for propagation — user should have access again
require.Eventually(t, func() bool {
_, err = userS3Client.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(bucketName),
})
return err == nil
}, 10*time.Second, 500*time.Millisecond, "User in re-enabled group should have access again")
})
}
// TestIAMGroupUserDeletionSideEffect tests that deleting a user removes them from all groups.
func TestIAMGroupUserDeletionSideEffect(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
groupName := "test-deletion-group"
userName := "test-deletion-user"
// Create group and user
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(groupName)})
require.NoError(t, err)
defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
t.Cleanup(func() {
// Best-effort: user may already be deleted by the test
iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
})
// Add user to group
_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(groupName),
UserName: aws.String(userName),
})
require.NoError(t, err)
// Verify user is in group
getResp, err := iamClient.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)})
require.NoError(t, err)
assert.Len(t, getResp.Users, 1, "Group should have 1 member before deletion")
// Delete the user
_, err = iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
// Verify user was removed from the group
getResp, err = iamClient.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)})
require.NoError(t, err)
assert.Empty(t, getResp.Users, "Group should have no members after user deletion")
}
// TestIAMGroupMultipleGroups tests that a user can belong to multiple groups
// and inherits policies from all of them.
func TestIAMGroupMultipleGroups(t *testing.T) {
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
iamClient, err := framework.CreateIAMClientWithJWT("admin-user", "TestAdminRole")
require.NoError(t, err)
group1 := "test-multi-group-1"
group2 := "test-multi-group-2"
userName := "test-multi-group-user"
// Create two groups
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(group1)})
require.NoError(t, err)
defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(group1)})
_, err = iamClient.CreateGroup(&iam.CreateGroupInput{GroupName: aws.String(group2)})
require.NoError(t, err)
defer iamClient.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(group2)})
// Create user
_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
defer func() {
iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(group1), UserName: aws.String(userName),
})
iamClient.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
GroupName: aws.String(group2), UserName: aws.String(userName),
})
iamClient.DeleteUser(&iam.DeleteUserInput{UserName: aws.String(userName)})
}()
// Add user to both groups
_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(group1), UserName: aws.String(userName),
})
require.NoError(t, err)
_, err = iamClient.AddUserToGroup(&iam.AddUserToGroupInput{
GroupName: aws.String(group2), UserName: aws.String(userName),
})
require.NoError(t, err)
// Verify user appears in both groups
resp, err := iamClient.ListGroupsForUser(&iam.ListGroupsForUserInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
groupNames := make(map[string]bool)
for _, g := range resp.Groups {
groupNames[*g.GroupName] = true
}
assert.True(t, groupNames[group1], "User should be in group 1")
assert.True(t, groupNames[group2], "User should be in group 2")
}
// --- Response types for raw IAM API calls ---
type CreateGroupResponse struct {
XMLName xml.Name `xml:"CreateGroupResponse"`
CreateGroupResult struct {
Group struct {
GroupName string `xml:"GroupName"`
} `xml:"Group"`
} `xml:"CreateGroupResult"`
}
type ListGroupsResponse struct {
XMLName xml.Name `xml:"ListGroupsResponse"`
ListGroupsResult struct {
Groups []struct {
GroupName string `xml:"GroupName"`
} `xml:"Groups>member"`
} `xml:"ListGroupsResult"`
}
// callIAMAPIAuthenticated sends an authenticated raw IAM API request using the
// framework's JWT token. This is needed for custom extensions not in the AWS SDK
// (like UpdateGroup with Disabled parameter).
func callIAMAPIAuthenticated(_ *testing.T, framework *S3IAMTestFramework, action string, params url.Values) (*http.Response, error) {
params.Set("Action", action)
req, err := http.NewRequest(http.MethodPost, TestIAMEndpoint+"/",
strings.NewReader(params.Encode()))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
token, err := framework.generateSTSSessionToken("admin-user", "TestAdminRole", time.Hour, "", nil)
if err != nil {
return nil, err
}
client := &http.Client{
Timeout: 30 * time.Second,
Transport: &BearerTokenTransport{Token: token},
}
return client.Do(req)
}
// TestIAMGroupRawAPI tests group operations using raw HTTP IAM API calls,
// verifying XML response format for group operations.
func TestIAMGroupRawAPI(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
if !isSeaweedFSRunning(t) {
t.Skip("SeaweedFS is not running at", TestIAMEndpoint)
}
framework := NewS3IAMTestFramework(t)
defer framework.Cleanup()
groupName := "test-raw-api-group"
t.Run("create_group_raw", func(t *testing.T) {
resp, err := callIAMAPIAuthenticated(t, framework, "CreateGroup", url.Values{
"GroupName": {groupName},
})
require.NoError(t, err)
defer resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode)
body, err := io.ReadAll(resp.Body)
require.NoError(t, err)
var createResp CreateGroupResponse
err = xml.Unmarshal(body, &createResp)
require.NoError(t, err)
assert.Equal(t, groupName, createResp.CreateGroupResult.Group.GroupName)
})
t.Run("list_groups_raw", func(t *testing.T) {
resp, err := callIAMAPIAuthenticated(t, framework, "ListGroups", url.Values{})
require.NoError(t, err)
defer resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode)
body, err := io.ReadAll(resp.Body)
require.NoError(t, err)
var listResp ListGroupsResponse
err = xml.Unmarshal(body, &listResp)
require.NoError(t, err)
found := false
for _, g := range listResp.ListGroupsResult.Groups {
if g.GroupName == groupName {
found = true
break
}
}
assert.True(t, found, "Created group should appear in raw ListGroups")
})
t.Run("delete_group_raw", func(t *testing.T) {
resp, err := callIAMAPIAuthenticated(t, framework, "DeleteGroup", url.Values{
"GroupName": {groupName},
})
require.NoError(t, err)
defer resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode)
})
}
// createS3Client creates an S3 client with static credentials
func createS3Client(t *testing.T, accessKey, secretKey string) *s3.S3 {
sess, err := session.NewSession(&aws.Config{
Region: aws.String("us-east-1"),
Endpoint: aws.String(TestS3Endpoint),
Credentials: credentials.NewStaticCredentials(accessKey, secretKey, ""),
DisableSSL: aws.Bool(true),
S3ForcePathStyle: aws.Bool(true),
})
require.NoError(t, err)
return s3.New(sess)
}
@@ -1,573 +0,0 @@
package example
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"strings"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
v1credentials "github.com/aws/aws-sdk-go/aws/credentials"
v1signer "github.com/aws/aws-sdk-go/aws/signer/v4"
v1s3 "github.com/aws/aws-sdk-go/service/s3"
v2aws "github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/credentials"
v2s3 "github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newS3V2Client creates an AWS SDK v2 S3 client from the test cluster.
func newS3V2Client(cluster *TestCluster) *v2s3.Client {
return v2s3.New(v2s3.Options{
Region: testRegion,
BaseEndpoint: v2aws.String(cluster.s3Endpoint),
Credentials: v2aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(testAccessKey, testSecretKey, "")),
UsePathStyle: true,
})
}
func TestGetObjectAttributes(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
cluster, err := startMiniCluster(t)
require.NoError(t, err)
defer cluster.Stop()
t.Run("Basic", func(t *testing.T) {
testGetObjectAttributesBasic(t, cluster)
})
t.Run("MultipartObject", func(t *testing.T) {
testGetObjectAttributesMultipart(t, cluster)
})
t.Run("SelectiveAttributes", func(t *testing.T) {
testGetObjectAttributesSelective(t, cluster)
})
t.Run("InvalidAttribute", func(t *testing.T) {
testGetObjectAttributesInvalid(t, cluster)
})
t.Run("NonExistentObject", func(t *testing.T) {
testGetObjectAttributesNotFound(t, cluster)
})
t.Run("VersionedObject", func(t *testing.T) {
testGetObjectAttributesVersioned(t, cluster)
})
t.Run("ConditionalHeaders", func(t *testing.T) {
testGetObjectAttributesConditionalHeaders(t, cluster)
})
t.Run("VersionedConditionalHeaders", func(t *testing.T) {
testGetObjectAttributesVersionedConditionalHeaders(t, cluster)
})
}
func testGetObjectAttributesBasic(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-basic-")
objectKey := "test-object.txt"
objectData := "Hello, GetObjectAttributes!"
_, err := cluster.s3Client.PutObject(&v1s3.PutObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
Body: bytes.NewReader([]byte(objectData)),
})
require.NoError(t, err)
client := newS3V2Client(cluster)
resp, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String(objectKey),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesEtag,
types.ObjectAttributesStorageClass,
types.ObjectAttributesObjectSize,
types.ObjectAttributesObjectParts,
},
})
require.NoError(t, err)
// ETag should be present and non-empty
require.NotNil(t, resp.ETag)
assert.NotEmpty(t, *resp.ETag)
assert.False(t, strings.Contains(*resp.ETag, `"`), "ETag in XML body should not have quotes")
// ObjectSize should match
require.NotNil(t, resp.ObjectSize)
assert.Equal(t, int64(len(objectData)), *resp.ObjectSize)
// StorageClass should be STANDARD (default)
assert.Equal(t, "STANDARD", string(resp.StorageClass))
// ObjectParts should be nil for non-multipart objects
assert.Nil(t, resp.ObjectParts)
// LastModified header should be present
assert.NotNil(t, resp.LastModified)
t.Logf("Basic GetObjectAttributes passed: ETag=%s, Size=%d, StorageClass=%s",
*resp.ETag, *resp.ObjectSize, resp.StorageClass)
}
func testGetObjectAttributesMultipart(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-mp-")
objectKey := "test-multipart.bin"
// Create a 2-part multipart upload
part1Data := bytes.Repeat([]byte("A"), 5*1024*1024) // 5MB (minimum part size)
part2Data := bytes.Repeat([]byte("B"), 3*1024*1024) // 3MB
initResp, err := cluster.s3Client.CreateMultipartUpload(&v1s3.CreateMultipartUploadInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
})
require.NoError(t, err)
uploadID := initResp.UploadId
part1Resp, err := cluster.s3Client.UploadPart(&v1s3.UploadPartInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
PartNumber: aws.Int64(1),
UploadId: uploadID,
Body: bytes.NewReader(part1Data),
})
require.NoError(t, err)
part2Resp, err := cluster.s3Client.UploadPart(&v1s3.UploadPartInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
PartNumber: aws.Int64(2),
UploadId: uploadID,
Body: bytes.NewReader(part2Data),
})
require.NoError(t, err)
_, err = cluster.s3Client.CompleteMultipartUpload(&v1s3.CompleteMultipartUploadInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
UploadId: uploadID,
MultipartUpload: &v1s3.CompletedMultipartUpload{
Parts: []*v1s3.CompletedPart{
{ETag: part1Resp.ETag, PartNumber: aws.Int64(1)},
{ETag: part2Resp.ETag, PartNumber: aws.Int64(2)},
},
},
})
require.NoError(t, err)
// Wait briefly for metadata to settle
time.Sleep(200 * time.Millisecond)
client := newS3V2Client(cluster)
resp, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String(objectKey),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesObjectParts,
types.ObjectAttributesObjectSize,
},
})
require.NoError(t, err)
require.NotNil(t, resp.ObjectSize)
assert.Equal(t, int64(len(part1Data)+len(part2Data)), *resp.ObjectSize)
require.NotNil(t, resp.ObjectParts, "ObjectParts should be present for multipart objects")
assert.Equal(t, int32(2), *resp.ObjectParts.TotalPartsCount)
require.Len(t, resp.ObjectParts.Parts, 2)
assert.Equal(t, int32(1), *resp.ObjectParts.Parts[0].PartNumber)
assert.Equal(t, int64(len(part1Data)), *resp.ObjectParts.Parts[0].Size)
assert.Equal(t, int32(2), *resp.ObjectParts.Parts[1].PartNumber)
assert.Equal(t, int64(len(part2Data)), *resp.ObjectParts.Parts[1].Size)
// Test pagination: MaxParts=1
resp2, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String(objectKey),
MaxParts: v2aws.Int32(1),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesObjectParts,
},
})
require.NoError(t, err)
require.NotNil(t, resp2.ObjectParts)
assert.Len(t, resp2.ObjectParts.Parts, 1)
assert.True(t, *resp2.ObjectParts.IsTruncated)
assert.Equal(t, int32(2), *resp2.ObjectParts.TotalPartsCount)
t.Logf("Multipart GetObjectAttributes passed: %d parts, total size %d",
*resp.ObjectParts.TotalPartsCount, *resp.ObjectSize)
}
func testGetObjectAttributesSelective(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-sel-")
objectKey := "test-selective.txt"
objectData := "Selective attributes test"
_, err := cluster.s3Client.PutObject(&v1s3.PutObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
Body: bytes.NewReader([]byte(objectData)),
})
require.NoError(t, err)
client := newS3V2Client(cluster)
// Request only ETag
resp, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String(objectKey),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesEtag,
},
})
require.NoError(t, err)
require.NotNil(t, resp.ETag)
assert.NotEmpty(t, *resp.ETag)
assert.Nil(t, resp.ObjectSize, "ObjectSize should not be present when not requested")
assert.Empty(t, string(resp.StorageClass), "StorageClass should not be present when not requested")
assert.Nil(t, resp.ObjectParts, "ObjectParts should not be present when not requested")
t.Logf("Selective GetObjectAttributes passed: ETag=%s", *resp.ETag)
}
func testGetObjectAttributesInvalid(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-inv-")
objectKey := "test-object.txt"
_, err := cluster.s3Client.PutObject(&v1s3.PutObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
Body: bytes.NewReader([]byte("test")),
})
require.NoError(t, err)
// Use raw HTTP to send an invalid attribute name since the SDK validates
reqURL := fmt.Sprintf("%s/%s/%s?attributes", cluster.s3Endpoint, bucketName, objectKey)
req, err := http.NewRequest("GET", reqURL, nil)
require.NoError(t, err)
req.Header.Set("X-Amz-Object-Attributes", "InvalidAttr")
signer := v1signer.NewSigner(v1credentials.NewStaticCredentials(testAccessKey, testSecretKey, ""))
_, err = signer.Sign(req, nil, "s3", testRegion, time.Now())
require.NoError(t, err)
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
require.NoError(t, err)
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 400, resp.StatusCode)
t.Logf("Invalid attribute test passed: got %d", resp.StatusCode)
}
func testGetObjectAttributesNotFound(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-nf-")
client := newS3V2Client(cluster)
_, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("nonexistent-key"),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesEtag,
},
})
require.Error(t, err)
assert.Contains(t, err.Error(), "NoSuchKey")
t.Logf("NotFound GetObjectAttributes passed")
}
func testGetObjectAttributesVersioned(t *testing.T, cluster *TestCluster) {
client := newS3V2Client(cluster)
bucketName := createTestBucket(t, cluster, "test-goa-ver-")
// Enable versioning
_, err := client.PutBucketVersioning(context.Background(), &v2s3.PutBucketVersioningInput{
Bucket: v2aws.String(bucketName),
VersioningConfiguration: &types.VersioningConfiguration{
Status: types.BucketVersioningStatusEnabled,
},
})
require.NoError(t, err)
time.Sleep(200 * time.Millisecond)
// Put two versions of the same object
v1Data := "version 1 content"
putResp1, err := client.PutObject(context.Background(), &v2s3.PutObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("versioned-key"),
Body: strings.NewReader(v1Data),
})
require.NoError(t, err)
require.NotNil(t, putResp1.VersionId)
versionId1 := *putResp1.VersionId
v2Data := "version 2 content - longer"
putResp2, err := client.PutObject(context.Background(), &v2s3.PutObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("versioned-key"),
Body: strings.NewReader(v2Data),
})
require.NoError(t, err)
require.NotNil(t, putResp2.VersionId)
versionId2 := *putResp2.VersionId
assert.NotEqual(t, versionId1, versionId2, "versions should differ")
// GetObjectAttributes for latest version (v2)
resp, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("versioned-key"),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesObjectSize,
types.ObjectAttributesEtag,
},
})
require.NoError(t, err)
require.NotNil(t, resp.ObjectSize)
assert.Equal(t, int64(len(v2Data)), *resp.ObjectSize)
require.NotNil(t, resp.VersionId)
assert.Equal(t, versionId2, *resp.VersionId)
// GetObjectAttributes for specific older version (v1)
resp1, err := client.GetObjectAttributes(context.Background(), &v2s3.GetObjectAttributesInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("versioned-key"),
VersionId: v2aws.String(versionId1),
ObjectAttributes: []types.ObjectAttributes{
types.ObjectAttributesObjectSize,
types.ObjectAttributesEtag,
},
})
require.NoError(t, err)
require.NotNil(t, resp1.ObjectSize)
assert.Equal(t, int64(len(v1Data)), *resp1.ObjectSize)
require.NotNil(t, resp1.VersionId)
assert.Equal(t, versionId1, *resp1.VersionId)
t.Logf("Versioned GetObjectAttributes passed: v1 size=%d (id=%s), v2 size=%d (id=%s)",
*resp1.ObjectSize, versionId1, *resp.ObjectSize, versionId2)
}
// signedGetObjectAttributes creates a signed GET request for ?attributes with custom headers.
func signedGetObjectAttributes(t *testing.T, cluster *TestCluster, bucketName, objectKey string, extraHeaders map[string]string) *http.Response {
reqURL := fmt.Sprintf("%s/%s/%s?attributes", cluster.s3Endpoint, bucketName, objectKey)
req, err := http.NewRequest("GET", reqURL, nil)
require.NoError(t, err)
req.Header.Set("X-Amz-Object-Attributes", "ETag,ObjectSize")
for k, v := range extraHeaders {
req.Header.Set(k, v)
}
signer := v1signer.NewSigner(v1credentials.NewStaticCredentials(testAccessKey, testSecretKey, ""))
_, err = signer.Sign(req, nil, "s3", testRegion, time.Now())
require.NoError(t, err)
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
require.NoError(t, err)
return resp
}
func testGetObjectAttributesConditionalHeaders(t *testing.T, cluster *TestCluster) {
bucketName := createTestBucket(t, cluster, "test-goa-cond-")
objectKey := "cond-test.txt"
_, err := cluster.s3Client.PutObject(&v1s3.PutObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
Body: bytes.NewReader([]byte("conditional headers test")),
})
require.NoError(t, err)
// Get the ETag and Last-Modified for the object
headResp, err := cluster.s3Client.HeadObject(&v1s3.HeadObjectInput{
Bucket: aws.String(bucketName),
Key: aws.String(objectKey),
})
require.NoError(t, err)
etag := aws.StringValue(headResp.ETag)
lastModified := headResp.LastModified
require.NotNil(t, lastModified)
pastDate := lastModified.Add(-1 * time.Hour).UTC().Format(http.TimeFormat)
futureDate := lastModified.Add(1 * time.Hour).UTC().Format(http.TimeFormat)
// RFC 7232: If-Match true + If-Unmodified-Since false => 200 OK
// If-Unmodified-Since is ignored when If-Match is present
t.Run("IfMatch_true_IfUnmodifiedSince_false", func(t *testing.T) {
resp := signedGetObjectAttributes(t, cluster, bucketName, objectKey, map[string]string{
"If-Match": etag,
"If-Unmodified-Since": pastDate, // object was modified after this => false
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 200, resp.StatusCode,
"If-Match=true should return 200 even when If-Unmodified-Since=false (RFC 7232 Section 3.4)")
})
// RFC 7232: If-None-Match false + If-Modified-Since true => 304 Not Modified
// If-Modified-Since is ignored when If-None-Match is present
t.Run("IfNoneMatch_false_IfModifiedSince_true", func(t *testing.T) {
resp := signedGetObjectAttributes(t, cluster, bucketName, objectKey, map[string]string{
"If-None-Match": etag,
"If-Modified-Since": pastDate, // object was modified after this => true
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 304, resp.StatusCode,
"If-None-Match=false (ETag match) should return 304 even when If-Modified-Since=true (RFC 7232 Section 3.3)")
})
// If-Match succeeds, If-Unmodified-Since also succeeds => 200
t.Run("IfMatch_true_IfUnmodifiedSince_true", func(t *testing.T) {
resp := signedGetObjectAttributes(t, cluster, bucketName, objectKey, map[string]string{
"If-Match": etag,
"If-Unmodified-Since": futureDate,
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 200, resp.StatusCode)
})
// If-None-Match passes (ETag differs), If-Modified-Since ignored => 200
// Per RFC 7232, If-Modified-Since is ignored when If-None-Match is present
t.Run("IfNoneMatch_true_IfModifiedSince_ignored", func(t *testing.T) {
resp := signedGetObjectAttributes(t, cluster, bucketName, objectKey, map[string]string{
"If-None-Match": `"nonexistent-etag"`,
"If-Modified-Since": futureDate, // would fail alone, but is ignored
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 200, resp.StatusCode,
"If-None-Match=true means If-Modified-Since is ignored, should return 200 (RFC 7232 Section 3.3)")
})
// If-Match fails => 412 regardless of If-Unmodified-Since
t.Run("IfMatch_false", func(t *testing.T) {
resp := signedGetObjectAttributes(t, cluster, bucketName, objectKey, map[string]string{
"If-Match": `"wrong-etag"`,
"If-Unmodified-Since": futureDate,
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 412, resp.StatusCode)
})
t.Logf("Conditional headers tests passed")
}
// signedGetObjectAttributesVersioned creates a signed GET request for ?attributes&versionId=... with custom headers.
func signedGetObjectAttributesVersioned(t *testing.T, cluster *TestCluster, bucketName, objectKey, versionId string, extraHeaders map[string]string) *http.Response {
reqURL := fmt.Sprintf("%s/%s/%s?attributes&versionId=%s", cluster.s3Endpoint, bucketName, objectKey, versionId)
req, err := http.NewRequest("GET", reqURL, nil)
require.NoError(t, err)
req.Header.Set("X-Amz-Object-Attributes", "ETag,ObjectSize")
for k, v := range extraHeaders {
req.Header.Set(k, v)
}
signer := v1signer.NewSigner(v1credentials.NewStaticCredentials(testAccessKey, testSecretKey, ""))
_, err = signer.Sign(req, nil, "s3", testRegion, time.Now())
require.NoError(t, err)
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
require.NoError(t, err)
return resp
}
func testGetObjectAttributesVersionedConditionalHeaders(t *testing.T, cluster *TestCluster) {
client := newS3V2Client(cluster)
bucketName := createTestBucket(t, cluster, "test-goa-vcond-")
// Enable versioning
_, err := client.PutBucketVersioning(context.Background(), &v2s3.PutBucketVersioningInput{
Bucket: v2aws.String(bucketName),
VersioningConfiguration: &types.VersioningConfiguration{
Status: types.BucketVersioningStatusEnabled,
},
})
require.NoError(t, err)
time.Sleep(200 * time.Millisecond)
// Put two versions with different content (different ETags)
v1Data := "version 1 - original"
putResp1, err := client.PutObject(context.Background(), &v2s3.PutObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("vcond-key"),
Body: strings.NewReader(v1Data),
})
require.NoError(t, err)
require.NotNil(t, putResp1.VersionId)
vid1 := *putResp1.VersionId
v2Data := "version 2 - updated content"
putResp2, err := client.PutObject(context.Background(), &v2s3.PutObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("vcond-key"),
Body: strings.NewReader(v2Data),
})
require.NoError(t, err)
require.NotNil(t, putResp2.VersionId)
vid2 := *putResp2.VersionId
// Get ETags for each version
headV1, err := client.HeadObject(context.Background(), &v2s3.HeadObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("vcond-key"),
VersionId: v2aws.String(vid1),
})
require.NoError(t, err)
etagV1 := *headV1.ETag
headV2, err := client.HeadObject(context.Background(), &v2s3.HeadObjectInput{
Bucket: v2aws.String(bucketName),
Key: v2aws.String("vcond-key"),
VersionId: v2aws.String(vid2),
})
require.NoError(t, err)
etagV2 := *headV2.ETag
require.NotEqual(t, etagV1, etagV2, "versions should have different ETags")
// If-Match with v1's ETag + versionId=v1 => 200
// Before the fix, this would fail with 412 because conditional headers
// were evaluated against the latest version (v2) whose ETag differs
t.Run("IfMatch_v1_etag_versionId_v1", func(t *testing.T) {
resp := signedGetObjectAttributesVersioned(t, cluster, bucketName, "vcond-key", vid1, map[string]string{
"If-Match": etagV1,
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 200, resp.StatusCode,
"If-Match with v1 ETag targeting versionId=v1 should return 200")
})
// If-Match with v2's ETag + versionId=v1 => 412
// The ETag doesn't match v1, so this should fail
t.Run("IfMatch_v2_etag_versionId_v1", func(t *testing.T) {
resp := signedGetObjectAttributesVersioned(t, cluster, bucketName, "vcond-key", vid1, map[string]string{
"If-Match": etagV2,
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 412, resp.StatusCode,
"If-Match with v2 ETag targeting versionId=v1 should return 412")
})
// If-None-Match with v1's ETag + versionId=v1 => 304
t.Run("IfNoneMatch_v1_etag_versionId_v1", func(t *testing.T) {
resp := signedGetObjectAttributesVersioned(t, cluster, bucketName, "vcond-key", vid1, map[string]string{
"If-None-Match": etagV1,
})
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
assert.Equal(t, 304, resp.StatusCode,
"If-None-Match with v1 ETag targeting versionId=v1 should return 304")
})
t.Logf("Versioned conditional headers tests passed: vid1=%s, vid2=%s", vid1, vid2)
}
-212
View File
@@ -422,218 +422,6 @@ func TestS3MultipartOperationsInheritPutObjectPermissions(t *testing.T) {
require.Equal(t, 0, len(listUploadsOut.Uploads))
}
// TestS3IAMManagedPolicyLifecycle is an end-to-end integration test covering the
// user-reported use case in https://github.com/seaweedfs/seaweedfs/issues/8506
// where managed policy operations (GetPolicy, ListPolicies, DeletePolicy,
// AttachUserPolicy, DetachUserPolicy) returned 500 errors.
func TestS3IAMManagedPolicyLifecycle(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
cluster, err := startMiniCluster(t)
require.NoError(t, err)
defer cluster.Stop()
iamClient := newIAMClient(t, cluster.s3Endpoint)
// Step 1: Create a user (this already worked per the issue)
userName := uniqueName("lifecycle-user")
_, err = iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err, "CreateUser should succeed")
// Step 2: Create a managed policy via IAM API
policyName := uniqueName("lifecycle-policy")
policyArn := fmt.Sprintf("arn:aws:iam:::policy/%s", policyName)
policyDoc := `{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:PutObject"],
"Resource": "arn:aws:s3:::*"
}]
}`
createOut, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName),
PolicyDocument: aws.String(policyDoc),
})
require.NoError(t, err, "CreatePolicy should succeed")
require.NotNil(t, createOut.Policy)
require.Equal(t, policyName, *createOut.Policy.PolicyName)
// Step 3: ListPolicies — should include the created policy (was returning 500)
listOut, err := iamClient.ListPolicies(&iam.ListPoliciesInput{})
require.NoError(t, err, "ListPolicies should succeed (was returning 500)")
require.True(t, managedPolicyContains(listOut.Policies, policyName),
"ListPolicies should contain the newly created policy")
// Step 4: GetPolicy by ARN — should return the policy (was returning 500)
getOut, err := iamClient.GetPolicy(&iam.GetPolicyInput{PolicyArn: aws.String(policyArn)})
require.NoError(t, err, "GetPolicy should succeed (was returning 500)")
require.NotNil(t, getOut.Policy)
require.Equal(t, policyName, *getOut.Policy.PolicyName)
require.Equal(t, policyArn, *getOut.Policy.Arn)
// Step 5: AttachUserPolicy — should succeed (was returning 500)
_, err = iamClient.AttachUserPolicy(&iam.AttachUserPolicyInput{
UserName: aws.String(userName),
PolicyArn: aws.String(policyArn),
})
require.NoError(t, err, "AttachUserPolicy should succeed (was returning 500)")
// Step 6: ListAttachedUserPolicies — verify the policy is attached
attachedOut, err := iamClient.ListAttachedUserPolicies(&iam.ListAttachedUserPoliciesInput{
UserName: aws.String(userName),
})
require.NoError(t, err, "ListAttachedUserPolicies should succeed")
require.True(t, attachedPolicyContains(attachedOut.AttachedPolicies, policyName),
"Policy should appear in user's attached policies")
// Step 7: Idempotent re-attach should not fail
_, err = iamClient.AttachUserPolicy(&iam.AttachUserPolicyInput{
UserName: aws.String(userName),
PolicyArn: aws.String(policyArn),
})
require.NoError(t, err, "Re-attaching same policy should be idempotent")
// Step 8: DeletePolicy while attached — should fail with DeleteConflict (AWS behavior)
_, err = iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: aws.String(policyArn)})
require.Error(t, err, "DeletePolicy should fail while policy is attached")
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeDeleteConflictException, awsErr.Code(),
"Should return DeleteConflict when deleting attached policy")
// Step 9: DetachUserPolicy
_, err = iamClient.DetachUserPolicy(&iam.DetachUserPolicyInput{
UserName: aws.String(userName),
PolicyArn: aws.String(policyArn),
})
require.NoError(t, err, "DetachUserPolicy should succeed")
// Verify detached
attachedOut, err = iamClient.ListAttachedUserPolicies(&iam.ListAttachedUserPoliciesInput{
UserName: aws.String(userName),
})
require.NoError(t, err)
require.False(t, attachedPolicyContains(attachedOut.AttachedPolicies, policyName),
"Policy should no longer appear in user's attached policies after detach")
// Step 10: DeletePolicy — should now succeed (was returning XML parsing error)
_, err = iamClient.DeletePolicy(&iam.DeletePolicyInput{PolicyArn: aws.String(policyArn)})
require.NoError(t, err, "DeletePolicy should succeed after detach (was returning XML parsing error)")
// Step 11: Verify the policy is gone
listOut, err = iamClient.ListPolicies(&iam.ListPoliciesInput{})
require.NoError(t, err)
require.False(t, managedPolicyContains(listOut.Policies, policyName),
"Deleted policy should not appear in ListPolicies")
_, err = iamClient.GetPolicy(&iam.GetPolicyInput{PolicyArn: aws.String(policyArn)})
require.Error(t, err, "GetPolicy should fail for deleted policy")
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
}
// TestS3IAMManagedPolicyErrorCases covers error cases from the user-reported issue:
// invalid ARNs, missing policies, and missing users.
func TestS3IAMManagedPolicyErrorCases(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
cluster, err := startMiniCluster(t)
require.NoError(t, err)
defer cluster.Stop()
iamClient := newIAMClient(t, cluster.s3Endpoint)
t.Run("GetPolicy with nonexistent ARN returns NoSuchEntity", func(t *testing.T) {
_, err := iamClient.GetPolicy(&iam.GetPolicyInput{
PolicyArn: aws.String("arn:aws:iam:::policy/does-not-exist"),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
t.Run("DeletePolicy with nonexistent ARN returns NoSuchEntity", func(t *testing.T) {
_, err := iamClient.DeletePolicy(&iam.DeletePolicyInput{
PolicyArn: aws.String("arn:aws:iam:::policy/does-not-exist"),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
t.Run("AttachUserPolicy with nonexistent policy returns NoSuchEntity", func(t *testing.T) {
userName := uniqueName("err-user")
_, err := iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
_, err = iamClient.AttachUserPolicy(&iam.AttachUserPolicyInput{
UserName: aws.String(userName),
PolicyArn: aws.String("arn:aws:iam:::policy/does-not-exist"),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
t.Run("AttachUserPolicy with nonexistent user returns NoSuchEntity", func(t *testing.T) {
policyName := uniqueName("err-policy")
_, err := iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName),
PolicyDocument: aws.String(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}]}`),
})
require.NoError(t, err)
_, err = iamClient.AttachUserPolicy(&iam.AttachUserPolicyInput{
UserName: aws.String("nonexistent-user"),
PolicyArn: aws.String(fmt.Sprintf("arn:aws:iam:::policy/%s", policyName)),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
t.Run("DetachUserPolicy that is not attached returns NoSuchEntity", func(t *testing.T) {
userName := uniqueName("detach-user")
_, err := iamClient.CreateUser(&iam.CreateUserInput{UserName: aws.String(userName)})
require.NoError(t, err)
policyName := uniqueName("detach-policy")
_, err = iamClient.CreatePolicy(&iam.CreatePolicyInput{
PolicyName: aws.String(policyName),
PolicyDocument: aws.String(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}]}`),
})
require.NoError(t, err)
_, err = iamClient.DetachUserPolicy(&iam.DetachUserPolicyInput{
UserName: aws.String(userName),
PolicyArn: aws.String(fmt.Sprintf("arn:aws:iam:::policy/%s", policyName)),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
t.Run("ListAttachedUserPolicies for nonexistent user returns NoSuchEntity", func(t *testing.T) {
_, err := iamClient.ListAttachedUserPolicies(&iam.ListAttachedUserPoliciesInput{
UserName: aws.String("nonexistent-user"),
})
require.Error(t, err)
var awsErr awserr.Error
require.True(t, errors.As(err, &awsErr))
require.Equal(t, iam.ErrCodeNoSuchEntityException, awsErr.Code())
})
}
func execShell(t *testing.T, weedCmd, master, filer, shellCmd string) string {
// weed shell -master=... -filer=...
args := []string{"shell", "-master=" + master, "-filer=" + filer}
+7 -4
View File
@@ -23,6 +23,8 @@ func TestAdminStatusAndHealthz(t *testing.T) {
if err != nil {
t.Fatalf("create status request: %v", err)
}
statusReq.Header.Set(request_id.AmzRequestIDHeader, "test-request-id-1")
statusResp := framework.DoRequest(t, client, statusReq)
statusBody := framework.ReadAllAndClose(t, statusResp)
@@ -32,8 +34,8 @@ func TestAdminStatusAndHealthz(t *testing.T) {
if got := statusResp.Header.Get("Server"); !strings.Contains(got, "SeaweedFS Volume") {
t.Fatalf("expected /status Server header to contain SeaweedFS Volume, got %q", got)
}
if got := statusResp.Header.Get(request_id.AmzRequestIDHeader); got == "" {
t.Fatal("expected server-generated request id in response header")
if got := statusResp.Header.Get(request_id.AmzRequestIDHeader); got != "test-request-id-1" {
t.Fatalf("expected echoed request id, got %q", got)
}
var payload map[string]interface{}
@@ -47,6 +49,7 @@ func TestAdminStatusAndHealthz(t *testing.T) {
}
healthReq := mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/healthz")
healthReq.Header.Set(request_id.AmzRequestIDHeader, "test-request-id-2")
healthResp := framework.DoRequest(t, client, healthReq)
_ = framework.ReadAllAndClose(t, healthResp)
if healthResp.StatusCode != http.StatusOK {
@@ -55,8 +58,8 @@ func TestAdminStatusAndHealthz(t *testing.T) {
if got := healthResp.Header.Get("Server"); !strings.Contains(got, "SeaweedFS Volume") {
t.Fatalf("expected /healthz Server header to contain SeaweedFS Volume, got %q", got)
}
if got := healthResp.Header.Get(request_id.AmzRequestIDHeader); got == "" {
t.Fatal("expected /healthz server-generated request id in response header")
if got := healthResp.Header.Get(request_id.AmzRequestIDHeader); got != "test-request-id-2" {
t.Fatalf("expected /healthz echoed request id, got %q", got)
}
uiResp := framework.DoRequest(t, client, mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/ui/index.html"))
-6
View File
@@ -80,11 +80,6 @@ type AccessKeyInfo struct {
CreatedAt time.Time `json:"created_at"`
}
type CreateAccessKeyRequest struct {
AccessKey string `json:"access_key"`
SecretKey string `json:"secret_key"`
}
type UpdateAccessKeyStatusRequest struct {
Status string `json:"status" binding:"required"`
}
@@ -95,7 +90,6 @@ type UserDetails struct {
Actions []string `json:"actions"`
PolicyNames []string `json:"policy_names"`
AccessKeys []AccessKeyInfo `json:"access_keys"`
Groups []string `json:"groups"`
}
type FilerNode struct {
+1 -2
View File
@@ -4,13 +4,12 @@ import (
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/cluster"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"github.com/seaweedfs/seaweedfs/weed/wdclient/exclusive_locks"
)
const (
adminLockName = cluster.AdminShellLockName
adminLockName = "shell"
adminLockClientName = "admin-plugin"
)
-61
View File
@@ -1,61 +0,0 @@
package dash
import (
"time"
"github.com/seaweedfs/seaweedfs/weed/cluster"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
"github.com/seaweedfs/seaweedfs/weed/wdclient/exclusive_locks"
)
const adminPresenceClientName = "admin-server"
type adminPresenceLock struct {
locker *exclusive_locks.ExclusiveLocker
stopCh chan struct{}
}
func newAdminPresenceLock(masterClient *wdclient.MasterClient) *adminPresenceLock {
if masterClient == nil {
return nil
}
return &adminPresenceLock{
locker: exclusive_locks.NewExclusiveLocker(masterClient, cluster.AdminServerPresenceLockName),
stopCh: make(chan struct{}),
}
}
func (l *adminPresenceLock) Start() {
if l == nil || l.locker == nil {
return
}
l.locker.SetMessage("admin server connected")
go func() {
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
if !l.locker.IsLocked() {
l.locker.RequestLock(adminPresenceClientName)
}
select {
case <-l.stopCh:
return
case <-ticker.C:
}
}
}()
}
func (l *adminPresenceLock) Stop() {
if l == nil {
return
}
select {
case <-l.stopCh:
default:
close(l.stopCh)
}
if l.locker != nil {
l.locker.ReleaseLock()
}
}
+36 -253
View File
@@ -2,9 +2,7 @@ package dash
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"sort"
"strings"
@@ -13,7 +11,6 @@ import (
"github.com/seaweedfs/seaweedfs/weed/admin/maintenance"
adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
"github.com/seaweedfs/seaweedfs/weed/cluster"
clustermaintenance "github.com/seaweedfs/seaweedfs/weed/cluster/maintenance"
"github.com/seaweedfs/seaweedfs/weed/credential"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -102,7 +99,6 @@ type AdminServer struct {
maintenanceManager *maintenance.MaintenanceManager
plugin *adminplugin.Plugin
pluginLock *AdminLockManager
adminPresenceLock *adminPresenceLock
expireJobHandler func(jobID string, reason string) (*adminplugin.TrackedJob, bool, error)
// Topic retention purger
@@ -122,7 +118,7 @@ type AdminServer struct {
// Type definitions moved to types.go
func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string, icebergPort int) *AdminServer {
func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string, icebergPort int, idleSleepDuration time.Duration) *AdminServer {
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.admin")
// Create master client with multiple master support
@@ -141,10 +137,6 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
go masterClient.KeepConnectedToMaster(ctx)
lockManager := NewAdminLockManager(masterClient, adminLockClientName)
presenceLock := newAdminPresenceLock(masterClient)
if presenceLock != nil {
presenceLock.Start()
}
server := &AdminServer{
masterClient: masterClient,
@@ -158,7 +150,6 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
s3TablesManager: newS3TablesManager(),
icebergPort: icebergPort,
pluginLock: lockManager,
adminPresenceLock: presenceLock,
}
// Initialize topic retention purger
@@ -237,19 +228,24 @@ func NewAdminServer(masters string, templateFS http.FileSystem, dataDir string,
}()
}
pluginOpts := adminplugin.Options{
DataDir: dataDir,
plugin, err := adminplugin.New(adminplugin.Options{
DataDir: dataDir,
IdleSleepDuration: idleSleepDuration,
ClusterContextProvider: func(_ context.Context) (*plugin_pb.ClusterContext, error) {
return server.buildDefaultPluginClusterContext(), nil
},
LockManager: lockManager,
ConfigDefaultsProvider: server.enrichConfigDefaults,
}
plugin, err := adminplugin.New(pluginOpts)
LockManager: lockManager,
})
if err != nil && dataDir != "" {
glog.Warningf("Failed to initialize plugin with dataDir=%q: %v. Falling back to in-memory plugin state.", dataDir, err)
pluginOpts.DataDir = ""
plugin, err = adminplugin.New(pluginOpts)
plugin, err = adminplugin.New(adminplugin.Options{
DataDir: "",
IdleSleepDuration: idleSleepDuration,
ClusterContextProvider: func(_ context.Context) (*plugin_pb.ClusterContext, error) {
return server.buildDefaultPluginClusterContext(), nil
},
LockManager: lockManager,
})
}
if err != nil {
glog.Errorf("Failed to initialize plugin: %v", err)
@@ -273,89 +269,6 @@ func (s *AdminServer) loadTaskConfigurationsFromPersistence() {
configUpdateRegistry.UpdateAllConfigs(s.configPersistence)
}
// enrichConfigDefaults is called by the plugin when bootstrapping a job type's
// default config from its descriptor. For admin_script, it fetches maintenance
// scripts from the master and uses them as the script default.
//
// MIGRATION: This exists to help users migrate from master.toml [master.maintenance]
// to the admin script plugin worker. Remove after March 2027.
func (s *AdminServer) enrichConfigDefaults(cfg *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig {
if cfg.JobType != "admin_script" {
return cfg
}
var maintenanceScripts string
var sleepMinutes uint32
err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})
if err != nil {
return err
}
maintenanceScripts = resp.MaintenanceScripts
sleepMinutes = resp.MaintenanceSleepMinutes
return nil
})
if err != nil {
glog.V(1).Infof("Could not fetch master configuration for admin_script defaults: %v", err)
return cfg
}
script := cleanMaintenanceScript(maintenanceScripts)
if script == "" {
return cfg
}
interval := int64(sleepMinutes)
if interval <= 0 {
interval = clustermaintenance.DefaultMaintenanceSleepMinutes
}
glog.V(0).Infof("Enriching admin_script defaults from master maintenance scripts (interval=%dm)", interval)
if cfg.AdminConfigValues == nil {
cfg.AdminConfigValues = make(map[string]*plugin_pb.ConfigValue)
}
cfg.AdminConfigValues["script"] = &plugin_pb.ConfigValue{
Kind: &plugin_pb.ConfigValue_StringValue{StringValue: script},
}
cfg.AdminConfigValues["run_interval_minutes"] = &plugin_pb.ConfigValue{
Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: interval},
}
cfg.UpdatedBy = "master_migration"
return cfg
}
// cleanMaintenanceScript strips lock/unlock commands and normalizes a
// maintenance script string for use with the admin script plugin worker.
//
// MIGRATION: Used by enrichConfigDefaults. Remove after March 2027.
func cleanMaintenanceScript(script string) string {
script = strings.ReplaceAll(script, "\r\n", "\n")
var lines []string
for _, line := range strings.Split(script, "\n") {
trimmed := strings.TrimSpace(line)
if trimmed == "" || strings.HasPrefix(trimmed, "#") {
continue
}
// Strip inline comments (e.g., "lock # migration note")
if idx := strings.Index(trimmed, "#"); idx >= 0 {
trimmed = strings.TrimSpace(trimmed[:idx])
if trimmed == "" {
continue
}
}
firstToken := strings.ToLower(strings.Fields(trimmed)[0])
if firstToken == "lock" || firstToken == "unlock" {
continue
}
lines = append(lines, trimmed)
}
return strings.Join(lines, "\n")
}
// GetCredentialManager returns the credential manager
func (s *AdminServer) GetCredentialManager() *credential.CredentialManager {
return s.credentialManager
@@ -373,21 +286,8 @@ func (s *AdminServer) GetCredentialManager() *credential.CredentialManager {
// InvalidateCache method moved to cluster_topology.go
// GetS3BucketsData retrieves Object Store buckets with pagination and sorting
func (s *AdminServer) GetS3BucketsData(page, pageSize int, sortBy, sortOrder string) (S3BucketsData, error) {
if page < 1 {
page = 1
}
if pageSize < 1 || pageSize > 1000 {
pageSize = 100
}
if sortBy == "" {
sortBy = "name"
}
if sortOrder == "" {
sortOrder = "asc"
}
// GetS3BucketsData retrieves all Object Store buckets and aggregates total storage metrics
func (s *AdminServer) GetS3BucketsData() (S3BucketsData, error) {
buckets, err := s.GetS3Buckets()
if err != nil {
return S3BucketsData{}, err
@@ -398,97 +298,14 @@ func (s *AdminServer) GetS3BucketsData(page, pageSize int, sortBy, sortOrder str
totalSize += bucket.PhysicalSize
}
totalBuckets := len(buckets)
// Sort buckets
s.sortBuckets(buckets, sortBy, sortOrder)
// Calculate pagination
totalPages := (totalBuckets + pageSize - 1) / pageSize
if totalPages == 0 {
totalPages = 1
}
if page > totalPages {
page = totalPages
}
startIndex := (page - 1) * pageSize
endIndex := startIndex + pageSize
if startIndex >= totalBuckets {
buckets = []S3Bucket{}
} else {
if endIndex > totalBuckets {
endIndex = totalBuckets
}
buckets = buckets[startIndex:endIndex]
}
return S3BucketsData{
Buckets: buckets,
TotalBuckets: totalBuckets,
TotalBuckets: len(buckets),
TotalSize: totalSize,
LastUpdated: time.Now(),
CurrentPage: page,
TotalPages: totalPages,
PageSize: pageSize,
SortBy: sortBy,
SortOrder: sortOrder,
}, nil
}
// sortBuckets sorts the bucket slice in place by the given field and order
func (s *AdminServer) sortBuckets(buckets []S3Bucket, sortBy, sortOrder string) {
desc := sortOrder == "desc"
sort.Slice(buckets, func(i, j int) bool {
a, b := buckets[i], buckets[j]
switch sortBy {
case "owner":
if a.Owner != b.Owner {
if desc {
return a.Owner > b.Owner
}
return a.Owner < b.Owner
}
case "created":
if !a.CreatedAt.Equal(b.CreatedAt) {
if desc {
return a.CreatedAt.After(b.CreatedAt)
}
return a.CreatedAt.Before(b.CreatedAt)
}
case "objects":
if a.ObjectCount != b.ObjectCount {
if desc {
return a.ObjectCount > b.ObjectCount
}
return a.ObjectCount < b.ObjectCount
}
case "logical_size":
if a.LogicalSize != b.LogicalSize {
if desc {
return a.LogicalSize > b.LogicalSize
}
return a.LogicalSize < b.LogicalSize
}
case "physical_size":
if a.PhysicalSize != b.PhysicalSize {
if desc {
return a.PhysicalSize > b.PhysicalSize
}
return a.PhysicalSize < b.PhysicalSize
}
}
// Tie-breaker: sort by name (also the default/primary for sortBy=="name")
if a.Name != b.Name {
if desc {
return a.Name > b.Name
}
return a.Name < b.Name
}
return false
})
}
// GetS3Buckets retrieves all Object Store buckets from the filer and collects size/object data from collections
func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
var buckets []S3Bucket
@@ -504,48 +321,28 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
// Now list buckets from the filer and match with collection data
err = s.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
// Paginate through all buckets in the buckets directory
const listPageSize = 1000
startFrom := ""
var snapshotTsNs int64
// List buckets by looking at the buckets directory
stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
Directory: filerConfig.BucketsPath,
Prefix: "",
StartFromFileName: "",
InclusiveStartFrom: false,
Limit: 1000,
})
if err != nil {
return err
}
for {
stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
Directory: filerConfig.BucketsPath,
Prefix: "",
StartFromFileName: startFrom,
InclusiveStartFrom: false,
Limit: listPageSize,
SnapshotTsNs: snapshotTsNs,
})
resp, err := stream.Recv()
if err != nil {
if err.Error() == "EOF" {
break
}
return err
}
pageCount := 0
lastName := ""
for {
resp, err := stream.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return err
}
if snapshotTsNs == 0 && resp.SnapshotTsNs != 0 {
snapshotTsNs = resp.SnapshotTsNs
}
if resp.Entry == nil {
continue
}
lastName = resp.Entry.Name
pageCount++
if !resp.Entry.IsDirectory {
continue
}
if resp.Entry.IsDirectory {
bucketName := resp.Entry.Name
if strings.HasPrefix(bucketName, ".") {
// Skip internal/system directories from Object Store bucket listing.
@@ -598,18 +395,13 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
}
}
var createdAt, lastModified time.Time
if resp.Entry.Attributes != nil {
createdAt = time.Unix(resp.Entry.Attributes.Crtime, 0)
lastModified = time.Unix(resp.Entry.Attributes.Mtime, 0)
}
bucket := S3Bucket{
Name: bucketName,
CreatedAt: createdAt,
CreatedAt: time.Unix(resp.Entry.Attributes.Crtime, 0),
LogicalSize: logicalSize,
PhysicalSize: physicalSize,
ObjectCount: objectCount,
LastModified: lastModified,
LastModified: time.Unix(resp.Entry.Attributes.Mtime, 0),
Quota: quota,
QuotaEnabled: quotaEnabled,
VersioningStatus: versioningStatus,
@@ -620,12 +412,6 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
}
buckets = append(buckets, bucket)
}
// If we received fewer entries than the page size, we've listed everything
if pageCount < listPageSize {
break
}
startFrom = lastName
}
return nil
@@ -1502,9 +1288,6 @@ func (s *AdminServer) Shutdown() {
// Stop maintenance manager
s.StopMaintenanceManager()
if s.adminPresenceLock != nil {
s.adminPresenceLock.Stop()
}
if s.plugin != nil {
s.plugin.Shutdown()
-77
View File
@@ -1,77 +0,0 @@
// MIGRATION: Tests for enrichConfigDefaults helpers. Remove after March 2027.
package dash
import "testing"
func TestCleanMaintenanceScript(t *testing.T) {
tests := []struct {
name string
input string
expected string
}{
{
name: "empty",
input: "",
expected: "",
},
{
name: "only lock unlock",
input: " lock\n unlock\n",
expected: "",
},
{
name: "strips lock and unlock",
input: " lock\n ec.balance -apply\n volume.fix.replication -apply\n unlock\n",
expected: "ec.balance -apply\nvolume.fix.replication -apply",
},
{
name: "case insensitive lock",
input: "Lock\nec.balance -apply\nUNLOCK",
expected: "ec.balance -apply",
},
{
name: "preserves comments removal",
input: "lock\n# a comment\nec.balance -apply\nunlock",
expected: "ec.balance -apply",
},
{
name: "no lock unlock present",
input: "ec.balance -apply\nvolume.fix.replication -apply",
expected: "ec.balance -apply\nvolume.fix.replication -apply",
},
{
name: "windows line endings",
input: "lock\r\nec.balance -apply\r\nunlock\r\n",
expected: "ec.balance -apply",
},
{
name: "lock with inline comment",
input: "lock # migration\nec.balance -apply\nunlock # done",
expected: "ec.balance -apply",
},
{
name: "command with inline comment preserved",
input: "lock\nec.balance -apply # rebalance shards\nunlock",
expected: "ec.balance -apply",
},
{
name: "only inline comment after stripping",
input: "# full line comment\n # indented comment\n",
expected: "",
},
{
name: "typical master default",
input: "\n lock\n ec.encode -fullPercent=95 -quietFor=1h\n ec.rebuild -apply\n ec.balance -apply\n fs.log.purge -daysAgo=7\n volume.deleteEmpty -quietFor=24h -apply\n volume.balance -apply\n volume.fix.replication -apply\n s3.clean.uploads -timeAgo=24h\n unlock\n",
expected: "ec.encode -fullPercent=95 -quietFor=1h\nec.rebuild -apply\nec.balance -apply\nfs.log.purge -daysAgo=7\nvolume.deleteEmpty -quietFor=24h -apply\nvolume.balance -apply\nvolume.fix.replication -apply\ns3.clean.uploads -timeAgo=24h",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := cleanMaintenanceScript(tt.input)
if got != tt.expected {
t.Errorf("cleanMaintenanceScript(%q) = %q, want %q", tt.input, got, tt.expected)
}
})
}
}
+1 -10
View File
@@ -26,15 +26,6 @@ type S3BucketsData struct {
TotalBuckets int `json:"total_buckets"`
TotalSize int64 `json:"total_size"`
LastUpdated time.Time `json:"last_updated"`
// Pagination
CurrentPage int `json:"current_page"`
TotalPages int `json:"total_pages"`
PageSize int `json:"page_size"`
// Sorting
SortBy string `json:"sort_by"`
SortOrder string `json:"sort_order"`
}
type CreateBucketRequest struct {
@@ -57,7 +48,7 @@ type CreateBucketRequest struct {
func (s *AdminServer) ShowS3Buckets(w http.ResponseWriter, r *http.Request) {
username := UsernameFromContext(r.Context())
data, err := s.GetS3BucketsData(1, 100, "name", "asc")
data, err := s.GetS3BucketsData()
if err != nil {
writeJSONError(w, http.StatusInternalServerError, "Failed to get Object Store buckets: "+err.Error())
return
-250
View File
@@ -1,250 +0,0 @@
package dash
import (
"context"
"errors"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/credential"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
)
// cloneGroup creates a deep copy of an iam_pb.Group to avoid mutating stored state.
func cloneGroup(g *iam_pb.Group) *iam_pb.Group {
clone := &iam_pb.Group{
Name: g.Name,
Disabled: g.Disabled,
}
if g.Members != nil {
clone.Members = make([]string, len(g.Members))
copy(clone.Members, g.Members)
}
if g.PolicyNames != nil {
clone.PolicyNames = make([]string, len(g.PolicyNames))
copy(clone.PolicyNames, g.PolicyNames)
}
return clone
}
func (s *AdminServer) GetGroups(ctx context.Context) ([]GroupData, error) {
if s.credentialManager == nil {
return nil, fmt.Errorf("credential manager not available")
}
groupNames, err := s.credentialManager.ListGroups(ctx)
if err != nil {
return nil, fmt.Errorf("failed to list groups: %w", err)
}
var groups []GroupData
for _, name := range groupNames {
g, err := s.credentialManager.GetGroup(ctx, name)
if err != nil {
if errors.Is(err, credential.ErrGroupNotFound) {
glog.V(1).Infof("Group %s listed but not found, skipping", name)
continue
}
return nil, fmt.Errorf("failed to get group %s: %w", name, err)
}
status := "enabled"
if g.Disabled {
status = "disabled"
}
groups = append(groups, GroupData{
Name: g.Name,
MemberCount: len(g.Members),
PolicyCount: len(g.PolicyNames),
Status: status,
Members: g.Members,
PolicyNames: g.PolicyNames,
})
}
return groups, nil
}
func (s *AdminServer) GetGroupDetails(ctx context.Context, name string) (*GroupData, error) {
if s.credentialManager == nil {
return nil, fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, name)
if err != nil {
return nil, fmt.Errorf("failed to get group: %w", err)
}
status := "enabled"
if g.Disabled {
status = "disabled"
}
return &GroupData{
Name: g.Name,
MemberCount: len(g.Members),
PolicyCount: len(g.PolicyNames),
Status: status,
Members: g.Members,
PolicyNames: g.PolicyNames,
}, nil
}
func (s *AdminServer) CreateGroup(ctx context.Context, name string) (*GroupData, error) {
if s.credentialManager == nil {
return nil, fmt.Errorf("credential manager not available")
}
group := &iam_pb.Group{Name: name}
if err := s.credentialManager.CreateGroup(ctx, group); err != nil {
return nil, fmt.Errorf("failed to create group: %w", err)
}
glog.V(1).Infof("Created group %s", group.Name)
return &GroupData{
Name: group.Name,
Status: "enabled",
}, nil
}
func (s *AdminServer) DeleteGroup(ctx context.Context, name string) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
// Check for members and attached policies before deleting (same guards as IAM handlers)
g, err := s.credentialManager.GetGroup(ctx, name)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
if len(g.Members) > 0 {
return fmt.Errorf("cannot delete group %s: group has %d member(s): %w", name, len(g.Members), credential.ErrGroupNotEmpty)
}
if len(g.PolicyNames) > 0 {
return fmt.Errorf("cannot delete group %s: group has %d attached policy(ies): %w", name, len(g.PolicyNames), credential.ErrGroupNotEmpty)
}
if err := s.credentialManager.DeleteGroup(ctx, name); err != nil {
return fmt.Errorf("failed to delete group: %w", err)
}
glog.V(1).Infof("Deleted group %s", name)
return nil
}
func (s *AdminServer) AddGroupMember(ctx context.Context, groupName, username string) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, groupName)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
g = cloneGroup(g)
if _, err := s.credentialManager.GetUser(ctx, username); err != nil {
return fmt.Errorf("user %s not found: %w", username, err)
}
for _, m := range g.Members {
if m == username {
return nil // already a member
}
}
g.Members = append(g.Members, username)
if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
return fmt.Errorf("failed to update group: %w", err)
}
glog.V(1).Infof("Added user %s to group %s", username, groupName)
return nil
}
func (s *AdminServer) RemoveGroupMember(ctx context.Context, groupName, username string) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, groupName)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
g = cloneGroup(g)
found := false
var newMembers []string
for _, m := range g.Members {
if m == username {
found = true
} else {
newMembers = append(newMembers, m)
}
}
if !found {
return fmt.Errorf("user %s is not a member of group %s: %w", username, groupName, credential.ErrUserNotInGroup)
}
g.Members = newMembers
if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
return fmt.Errorf("failed to update group: %w", err)
}
glog.V(1).Infof("Removed user %s from group %s", username, groupName)
return nil
}
func (s *AdminServer) AttachGroupPolicy(ctx context.Context, groupName, policyName string) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, groupName)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
g = cloneGroup(g)
if _, err := s.credentialManager.GetPolicy(ctx, policyName); err != nil {
return fmt.Errorf("policy %s not found: %w", policyName, err)
}
for _, p := range g.PolicyNames {
if p == policyName {
return nil // already attached
}
}
g.PolicyNames = append(g.PolicyNames, policyName)
if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
return fmt.Errorf("failed to update group: %w", err)
}
glog.V(1).Infof("Attached policy %s to group %s", policyName, groupName)
return nil
}
func (s *AdminServer) DetachGroupPolicy(ctx context.Context, groupName, policyName string) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, groupName)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
g = cloneGroup(g)
found := false
var newPolicies []string
for _, p := range g.PolicyNames {
if p == policyName {
found = true
} else {
newPolicies = append(newPolicies, p)
}
}
if !found {
return fmt.Errorf("policy %s is not attached to group %s: %w", policyName, groupName, credential.ErrPolicyNotAttached)
}
g.PolicyNames = newPolicies
if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
return fmt.Errorf("failed to update group: %w", err)
}
glog.V(1).Infof("Detached policy %s from group %s", policyName, groupName)
return nil
}
func (s *AdminServer) SetGroupStatus(ctx context.Context, groupName string, enabled bool) error {
if s.credentialManager == nil {
return fmt.Errorf("credential manager not available")
}
g, err := s.credentialManager.GetGroup(ctx, groupName)
if err != nil {
return fmt.Errorf("failed to get group: %w", err)
}
g = cloneGroup(g)
g.Disabled = !enabled
if err := s.credentialManager.UpdateGroup(ctx, g); err != nil {
return fmt.Errorf("failed to update group: %w", err)
}
glog.V(1).Infof("Set group %s status to enabled=%v", groupName, enabled)
return nil
}
+5 -8
View File
@@ -324,7 +324,7 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
}
// Only process directories that are versions (start with "v")
if versionResp.Entry != nil && versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
if versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
versionDir := filepath.Join(topicDir, versionResp.Entry.Name)
// List all partition directories under the version directory (e.g., 0315-0630)
@@ -352,7 +352,7 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
}
// Only process directories that are partitions (format: NNNN-NNNN)
if partitionResp.Entry != nil && partitionResp.Entry.IsDirectory {
if partitionResp.Entry.IsDirectory {
// Parse partition range to get partition start ID (e.g., "0315-0630" -> 315)
var partitionStart, partitionStop int32
if n, err := fmt.Sscanf(partitionResp.Entry.Name, "%04d-%04d", &partitionStart, &partitionStop); n != 2 || err != nil {
@@ -387,11 +387,11 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
}
// Only process .offset files
if offsetResp.Entry != nil && !offsetResp.Entry.IsDirectory && strings.HasSuffix(offsetResp.Entry.Name, ".offset") {
if !offsetResp.Entry.IsDirectory && strings.HasSuffix(offsetResp.Entry.Name, ".offset") {
consumerGroup := strings.TrimSuffix(offsetResp.Entry.Name, ".offset")
// Read the offset value from the file
offsetData, err := filer.ReadInsideFiler(context.Background(), client, partitionDir, offsetResp.Entry.Name)
offsetData, err := filer.ReadInsideFiler(client, partitionDir, offsetResp.Entry.Name)
if err != nil {
glog.Warningf("Failed to read offset file %s: %v", offsetResp.Entry.Name, err)
continue
@@ -401,10 +401,7 @@ func (s *AdminServer) GetConsumerGroupOffsets(namespace, topicName string) ([]Co
offset := int64(util.BytesToUint64(offsetData))
// Get the file modification time
var lastUpdated time.Time
if offsetResp.Entry.Attributes != nil {
lastUpdated = time.Unix(offsetResp.Entry.Attributes.Mtime, 0)
}
lastUpdated := time.Unix(offsetResp.Entry.Attributes.Mtime, 0)
offsets = append(offsets, ConsumerGroupOffsetInfo{
ConsumerGroup: consumerGroup,
-3
View File
@@ -867,9 +867,6 @@ func applyDescriptorDefaultsToPersistedConfig(
if runtime.PerWorkerExecutionConcurrency <= 0 {
runtime.PerWorkerExecutionConcurrency = defaults.PerWorkerExecutionConcurrency
}
if runtime.JobTypeMaxRuntimeSeconds <= 0 {
runtime.JobTypeMaxRuntimeSeconds = defaults.JobTypeMaxRuntimeSeconds
}
if runtime.RetryBackoffSeconds <= 0 {
runtime.RetryBackoffSeconds = defaults.RetryBackoffSeconds
}
+2 -9
View File
@@ -151,21 +151,17 @@ func (p *TopicRetentionPurger) purgeTopicData(topicRetention TopicRetentionConfi
}
// Only process directories that are versions (start with "v")
if versionResp.Entry != nil && versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
if versionResp.Entry.IsDirectory && strings.HasPrefix(versionResp.Entry.Name, "v") {
versionTime, err := p.parseVersionTime(versionResp.Entry.Name)
if err != nil {
glog.Warningf("Failed to parse version time from %s: %v", versionResp.Entry.Name, err)
continue
}
var modTime time.Time
if versionResp.Entry.Attributes != nil {
modTime = time.Unix(versionResp.Entry.Attributes.Mtime, 0)
}
versionDirs = append(versionDirs, VersionDirInfo{
Name: versionResp.Entry.Name,
VersionTime: versionTime,
ModTime: modTime,
ModTime: time.Unix(versionResp.Entry.Attributes.Mtime, 0),
})
}
}
@@ -264,9 +260,6 @@ func (p *TopicRetentionPurger) deleteDirectoryRecursively(client filer_pb.Seawee
return fmt.Errorf("failed to receive entries: %w", err)
}
if resp.Entry == nil {
continue
}
entryPath := filepath.Join(dirPath, resp.Entry.Name)
if resp.Entry.IsDirectory {
-24
View File
@@ -589,30 +589,6 @@ type UpdateServiceAccountRequest struct {
Expiration string `json:"expiration,omitempty"`
}
// Group management structures
type GroupData struct {
Name string `json:"name"`
MemberCount int `json:"member_count"`
PolicyCount int `json:"policy_count"`
Status string `json:"status"` // "enabled" or "disabled"
Members []string `json:"members"`
PolicyNames []string `json:"policy_names"`
}
type GroupsPageData struct {
Username string `json:"username"`
Groups []GroupData `json:"groups"`
TotalGroups int `json:"total_groups"`
ActiveGroups int `json:"active_groups"`
AvailableUsers []string `json:"available_users"`
AvailablePolicies []string `json:"available_policies"`
LastUpdated time.Time `json:"last_updated"`
}
type CreateGroupRequest struct {
Name string `json:"name"`
}
// STS Configuration display types
type STSConfigData struct {
Enabled bool `json:"enabled"`
+5 -64
View File
@@ -4,21 +4,13 @@ import (
"context"
"crypto/rand"
"encoding/base64"
"errors"
"fmt"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/credential"
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
)
var (
ErrAccessKeyInUse = errors.New("access key already in use")
ErrUserNotFound = errors.New("user not found")
ErrInvalidInput = errors.New("invalid input")
)
// CreateObjectStoreUser creates a new user using the credential manager
func (s *AdminServer) CreateObjectStoreUser(req CreateUserRequest) (*ObjectStoreUser, error) {
if s.credentialManager == nil {
@@ -195,24 +187,6 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
details.Email = identity.Account.EmailAddress
}
// Look up groups the user belongs to
groupNames, err := s.credentialManager.ListGroups(ctx)
if err != nil {
return nil, fmt.Errorf("failed to list groups: %w", err)
}
for _, gName := range groupNames {
g, err := s.credentialManager.GetGroup(ctx, gName)
if err != nil {
return nil, fmt.Errorf("failed to get group %s: %w", gName, err)
}
for _, member := range g.Members {
if member == username {
details.Groups = append(details.Groups, gName)
break
}
}
}
// Convert credentials to access key info
for _, cred := range identity.Credentials {
details.AccessKeys = append(details.AccessKeys, AccessKeyInfo{
@@ -227,7 +201,7 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
}
// CreateAccessKey creates a new access key for a user
func (s *AdminServer) CreateAccessKey(username string, req *CreateAccessKeyRequest) (*AccessKeyInfo, error) {
func (s *AdminServer) CreateAccessKey(username string) (*AccessKeyInfo, error) {
if s.credentialManager == nil {
return nil, fmt.Errorf("credential manager not available")
}
@@ -238,41 +212,14 @@ func (s *AdminServer) CreateAccessKey(username string, req *CreateAccessKeyReque
_, err := s.credentialManager.GetUser(ctx, username)
if err != nil {
if err == credential.ErrUserNotFound {
return nil, fmt.Errorf("user %s: %w", username, ErrUserNotFound)
return nil, fmt.Errorf("user %s not found", username)
}
return nil, fmt.Errorf("failed to get user: %w", err)
}
if req == nil {
req = &CreateAccessKeyRequest{}
}
// Validate provided keys
if req.AccessKey != "" && (len(req.AccessKey) < 4 || len(req.AccessKey) > 128) {
return nil, fmt.Errorf("access key must be between 4 and 128 characters: %w", ErrInvalidInput)
}
if req.SecretKey != "" && (len(req.SecretKey) < 8 || len(req.SecretKey) > 128) {
return nil, fmt.Errorf("secret key must be between 8 and 128 characters: %w", ErrInvalidInput)
}
// Use provided keys or generate new ones
accessKey := req.AccessKey
if accessKey == "" {
accessKey = generateAccessKey()
}
secretKey := req.SecretKey
if secretKey == "" {
secretKey = generateSecretKey()
}
// Verify access key is globally unique
existingUser, err := s.credentialManager.GetUserByAccessKey(ctx, accessKey)
if existingUser != nil {
return nil, ErrAccessKeyInUse
}
if err != nil && !errors.Is(err, credential.ErrAccessKeyNotFound) && !isNotFoundError(err) {
return nil, fmt.Errorf("failed to check access key uniqueness: %w", err)
}
// Generate new access key
accessKey := generateAccessKey()
secretKey := generateSecretKey()
credential := &iam_pb.Credential{
AccessKey: accessKey,
@@ -417,12 +364,6 @@ func (s *AdminServer) UpdateUserPolicies(username string, actions []string) erro
return nil
}
// isNotFoundError checks for "not found" in the error message as a fallback
// for stores (e.g. gRPC) that don't return the credential.ErrAccessKeyNotFound sentinel.
func isNotFoundError(err error) bool {
return err != nil && strings.Contains(strings.ToLower(err.Error()), "not found")
}
// Helper functions for generating keys and IDs
func generateAccessKey() string {
// Generate 20-character access key (AWS standard)
-1
View File
@@ -457,7 +457,6 @@ func (s *AdminServer) GetClusterVolumeServers() (*ClusterVolumeServersData, erro
// Process disk information
for _, diskInfo := range node.DiskInfos {
vs.MaxVolumes += int(diskInfo.MaxVolumeCount)
vs.DiskCapacity += int64(diskInfo.MaxVolumeCount) * int64(volumeSizeLimitMB) * 1024 * 1024 // Use actual volume size limit
// Count regular volumes and calculate disk usage
+5 -46
View File
@@ -3,7 +3,6 @@ package handlers
import (
"net/http"
"net/url"
"strconv"
"time"
"github.com/gorilla/mux"
@@ -29,7 +28,6 @@ type AdminHandlers struct {
pluginHandlers *PluginHandlers
mqHandlers *MessageQueueHandlers
serviceAccountHandlers *ServiceAccountHandlers
groupHandlers *GroupHandlers
}
// NewAdminHandlers creates a new instance of AdminHandlers
@@ -42,7 +40,6 @@ func NewAdminHandlers(adminServer *dash.AdminServer, store sessions.Store) *Admi
pluginHandlers := NewPluginHandlers(adminServer)
mqHandlers := NewMessageQueueHandlers(adminServer)
serviceAccountHandlers := NewServiceAccountHandlers(adminServer)
groupHandlers := NewGroupHandlers(adminServer)
return &AdminHandlers{
adminServer: adminServer,
sessionStore: store,
@@ -54,7 +51,6 @@ func NewAdminHandlers(adminServer *dash.AdminServer, store sessions.Store) *Admi
pluginHandlers: pluginHandlers,
mqHandlers: mqHandlers,
serviceAccountHandlers: serviceAccountHandlers,
groupHandlers: groupHandlers,
}
}
@@ -108,7 +104,6 @@ func (h *AdminHandlers) registerUIRoutes(r *mux.Router) {
r.HandleFunc("/object-store/buckets/{bucket}", h.ShowBucketDetails).Methods(http.MethodGet)
r.HandleFunc("/object-store/users", h.userHandlers.ShowObjectStoreUsers).Methods(http.MethodGet)
r.HandleFunc("/object-store/policies", h.policyHandlers.ShowPolicies).Methods(http.MethodGet)
r.HandleFunc("/object-store/groups", h.groupHandlers.ShowGroups).Methods(http.MethodGet)
r.HandleFunc("/object-store/service-accounts", h.serviceAccountHandlers.ShowServiceAccounts).Methods(http.MethodGet)
r.HandleFunc("/object-store/s3tables/buckets", h.ShowS3TablesBuckets).Methods(http.MethodGet)
r.HandleFunc("/object-store/s3tables/buckets/{bucket}/namespaces", h.ShowS3TablesNamespaces).Methods(http.MethodGet)
@@ -190,19 +185,6 @@ func (h *AdminHandlers) registerAPIRoutes(api *mux.Router, enforceWrite bool) {
saApi.Handle("/{id}", wrapWrite(h.serviceAccountHandlers.UpdateServiceAccount)).Methods(http.MethodPut)
saApi.Handle("/{id}", wrapWrite(h.serviceAccountHandlers.DeleteServiceAccount)).Methods(http.MethodDelete)
groupsApi := api.PathPrefix("/groups").Subrouter()
groupsApi.HandleFunc("", h.groupHandlers.GetGroups).Methods(http.MethodGet)
groupsApi.Handle("", wrapWrite(h.groupHandlers.CreateGroup)).Methods(http.MethodPost)
groupsApi.HandleFunc("/{name}", h.groupHandlers.GetGroupDetails).Methods(http.MethodGet)
groupsApi.Handle("/{name}", wrapWrite(h.groupHandlers.DeleteGroup)).Methods(http.MethodDelete)
groupsApi.Handle("/{name}/status", wrapWrite(h.groupHandlers.SetGroupStatus)).Methods(http.MethodPut)
groupsApi.HandleFunc("/{name}/members", h.groupHandlers.GetGroupMembers).Methods(http.MethodGet)
groupsApi.Handle("/{name}/members", wrapWrite(h.groupHandlers.AddGroupMember)).Methods(http.MethodPost)
groupsApi.Handle("/{name}/members/{username}", wrapWrite(h.groupHandlers.RemoveGroupMember)).Methods(http.MethodDelete)
groupsApi.HandleFunc("/{name}/policies", h.groupHandlers.GetGroupPolicies).Methods(http.MethodGet)
groupsApi.Handle("/{name}/policies", wrapWrite(h.groupHandlers.AttachGroupPolicy)).Methods(http.MethodPost)
groupsApi.Handle("/{name}/policies/{policyName}", wrapWrite(h.groupHandlers.DetachGroupPolicy)).Methods(http.MethodDelete)
policyApi := api.PathPrefix("/object-store/policies").Subrouter()
policyApi.HandleFunc("", h.policyHandlers.GetPolicies).Methods(http.MethodGet)
policyApi.Handle("", wrapWrite(h.policyHandlers.CreatePolicy)).Methods(http.MethodPost)
@@ -294,26 +276,8 @@ func (h *AdminHandlers) ShowDashboard(w http.ResponseWriter, r *http.Request) {
// ShowS3Buckets renders the Object Store buckets management page
func (h *AdminHandlers) ShowS3Buckets(w http.ResponseWriter, r *http.Request) {
// Get pagination and sorting parameters from query string
page := 1
if p := r.URL.Query().Get("page"); p != "" {
if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
page = parsed
}
}
pageSize := 100
if ps := r.URL.Query().Get("pageSize"); ps != "" {
if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
pageSize = parsed
}
}
sortBy := defaultQuery(r.URL.Query().Get("sortBy"), "name")
sortOrder := defaultQuery(r.URL.Query().Get("sortOrder"), "asc")
// Get Object Store buckets data with pagination
s3Data := h.getS3BucketsData(r, page, pageSize, sortBy, sortOrder)
// Get Object Store buckets data from the server
s3Data := h.getS3BucketsData(r)
username := h.getUsername(r)
// Render HTML template
@@ -480,15 +444,15 @@ func (h *AdminHandlers) ShowBucketDetails(w http.ResponseWriter, r *http.Request
writeJSON(w, http.StatusOK, details)
}
// getS3BucketsData retrieves Object Store buckets data from the server with pagination
func (h *AdminHandlers) getS3BucketsData(r *http.Request, page, pageSize int, sortBy, sortOrder string) dash.S3BucketsData {
// getS3BucketsData retrieves Object Store buckets data from the server
func (h *AdminHandlers) getS3BucketsData(r *http.Request) dash.S3BucketsData {
username := dash.UsernameFromContext(r.Context())
if username == "" {
username = "admin"
}
// Get Object Store buckets data
data, err := h.adminServer.GetS3BucketsData(page, pageSize, sortBy, sortOrder)
data, err := h.adminServer.GetS3BucketsData()
if err != nil {
// Return empty data on error
return dash.S3BucketsData{
@@ -497,11 +461,6 @@ func (h *AdminHandlers) getS3BucketsData(r *http.Request, page, pageSize int, so
TotalBuckets: 0,
TotalSize: 0,
LastUpdated: time.Now(),
CurrentPage: 1,
TotalPages: 1,
PageSize: pageSize,
SortBy: sortBy,
SortOrder: sortOrder,
}
}
-271
View File
@@ -1,271 +0,0 @@
package handlers
import (
"bytes"
"errors"
"net/http"
"time"
"github.com/gorilla/mux"
"github.com/seaweedfs/seaweedfs/weed/admin/dash"
"github.com/seaweedfs/seaweedfs/weed/admin/view/app"
"github.com/seaweedfs/seaweedfs/weed/admin/view/layout"
"github.com/seaweedfs/seaweedfs/weed/credential"
"github.com/seaweedfs/seaweedfs/weed/glog"
)
func groupErrorToHTTPStatus(err error) int {
if errors.Is(err, credential.ErrGroupNotFound) {
return http.StatusNotFound
}
if errors.Is(err, credential.ErrGroupAlreadyExists) {
return http.StatusConflict
}
if errors.Is(err, credential.ErrUserNotInGroup) {
return http.StatusBadRequest
}
if errors.Is(err, credential.ErrPolicyNotAttached) {
return http.StatusBadRequest
}
if errors.Is(err, credential.ErrUserNotFound) {
return http.StatusNotFound
}
if errors.Is(err, credential.ErrPolicyNotFound) {
return http.StatusNotFound
}
if errors.Is(err, credential.ErrGroupNotEmpty) {
return http.StatusConflict
}
return http.StatusInternalServerError
}
type GroupHandlers struct {
adminServer *dash.AdminServer
}
func NewGroupHandlers(adminServer *dash.AdminServer) *GroupHandlers {
return &GroupHandlers{adminServer: adminServer}
}
func (h *GroupHandlers) ShowGroups(w http.ResponseWriter, r *http.Request) {
data, err := h.getGroupsPageData(r)
if err != nil {
glog.Errorf("Failed to get groups data: %v", err)
writeJSONError(w, http.StatusInternalServerError, "Failed to load groups: "+err.Error())
return
}
var buf bytes.Buffer
component := app.Groups(data)
viewCtx := layout.NewViewContext(r, dash.UsernameFromContext(r.Context()), dash.CSRFTokenFromContext(r.Context()))
layoutComponent := layout.Layout(viewCtx, component)
if err := layoutComponent.Render(r.Context(), &buf); err != nil {
glog.Errorf("Failed to render groups template: %v", err)
w.WriteHeader(http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/html")
_, _ = w.Write(buf.Bytes())
}
func (h *GroupHandlers) GetGroups(w http.ResponseWriter, r *http.Request) {
groups, err := h.adminServer.GetGroups(r.Context())
if err != nil {
glog.Errorf("Failed to get groups: %v", err)
writeJSONError(w, http.StatusInternalServerError, "Failed to get groups")
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"groups": groups})
}
func (h *GroupHandlers) CreateGroup(w http.ResponseWriter, r *http.Request) {
var req dash.CreateGroupRequest
if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
return
}
if req.Name == "" {
writeJSONError(w, http.StatusBadRequest, "Group name is required")
return
}
group, err := h.adminServer.CreateGroup(r.Context(), req.Name)
if err != nil {
glog.Errorf("Failed to create group: %v", err)
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to create group: "+err.Error())
return
}
writeJSON(w, http.StatusOK, group)
}
func (h *GroupHandlers) GetGroupDetails(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
group, err := h.adminServer.GetGroupDetails(r.Context(), name)
if err != nil {
glog.Errorf("Failed to get group details: %v", err)
status := groupErrorToHTTPStatus(err)
msg := "Failed to retrieve group"
if status == http.StatusNotFound {
msg = "Group not found"
}
writeJSONError(w, status, msg)
return
}
writeJSON(w, http.StatusOK, group)
}
func (h *GroupHandlers) DeleteGroup(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
if err := h.adminServer.DeleteGroup(r.Context(), name); err != nil {
glog.Errorf("Failed to delete group: %v", err)
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to delete group: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Group deleted successfully"})
}
func (h *GroupHandlers) GetGroupMembers(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
group, err := h.adminServer.GetGroupDetails(r.Context(), name)
if err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to get group: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"members": group.Members})
}
func (h *GroupHandlers) AddGroupMember(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
var req struct {
Username string `json:"username"`
}
if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
return
}
if req.Username == "" {
writeJSONError(w, http.StatusBadRequest, "Username is required")
return
}
if err := h.adminServer.AddGroupMember(r.Context(), name, req.Username); err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to add member: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Member added successfully"})
}
func (h *GroupHandlers) RemoveGroupMember(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
username := mux.Vars(r)["username"]
if err := h.adminServer.RemoveGroupMember(r.Context(), name, username); err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to remove member: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Member removed successfully"})
}
func (h *GroupHandlers) GetGroupPolicies(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
group, err := h.adminServer.GetGroupDetails(r.Context(), name)
if err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to get group: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"policies": group.PolicyNames})
}
func (h *GroupHandlers) AttachGroupPolicy(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
var req struct {
PolicyName string `json:"policy_name"`
}
if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
return
}
if req.PolicyName == "" {
writeJSONError(w, http.StatusBadRequest, "Policy name is required")
return
}
if err := h.adminServer.AttachGroupPolicy(r.Context(), name, req.PolicyName); err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to attach policy: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Policy attached successfully"})
}
func (h *GroupHandlers) DetachGroupPolicy(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
policyName := mux.Vars(r)["policyName"]
if err := h.adminServer.DetachGroupPolicy(r.Context(), name, policyName); err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to detach policy: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Policy detached successfully"})
}
func (h *GroupHandlers) SetGroupStatus(w http.ResponseWriter, r *http.Request) {
name := mux.Vars(r)["name"]
var req struct {
Enabled *bool `json:"enabled"`
}
if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
return
}
if req.Enabled == nil {
writeJSONError(w, http.StatusBadRequest, "enabled field is required")
return
}
if err := h.adminServer.SetGroupStatus(r.Context(), name, *req.Enabled); err != nil {
writeJSONError(w, groupErrorToHTTPStatus(err), "Failed to update group status: "+err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "Group status updated"})
}
func (h *GroupHandlers) getGroupsPageData(r *http.Request) (dash.GroupsPageData, error) {
username := dash.UsernameFromContext(r.Context())
if username == "" {
username = "admin"
}
groups, err := h.adminServer.GetGroups(r.Context())
if err != nil {
return dash.GroupsPageData{}, err
}
activeCount := 0
for _, g := range groups {
if g.Status == "enabled" {
activeCount++
}
}
// Get available users for dropdown
var availableUsers []string
users, err := h.adminServer.GetObjectStoreUsers(r.Context())
if err == nil {
for _, user := range users {
availableUsers = append(availableUsers, user.Username)
}
}
// Get available policies for dropdown
var availablePolicies []string
policies, err := h.adminServer.GetPolicies()
if err == nil {
for _, p := range policies {
availablePolicies = append(availablePolicies, p.Name)
}
}
return dash.GroupsPageData{
Username: username,
Groups: groups,
TotalGroups: len(groups),
ActiveGroups: activeCount,
AvailableUsers: availableUsers,
AvailablePolicies: availablePolicies,
LastUpdated: time.Now(),
}, nil
}
+2 -24
View File
@@ -1,9 +1,7 @@
package handlers
import (
"errors"
"fmt"
"io"
"net/http"
"time"
@@ -157,30 +155,10 @@ func (h *UserHandlers) CreateAccessKey(w http.ResponseWriter, r *http.Request) {
return
}
var req *dash.CreateAccessKeyRequest
var body dash.CreateAccessKeyRequest
if err := decodeJSONBody(newJSONMaxReader(w, r), &body); err != nil {
if !errors.Is(err, io.EOF) {
writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
return
}
// Empty body: auto-generate both keys
} else {
req = &body
}
accessKey, err := h.adminServer.CreateAccessKey(username, req)
accessKey, err := h.adminServer.CreateAccessKey(username)
if err != nil {
glog.Errorf("Failed to create access key for user %s: %v", username, err)
if errors.Is(err, dash.ErrAccessKeyInUse) {
writeJSONError(w, http.StatusConflict, err.Error())
} else if errors.Is(err, dash.ErrUserNotFound) {
writeJSONError(w, http.StatusNotFound, err.Error())
} else if errors.Is(err, dash.ErrInvalidInput) {
writeJSONError(w, http.StatusBadRequest, err.Error())
} else {
writeJSONError(w, http.StatusInternalServerError, "Failed to create access key: "+err.Error())
}
writeJSONError(w, http.StatusInternalServerError, "Failed to create access key: "+err.Error())
return
}
@@ -5,7 +5,6 @@ import (
"github.com/seaweedfs/seaweedfs/weed/admin/topology"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/worker/tasks"
"github.com/seaweedfs/seaweedfs/weed/worker/types"
@@ -230,12 +229,6 @@ func (s *MaintenanceIntegration) ScanWithTaskDetectors(volumeMetrics []*types.Vo
continue
}
// Cancel stale pending tasks for this type before re-detection
maintenanceType := s.taskTypeMap[taskType]
if cancelled := s.maintenanceQueue.CancelPendingTasksByType(maintenanceType); cancelled > 0 {
glog.Infof("Cancelled %d stale pending %s tasks before re-detection", cancelled, taskType)
}
glog.V(2).Infof("Running detection for task type: %s", taskType)
results, err := detector.ScanForTasks(filteredMetrics, clusterInfo)
@@ -535,15 +528,10 @@ func (s *MaintenanceIntegration) SyncTask(task *MaintenanceTask) {
// Volume size is not currently used for Balance/Vacuum impact and is not stored in MaintenanceTask
sourceImpact, targetImpact := topology.CalculateTaskStorageImpact(topology.TaskType(string(taskType)), 0)
// Use unified sources and targets from TaskParams.
// Task protos store ServerAddresses (with gRPC port, e.g., "host:port.grpcPort")
// but the topology indexes disks by NodeId (e.g., "host:port").
// Strip the gRPC port suffix via ToHttpAddress() to match the topology key.
// Use unified sources and targets from TaskParams
for _, src := range task.TypedParams.Sources {
resolvedSrc := pb.ServerAddress(src.Node).ToHttpAddress()
glog.V(2).Infof("SyncTask %s: source proto Node=%q resolved to %q, diskId=%d", task.ID, src.Node, resolvedSrc, src.DiskId)
sources = append(sources, topology.TaskSource{
SourceServer: resolvedSrc,
SourceServer: src.Node,
SourceDisk: src.DiskId,
StorageChange: sourceImpact,
})
@@ -551,10 +539,8 @@ func (s *MaintenanceIntegration) SyncTask(task *MaintenanceTask) {
estimatedSize += int64(src.EstimatedSize)
}
for _, target := range task.TypedParams.Targets {
resolvedTarget := pb.ServerAddress(target.Node).ToHttpAddress()
glog.V(2).Infof("SyncTask %s: target proto Node=%q resolved to %q, diskId=%d", task.ID, target.Node, resolvedTarget, target.DiskId)
destinations = append(destinations, topology.TaskDestination{
TargetServer: resolvedTarget,
TargetServer: target.Node,
TargetDisk: target.DiskId,
StorageChange: targetImpact,
})
@@ -297,7 +297,7 @@ func (mm *MaintenanceManager) logTopologyStatus() {
errorCount := mm.errorCount
mm.mutex.RUnlock()
glog.V(1).Infof("Topology status: %d nodes, %d disks, %d workers, %d pending tasks, %d running tasks, errors: %d",
glog.V(0).Infof("Topology status: %d nodes, %d disks, %d workers, %d pending tasks, %d running tasks, errors: %d",
nodeCount, diskCount, workerCount,
stats.TasksByStatus[TaskStatusPending],
stats.TasksByStatus[TaskStatusInProgress]+stats.TasksByStatus[TaskStatusAssigned],
+2 -49
View File
@@ -4,7 +4,6 @@ import (
"crypto/rand"
"fmt"
"sort"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -231,46 +230,6 @@ func (mq *MaintenanceQueue) hasDuplicateTask(newTask *MaintenanceTask) bool {
return false
}
// CancelPendingTasksByType cancels all pending tasks of a given type.
// This is called before each detection cycle to ensure stale proposals
// from previous cycles are cleaned up before creating new ones.
func (mq *MaintenanceQueue) CancelPendingTasksByType(taskType MaintenanceTaskType) int {
mq.mutex.Lock()
var remaining []*MaintenanceTask
var cancelledSnapshots []*MaintenanceTask
cancelled := 0
for _, task := range mq.pendingTasks {
if task.Type == taskType {
task.Status = TaskStatusCancelled
now := time.Now()
task.CompletedAt = &now
cancelled++
cancelledSnapshots = append(cancelledSnapshots, snapshotTask(task))
glog.V(1).Infof("Cancelled stale pending task %s (%s) for volume %d before re-detection",
task.ID, task.Type, task.VolumeID)
// Release capacity in ActiveTopology and remove pending operation
if mq.integration != nil {
if at := mq.integration.GetActiveTopology(); at != nil {
_ = at.CompleteTask(task.ID)
}
}
mq.removePendingOperation(task.ID)
} else {
remaining = append(remaining, task)
}
}
mq.pendingTasks = remaining
mq.mutex.Unlock()
// Persist cancelled state outside the lock to avoid blocking
for _, snapshot := range cancelledSnapshots {
mq.saveTaskState(snapshot)
}
return cancelled
}
// AddTasksFromResults converts detection results to tasks and adds them to the queue
func (mq *MaintenanceQueue) AddTasksFromResults(results []*TaskDetectionResult) {
for _, result := range results {
@@ -496,8 +455,8 @@ func (mq *MaintenanceQueue) CompleteTask(taskID string, error string) {
task.Status = TaskStatusFailed
task.Error = error
// Check if task should be retried (skip retry for permanent errors)
if task.RetryCount < task.MaxRetries && !isNonRetriableError(error) {
// Check if task should be retried
if task.RetryCount < task.MaxRetries {
// Record unassignment due to failure/retry
if task.WorkerID != "" && len(task.AssignmentHistory) > 0 {
lastAssignment := task.AssignmentHistory[len(task.AssignmentHistory)-1]
@@ -600,12 +559,6 @@ func (mq *MaintenanceQueue) CompleteTask(taskID string, error string) {
}
}
// isNonRetriableError returns true for errors that will never succeed on retry,
// such as when the volume doesn't exist on the source server.
func isNonRetriableError(errMsg string) bool {
return strings.Contains(errMsg, "not found")
}
// UpdateTaskProgress updates the progress of a running task
func (mq *MaintenanceQueue) UpdateTaskProgress(taskID string, progress float64) {
mq.mutex.Lock()
-47
View File
@@ -200,53 +200,6 @@ func (s *ConfigStore) SaveJobTypeConfig(config *plugin_pb.PersistedJobTypeConfig
return nil
}
// SaveJobTypeConfigIfNotExists atomically checks whether a config for the
// given job type already exists and only persists config when none is found.
// Returns true if the config was saved, false if a config already existed.
func (s *ConfigStore) SaveJobTypeConfigIfNotExists(config *plugin_pb.PersistedJobTypeConfig) (bool, error) {
if config == nil {
return false, fmt.Errorf("job type config is nil")
}
if config.JobType == "" {
return false, fmt.Errorf("job type config has empty job_type")
}
sanitizedJobType, err := sanitizeJobType(config.JobType)
if err != nil {
return false, err
}
config.JobType = sanitizedJobType
clone := proto.Clone(config).(*plugin_pb.PersistedJobTypeConfig)
s.mu.Lock()
defer s.mu.Unlock()
if !s.configured {
if _, exists := s.memConfigs[config.JobType]; exists {
return false, nil
}
s.memConfigs[config.JobType] = clone
return true, nil
}
pbPath := filepath.Join(s.baseDir, jobTypesDirName, config.JobType, configPBFileName)
if _, statErr := os.Stat(pbPath); statErr == nil {
return false, nil
}
jobTypeDir, err := s.ensureJobTypeDir(config.JobType)
if err != nil {
return false, err
}
jsonPath := filepath.Join(jobTypeDir, configJSONFileName)
if err := writeProtoFiles(clone, filepath.Join(jobTypeDir, configPBFileName), jsonPath); err != nil {
return false, fmt.Errorf("save job type config for %s: %w", config.JobType, err)
}
return true, nil
}
func (s *ConfigStore) LoadJobTypeConfig(jobType string) (*plugin_pb.PersistedJobTypeConfig, error) {
if _, err := sanitizeJobType(jobType); err != nil {
return nil, err
-75
View File
@@ -208,81 +208,6 @@ func TestConfigStoreMonitorStateRoundTrip(t *testing.T) {
}
}
func TestConfigStoreSaveJobTypeConfigIfNotExists(t *testing.T) {
t.Parallel()
t.Run("in-memory", func(t *testing.T) {
t.Parallel()
store, err := NewConfigStore("")
if err != nil {
t.Fatalf("NewConfigStore: %v", err)
}
testSaveJobTypeConfigIfNotExists(t, store)
})
t.Run("on-disk", func(t *testing.T) {
t.Parallel()
store, err := NewConfigStore(t.TempDir())
if err != nil {
t.Fatalf("NewConfigStore: %v", err)
}
testSaveJobTypeConfigIfNotExists(t, store)
})
}
func testSaveJobTypeConfigIfNotExists(t *testing.T, store *ConfigStore) {
t.Helper()
cfg := &plugin_pb.PersistedJobTypeConfig{
JobType: "admin_script",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true},
}
// First call should save.
saved, err := store.SaveJobTypeConfigIfNotExists(cfg)
if err != nil {
t.Fatalf("first SaveJobTypeConfigIfNotExists: %v", err)
}
if !saved {
t.Fatal("expected first call to save the config")
}
// Second call with same job type should not save.
saved, err = store.SaveJobTypeConfigIfNotExists(&plugin_pb.PersistedJobTypeConfig{
JobType: "admin_script",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: false},
})
if err != nil {
t.Fatalf("second SaveJobTypeConfigIfNotExists: %v", err)
}
if saved {
t.Fatal("expected second call to be a no-op")
}
// Verify the original config was preserved.
loaded, err := store.LoadJobTypeConfig("admin_script")
if err != nil {
t.Fatalf("LoadJobTypeConfig: %v", err)
}
if loaded == nil {
t.Fatal("expected config to exist")
}
if !loaded.AdminRuntime.Enabled {
t.Fatal("expected original config (Enabled=true) to be preserved")
}
// Different job type should still save.
saved, err = store.SaveJobTypeConfigIfNotExists(&plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
})
if err != nil {
t.Fatalf("SaveJobTypeConfigIfNotExists for different type: %v", err)
}
if !saved {
t.Fatal("expected save for a different job type")
}
}
func TestConfigStoreJobDetailRoundTrip(t *testing.T) {
t.Parallel()
+24 -43
View File
@@ -32,13 +32,9 @@ type Options struct {
OutgoingBufferSize int
SendTimeout time.Duration
SchedulerTick time.Duration
IdleSleepDuration time.Duration
ClusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error)
LockManager LockManager
// ConfigDefaultsProvider is an optional callback invoked when a job type's
// config is being bootstrapped from its descriptor defaults. It can enrich
// or replace the default config before it is persisted. If nil, descriptor
// defaults are used as-is.
ConfigDefaultsProvider func(config *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig
}
// JobTypeInfo contains metadata about a plugin job type.
@@ -58,13 +54,16 @@ type Plugin struct {
sendTimeout time.Duration
schedulerTick time.Duration
idleSleepDuration time.Duration
clusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error)
configDefaultsProvider func(config *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig
lockManager LockManager
schedulerMu sync.Mutex
nextDetectionAt map[string]time.Time
detectionInFlight map[string]bool
schedulerMu sync.Mutex
schedulerPhase string
currentJobType string
iterationStartedAt time.Time
lastIterationEndedAt time.Time
lastIterationWorkDetected bool
detectorLeaseMu sync.Mutex
detectorLeases map[string]string
@@ -74,11 +73,6 @@ type Plugin struct {
adminScriptRunMu sync.RWMutex
schedulerDetectionMu sync.Mutex
schedulerDetection map[string]*schedulerDetectionInfo
schedulerRunMu sync.Mutex
schedulerRun map[string]*schedulerRunInfo
schedulerLoopMu sync.Mutex
schedulerLoopState schedulerLoopState
schedulerWakeCh chan struct{}
dedupeMu sync.Mutex
recentDedupeByType map[string]map[string]time.Time
@@ -157,6 +151,10 @@ func New(options Options) (*Plugin, error) {
if schedulerTick <= 0 {
schedulerTick = defaultSchedulerTick
}
idleSleepDuration := options.IdleSleepDuration
if idleSleepDuration <= 0 {
idleSleepDuration = defaultIdleSleepDuration
}
plugin := &Plugin{
store: store,
@@ -164,24 +162,21 @@ func New(options Options) (*Plugin, error) {
outgoingBuffer: bufferSize,
sendTimeout: sendTimeout,
schedulerTick: schedulerTick,
idleSleepDuration: idleSleepDuration,
clusterContextProvider: options.ClusterContextProvider,
configDefaultsProvider: options.ConfigDefaultsProvider,
lockManager: options.LockManager,
schedulerPhase: "idle",
sessions: make(map[string]*streamSession),
pendingSchema: make(map[string]chan *plugin_pb.ConfigSchemaResponse),
pendingDetection: make(map[string]*pendingDetectionState),
pendingExecution: make(map[string]chan *plugin_pb.JobCompleted),
nextDetectionAt: make(map[string]time.Time),
detectionInFlight: make(map[string]bool),
detectorLeases: make(map[string]string),
schedulerExecReservations: make(map[string]int),
schedulerDetection: make(map[string]*schedulerDetectionInfo),
schedulerRun: make(map[string]*schedulerRunInfo),
recentDedupeByType: make(map[string]map[string]time.Time),
jobs: make(map[string]*TrackedJob),
activities: make([]JobActivity, 0, 256),
persistTicker: time.NewTicker(2 * time.Second),
schedulerWakeCh: make(chan struct{}, 1),
shutdownCh: make(chan struct{}),
}
plugin.ctx, plugin.ctxCancel = context.WithCancel(context.Background())
@@ -385,14 +380,9 @@ func (r *Plugin) LoadJobTypeConfig(jobType string) (*plugin_pb.PersistedJobTypeC
}
func (r *Plugin) SaveJobTypeConfig(config *plugin_pb.PersistedJobTypeConfig) error {
if err := r.store.SaveJobTypeConfig(config); err != nil {
return err
}
r.wakeScheduler()
return nil
return r.store.SaveJobTypeConfig(config)
}
func (r *Plugin) LoadDescriptor(jobType string) (*plugin_pb.JobTypeDescriptor, error) {
return r.store.LoadDescriptor(jobType)
}
@@ -931,7 +921,6 @@ func (r *Plugin) handleWorkerMessage(workerID string, message *plugin_pb.WorkerT
switch body := message.Body.(type) {
case *plugin_pb.WorkerToAdminMessage_Hello:
r.registry.UpsertFromHello(body.Hello)
r.wakeScheduler()
case *plugin_pb.WorkerToAdminMessage_Heartbeat:
r.registry.UpdateHeartbeat(workerID, body.Heartbeat)
case *plugin_pb.WorkerToAdminMessage_ConfigSchemaResponse:
@@ -1001,6 +990,14 @@ func (r *Plugin) ensureJobTypeConfigFromDescriptor(jobType string, descriptor *p
return nil
}
existing, err := r.store.LoadJobTypeConfig(jobType)
if err != nil {
return err
}
if existing != nil {
return nil
}
workerDefaults := CloneConfigValueMap(descriptor.WorkerDefaultValues)
if len(workerDefaults) == 0 && descriptor.WorkerConfigForm != nil {
workerDefaults = CloneConfigValueMap(descriptor.WorkerConfigForm.DefaultValues)
@@ -1023,7 +1020,6 @@ func (r *Plugin) ensureJobTypeConfigFromDescriptor(jobType string, descriptor *p
PerWorkerExecutionConcurrency: defaults.PerWorkerExecutionConcurrency,
RetryLimit: defaults.RetryLimit,
RetryBackoffSeconds: defaults.RetryBackoffSeconds,
JobTypeMaxRuntimeSeconds: defaults.JobTypeMaxRuntimeSeconds,
}
}
@@ -1037,22 +1033,7 @@ func (r *Plugin) ensureJobTypeConfigFromDescriptor(jobType string, descriptor *p
UpdatedBy: "plugin",
}
// Check existence first to avoid calling configDefaultsProvider unnecessarily
// (e.g., it may make a blocking gRPC call to fetch master config).
existing, err := r.store.LoadJobTypeConfig(jobType)
if err != nil {
return err
}
if existing != nil {
return nil
}
if r.configDefaultsProvider != nil {
cfg = r.configDefaultsProvider(cfg)
}
_, err = r.store.SaveJobTypeConfigIfNotExists(cfg)
return err
return r.store.SaveJobTypeConfig(cfg)
}
func (r *Plugin) handleDetectionProposals(workerID string, message *plugin_pb.DetectionProposals) {
-73
View File
@@ -861,79 +861,6 @@ func (r *Plugin) trackExecutionQueued(job *plugin_pb.JobSpec) {
})
}
func (r *Plugin) cancelQueuedJob(job *plugin_pb.JobSpec, cause error) {
reason := "job canceled"
if cause != nil {
reason = cause.Error()
}
r.markJobCanceled(job, reason)
}
func (r *Plugin) markJobCanceled(job *plugin_pb.JobSpec, reason string) {
if job == nil || strings.TrimSpace(job.JobId) == "" {
return
}
now := time.Now().UTC()
if strings.TrimSpace(reason) == "" {
reason = "job canceled"
}
r.jobsMu.Lock()
tracked := r.jobs[job.JobId]
if tracked == nil {
tracked = &TrackedJob{
JobID: job.JobId,
CreatedAt: timeToPtr(now),
}
r.jobs[job.JobId] = tracked
}
if job.JobType != "" {
tracked.JobType = job.JobType
}
tracked.State = StateCanceled
tracked.Stage = "canceled"
tracked.Message = reason
tracked.ErrorMessage = reason
tracked.Progress = 0
if tracked.CreatedAt == nil || tracked.CreatedAt.IsZero() {
tracked.CreatedAt = timeToPtr(now)
}
tracked.UpdatedAt = timeToPtr(now)
tracked.CompletedAt = timeToPtr(now)
trackedSnapshot := cloneTrackedJob(*tracked)
r.pruneTrackedJobsLocked()
r.dirtyJobs = true
r.jobsMu.Unlock()
r.persistJobDetailSnapshot(job.JobId, func(detail *TrackedJob) {
detail.JobID = job.JobId
if job.JobType != "" {
detail.JobType = job.JobType
}
detail.State = trackedSnapshot.State
detail.Stage = trackedSnapshot.Stage
detail.Message = trackedSnapshot.Message
detail.ErrorMessage = trackedSnapshot.ErrorMessage
detail.Progress = trackedSnapshot.Progress
if detail.CreatedAt == nil || detail.CreatedAt.IsZero() {
detail.CreatedAt = trackedSnapshot.CreatedAt
}
detail.UpdatedAt = trackedSnapshot.UpdatedAt
detail.CompletedAt = trackedSnapshot.CompletedAt
})
r.appendActivity(JobActivity{
JobID: job.JobId,
JobType: job.JobType,
Source: "admin_scheduler",
Message: reason,
Stage: "canceled",
OccurredAt: timeToPtr(now),
})
}
func (r *Plugin) trackExecutionCompletion(completed *plugin_pb.JobCompleted) *TrackedJob {
if completed == nil || strings.TrimSpace(completed.JobId) == "" {
return nil
+185 -478
View File
@@ -13,17 +13,14 @@ import (
"google.golang.org/protobuf/types/known/timestamppb"
)
var (
errExecutorAtCapacity = errors.New("executor is at capacity")
errSchedulerShutdown = errors.New("scheduler shutdown")
)
var errExecutorAtCapacity = errors.New("executor is at capacity")
const (
defaultSchedulerTick = 5 * time.Second
defaultScheduledDetectionInterval = 300 * time.Second
defaultIdleSleepDuration = 17 * time.Minute
defaultMaxJobTypeDuration = 30 * time.Minute
defaultScheduledDetectionTimeout = 45 * time.Second
defaultScheduledExecutionTimeout = 90 * time.Second
defaultScheduledJobTypeMaxRuntime = 30 * time.Minute
defaultScheduledMaxResults int32 = 1000
defaultScheduledExecutionConcurrency = 1
defaultScheduledPerWorkerConcurrency = 1
@@ -32,14 +29,12 @@ const (
defaultClusterContextTimeout = 10 * time.Second
defaultWaitingBacklogFloor = 8
defaultWaitingBacklogMultiplier = 4
maxEstimatedRuntimeCap = 8 * time.Hour
)
type schedulerPolicy struct {
DetectionInterval time.Duration
DetectionTimeout time.Duration
ExecutionTimeout time.Duration
JobTypeMaxRuntime time.Duration
MaxJobTypeDuration time.Duration
RetryBackoff time.Duration
MaxResults int32
ExecutionConcurrency int
@@ -48,72 +43,62 @@ type schedulerPolicy struct {
ExecutorReserveBackoff time.Duration
}
func (r *Plugin) setSchedulerPhase(phase, jobType string) {
r.schedulerMu.Lock()
r.schedulerPhase = phase
r.currentJobType = jobType
r.schedulerMu.Unlock()
}
func (r *Plugin) schedulerLoop() {
defer r.wg.Done()
// Try once immediately on startup.
workDetected := r.runSchedulerIteration()
for {
if workDetected {
// Immediate re-iteration when work was found.
select {
case <-r.shutdownCh:
return
default:
}
} else {
r.setSchedulerPhase("idle", "")
if !waitForShutdownOrTimer(r.shutdownCh, r.idleSleepDuration) {
return
}
}
select {
case <-r.shutdownCh:
return
default:
}
hadJobs := r.runSchedulerIteration()
r.recordSchedulerIterationComplete(hadJobs)
if hadJobs {
continue
}
r.setSchedulerLoopState("", "sleeping")
idleSleep := defaultSchedulerIdleSleep
if nextRun := r.earliestNextDetectionAt(); !nextRun.IsZero() {
if until := time.Until(nextRun); until <= 0 {
idleSleep = 0
} else if until < idleSleep {
idleSleep = until
}
}
if idleSleep <= 0 {
continue
}
timer := time.NewTimer(idleSleep)
select {
case <-r.shutdownCh:
timer.Stop()
return
case <-r.schedulerWakeCh:
if !timer.Stop() {
<-timer.C
}
continue
case <-timer.C:
}
workDetected = r.runSchedulerIteration()
}
}
func (r *Plugin) runSchedulerIteration() bool {
r.expireStaleJobs(time.Now().UTC())
now := time.Now().UTC()
r.schedulerMu.Lock()
r.iterationStartedAt = now
r.schedulerMu.Unlock()
r.expireStaleJobs(now)
jobTypes := r.registry.DetectableJobTypes()
if len(jobTypes) == 0 {
r.setSchedulerLoopState("", "idle")
r.finishIteration(false)
return false
}
r.setSchedulerLoopState("", "waiting_for_lock")
releaseLock, err := r.acquireAdminLock("plugin scheduler iteration")
if err != nil {
glog.Warningf("Plugin scheduler failed to acquire lock: %v", err)
r.setSchedulerLoopState("", "idle")
return false
}
if releaseLock != nil {
defer releaseLock()
}
active := make(map[string]struct{}, len(jobTypes))
hadJobs := false
enabledJobTypes := make([]string, 0, len(jobTypes))
policies := make(map[string]schedulerPolicy, len(jobTypes))
for _, jobType := range jobTypes {
active[jobType] = struct{}{}
@@ -127,42 +112,90 @@ func (r *Plugin) runSchedulerIteration() bool {
r.clearSchedulerJobType(jobType)
continue
}
initialDelay := time.Duration(0)
if runInfo := r.snapshotSchedulerRun(jobType); runInfo.lastRunStartedAt.IsZero() {
initialDelay = 5 * time.Second
}
if !r.markDetectionDue(jobType, policy.DetectionInterval, initialDelay) {
continue
}
detected := r.runJobTypeIteration(jobType, policy)
if detected {
hadJobs = true
}
enabledJobTypes = append(enabledJobTypes, jobType)
policies[jobType] = policy
}
r.pruneSchedulerState(active)
r.pruneDetectorLeases(active)
r.setSchedulerLoopState("", "idle")
return hadJobs
if len(enabledJobTypes) == 0 {
r.finishIteration(false)
return false
}
// Acquire the lock ONCE for the entire iteration.
r.setSchedulerPhase("acquiring_lock", "")
releaseLock, lockErr := r.acquireAdminLock("plugin scheduler iteration")
if lockErr != nil {
glog.Warningf("Plugin scheduler failed to acquire lock: %v", lockErr)
r.appendActivity(JobActivity{
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduler iteration aborted: failed to acquire lock: %v", lockErr),
Stage: "failed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.finishIteration(false)
return false
}
defer releaseLock()
// Load cluster context ONCE for all job types.
clusterContext, err := r.loadSchedulerClusterContext()
if err != nil {
r.appendActivity(JobActivity{
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduler iteration aborted: %v", err),
Stage: "failed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.finishIteration(false)
return false
}
// Process each job type sequentially.
anyWorkDetected := false
for _, jobType := range enabledJobTypes {
select {
case <-r.shutdownCh:
r.finishIteration(anyWorkDetected)
return anyWorkDetected
default:
}
policy := policies[jobType]
r.setSchedulerPhase("processing", jobType)
if r.runJobTypeIteration(jobType, policy, clusterContext) {
anyWorkDetected = true
}
}
r.finishIteration(anyWorkDetected)
return anyWorkDetected
}
func (r *Plugin) wakeScheduler() {
if r == nil {
return
}
select {
case r.schedulerWakeCh <- struct{}{}:
default:
}
func (r *Plugin) finishIteration(workDetected bool) {
now := time.Now().UTC()
r.schedulerMu.Lock()
r.lastIterationEndedAt = now
r.lastIterationWorkDetected = workDetected
r.currentJobType = ""
r.schedulerPhase = "idle"
r.schedulerMu.Unlock()
}
func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) bool {
r.recordSchedulerRunStart(jobType)
r.clearWaitingJobQueue(jobType)
r.setSchedulerLoopState(jobType, "detecting")
r.markJobTypeInFlight(jobType)
defer r.finishDetection(jobType)
func (r *Plugin) runJobTypeIteration(
jobType string,
policy schedulerPolicy,
clusterContext *plugin_pb.ClusterContext,
) bool {
budget := policy.MaxJobTypeDuration
if budget <= 0 {
budget = defaultMaxJobTypeDuration
}
ctx, cancel := context.WithTimeout(r.ctx, budget)
defer cancel()
start := time.Now().UTC()
r.appendActivity(JobActivity{
@@ -182,70 +215,25 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
Stage: "skipped_waiting_backlog",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, "skipped")
return false
}
maxRuntime := policy.JobTypeMaxRuntime
if maxRuntime <= 0 {
maxRuntime = defaultScheduledJobTypeMaxRuntime
}
jobCtx, cancel := context.WithTimeout(context.Background(), maxRuntime)
defer cancel()
clusterContext, err := r.loadSchedulerClusterContext(jobCtx)
if err != nil {
r.recordSchedulerDetectionError(jobType, err)
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduled detection aborted: %v", err),
Stage: "failed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, "error")
return false
}
detectionTimeout := policy.DetectionTimeout
remaining := time.Until(start.Add(maxRuntime))
if remaining <= 0 {
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: "scheduled run timed out before detection",
Stage: "timeout",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, "timeout")
return false
}
if detectionTimeout <= 0 {
detectionTimeout = defaultScheduledDetectionTimeout
}
if detectionTimeout > remaining {
detectionTimeout = remaining
}
detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout)
proposals, err := r.RunDetection(detectCtx, jobType, clusterContext, policy.MaxResults)
cancelDetect()
detCtx, detCancel := context.WithTimeout(ctx, detectionTimeout)
proposals, err := r.RunDetection(detCtx, jobType, clusterContext, policy.MaxResults)
detCancel()
if err != nil {
r.recordSchedulerDetectionError(jobType, err)
stage := "failed"
status := "error"
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
stage = "timeout"
status = "timeout"
}
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduled detection failed: %v", err),
Stage: stage,
Stage: "failed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, status)
return false
}
@@ -258,8 +246,6 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
})
r.recordSchedulerDetectionSuccess(jobType, len(proposals))
detected := len(proposals) > 0
filteredByActive, skippedActive := r.filterProposalsWithActiveJobs(jobType, proposals)
if skippedActive > 0 {
r.appendActivity(JobActivity{
@@ -272,8 +258,7 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
}
if len(filteredByActive) == 0 {
r.recordSchedulerRunComplete(jobType, "success")
return detected
return false
}
filtered := r.filterScheduledProposals(filteredByActive)
@@ -288,82 +273,11 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
}
if len(filtered) == 0 {
r.recordSchedulerRunComplete(jobType, "success")
return detected
return false
}
r.setSchedulerLoopState(jobType, "executing")
// Scan proposals for the maximum estimated_runtime_seconds so the
// execution phase gets enough time for large jobs (e.g. vacuum on
// big volumes). If any proposal needs more time than the remaining
// JobTypeMaxRuntime, extend the execution context accordingly.
var maxEstimatedRuntime time.Duration
for _, p := range filtered {
if p.Parameters != nil {
if est, ok := p.Parameters["estimated_runtime_seconds"]; ok {
if v := est.GetInt64Value(); v > 0 {
if d := time.Duration(v) * time.Second; d > maxEstimatedRuntime {
maxEstimatedRuntime = d
}
}
}
}
}
if maxEstimatedRuntime > maxEstimatedRuntimeCap {
maxEstimatedRuntime = maxEstimatedRuntimeCap
}
remaining = time.Until(start.Add(maxRuntime))
if remaining <= 0 {
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: "scheduled execution skipped: job type max runtime reached",
Stage: "timeout",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, "timeout")
return detected
}
// If the longest estimated job exceeds the remaining JobTypeMaxRuntime,
// create a new execution context with enough headroom instead of using
// jobCtx which would cancel too early.
execCtx := jobCtx
execCancel := context.CancelFunc(func() {})
if maxEstimatedRuntime > 0 && maxEstimatedRuntime > remaining {
execCtx, execCancel = context.WithTimeout(context.Background(), maxEstimatedRuntime)
remaining = maxEstimatedRuntime
}
defer execCancel()
execPolicy := policy
if execPolicy.ExecutionTimeout <= 0 {
execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
}
if execPolicy.ExecutionTimeout > remaining {
execPolicy.ExecutionTimeout = remaining
}
successCount, errorCount, canceledCount := r.dispatchScheduledProposals(execCtx, jobType, filtered, clusterContext, execPolicy)
status := "success"
if execCtx.Err() != nil {
status = "timeout"
} else if errorCount > 0 || canceledCount > 0 {
status = "error"
}
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduled execution finished: success=%d error=%d canceled=%d", successCount, errorCount, canceledCount),
Stage: "executed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
r.recordSchedulerRunComplete(jobType, status)
return detected
r.dispatchScheduledProposals(ctx, jobType, filtered, clusterContext, policy)
return true
}
func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, error) {
@@ -385,10 +299,9 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
}
policy := schedulerPolicy{
DetectionInterval: durationFromSeconds(adminRuntime.DetectionIntervalSeconds, defaultScheduledDetectionInterval),
DetectionTimeout: durationFromSeconds(adminRuntime.DetectionTimeoutSeconds, defaultScheduledDetectionTimeout),
ExecutionTimeout: defaultScheduledExecutionTimeout,
JobTypeMaxRuntime: durationFromSeconds(adminRuntime.JobTypeMaxRuntimeSeconds, defaultScheduledJobTypeMaxRuntime),
MaxJobTypeDuration: defaultMaxJobTypeDuration,
RetryBackoff: durationFromSeconds(adminRuntime.RetryBackoffSeconds, defaultScheduledRetryBackoff),
MaxResults: adminRuntime.MaxJobsPerDetection,
ExecutionConcurrency: int(adminRuntime.GlobalExecutionConcurrency),
@@ -397,9 +310,6 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
ExecutorReserveBackoff: 200 * time.Millisecond,
}
if policy.DetectionInterval < r.schedulerTick {
policy.DetectionInterval = r.schedulerTick
}
if policy.MaxResults <= 0 {
policy.MaxResults = defaultScheduledMaxResults
}
@@ -418,9 +328,6 @@ func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, err
if policy.RetryLimit < 0 {
policy.RetryLimit = 0
}
if policy.JobTypeMaxRuntime <= 0 {
policy.JobTypeMaxRuntime = defaultScheduledJobTypeMaxRuntime
}
// Plugin protocol currently has only detection timeout in admin settings.
execTimeout := time.Duration(adminRuntime.DetectionTimeoutSeconds*2) * time.Second
@@ -439,28 +346,18 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
}
r.schedulerMu.Lock()
nextDetectionAt := make(map[string]time.Time, len(r.nextDetectionAt))
for jobType, nextRun := range r.nextDetectionAt {
nextDetectionAt[jobType] = nextRun
}
detectionInFlight := make(map[string]bool, len(r.detectionInFlight))
for jobType, inFlight := range r.detectionInFlight {
detectionInFlight[jobType] = inFlight
}
currentJobType := r.currentJobType
r.schedulerMu.Unlock()
states := make([]SchedulerJobTypeState, 0, len(jobTypes))
for _, jobTypeInfo := range jobTypes {
jobType := jobTypeInfo.JobType
state := SchedulerJobTypeState{
JobType: jobType,
DetectionInFlight: detectionInFlight[jobType],
JobType: jobType,
}
if nextRun, ok := nextDetectionAt[jobType]; ok && !nextRun.IsZero() {
nextRunUTC := nextRun.UTC()
state.NextDetectionAt = &nextRunUTC
}
// Mark as in-flight if this job type is being processed right now.
state.DetectionInFlight = jobType == currentJobType && currentJobType != ""
policy, enabled, loadErr := r.loadSchedulerPolicy(jobType)
@@ -469,10 +366,9 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
} else {
state.Enabled = enabled
if enabled {
state.DetectionIntervalSeconds = secondsFromDuration(policy.DetectionInterval)
state.DetectionTimeoutSeconds = secondsFromDuration(policy.DetectionTimeout)
state.ExecutionTimeoutSeconds = secondsFromDuration(policy.ExecutionTimeout)
state.JobTypeMaxRuntimeSeconds = secondsFromDuration(policy.JobTypeMaxRuntime)
state.MaxJobTypeDurationSeconds = secondsFromDuration(policy.MaxJobTypeDuration)
state.MaxJobsPerDetection = policy.MaxResults
state.GlobalExecutionConcurrency = policy.ExecutionConcurrency
state.PerWorkerExecutionConcurrency = policy.PerWorkerConcurrency
@@ -481,19 +377,6 @@ func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
}
}
runInfo := r.snapshotSchedulerRun(jobType)
if !runInfo.lastRunStartedAt.IsZero() {
at := runInfo.lastRunStartedAt
state.LastRunStartedAt = &at
}
if !runInfo.lastRunCompletedAt.IsZero() {
at := runInfo.lastRunCompletedAt
state.LastRunCompletedAt = &at
}
if runInfo.lastRunStatus != "" {
state.LastRunStatus = runInfo.lastRunStatus
}
leasedWorkerID := r.getDetectorLease(jobType)
if leasedWorkerID != "" {
state.DetectorWorkerID = leasedWorkerID
@@ -545,84 +428,10 @@ func deriveSchedulerAdminRuntime(
PerWorkerExecutionConcurrency: defaults.PerWorkerExecutionConcurrency,
RetryLimit: defaults.RetryLimit,
RetryBackoffSeconds: defaults.RetryBackoffSeconds,
JobTypeMaxRuntimeSeconds: defaults.JobTypeMaxRuntimeSeconds,
}
}
func (r *Plugin) markDetectionDue(jobType string, interval, initialDelay time.Duration) bool {
now := time.Now().UTC()
r.schedulerMu.Lock()
defer r.schedulerMu.Unlock()
if r.detectionInFlight[jobType] {
return false
}
nextRun, exists := r.nextDetectionAt[jobType]
if exists && now.Before(nextRun) {
return false
}
if !exists && initialDelay > 0 {
r.nextDetectionAt[jobType] = now.Add(initialDelay)
return false
}
r.nextDetectionAt[jobType] = now.Add(interval)
r.detectionInFlight[jobType] = true
return true
}
func (r *Plugin) earliestNextDetectionAt() time.Time {
if r == nil {
return time.Time{}
}
r.schedulerMu.Lock()
defer r.schedulerMu.Unlock()
var earliest time.Time
for _, nextRun := range r.nextDetectionAt {
if nextRun.IsZero() {
continue
}
if earliest.IsZero() || nextRun.Before(earliest) {
earliest = nextRun
}
}
return earliest
}
func (r *Plugin) markJobTypeInFlight(jobType string) {
r.schedulerMu.Lock()
r.detectionInFlight[jobType] = true
r.schedulerMu.Unlock()
}
func (r *Plugin) finishDetection(jobType string) {
r.schedulerMu.Lock()
delete(r.detectionInFlight, jobType)
r.schedulerMu.Unlock()
}
func (r *Plugin) pruneSchedulerState(activeJobTypes map[string]struct{}) {
r.schedulerMu.Lock()
defer r.schedulerMu.Unlock()
for jobType := range r.nextDetectionAt {
if _, ok := activeJobTypes[jobType]; !ok {
delete(r.nextDetectionAt, jobType)
delete(r.detectionInFlight, jobType)
}
}
}
func (r *Plugin) clearSchedulerJobType(jobType string) {
r.schedulerMu.Lock()
delete(r.nextDetectionAt, jobType)
delete(r.detectionInFlight, jobType)
r.schedulerMu.Unlock()
r.clearDetectorLease(jobType, "")
}
@@ -637,18 +446,15 @@ func (r *Plugin) pruneDetectorLeases(activeJobTypes map[string]struct{}) {
}
}
func (r *Plugin) loadSchedulerClusterContext(ctx context.Context) (*plugin_pb.ClusterContext, error) {
func (r *Plugin) loadSchedulerClusterContext() (*plugin_pb.ClusterContext, error) {
if r.clusterContextProvider == nil {
return nil, fmt.Errorf("cluster context provider is not configured")
}
if ctx == nil {
ctx = context.Background()
}
clusterCtx, cancel := context.WithTimeout(ctx, defaultClusterContextTimeout)
ctx, cancel := context.WithTimeout(context.Background(), defaultClusterContextTimeout)
defer cancel()
clusterContext, err := r.clusterContextProvider(clusterCtx)
clusterContext, err := r.clusterContextProvider(ctx)
if err != nil {
return nil, err
}
@@ -659,16 +465,12 @@ func (r *Plugin) loadSchedulerClusterContext(ctx context.Context) (*plugin_pb.Cl
}
func (r *Plugin) dispatchScheduledProposals(
ctx context.Context,
parentCtx context.Context,
jobType string,
proposals []*plugin_pb.JobProposal,
clusterContext *plugin_pb.ClusterContext,
policy schedulerPolicy,
) (int, int, int) {
if ctx == nil {
ctx = context.Background()
}
) {
jobQueue := make(chan *plugin_pb.JobSpec, len(proposals))
for index, proposal := range proposals {
job := buildScheduledJobSpec(jobType, proposal, index)
@@ -676,7 +478,10 @@ func (r *Plugin) dispatchScheduledProposals(
select {
case <-r.shutdownCh:
close(jobQueue)
return 0, 0, 0
return
case <-parentCtx.Done():
close(jobQueue)
return
default:
jobQueue <- job
}
@@ -687,7 +492,6 @@ func (r *Plugin) dispatchScheduledProposals(
var statsMu sync.Mutex
successCount := 0
errorCount := 0
canceledCount := 0
workerCount := policy.ExecutionConcurrency
if workerCount < 1 {
@@ -699,44 +503,30 @@ func (r *Plugin) dispatchScheduledProposals(
go func() {
defer wg.Done()
jobLoop:
for job := range jobQueue {
select {
case <-r.shutdownCh:
return
case <-parentCtx.Done():
return
default:
}
if ctx.Err() != nil {
r.cancelQueuedJob(job, ctx.Err())
statsMu.Lock()
canceledCount++
statsMu.Unlock()
continue
}
for {
select {
case <-r.shutdownCh:
return
case <-parentCtx.Done():
return
default:
}
if ctx.Err() != nil {
r.cancelQueuedJob(job, ctx.Err())
statsMu.Lock()
canceledCount++
statsMu.Unlock()
continue jobLoop
}
executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy)
executor, release, reserveErr := r.reserveScheduledExecutor(parentCtx, jobType, policy)
if reserveErr != nil {
if ctx.Err() != nil {
r.cancelQueuedJob(job, ctx.Err())
statsMu.Lock()
canceledCount++
statsMu.Unlock()
continue jobLoop
select {
case <-r.shutdownCh:
return
default:
}
statsMu.Lock()
errorCount++
@@ -751,23 +541,16 @@ func (r *Plugin) dispatchScheduledProposals(
break
}
err := r.executeScheduledJobWithExecutor(ctx, executor, job, clusterContext, policy)
err := r.executeScheduledJobWithExecutor(parentCtx, executor, job, clusterContext, policy)
release()
if errors.Is(err, errExecutorAtCapacity) {
r.trackExecutionQueued(job)
if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
if !waitForShutdownOrTimer(r.shutdownCh, policy.ExecutorReserveBackoff) {
return
}
continue
}
if err != nil {
if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
r.cancelQueuedJob(job, err)
statsMu.Lock()
canceledCount++
statsMu.Unlock()
continue jobLoop
}
statsMu.Lock()
errorCount++
statsMu.Unlock()
@@ -793,16 +576,13 @@ func (r *Plugin) dispatchScheduledProposals(
wg.Wait()
drainErr := ctx.Err()
if drainErr == nil {
drainErr = errSchedulerShutdown
}
for job := range jobQueue {
r.cancelQueuedJob(job, drainErr)
canceledCount++
}
return successCount, errorCount, canceledCount
r.appendActivity(JobActivity{
JobType: jobType,
Source: "admin_scheduler",
Message: fmt.Sprintf("scheduled execution finished: success=%d error=%d", successCount, errorCount),
Stage: "executed",
OccurredAt: timeToPtr(time.Now().UTC()),
})
}
func (r *Plugin) reserveScheduledExecutor(
@@ -810,26 +590,18 @@ func (r *Plugin) reserveScheduledExecutor(
jobType string,
policy schedulerPolicy,
) (*WorkerSession, func(), error) {
if ctx == nil {
ctx = context.Background()
}
deadline := time.Now().Add(policy.ExecutionTimeout)
if policy.ExecutionTimeout <= 0 {
deadline = time.Now().Add(10 * time.Minute) // Default cap
}
if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
deadline = ctxDeadline
}
for {
select {
case <-r.shutdownCh:
return nil, nil, fmt.Errorf("plugin is shutting down")
default:
}
if ctx.Err() != nil {
case <-ctx.Done():
return nil, nil, ctx.Err()
default:
}
if time.Now().After(deadline) {
@@ -838,11 +610,8 @@ func (r *Plugin) reserveScheduledExecutor(
executors, err := r.registry.ListExecutors(jobType)
if err != nil {
if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
if ctx.Err() != nil {
return nil, nil, ctx.Err()
}
return nil, nil, fmt.Errorf("plugin is shutting down")
if !waitForShutdownOrCtx(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
return nil, nil, fmt.Errorf("plugin is shutting down or context canceled")
}
continue
}
@@ -855,11 +624,8 @@ func (r *Plugin) reserveScheduledExecutor(
return executor, release, nil
}
if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
if ctx.Err() != nil {
return nil, nil, ctx.Err()
}
return nil, nil, fmt.Errorf("plugin is shutting down")
if !waitForShutdownOrCtx(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
return nil, nil, fmt.Errorf("plugin is shutting down or context canceled")
}
}
}
@@ -943,7 +709,7 @@ func schedulerWorkerExecutionLimit(executor *WorkerSession, jobType string, poli
}
func (r *Plugin) executeScheduledJobWithExecutor(
ctx context.Context,
parentCtx context.Context,
executor *WorkerSession,
job *plugin_pb.JobSpec,
clusterContext *plugin_pb.ClusterContext,
@@ -959,34 +725,12 @@ func (r *Plugin) executeScheduledJobWithExecutor(
select {
case <-r.shutdownCh:
return fmt.Errorf("plugin is shutting down")
case <-parentCtx.Done():
return parentCtx.Err()
default:
}
if ctx != nil && ctx.Err() != nil {
return ctx.Err()
}
parent := ctx
if parent == nil {
parent = context.Background()
}
// Use the job's estimated runtime if provided and larger than the
// default execution timeout. This lets handlers like vacuum scale
// the timeout based on volume size so large volumes are not killed.
timeout := policy.ExecutionTimeout
if job.Parameters != nil {
if est, ok := job.Parameters["estimated_runtime_seconds"]; ok {
if v := est.GetInt64Value(); v > 0 {
estimated := time.Duration(v) * time.Second
if estimated > maxEstimatedRuntimeCap {
estimated = maxEstimatedRuntimeCap
}
if estimated > timeout {
timeout = estimated
}
}
}
}
execCtx, cancel := context.WithTimeout(parent, timeout)
execCtx, cancel := context.WithTimeout(parentCtx, policy.ExecutionTimeout)
_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
cancel()
if err == nil {
@@ -1006,11 +750,8 @@ func (r *Plugin) executeScheduledJobWithExecutor(
Stage: "retry",
OccurredAt: timeToPtr(time.Now().UTC()),
})
if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.RetryBackoff) {
if ctx != nil && ctx.Err() != nil {
return ctx.Err()
}
return fmt.Errorf("plugin is shutting down")
if !waitForShutdownOrCtx(r.shutdownCh, parentCtx, policy.RetryBackoff) {
return fmt.Errorf("plugin is shutting down or context canceled")
}
}
}
@@ -1055,53 +796,6 @@ func (r *Plugin) countWaitingTrackedJobs(jobType string) int {
return waiting
}
func (r *Plugin) clearWaitingJobQueue(jobType string) int {
normalizedJobType := strings.TrimSpace(jobType)
if normalizedJobType == "" {
return 0
}
jobIDs := make([]string, 0)
seen := make(map[string]struct{})
r.jobsMu.RLock()
for _, job := range r.jobs {
if job == nil {
continue
}
if strings.TrimSpace(job.JobType) != normalizedJobType {
continue
}
if !isWaitingTrackedJobState(job.State) {
continue
}
jobID := strings.TrimSpace(job.JobID)
if jobID == "" {
continue
}
if _, ok := seen[jobID]; ok {
continue
}
seen[jobID] = struct{}{}
jobIDs = append(jobIDs, jobID)
}
r.jobsMu.RUnlock()
if len(jobIDs) == 0 {
return 0
}
reason := fmt.Sprintf("cleared queued job before %s run", normalizedJobType)
for _, jobID := range jobIDs {
r.markJobCanceled(&plugin_pb.JobSpec{
JobId: jobID,
JobType: normalizedJobType,
}, reason)
}
return len(jobIDs)
}
func waitingBacklogThreshold(policy schedulerPolicy) int {
concurrency := policy.ExecutionConcurrency
if concurrency <= 0 {
@@ -1183,12 +877,25 @@ func secondsFromDuration(duration time.Duration) int32 {
return int32(duration / time.Second)
}
func waitForShutdownOrTimerWithContext(shutdown <-chan struct{}, ctx context.Context, duration time.Duration) bool {
func waitForShutdownOrTimer(shutdown <-chan struct{}, duration time.Duration) bool {
if duration <= 0 {
return true
}
if ctx == nil {
ctx = context.Background()
timer := time.NewTimer(duration)
defer timer.Stop()
select {
case <-shutdown:
return false
case <-timer.C:
return true
}
}
func waitForShutdownOrCtx(shutdown <-chan struct{}, ctx context.Context, duration time.Duration) bool {
if duration <= 0 {
return true
}
timer := time.NewTimer(duration)
+337 -23
View File
@@ -3,6 +3,7 @@ package plugin
import (
"context"
"fmt"
"sync"
"testing"
"time"
@@ -29,7 +30,6 @@ func TestLoadSchedulerPolicyUsesAdminConfig(t *testing.T) {
PerWorkerExecutionConcurrency: 2,
RetryLimit: 4,
RetryBackoffSeconds: 7,
JobTypeMaxRuntimeSeconds: 1800,
},
})
if err != nil {
@@ -55,8 +55,8 @@ func TestLoadSchedulerPolicyUsesAdminConfig(t *testing.T) {
if policy.RetryLimit != 4 {
t.Fatalf("unexpected retry limit: got=%d", policy.RetryLimit)
}
if policy.JobTypeMaxRuntime != 30*time.Minute {
t.Fatalf("unexpected max runtime: got=%v", policy.JobTypeMaxRuntime)
if policy.MaxJobTypeDuration != defaultMaxJobTypeDuration {
t.Fatalf("unexpected max job type duration: got=%v", policy.MaxJobTypeDuration)
}
}
@@ -80,7 +80,6 @@ func TestLoadSchedulerPolicyUsesDescriptorDefaultsWhenConfigMissing(t *testing.T
PerWorkerExecutionConcurrency: 2,
RetryLimit: 3,
RetryBackoffSeconds: 6,
JobTypeMaxRuntimeSeconds: 1200,
},
})
if err != nil {
@@ -103,9 +102,6 @@ func TestLoadSchedulerPolicyUsesDescriptorDefaultsWhenConfigMissing(t *testing.T
if policy.PerWorkerConcurrency != 2 {
t.Fatalf("unexpected per-worker concurrency: got=%d", policy.PerWorkerConcurrency)
}
if policy.JobTypeMaxRuntime != 20*time.Minute {
t.Fatalf("unexpected max runtime: got=%v", policy.JobTypeMaxRuntime)
}
}
func TestReserveScheduledExecutorRespectsPerWorkerLimit(t *testing.T) {
@@ -403,7 +399,6 @@ func TestListSchedulerStatesIncludesPolicyAndState(t *testing.T) {
PerWorkerExecutionConcurrency: 2,
RetryLimit: 1,
RetryBackoffSeconds: 9,
JobTypeMaxRuntimeSeconds: 900,
},
})
if err != nil {
@@ -417,10 +412,9 @@ func TestListSchedulerStatesIncludesPolicyAndState(t *testing.T) {
},
})
nextDetectionAt := time.Now().UTC().Add(2 * time.Minute).Round(time.Second)
// Mark this job type as currently processing to test DetectionInFlight.
pluginSvc.schedulerMu.Lock()
pluginSvc.nextDetectionAt[jobType] = nextDetectionAt
pluginSvc.detectionInFlight[jobType] = true
pluginSvc.currentJobType = jobType
pluginSvc.schedulerMu.Unlock()
states, err := pluginSvc.ListSchedulerStates()
@@ -439,16 +433,7 @@ func TestListSchedulerStatesIncludesPolicyAndState(t *testing.T) {
t.Fatalf("unexpected policy error: %s", state.PolicyError)
}
if !state.DetectionInFlight {
t.Fatalf("expected detection in flight")
}
if state.NextDetectionAt == nil {
t.Fatalf("expected next detection time")
}
if state.NextDetectionAt.Unix() != nextDetectionAt.Unix() {
t.Fatalf("unexpected next detection time: got=%v want=%v", state.NextDetectionAt, nextDetectionAt)
}
if state.DetectionIntervalSeconds != 45 {
t.Fatalf("unexpected detection interval: got=%d", state.DetectionIntervalSeconds)
t.Fatalf("expected detection in flight when current job type matches")
}
if state.DetectionTimeoutSeconds != 30 {
t.Fatalf("unexpected detection timeout: got=%d", state.DetectionTimeoutSeconds)
@@ -456,8 +441,8 @@ func TestListSchedulerStatesIncludesPolicyAndState(t *testing.T) {
if state.ExecutionTimeoutSeconds != 90 {
t.Fatalf("unexpected execution timeout: got=%d", state.ExecutionTimeoutSeconds)
}
if state.JobTypeMaxRuntimeSeconds != 900 {
t.Fatalf("unexpected job type max runtime: got=%d", state.JobTypeMaxRuntimeSeconds)
if state.MaxJobTypeDurationSeconds != int32(defaultMaxJobTypeDuration/time.Second) {
t.Fatalf("unexpected max job type duration: got=%d", state.MaxJobTypeDurationSeconds)
}
if state.MaxJobsPerDetection != 80 {
t.Fatalf("unexpected max jobs per detection: got=%d", state.MaxJobsPerDetection)
@@ -480,6 +465,23 @@ func TestListSchedulerStatesIncludesPolicyAndState(t *testing.T) {
if state.ExecutorWorkerCount != 1 {
t.Fatalf("unexpected executor worker count: got=%d", state.ExecutorWorkerCount)
}
// Clear the current job type and verify DetectionInFlight is false.
pluginSvc.schedulerMu.Lock()
pluginSvc.currentJobType = ""
pluginSvc.schedulerMu.Unlock()
states2, err := pluginSvc.ListSchedulerStates()
if err != nil {
t.Fatalf("ListSchedulerStates (2): %v", err)
}
state2 := findSchedulerState(states2, jobType)
if state2 == nil {
t.Fatalf("missing scheduler state for %s (2)", jobType)
}
if state2.DetectionInFlight {
t.Fatalf("expected detection not in flight when current job type is empty")
}
}
func TestListSchedulerStatesShowsDisabledWhenNoPolicy(t *testing.T) {
@@ -594,3 +596,315 @@ func TestPickDetectorReassignsWhenLeaseIsStale(t *testing.T) {
t.Fatalf("expected detector lease to be updated to worker-a, got=%s", lease)
}
}
// mockLockManager records lock/release calls for testing.
type mockLockManager struct {
mu sync.Mutex
acquireCount int
releaseCount int
failAcquire bool
}
func (m *mockLockManager) Acquire(reason string) (func(), error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.failAcquire {
return nil, fmt.Errorf("mock lock acquisition failed")
}
m.acquireCount++
return func() {
m.mu.Lock()
m.releaseCount++
m.mu.Unlock()
}, nil
}
func (m *mockLockManager) Status() interface{} {
return nil
}
func (m *mockLockManager) getAcquireCount() int {
m.mu.Lock()
defer m.mu.Unlock()
return m.acquireCount
}
func (m *mockLockManager) getReleaseCount() int {
m.mu.Lock()
defer m.mu.Unlock()
return m.releaseCount
}
func TestIterationAcquiresLockOnce(t *testing.T) {
t.Parallel()
lock := &mockLockManager{}
pluginSvc, err := New(Options{
LockManager: lock,
})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
// Register two enabled job types.
for _, jt := range []string{"vacuum", "balance"} {
err = pluginSvc.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
JobType: jt,
AdminRuntime: &plugin_pb.AdminRuntimeConfig{
Enabled: true,
DetectionTimeoutSeconds: 5,
MaxJobsPerDetection: 10,
GlobalExecutionConcurrency: 1,
PerWorkerExecutionConcurrency: 1,
},
})
if err != nil {
t.Fatalf("SaveJobTypeConfig(%s): %v", jt, err)
}
pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
WorkerId: "worker-" + jt,
Capabilities: []*plugin_pb.JobTypeCapability{
{JobType: jt, CanDetect: true, CanExecute: true},
},
})
}
// runSchedulerIteration requires a cluster context provider.
pluginSvc.clusterContextProvider = func(_ context.Context) (*plugin_pb.ClusterContext, error) {
return &plugin_pb.ClusterContext{}, nil
}
pluginSvc.runSchedulerIteration()
// Lock should have been acquired exactly once (not per-job-type).
if lock.getAcquireCount() != 1 {
t.Fatalf("expected 1 lock acquisition, got %d", lock.getAcquireCount())
}
if lock.getReleaseCount() != 1 {
t.Fatalf("expected 1 lock release, got %d", lock.getReleaseCount())
}
}
func TestIterationReturnsfalseWhenLockFails(t *testing.T) {
t.Parallel()
lock := &mockLockManager{failAcquire: true}
pluginSvc, err := New(Options{
LockManager: lock,
})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
err = pluginSvc.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{
Enabled: true,
DetectionTimeoutSeconds: 5,
MaxJobsPerDetection: 10,
GlobalExecutionConcurrency: 1,
PerWorkerExecutionConcurrency: 1,
},
})
if err != nil {
t.Fatalf("SaveJobTypeConfig: %v", err)
}
pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
WorkerId: "worker-a",
Capabilities: []*plugin_pb.JobTypeCapability{
{JobType: "vacuum", CanDetect: true, CanExecute: true},
},
})
pluginSvc.clusterContextProvider = func(_ context.Context) (*plugin_pb.ClusterContext, error) {
return &plugin_pb.ClusterContext{}, nil
}
result := pluginSvc.runSchedulerIteration()
if result {
t.Fatalf("expected false when lock acquisition fails")
}
}
func TestSchedulerPhaseTracking(t *testing.T) {
t.Parallel()
pluginSvc, err := New(Options{})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
// Initial phase should be idle.
pluginSvc.schedulerMu.Lock()
phase := pluginSvc.schedulerPhase
pluginSvc.schedulerMu.Unlock()
if phase != "idle" {
t.Fatalf("expected initial phase to be idle, got=%s", phase)
}
pluginSvc.setSchedulerPhase("processing", "vacuum")
pluginSvc.schedulerMu.Lock()
phase = pluginSvc.schedulerPhase
jobType := pluginSvc.currentJobType
pluginSvc.schedulerMu.Unlock()
if phase != "processing" {
t.Fatalf("expected phase processing, got=%s", phase)
}
if jobType != "vacuum" {
t.Fatalf("expected current job type vacuum, got=%s", jobType)
}
pluginSvc.finishIteration(true)
pluginSvc.schedulerMu.Lock()
phase = pluginSvc.schedulerPhase
jobType = pluginSvc.currentJobType
workDetected := pluginSvc.lastIterationWorkDetected
lastEnded := pluginSvc.lastIterationEndedAt
pluginSvc.schedulerMu.Unlock()
if phase != "idle" {
t.Fatalf("expected phase idle after finish, got=%s", phase)
}
if jobType != "" {
t.Fatalf("expected empty job type after finish, got=%s", jobType)
}
if !workDetected {
t.Fatalf("expected last iteration work detected to be true")
}
if lastEnded.IsZero() {
t.Fatalf("expected last iteration ended at to be set")
}
}
func TestGetSchedulerStatusIncludesIterationFields(t *testing.T) {
t.Parallel()
pluginSvc, err := New(Options{
IdleSleepDuration: 10 * time.Minute,
})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
now := time.Now().UTC()
pluginSvc.schedulerMu.Lock()
pluginSvc.schedulerPhase = "processing"
pluginSvc.currentJobType = "vacuum"
pluginSvc.iterationStartedAt = now.Add(-5 * time.Second)
pluginSvc.lastIterationEndedAt = now.Add(-20 * time.Second)
pluginSvc.lastIterationWorkDetected = true
pluginSvc.schedulerMu.Unlock()
status := pluginSvc.GetSchedulerStatus()
if status.Phase != "processing" {
t.Fatalf("expected phase processing, got=%s", status.Phase)
}
if status.CurrentJobType != "vacuum" {
t.Fatalf("expected current job type vacuum, got=%s", status.CurrentJobType)
}
if status.IdleSleepSeconds != 600 {
t.Fatalf("expected idle sleep 600s, got=%d", status.IdleSleepSeconds)
}
if status.IterationStartedAt == nil {
t.Fatalf("expected iteration started at to be set")
}
if status.LastIterationEndedAt == nil {
t.Fatalf("expected last iteration ended at to be set")
}
if !status.LastIterationWorkDetected {
t.Fatalf("expected last iteration work detected to be true")
}
// SchedulerTickSeconds should match IdleSleepSeconds for backward compat.
if status.SchedulerTickSeconds != 600 {
t.Fatalf("expected scheduler tick seconds to match idle sleep, got=%d", status.SchedulerTickSeconds)
}
}
func TestGracefulShutdownDuringIteration(t *testing.T) {
t.Parallel()
pluginSvc, err := New(Options{
IdleSleepDuration: time.Millisecond,
ClusterContextProvider: func(_ context.Context) (*plugin_pb.ClusterContext, error) {
return &plugin_pb.ClusterContext{}, nil
},
})
if err != nil {
t.Fatalf("New: %v", err)
}
// Register an enabled job type so the scheduler loop has work to consider.
err = pluginSvc.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{
Enabled: true,
DetectionTimeoutSeconds: 5,
MaxJobsPerDetection: 10,
GlobalExecutionConcurrency: 1,
PerWorkerExecutionConcurrency: 1,
},
})
if err != nil {
t.Fatalf("SaveJobTypeConfig: %v", err)
}
pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{
WorkerId: "worker-a",
Capabilities: []*plugin_pb.JobTypeCapability{
{JobType: "vacuum", CanDetect: true, CanExecute: true},
},
})
// Shutdown while the scheduler loop is actively iterating.
done := make(chan struct{})
go func() {
pluginSvc.Shutdown()
close(done)
}()
select {
case <-done:
// Good — clean shutdown.
case <-time.After(5 * time.Second):
t.Fatalf("shutdown did not complete in time")
}
}
func TestIdleSleepDurationDefault(t *testing.T) {
t.Parallel()
pluginSvc, err := New(Options{})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
if pluginSvc.idleSleepDuration != defaultIdleSleepDuration {
t.Fatalf("expected default idle sleep %v, got=%v", defaultIdleSleepDuration, pluginSvc.idleSleepDuration)
}
}
func TestIdleSleepDurationCustom(t *testing.T) {
t.Parallel()
customDuration := 5 * time.Minute
pluginSvc, err := New(Options{
IdleSleepDuration: customDuration,
})
if err != nil {
t.Fatalf("New: %v", err)
}
defer pluginSvc.Shutdown()
if pluginSvc.idleSleepDuration != customDuration {
t.Fatalf("expected custom idle sleep %v, got=%v", customDuration, pluginSvc.idleSleepDuration)
}
}
-5
View File
@@ -1,5 +0,0 @@
package plugin
import "time"
const defaultSchedulerIdleSleep = 61 * time.Second
+50 -135
View File
@@ -7,17 +7,17 @@ import (
)
type SchedulerStatus struct {
Now time.Time `json:"now"`
SchedulerTickSeconds int `json:"scheduler_tick_seconds"`
IdleSleepSeconds int `json:"idle_sleep_seconds,omitempty"`
NextDetectionAt *time.Time `json:"next_detection_at,omitempty"`
CurrentJobType string `json:"current_job_type,omitempty"`
CurrentPhase string `json:"current_phase,omitempty"`
LastIterationHadJobs bool `json:"last_iteration_had_jobs,omitempty"`
LastIterationDoneAt *time.Time `json:"last_iteration_done_at,omitempty"`
Waiting []SchedulerWaitingStatus `json:"waiting,omitempty"`
InProcessJobs []SchedulerJobStatus `json:"in_process_jobs,omitempty"`
JobTypes []SchedulerJobTypeStatus `json:"job_types,omitempty"`
Now time.Time `json:"now"`
SchedulerTickSeconds int `json:"scheduler_tick_seconds"`
IdleSleepSeconds int `json:"idle_sleep_seconds"`
Phase string `json:"phase"`
CurrentJobType string `json:"current_job_type,omitempty"`
IterationStartedAt *time.Time `json:"iteration_started_at,omitempty"`
LastIterationEndedAt *time.Time `json:"last_iteration_ended_at,omitempty"`
LastIterationWorkDetected bool `json:"last_iteration_work_detected"`
Waiting []SchedulerWaitingStatus `json:"waiting,omitempty"`
InProcessJobs []SchedulerJobStatus `json:"in_process_jobs,omitempty"`
JobTypes []SchedulerJobTypeStatus `json:"job_types,omitempty"`
}
type SchedulerWaitingStatus struct {
@@ -42,15 +42,14 @@ type SchedulerJobStatus struct {
}
type SchedulerJobTypeStatus struct {
JobType string `json:"job_type"`
Enabled bool `json:"enabled"`
DetectionInFlight bool `json:"detection_in_flight"`
NextDetectionAt *time.Time `json:"next_detection_at,omitempty"`
DetectionIntervalSeconds int32 `json:"detection_interval_seconds,omitempty"`
LastDetectedAt *time.Time `json:"last_detected_at,omitempty"`
LastDetectedCount int `json:"last_detected_count,omitempty"`
LastDetectionError string `json:"last_detection_error,omitempty"`
LastDetectionSkipped string `json:"last_detection_skipped,omitempty"`
JobType string `json:"job_type"`
Enabled bool `json:"enabled"`
DetectionInFlight bool `json:"detection_in_flight"`
MaxJobTypeDurationSeconds int32 `json:"max_job_type_duration_seconds,omitempty"`
LastDetectedAt *time.Time `json:"last_detected_at,omitempty"`
LastDetectedCount int `json:"last_detected_count,omitempty"`
LastDetectionError string `json:"last_detection_error,omitempty"`
LastDetectionSkipped string `json:"last_detection_skipped,omitempty"`
}
type schedulerDetectionInfo struct {
@@ -62,19 +61,6 @@ type schedulerDetectionInfo struct {
lastSkippedReason string
}
type schedulerRunInfo struct {
lastRunStartedAt time.Time
lastRunCompletedAt time.Time
lastRunStatus string
}
type schedulerLoopState struct {
currentJobType string
currentPhase string
lastIterationHadJobs bool
lastIterationCompleted time.Time
}
func (r *Plugin) recordSchedulerDetectionSuccess(jobType string, count int) {
if r == nil {
return
@@ -141,101 +127,31 @@ func (r *Plugin) snapshotSchedulerDetection(jobType string) schedulerDetectionIn
return *info
}
func (r *Plugin) recordSchedulerRunStart(jobType string) {
if r == nil {
return
}
r.schedulerRunMu.Lock()
defer r.schedulerRunMu.Unlock()
info := r.schedulerRun[jobType]
if info == nil {
info = &schedulerRunInfo{}
r.schedulerRun[jobType] = info
}
info.lastRunStartedAt = time.Now().UTC()
info.lastRunStatus = ""
}
func (r *Plugin) recordSchedulerRunComplete(jobType, status string) {
if r == nil {
return
}
r.schedulerRunMu.Lock()
defer r.schedulerRunMu.Unlock()
info := r.schedulerRun[jobType]
if info == nil {
info = &schedulerRunInfo{}
r.schedulerRun[jobType] = info
}
info.lastRunCompletedAt = time.Now().UTC()
info.lastRunStatus = status
}
func (r *Plugin) snapshotSchedulerRun(jobType string) schedulerRunInfo {
if r == nil {
return schedulerRunInfo{}
}
r.schedulerRunMu.Lock()
defer r.schedulerRunMu.Unlock()
info := r.schedulerRun[jobType]
if info == nil {
return schedulerRunInfo{}
}
return *info
}
func (r *Plugin) setSchedulerLoopState(jobType, phase string) {
if r == nil {
return
}
r.schedulerLoopMu.Lock()
r.schedulerLoopState.currentJobType = jobType
r.schedulerLoopState.currentPhase = phase
r.schedulerLoopMu.Unlock()
}
func (r *Plugin) recordSchedulerIterationComplete(hadJobs bool) {
if r == nil {
return
}
r.schedulerLoopMu.Lock()
r.schedulerLoopState.lastIterationHadJobs = hadJobs
r.schedulerLoopState.lastIterationCompleted = time.Now().UTC()
r.schedulerLoopMu.Unlock()
}
func (r *Plugin) snapshotSchedulerLoopState() schedulerLoopState {
if r == nil {
return schedulerLoopState{}
}
r.schedulerLoopMu.Lock()
defer r.schedulerLoopMu.Unlock()
return r.schedulerLoopState
}
func (r *Plugin) GetSchedulerStatus() SchedulerStatus {
now := time.Now().UTC()
loopState := r.snapshotSchedulerLoopState()
r.schedulerMu.Lock()
phase := r.schedulerPhase
currentJobType := r.currentJobType
iterationStartedAt := r.iterationStartedAt
lastIterationEndedAt := r.lastIterationEndedAt
lastIterationWorkDetected := r.lastIterationWorkDetected
r.schedulerMu.Unlock()
status := SchedulerStatus{
Now: now,
SchedulerTickSeconds: int(secondsFromDuration(r.schedulerTick)),
InProcessJobs: r.listInProcessJobs(now),
IdleSleepSeconds: int(defaultSchedulerIdleSleep / time.Second),
CurrentJobType: loopState.currentJobType,
CurrentPhase: loopState.currentPhase,
LastIterationHadJobs: loopState.lastIterationHadJobs,
Now: now,
SchedulerTickSeconds: int(secondsFromDuration(r.idleSleepDuration)),
IdleSleepSeconds: int(secondsFromDuration(r.idleSleepDuration)),
Phase: phase,
CurrentJobType: currentJobType,
LastIterationWorkDetected: lastIterationWorkDetected,
InProcessJobs: r.listInProcessJobs(now),
}
nextDetectionAt := r.earliestNextDetectionAt()
if nextDetectionAt.IsZero() && loopState.currentPhase == "sleeping" && !loopState.lastIterationCompleted.IsZero() {
nextDetectionAt = loopState.lastIterationCompleted.Add(defaultSchedulerIdleSleep)
if !iterationStartedAt.IsZero() {
status.IterationStartedAt = timeToPtr(iterationStartedAt)
}
if !nextDetectionAt.IsZero() {
at := nextDetectionAt
status.NextDetectionAt = &at
}
if !loopState.lastIterationCompleted.IsZero() {
at := loopState.lastIterationCompleted
status.LastIterationDoneAt = &at
if !lastIterationEndedAt.IsZero() {
status.LastIterationEndedAt = timeToPtr(lastIterationEndedAt)
}
states, err := r.ListSchedulerStates()
@@ -251,11 +167,10 @@ func (r *Plugin) GetSchedulerStatus() SchedulerStatus {
info := r.snapshotSchedulerDetection(jobType)
jobStatus := SchedulerJobTypeStatus{
JobType: jobType,
Enabled: state.Enabled,
DetectionInFlight: state.DetectionInFlight,
NextDetectionAt: state.NextDetectionAt,
DetectionIntervalSeconds: state.DetectionIntervalSeconds,
JobType: jobType,
Enabled: state.Enabled,
DetectionInFlight: state.DetectionInFlight,
MaxJobTypeDurationSeconds: state.MaxJobTypeDurationSeconds,
}
if !info.lastDetectedAt.IsZero() {
jobStatus.LastDetectedAt = timeToPtr(info.lastDetectedAt)
@@ -271,18 +186,18 @@ func (r *Plugin) GetSchedulerStatus() SchedulerStatus {
if state.DetectionInFlight {
waiting = append(waiting, SchedulerWaitingStatus{
Reason: "detection_in_flight",
Reason: "processing",
JobType: jobType,
})
} else if state.Enabled && state.NextDetectionAt != nil && now.Before(*state.NextDetectionAt) {
waiting = append(waiting, SchedulerWaitingStatus{
Reason: "next_detection_at",
JobType: jobType,
Until: state.NextDetectionAt,
})
}
}
if phase == "idle" && !lastIterationEndedAt.IsZero() {
waiting = append(waiting, SchedulerWaitingStatus{
Reason: "idle_sleep",
})
}
sort.Slice(jobTypes, func(i, j int) bool {
return jobTypes[i].JobType < jobTypes[j].JobType
})
+15 -20
View File
@@ -82,26 +82,21 @@ type JobDetail struct {
}
type SchedulerJobTypeState struct {
JobType string `json:"job_type"`
Enabled bool `json:"enabled"`
PolicyError string `json:"policy_error,omitempty"`
DetectionInFlight bool `json:"detection_in_flight"`
NextDetectionAt *time.Time `json:"next_detection_at,omitempty"`
DetectionIntervalSeconds int32 `json:"detection_interval_seconds,omitempty"`
DetectionTimeoutSeconds int32 `json:"detection_timeout_seconds,omitempty"`
ExecutionTimeoutSeconds int32 `json:"execution_timeout_seconds,omitempty"`
JobTypeMaxRuntimeSeconds int32 `json:"job_type_max_runtime_seconds,omitempty"`
MaxJobsPerDetection int32 `json:"max_jobs_per_detection,omitempty"`
GlobalExecutionConcurrency int `json:"global_execution_concurrency,omitempty"`
PerWorkerExecutionConcurrency int `json:"per_worker_execution_concurrency,omitempty"`
RetryLimit int `json:"retry_limit,omitempty"`
RetryBackoffSeconds int32 `json:"retry_backoff_seconds,omitempty"`
DetectorAvailable bool `json:"detector_available"`
DetectorWorkerID string `json:"detector_worker_id,omitempty"`
ExecutorWorkerCount int `json:"executor_worker_count"`
LastRunStartedAt *time.Time `json:"last_run_started_at,omitempty"`
LastRunCompletedAt *time.Time `json:"last_run_completed_at,omitempty"`
LastRunStatus string `json:"last_run_status,omitempty"`
JobType string `json:"job_type"`
Enabled bool `json:"enabled"`
PolicyError string `json:"policy_error,omitempty"`
DetectionInFlight bool `json:"detection_in_flight"`
DetectionTimeoutSeconds int32 `json:"detection_timeout_seconds,omitempty"`
ExecutionTimeoutSeconds int32 `json:"execution_timeout_seconds,omitempty"`
MaxJobTypeDurationSeconds int32 `json:"max_job_type_duration_seconds,omitempty"`
MaxJobsPerDetection int32 `json:"max_jobs_per_detection,omitempty"`
GlobalExecutionConcurrency int `json:"global_execution_concurrency,omitempty"`
PerWorkerExecutionConcurrency int `json:"per_worker_execution_concurrency,omitempty"`
RetryLimit int `json:"retry_limit,omitempty"`
RetryBackoffSeconds int32 `json:"retry_backoff_seconds,omitempty"`
DetectorAvailable bool `json:"detector_available"`
DetectorWorkerID string `json:"detector_worker_id,omitempty"`
ExecutorWorkerCount int `json:"executor_worker_count"`
}
func timeToPtr(t time.Time) *time.Time {
+2 -30
View File
@@ -1,9 +1,5 @@
/* SeaweedFS Dashboard Custom Styles */
:root {
--navbar-height: 56px;
}
/* Link colors - muted */
a {
color: #5b7c99;
@@ -16,7 +12,7 @@ a:hover {
/* Sidebar Styles */
.sidebar {
position: fixed;
top: var(--navbar-height);
top: 56px;
bottom: 0;
left: 0;
z-index: 100;
@@ -55,32 +51,13 @@ main {
@media (max-width: 767.98px) {
.sidebar {
top: var(--navbar-height);
padding-top: 0;
width: 240px;
background-color: #f8f9fa !important;
z-index: 1050;
}
.sidebar.show ~ .sidebar-backdrop {
display: block;
top: 5rem;
}
main {
margin-left: 0;
}
}
/* Sidebar backdrop for mobile overlay */
.sidebar-backdrop {
display: none;
position: fixed;
top: var(--navbar-height);
left: 0;
right: 0;
bottom: 0;
background-color: rgba(0, 0, 0, 0.5);
z-index: 1040;
}
/* Custom card styles */
.border-left-primary {
border-left: 0.25rem solid #6b8caf !important;
@@ -285,11 +262,6 @@ main {
box-shadow: 0 0.15rem 1.75rem 0 rgba(58, 59, 69, 0.15) !important;
}
/* Navbar user icon color */
.navbar-toggler .fa-user {
color: rgba(255, 255, 255, 0.75);
}
/* Collapsible menu styles */
.nav-link[data-bs-toggle="collapse"] {
position: relative;
+73 -68
View File
@@ -31,9 +31,6 @@ function initializeDashboard() {
// Set up submenu behavior
setupSubmenuBehavior();
// Set up mobile sidebar behavior
setupMobileSidebar();
}
// HTMX event listeners
@@ -197,33 +194,6 @@ function setupSubmenuBehavior() {
}
// Mobile sidebar toggle and backdrop behavior
function setupMobileSidebar() {
const sidebar = document.getElementById('sidebarMenu');
const backdrop = document.getElementById('sidebarBackdrop');
if (!sidebar || !backdrop) return;
const hideSidebar = () => {
const bsCollapse = bootstrap.Collapse.getInstance(sidebar);
if (bsCollapse) {
bsCollapse.hide();
}
};
// Close sidebar when backdrop is clicked
backdrop.addEventListener('click', hideSidebar);
// Close sidebar when a nav link is clicked (on mobile)
const sidebarToggler = document.querySelector("button[data-bs-target='#sidebarMenu']");
sidebar.querySelectorAll('a.nav-link:not([data-bs-toggle="collapse"])').forEach(function (link) {
link.addEventListener('click', function () {
if (sidebarToggler && getComputedStyle(sidebarToggler).display !== 'none') {
hideSidebar();
}
});
});
}
// Loading indicator functions
function showLoadingIndicator() {
const indicator = document.getElementById('loading-indicator');
@@ -478,7 +448,7 @@ async function handleCreateBucket(event) {
if (response.ok) {
// Success
showAlert(`Bucket "${bucketData.name}" created successfully!`, 'success');
showAlert('success', `Bucket "${bucketData.name}" created successfully!`);
// Close modal
const modal = bootstrap.Modal.getInstance(document.getElementById('createBucketModal'));
@@ -493,11 +463,11 @@ async function handleCreateBucket(event) {
}, 1500);
} else {
// Error
showAlert(result.error || 'Failed to create bucket', 'danger');
showAlert('danger', result.error || 'Failed to create bucket');
}
} catch (error) {
console.error('Error creating bucket:', error);
showAlert('Network error occurred while creating bucket', 'danger');
showAlert('danger', 'Network error occurred while creating bucket');
}
}
@@ -538,7 +508,7 @@ async function deleteBucket() {
if (response.ok) {
// Success
showAlert(`Bucket "${bucketToDelete}" deleted successfully!`, 'success');
showAlert('success', `Bucket "${bucketToDelete}" deleted successfully!`);
// Close modal
const modal = bootstrap.Modal.getInstance(document.getElementById('deleteBucketModal'));
@@ -550,11 +520,11 @@ async function deleteBucket() {
}, 1500);
} else {
// Error
showAlert(result.error || 'Failed to delete bucket', 'danger');
showAlert('danger', result.error || 'Failed to delete bucket');
}
} catch (error) {
console.error('Error deleting bucket:', error);
showAlert('Network error occurred while deleting bucket', 'danger');
showAlert('danger', 'Network error occurred while deleting bucket');
}
bucketToDelete = '';
@@ -609,7 +579,38 @@ function exportBucketList() {
window.URL.revokeObjectURL(url);
}
// showAlert is provided by modal-alerts.js with signature: showAlert(message, type)
// Show alert message
function showAlert(type, message) {
// Remove existing alerts
const existingAlerts = document.querySelectorAll('.alert-floating');
existingAlerts.forEach(alert => alert.remove());
// Create new alert
const alert = document.createElement('div');
alert.className = `alert alert-${type} alert-dismissible fade show alert-floating`;
alert.style.cssText = `
position: fixed;
top: 20px;
right: 20px;
z-index: 9999;
min-width: 300px;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
`;
alert.innerHTML = `
${message}
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
`;
document.body.appendChild(alert);
// Auto-remove after 5 seconds
setTimeout(() => {
if (alert.parentNode) {
alert.remove();
}
}, 5000);
}
// Format date for display
function formatDate(date) {
@@ -620,7 +621,7 @@ function formatDate(date) {
function adminCopyToClipboard(text) {
if (navigator.clipboard && navigator.clipboard.writeText) {
navigator.clipboard.writeText(text).then(() => {
showAlert('Copied to clipboard!', 'success');
showAlert('success', 'Copied to clipboard!');
}).catch(err => {
console.error('Failed to copy text: ', err);
fallbackCopyText(text);
@@ -646,13 +647,13 @@ function fallbackCopyText(text) {
try {
const successful = document.execCommand('copy');
if (successful) {
showAlert('Copied to clipboard!', 'success');
showAlert('success', 'Copied to clipboard!');
} else {
showAlert('Failed to copy to clipboard', 'danger');
showAlert('danger', 'Failed to copy to clipboard');
}
} catch (err) {
console.error('Fallback copy failed: ', err);
showAlert('Failed to copy to clipboard', 'danger');
showAlert('danger', 'Failed to copy to clipboard');
}
document.body.removeChild(textArea);
@@ -733,7 +734,7 @@ function exportVolumes() {
function exportCollections() {
const table = document.getElementById('collectionsTable');
if (!table) {
showAlert('Collections table not found', 'error');
showAlert('error', 'Collections table not found');
return;
}
@@ -769,7 +770,7 @@ function exportCollections() {
function exportMasters() {
const table = document.getElementById('mastersTable');
if (!table) {
showAlert('Masters table not found', 'error');
showAlert('error', 'Masters table not found');
return;
}
@@ -803,7 +804,7 @@ function exportMasters() {
function exportFilers() {
const table = document.getElementById('filersTable');
if (!table) {
showAlert('Filers table not found', 'error');
showAlert('error', 'Filers table not found');
return;
}
@@ -839,7 +840,7 @@ function exportFilers() {
function exportUsers() {
const table = document.getElementById('usersTable');
if (!table) {
showAlert('Users table not found', 'error');
showAlert('error', 'Users table not found');
return;
}
@@ -989,7 +990,7 @@ function confirmDeleteSelected() {
const selectedPaths = getSelectedFilePaths();
if (selectedPaths.length === 0) {
showAlert('No files selected', 'warning');
showAlert('warning', 'No files selected');
return;
}
@@ -1010,7 +1011,7 @@ function confirmDeleteSelected() {
// Delete multiple selected files
async function deleteSelectedFiles(filePaths) {
if (!filePaths || filePaths.length === 0) {
showAlert('No files selected', 'warning');
showAlert('warning', 'No files selected');
return;
}
@@ -1034,9 +1035,9 @@ async function deleteSelectedFiles(filePaths) {
if (result.deleted > 0) {
if (result.failed === 0) {
showAlert(`Successfully deleted ${result.deleted} item(s)`, 'success');
showAlert('success', `Successfully deleted ${result.deleted} item(s)`);
} else {
showAlert(`Deleted ${result.deleted} item(s), failed to delete ${result.failed} item(s)`, 'warning');
showAlert('warning', `Deleted ${result.deleted} item(s), failed to delete ${result.failed} item(s)`);
if (result.errors && result.errors.length > 0) {
console.warn('Deletion errors:', result.errors);
}
@@ -1051,15 +1052,15 @@ async function deleteSelectedFiles(filePaths) {
if (result.errors && result.errors.length > 0) {
errorMessage += ': ' + result.errors.join(', ');
}
showAlert(errorMessage, 'error');
showAlert('error', errorMessage);
}
} else {
const error = await response.json();
showAlert(`Failed to delete files: ${error.error || 'Unknown error'}`, 'error');
showAlert('error', `Failed to delete files: ${error.error || 'Unknown error'}`);
}
} catch (error) {
console.error('Delete error:', error);
showAlert('Failed to delete files', 'error');
showAlert('error', 'Failed to delete files');
} finally {
// Re-enable the button
deleteBtn.disabled = false;
@@ -1280,7 +1281,7 @@ async function submitUploadFile() {
function exportFileList() {
const table = document.getElementById('fileTable');
if (!table) {
showAlert('File table not found', 'error');
showAlert('error', 'File table not found');
return;
}
@@ -1326,7 +1327,7 @@ async function viewFile(filePath) {
if (!response.ok) {
const error = await response.json();
showAlert(`Failed to view file: ${error.error || 'Unknown error'}`, 'error');
showAlert('error', `Failed to view file: ${error.error || 'Unknown error'}`);
return;
}
@@ -1335,7 +1336,7 @@ async function viewFile(filePath) {
} catch (error) {
console.error('View file error:', error);
showAlert('Failed to view file', 'error');
showAlert('error', 'Failed to view file');
}
}
@@ -1346,7 +1347,7 @@ async function showProperties(filePath) {
if (!response.ok) {
const error = await response.json();
showAlert(`Failed to get file properties: ${error.error || 'Unknown error'}`, 'error');
showAlert('error', `Failed to get file properties: ${error.error || 'Unknown error'}`);
return;
}
@@ -1355,7 +1356,7 @@ async function showProperties(filePath) {
} catch (error) {
console.error('Properties error:', error);
showAlert('Failed to get file properties', 'error');
showAlert('error', 'Failed to get file properties');
}
}
@@ -1382,16 +1383,16 @@ async function deleteFile(filePath) {
});
if (response.ok) {
showAlert(`Successfully deleted "${filePath}"`, 'success');
showAlert('success', `Successfully deleted "${filePath}"`);
// Reload the page to update the file list
window.location.reload();
} else {
const error = await response.json();
showAlert(`Failed to delete file: ${error.error || 'Unknown error'}`, 'error');
showAlert('error', `Failed to delete file: ${error.error || 'Unknown error'}`);
}
} catch (error) {
console.error('Delete error:', error);
showAlert('Failed to delete file', 'error');
showAlert('error', 'Failed to delete file');
}
}
@@ -1706,7 +1707,7 @@ async function handleUpdateQuota(event) {
if (response.ok) {
// Success
showAlert(`Quota for bucket "${bucketName}" updated successfully!`, 'success');
showAlert('success', `Quota for bucket "${bucketName}" updated successfully!`);
// Close modal
const modal = bootstrap.Modal.getInstance(document.getElementById('manageQuotaModal'));
@@ -1718,11 +1719,11 @@ async function handleUpdateQuota(event) {
}, 1500);
} else {
// Error
showAlert(result.error || 'Failed to update bucket quota', 'danger');
showAlert('danger', result.error || 'Failed to update bucket quota');
}
} catch (error) {
console.error('Error updating bucket quota:', error);
showAlert('Network error occurred while updating bucket quota', 'danger');
showAlert('danger', 'Network error occurred while updating bucket quota');
}
}
@@ -2196,6 +2197,10 @@ function showNewAccessKeyModal(accessKeyData) {
<i class="fas fa-check-circle me-2"></i>
<strong>Success!</strong> Your new access key has been created.
</div>
<div class="alert alert-warning">
<i class="fas fa-exclamation-triangle me-2"></i>
<strong>Important:</strong> This is the only time the secret key will be displayed. Please save it securely.
</div>
<div class="mb-3">
<label class="form-label"><strong>Access Key:</strong></label>
<div class="input-group">
@@ -2239,21 +2244,21 @@ function copyFromInput(inputId) {
try {
const successful = document.execCommand('copy');
if (successful) {
showAlert('Copied to clipboard!', 'success');
showAlert('success', 'Copied to clipboard!');
} else {
// Try modern clipboard API as fallback
navigator.clipboard.writeText(input.value).then(() => {
showAlert('Copied to clipboard!', 'success');
showAlert('success', 'Copied to clipboard!');
}).catch(() => {
showAlert('Failed to copy', 'danger');
showAlert('danger', 'Failed to copy');
});
}
} catch (err) {
// Try modern clipboard API as fallback
navigator.clipboard.writeText(input.value).then(() => {
showAlert('Copied to clipboard!', 'success');
showAlert('success', 'Copied to clipboard!');
}).catch(() => {
showAlert('Failed to copy', 'danger');
showAlert('danger', 'Failed to copy');
});
}
}
-23
View File
@@ -25,29 +25,6 @@ async function deleteUser(username) {
}, 'Are you sure you want to delete this user? This action cannot be undone.');
}
// Delete group function
async function deleteGroup(name) {
showDeleteConfirm(name, async function () {
try {
const encodedName = encodeURIComponent(name);
const response = await fetch(`/api/groups/${encodedName}`, {
method: 'DELETE'
});
if (response.ok) {
showAlert('Group deleted successfully', 'success');
setTimeout(() => window.location.reload(), 1000);
} else {
const error = await response.json().catch(() => ({}));
showAlert('Failed to delete group: ' + (error.error || 'Unknown error'), 'error');
}
} catch (error) {
console.error('Error deleting group:', error);
showAlert('Failed to delete group: ' + error.message, 'error');
}
}, 'Are you sure you want to delete this group? This action cannot be undone.');
}
// Delete access key function
async function deleteAccessKey(username, accessKey) {
showDeleteConfirm(accessKey, async function () {
+12 -56
View File
@@ -17,30 +17,20 @@ func (at *ActiveTopology) AssignTask(taskID string) error {
return fmt.Errorf("pending task %s not found", taskID)
}
// Skip capacity check if topology hasn't been populated yet
if len(at.disks) == 0 {
glog.Warningf("AssignTask %s: topology has no disks yet, skipping capacity check", taskID)
} else {
// Check if all destination disks have sufficient capacity to reserve
for _, dest := range task.Destinations {
targetKey := fmt.Sprintf("%s:%d", dest.TargetServer, dest.TargetDisk)
if targetDisk, exists := at.disks[targetKey]; exists {
availableCapacity := at.getEffectiveAvailableCapacityUnsafe(targetDisk)
// Check if all destination disks have sufficient capacity to reserve
for _, dest := range task.Destinations {
targetKey := fmt.Sprintf("%s:%d", dest.TargetServer, dest.TargetDisk)
if targetDisk, exists := at.disks[targetKey]; exists {
availableCapacity := at.getEffectiveAvailableCapacityUnsafe(targetDisk)
// Check if we have enough total capacity using the improved unified comparison
if !availableCapacity.CanAccommodate(dest.StorageChange) {
return fmt.Errorf("insufficient capacity on target disk %s:%d. Available: %+v, Required: %+v",
dest.TargetServer, dest.TargetDisk, availableCapacity, dest.StorageChange)
}
} else if dest.TargetServer != "" {
// Fail fast if destination disk is not found in topology
var existingKeys []string
for k := range at.disks {
existingKeys = append(existingKeys, k)
}
glog.Warningf("destination disk %s not found in topology. Existing disk keys: %v", targetKey, existingKeys)
return fmt.Errorf("destination disk %s not found in topology", targetKey)
// Check if we have enough total capacity using the improved unified comparison
if !availableCapacity.CanAccommodate(dest.StorageChange) {
return fmt.Errorf("insufficient capacity on target disk %s:%d. Available: %+v, Required: %+v",
dest.TargetServer, dest.TargetDisk, availableCapacity, dest.StorageChange)
}
} else if dest.TargetServer != "" {
// Fail fast if destination disk is not found in topology
return fmt.Errorf("destination disk %s not found in topology", targetKey)
}
}
@@ -272,40 +262,6 @@ func (at *ActiveTopology) HasAnyTask(volumeID uint32) bool {
return at.HasTask(volumeID, TaskTypeNone)
}
// GetTaskServerAdjustments returns per-server volume count adjustments for
// pending and assigned tasks of the given type. For each task, source servers
// are decremented and destination servers are incremented, reflecting the
// projected volume distribution once in-flight tasks complete.
func (at *ActiveTopology) GetTaskServerAdjustments(taskType TaskType) map[string]int {
at.mutex.RLock()
defer at.mutex.RUnlock()
adjustments := make(map[string]int)
for _, task := range at.pendingTasks {
if task.TaskType != taskType {
continue
}
for _, src := range task.Sources {
adjustments[src.SourceServer]--
}
for _, dst := range task.Destinations {
adjustments[dst.TargetServer]++
}
}
for _, task := range at.assignedTasks {
if task.TaskType != taskType {
continue
}
for _, src := range task.Sources {
adjustments[src.SourceServer]--
}
for _, dst := range task.Destinations {
adjustments[dst.TargetServer]++
}
}
return adjustments
}
// calculateSourceStorageImpact calculates storage impact for sources based on task type and cleanup type
func (at *ActiveTopology) calculateSourceStorageImpact(taskType TaskType, cleanupType SourceCleanupType, volumeSize int64) StorageSlotChange {
switch taskType {

Some files were not shown because too many files have changed in this diff Show More