mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-15 02:55:51 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Chris Lu
|
8d75290601 | ||
  Chris LuandGitHub
|
289ec5e2f5 | ||
|
Chris Lu
|
683e3d06a4 | ||
|
Chris LuandGitHub
|
2567be8040 | ||
|
14df5d1bb5 |
@@ -52,6 +52,7 @@ jobs:
|
||||
run: |
|
||||
cd test/kafka
|
||||
go mod download
|
||||
go mod tidy
|
||||
|
||||
- name: Run Kafka Gateway Unit Tests
|
||||
run: |
|
||||
@@ -96,6 +97,7 @@ jobs:
|
||||
run: |
|
||||
cd test/kafka
|
||||
go mod download
|
||||
go mod tidy
|
||||
|
||||
- name: Run Integration Tests
|
||||
run: |
|
||||
@@ -154,6 +156,7 @@ jobs:
|
||||
cd test/kafka
|
||||
# Use go mod download with timeout to prevent hanging
|
||||
timeout 90s go mod download || echo "Warning: Dependency download timed out, continuing with cached modules"
|
||||
go mod tidy
|
||||
|
||||
- name: Build and start SeaweedFS MQ
|
||||
run: |
|
||||
@@ -332,6 +335,7 @@ jobs:
|
||||
cd test/kafka
|
||||
# Use go mod download with timeout to prevent hanging
|
||||
timeout 90s go mod download || echo "Warning: Dependency download timed out, continuing with cached modules"
|
||||
go mod tidy
|
||||
|
||||
- name: Build and start SeaweedFS MQ
|
||||
run: |
|
||||
@@ -492,6 +496,7 @@ jobs:
|
||||
run: |
|
||||
cd test/kafka
|
||||
timeout 90s go mod download || echo "Warning: Dependency download timed out, continuing with cached modules"
|
||||
go mod tidy
|
||||
|
||||
- name: Build and start SeaweedFS MQ
|
||||
run: |
|
||||
@@ -649,6 +654,7 @@ jobs:
|
||||
run: |
|
||||
cd test/kafka
|
||||
timeout 90s go mod download || echo "Warning: Dependency download timed out, continuing with cached modules"
|
||||
go mod tidy
|
||||
|
||||
- name: Build and start SeaweedFS MQ
|
||||
run: |
|
||||
@@ -803,6 +809,7 @@ jobs:
|
||||
run: |
|
||||
cd test/kafka
|
||||
go mod download
|
||||
go mod tidy
|
||||
|
||||
- name: Run Protocol Tests
|
||||
run: |
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
apiVersion: v1
|
||||
description: SeaweedFS
|
||||
name: seaweedfs
|
||||
appVersion: "4.03"
|
||||
appVersion: "4.04"
|
||||
# Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
|
||||
version: 4.0.403
|
||||
version: 4.0.404
|
||||
|
||||
+5
-4
@@ -45,20 +45,21 @@ require (
|
||||
github.com/aws/aws-sdk-go v1.55.8 // indirect
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/config v1.31.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.5 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.6 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.6 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.6 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
|
||||
github.com/aws/smithy-go v1.24.0 // indirect
|
||||
|
||||
+10
-8
@@ -106,10 +106,10 @@ github.com/aws/aws-sdk-go-v2 v1.41.0 h1:tNvqh1s+v0vFYdA1xq0aOJH+Y5cRyZ5upu6roPgP
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.0/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.31.3 h1:RIb3yr/+PZ18YYNe6MDiG/3jVoJrPmdoCARwNkMGvco=
|
||||
github.com/aws/aws-sdk-go-v2/config v1.31.3/go.mod h1:jjgx1n7x0FAKl6TnakqrpkHWWKcX3xfWtdnIJs5K9CE=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.5 h1:xMo63RlqP3ZZydpJDMBsH9uJ10hgHYfQFIk1cHDXrR4=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.5/go.mod h1:hhbH6oRcou+LpXfA/0vPElh/e0M3aFeOblE1sssAAEk=
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.6 h1:hFLBGUKjmLAekvi1evLi5hVvFQtSo3GYwi+Bx4lpJf8=
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.6/go.mod h1:lcUL/gcd8WyjCrMnxez5OXkO3/rwcNmvfno62tnXNcI=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.6 h1:F9vWao2TwjV2MyiyVS+duza0NIRtAslgLUM0vTA1ZaE=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.6/go.mod h1:SgHzKjEVsdQr6Opor0ihgWtkWdfRAIwxYzSJ8O85VHY=
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 h1:80+uETIWS1BqjnN9uJ0dBUaETh+P1XwFy5vwHwK5r9k=
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16/go.mod h1:wOOsYuxYuB/7FlnVtzeBYRcjSRtQpAW0hCP7tIULMwo=
|
||||
github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 h1:0SzCLoPRSK3qSydsaFQWugP+lOBCTPwfcBOm6222+UA=
|
||||
@@ -118,8 +118,8 @@ github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 h1:rgGwPzb82iBYSvHMH
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16/go.mod h1:L/UxsGeKpGoIj6DxfhOWHWQ/kGKcd4I1VncE4++IyKA=
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 h1:1jtGzuV7c82xnqOVfx2F0xmJcOw5374L7N6juGW6x6U=
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16/go.mod h1:M2E5OQf+XLe+SZGmmpaI2yy+J326aFf6/+54PoxSANc=
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 h1:WKuaxf++XKWlHWu9ECbMlha8WOEGm0OUEZqm4K/Gcfk=
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4/go.mod h1:ZWy7j6v1vWGmPReu0iSGvRiise4YI5SkR3OHKTZ6Wuc=
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 h1:NLYTEyZmVZo0Qh183sC8nC+ydJXOOeIL/qI/sS3PdLY=
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15/go.mod h1:Z803iB3B0bc8oJV8zH2PERLRfQUJ2n2BXISpsA4+O1M=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 h1:0ryTNEdJbzUCEWkVXEXoqlXV72J5keC1GvILMOuD00E=
|
||||
@@ -132,8 +132,10 @@ github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 h1:wsSQ4SVz5YE1c
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15/go.mod h1:I7sditnFGtYMIqPRU1QoHZAUrXkGp4SczmlLwrNPlD0=
|
||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0 h1:IrbE3B8O9pm3lsg96AXIN5MXX4pECEuExh/A0Du3AuI=
|
||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0/go.mod h1:/sJLzHtiiZvs6C1RbxS/anSAFwZD6oC6M/kotQzOiLw=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 h1:eYnlt6QxnFINKzwxP5/Ucs1vkG7VT3Iezmvfgc2waUw=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.7/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 h1:HpI7aMmJ+mm1wkSHIA2t5EaFFv5EFYXePW30p1EIrbQ=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.4/go.mod h1:C5RdGMYGlfM0gYq/tifqgn4EbyX99V15P2V3R+VHbQU=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 h1:aM/Q24rIlS3bRAhTyFurowU8A0SMyGDtEOY/l/s/1Uw=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.8/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg=
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 h1:AHDr0DaHIAo8c9t1emrzAlVDFp+iMMKnPdYy6XO4MCE=
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12/go.mod h1:GQ73XawFFiWxyWXMHWfhiomvP3tXtdNar/fi8z18sx0=
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 h1:SciGFVNZ4mHdm7gpD1dgZYnCuVdX1s+lFTg4+4DOy70=
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -340,7 +341,7 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
|
||||
}
|
||||
|
||||
// Get versioning, object lock, and owner information from extended attributes
|
||||
versioningEnabled := false
|
||||
versioningStatus := ""
|
||||
objectLockEnabled := false
|
||||
objectLockMode := ""
|
||||
var objectLockDuration int32 = 0
|
||||
@@ -348,7 +349,7 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
|
||||
|
||||
if resp.Entry.Extended != nil {
|
||||
// Use shared utility to extract versioning information
|
||||
versioningEnabled = extractVersioningFromEntry(resp.Entry)
|
||||
versioningStatus = extractVersioningFromEntry(resp.Entry)
|
||||
|
||||
// Use shared utility to extract Object Lock information
|
||||
objectLockEnabled, objectLockMode, objectLockDuration = extractObjectLockInfoFromEntry(resp.Entry)
|
||||
@@ -367,7 +368,7 @@ func (s *AdminServer) GetS3Buckets() ([]S3Bucket, error) {
|
||||
LastModified: time.Unix(resp.Entry.Attributes.Mtime, 0),
|
||||
Quota: quota,
|
||||
QuotaEnabled: quotaEnabled,
|
||||
VersioningEnabled: versioningEnabled,
|
||||
VersioningStatus: versioningStatus,
|
||||
ObjectLockEnabled: objectLockEnabled,
|
||||
ObjectLockMode: objectLockMode,
|
||||
ObjectLockDuration: objectLockDuration,
|
||||
@@ -430,7 +431,7 @@ func (s *AdminServer) GetBucketDetails(bucketName string) (*BucketDetails, error
|
||||
details.Bucket.QuotaEnabled = quotaEnabled
|
||||
|
||||
// Get versioning, object lock, and owner information from extended attributes
|
||||
versioningEnabled := false
|
||||
versioningStatus := ""
|
||||
objectLockEnabled := false
|
||||
objectLockMode := ""
|
||||
var objectLockDuration int32 = 0
|
||||
@@ -438,7 +439,7 @@ func (s *AdminServer) GetBucketDetails(bucketName string) (*BucketDetails, error
|
||||
|
||||
if bucketResp.Entry.Extended != nil {
|
||||
// Use shared utility to extract versioning information
|
||||
versioningEnabled = extractVersioningFromEntry(bucketResp.Entry)
|
||||
versioningStatus = extractVersioningFromEntry(bucketResp.Entry)
|
||||
|
||||
// Use shared utility to extract Object Lock information
|
||||
objectLockEnabled, objectLockMode, objectLockDuration = extractObjectLockInfoFromEntry(bucketResp.Entry)
|
||||
@@ -449,7 +450,7 @@ func (s *AdminServer) GetBucketDetails(bucketName string) (*BucketDetails, error
|
||||
}
|
||||
}
|
||||
|
||||
details.Bucket.VersioningEnabled = versioningEnabled
|
||||
details.Bucket.VersioningStatus = versioningStatus
|
||||
details.Bucket.ObjectLockEnabled = objectLockEnabled
|
||||
details.Bucket.ObjectLockMode = objectLockMode
|
||||
details.Bucket.ObjectLockDuration = objectLockDuration
|
||||
@@ -491,6 +492,45 @@ func (s *AdminServer) listBucketObjects(client filer_pb.SeaweedFilerClient, buck
|
||||
|
||||
entry := resp.Entry
|
||||
if entry.IsDirectory {
|
||||
// Check if this is a .versions directory (represents a versioned object)
|
||||
if strings.HasSuffix(entry.Name, ".versions") {
|
||||
// This directory represents an object, add it as an object without the .versions suffix
|
||||
objectName := strings.TrimSuffix(entry.Name, ".versions")
|
||||
objectKey := objectName
|
||||
if directory != bucketBasePath {
|
||||
relativePath := directory[len(bucketBasePath)+1:]
|
||||
objectKey = fmt.Sprintf("%s/%s", relativePath, objectName)
|
||||
}
|
||||
|
||||
// Extract latest version metadata from extended attributes
|
||||
var size int64 = 0
|
||||
var mtime int64 = entry.Attributes.Mtime
|
||||
if entry.Extended != nil {
|
||||
// Get size of latest version
|
||||
if sizeBytes, ok := entry.Extended[s3_constants.ExtLatestVersionSizeKey]; ok && len(sizeBytes) == 8 {
|
||||
size = int64(util.BytesToUint64(sizeBytes))
|
||||
}
|
||||
// Get mtime of latest version
|
||||
if mtimeBytes, ok := entry.Extended[s3_constants.ExtLatestVersionMtimeKey]; ok && len(mtimeBytes) == 8 {
|
||||
mtime = int64(util.BytesToUint64(mtimeBytes))
|
||||
}
|
||||
}
|
||||
|
||||
obj := S3Object{
|
||||
Key: objectKey,
|
||||
Size: size,
|
||||
LastModified: time.Unix(mtime, 0),
|
||||
ETag: "",
|
||||
StorageClass: "STANDARD",
|
||||
}
|
||||
|
||||
details.Objects = append(details.Objects, obj)
|
||||
details.TotalCount++
|
||||
details.TotalSize += size
|
||||
// Don't recurse into .versions directories
|
||||
continue
|
||||
}
|
||||
|
||||
// Recursively list subdirectories
|
||||
subDir := fmt.Sprintf("%s/%s", directory, entry.Name)
|
||||
err := s.listBucketObjects(client, bucketBasePath, subDir, "", details)
|
||||
@@ -1902,9 +1942,8 @@ func extractObjectLockInfoFromEntry(entry *filer_pb.Entry) (bool, string, int32)
|
||||
}
|
||||
|
||||
// Function to extract versioning information from bucket entry using shared utilities
|
||||
func extractVersioningFromEntry(entry *filer_pb.Entry) bool {
|
||||
enabled, _ := s3api.LoadVersioningFromExtended(entry)
|
||||
return enabled
|
||||
func extractVersioningFromEntry(entry *filer_pb.Entry) string {
|
||||
return s3api.GetVersioningStatus(entry)
|
||||
}
|
||||
|
||||
// GetConfigPersistence returns the config persistence manager
|
||||
|
||||
@@ -125,6 +125,12 @@ func (s *AdminServer) CreateBucket(c *gin.Context) {
|
||||
// Convert quota to bytes
|
||||
quotaBytes := convertQuotaToBytes(req.QuotaSize, req.QuotaUnit)
|
||||
|
||||
// Validate quota: if enabled, size must be greater than 0
|
||||
if req.QuotaEnabled && quotaBytes <= 0 {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "Quota size must be greater than 0 when quota is enabled"})
|
||||
return
|
||||
}
|
||||
|
||||
// Sanitize owner: trim whitespace and enforce max length
|
||||
owner := strings.TrimSpace(req.Owner)
|
||||
if len(owner) > MaxOwnerNameLength {
|
||||
@@ -466,16 +472,19 @@ func (s *AdminServer) CreateS3BucketWithObjectLock(bucketName string, quotaBytes
|
||||
// Handle Object Lock configuration using shared utilities
|
||||
if objectLockEnabled {
|
||||
var duration int32 = 0
|
||||
var mode string = ""
|
||||
|
||||
if setDefaultRetention {
|
||||
// Validate Object Lock parameters only when setting default retention
|
||||
if err := s3api.ValidateObjectLockParameters(objectLockEnabled, objectLockMode, objectLockDuration); err != nil {
|
||||
return fmt.Errorf("invalid Object Lock parameters: %w", err)
|
||||
}
|
||||
duration = objectLockDuration
|
||||
mode = objectLockMode
|
||||
}
|
||||
|
||||
// Create Object Lock configuration using shared utility
|
||||
objectLockConfig := s3api.CreateObjectLockConfigurationFromParams(objectLockEnabled, objectLockMode, duration)
|
||||
objectLockConfig := s3api.CreateObjectLockConfigurationFromParams(objectLockEnabled, mode, duration)
|
||||
|
||||
// Store Object Lock configuration in extended attributes using shared utility
|
||||
if err := s3api.StoreObjectLockConfigurationInExtended(bucketEntry, objectLockConfig); err != nil {
|
||||
|
||||
@@ -78,7 +78,7 @@ type S3Bucket struct {
|
||||
LastModified time.Time `json:"last_modified"`
|
||||
Quota int64 `json:"quota"` // Quota in bytes, 0 means no quota
|
||||
QuotaEnabled bool `json:"quota_enabled"` // Whether quota is enabled
|
||||
VersioningEnabled bool `json:"versioning_enabled"` // Whether versioning is enabled
|
||||
VersioningStatus string `json:"versioning_status"` // Versioning status: "" (never enabled), "Enabled", or "Suspended"
|
||||
ObjectLockEnabled bool `json:"object_lock_enabled"` // Whether object lock is enabled
|
||||
ObjectLockMode string `json:"object_lock_mode"` // Object lock mode: "GOVERNANCE" or "COMPLIANCE"
|
||||
ObjectLockDuration int32 `json:"object_lock_duration"` // Default retention duration in days
|
||||
|
||||
@@ -164,14 +164,16 @@ templ S3Buckets(data dash.S3BucketsData) {
|
||||
}
|
||||
</td>
|
||||
<td>
|
||||
if bucket.VersioningEnabled {
|
||||
if bucket.VersioningStatus == "Enabled" {
|
||||
<span class="badge bg-success">
|
||||
<i class="fas fa-check me-1"></i>Enabled
|
||||
</span>
|
||||
} else {
|
||||
<span class="badge bg-secondary">
|
||||
<i class="fas fa-times me-1"></i>Disabled
|
||||
} else if bucket.VersioningStatus == "Suspended" {
|
||||
<span class="badge bg-warning">
|
||||
<i class="fas fa-pause me-1"></i>Suspended
|
||||
</span>
|
||||
} else {
|
||||
<span class="text-muted">Not configured</span>
|
||||
}
|
||||
</td>
|
||||
<td>
|
||||
@@ -185,9 +187,7 @@ templ S3Buckets(data dash.S3BucketsData) {
|
||||
</div>
|
||||
</div>
|
||||
} else {
|
||||
<span class="badge bg-secondary">
|
||||
<i class="fas fa-unlock me-1"></i>Disabled
|
||||
</span>
|
||||
<span class="text-muted">Not configured</span>
|
||||
}
|
||||
</td>
|
||||
<td>
|
||||
@@ -1044,9 +1044,11 @@ templ S3Buckets(data dash.S3BucketsData) {
|
||||
'<tr>' +
|
||||
'<td><strong>Versioning:</strong></td>' +
|
||||
'<td>' +
|
||||
(bucket.versioning_enabled ?
|
||||
(bucket.versioning_status === 'Enabled' ?
|
||||
'<span class="badge bg-success"><i class="fas fa-check me-1"></i>Enabled</span>' :
|
||||
'<span class="badge bg-secondary"><i class="fas fa-times me-1"></i>Disabled</span>'
|
||||
bucket.versioning_status === 'Suspended' ?
|
||||
'<span class="badge bg-warning"><i class="fas fa-pause me-1"></i>Suspended</span>' :
|
||||
'<span class="text-muted">Not configured</span>'
|
||||
) +
|
||||
'</td>' +
|
||||
'</tr>' +
|
||||
@@ -1055,8 +1057,10 @@ templ S3Buckets(data dash.S3BucketsData) {
|
||||
'<td>' +
|
||||
(bucket.object_lock_enabled ?
|
||||
'<span class="badge bg-warning"><i class="fas fa-lock me-1"></i>Enabled</span>' +
|
||||
'<br><small class="text-muted">' + escapeHtml(bucket.object_lock_mode) + ' • ' + bucket.object_lock_duration + ' days</small>' :
|
||||
'<span class="badge bg-secondary"><i class="fas fa-unlock me-1"></i>Disabled</span>'
|
||||
(bucket.object_lock_mode && bucket.object_lock_duration > 0 ?
|
||||
'<br><small class="text-muted">' + escapeHtml(bucket.object_lock_mode) + ' • ' + bucket.object_lock_duration + ' days</small>' :
|
||||
'') :
|
||||
'<span class="text-muted">Not configured</span>'
|
||||
) +
|
||||
'</td>' +
|
||||
'</tr>' +
|
||||
|
||||
File diff suppressed because one or more lines are too long
+19
-50
@@ -517,39 +517,26 @@ func initializeGrpcPortsOnIP(bindIp string) {
|
||||
continue
|
||||
}
|
||||
|
||||
// If gRPC port is 0, calculate it
|
||||
// If gRPC port is 0, calculate it from HTTP port
|
||||
if *config.grpcPort == 0 {
|
||||
calculatedPort := *config.httpPort + GrpcPortOffset
|
||||
// Check if calculated port is available (on both specific IP and all interfaces)
|
||||
// Also check if it was already allocated to another service in this function
|
||||
if !isPortOpenOnIP(bindIp, calculatedPort) || !isPortAvailable(calculatedPort) || allocatedGrpcPorts[calculatedPort] {
|
||||
glog.Warningf("Calculated gRPC port %d for %s is not available, finding alternative...", calculatedPort, config.name)
|
||||
newPort := findAvailablePortOnIP(bindIp, calculatedPort+1, 100, allocatedGrpcPorts)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available gRPC port for %s starting from %d, will use calculated %d and fail on binding", config.name, calculatedPort+1, calculatedPort)
|
||||
} else {
|
||||
calculatedPort = newPort
|
||||
glog.Infof("gRPC port %d for %s is available, using it instead of calculated %d", newPort, config.name, *config.httpPort+GrpcPortOffset)
|
||||
}
|
||||
}
|
||||
*config.grpcPort = calculatedPort
|
||||
allocatedGrpcPorts[calculatedPort] = true
|
||||
glog.V(1).Infof("%s gRPC port initialized to %d", config.name, calculatedPort)
|
||||
} else {
|
||||
// gRPC port was explicitly set, verify it's still available (check on both specific IP and all interfaces)
|
||||
// Also check if it was already allocated to another service in this function
|
||||
if !isPortOpenOnIP(bindIp, *config.grpcPort) || !isPortAvailable(*config.grpcPort) || allocatedGrpcPorts[*config.grpcPort] {
|
||||
glog.Warningf("Explicitly set gRPC port %d for %s is not available, finding alternative...", *config.grpcPort, config.name)
|
||||
newPort := findAvailablePortOnIP(bindIp, *config.grpcPort+1, 100, allocatedGrpcPorts)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available gRPC port for %s starting from %d, will use original %d and fail on binding", config.name, *config.grpcPort+1, *config.grpcPort)
|
||||
} else {
|
||||
glog.Infof("gRPC port %d for %s is available, using it instead of %d", newPort, config.name, *config.grpcPort)
|
||||
*config.grpcPort = newPort
|
||||
}
|
||||
}
|
||||
allocatedGrpcPorts[*config.grpcPort] = true
|
||||
*config.grpcPort = *config.httpPort + GrpcPortOffset
|
||||
}
|
||||
|
||||
// Verify the gRPC port is available (whether calculated or explicitly set)
|
||||
// Check on both specific IP and all interfaces, and check against already allocated ports
|
||||
if !isPortOpenOnIP(bindIp, *config.grpcPort) || !isPortAvailable(*config.grpcPort) || allocatedGrpcPorts[*config.grpcPort] {
|
||||
glog.Warningf("gRPC port %d for %s is not available, finding alternative...", *config.grpcPort, config.name)
|
||||
originalPort := *config.grpcPort
|
||||
newPort := findAvailablePortOnIP(bindIp, originalPort+1, 100, allocatedGrpcPorts)
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available gRPC port for %s starting from %d, will use %d and fail on binding", config.name, originalPort+1, originalPort)
|
||||
} else {
|
||||
glog.Infof("gRPC port %d for %s is available, using it instead of %d", newPort, config.name, originalPort)
|
||||
*config.grpcPort = newPort
|
||||
}
|
||||
}
|
||||
allocatedGrpcPorts[*config.grpcPort] = true
|
||||
glog.V(1).Infof("%s gRPC port set to %d", config.name, *config.grpcPort)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -934,26 +921,8 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) {
|
||||
|
||||
// gRPC port should have been initialized by ensureAllPortsAvailableOnIP in runMini
|
||||
// If it's still 0, that indicates a problem with the port initialization sequence
|
||||
// This defensive fallback handles edge cases where port initialization may have been skipped
|
||||
// or failed silently (e.g., due to configuration changes or error handling paths)
|
||||
if *miniAdminOptions.grpcPort == 0 {
|
||||
glog.Warningf("Admin gRPC port was not initialized before startAdminServer, attempting fallback initialization...")
|
||||
// Use the same availability checking logic as initializeGrpcPortsOnIP
|
||||
calculatedPort := *miniAdminOptions.port + GrpcPortOffset
|
||||
if !isPortOpenOnIP(getBindIp(), calculatedPort) || !isPortAvailable(calculatedPort) {
|
||||
glog.Warningf("Calculated fallback gRPC port %d is not available, finding alternative...", calculatedPort)
|
||||
newPort := findAvailablePortOnIP(getBindIp(), calculatedPort+1, 100, make(map[int]bool))
|
||||
if newPort == 0 {
|
||||
glog.Errorf("Could not find available gRPC port for Admin starting from %d, will use calculated %d and fail on binding", calculatedPort+1, calculatedPort)
|
||||
*miniAdminOptions.grpcPort = calculatedPort
|
||||
} else {
|
||||
glog.Infof("Fallback: using gRPC port %d for Admin", newPort)
|
||||
*miniAdminOptions.grpcPort = newPort
|
||||
}
|
||||
} else {
|
||||
*miniAdminOptions.grpcPort = calculatedPort
|
||||
glog.Infof("Fallback: Admin gRPC port initialized to %d", calculatedPort)
|
||||
}
|
||||
glog.Fatalf("Admin gRPC port was not initialized before startAdminServer. This indicates a problem with the port initialization sequence.")
|
||||
}
|
||||
|
||||
// Create data directory if specified
|
||||
|
||||
@@ -35,11 +35,6 @@ const (
|
||||
// gRPC keepalive settings - must be consistent between client and server
|
||||
GrpcKeepAliveTime = 60 * time.Second // ping interval when no activity
|
||||
GrpcKeepAliveTimeout = 20 * time.Second // ping timeout
|
||||
|
||||
// Connection recycling for Docker Swarm environments
|
||||
// Forces connections to be recycled periodically to handle DNS changes
|
||||
GrpcMaxConnectionAge = 5 * time.Minute // max time a connection may exist
|
||||
GrpcMaxConnectionAgeGrace = 30 * time.Second // grace period for RPCs to complete
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -63,10 +58,8 @@ func NewGrpcServer(opts ...grpc.ServerOption) *grpc.Server {
|
||||
var options []grpc.ServerOption
|
||||
options = append(options,
|
||||
grpc.KeepaliveParams(keepalive.ServerParameters{
|
||||
Time: GrpcKeepAliveTime, // server pings client if no activity for this long
|
||||
Timeout: GrpcKeepAliveTimeout, // ping timeout
|
||||
MaxConnectionAge: GrpcMaxConnectionAge, // max connection age for Docker Swarm DNS refresh
|
||||
MaxConnectionAgeGrace: GrpcMaxConnectionAgeGrace, // grace period for in-flight RPCs
|
||||
Time: GrpcKeepAliveTime, // server pings client if no activity for this long
|
||||
Timeout: GrpcKeepAliveTimeout, // ping timeout
|
||||
}),
|
||||
grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
|
||||
MinTime: GrpcKeepAliveTime, // min time a client should wait before sending a ping
|
||||
|
||||
@@ -28,7 +28,8 @@ func StoreVersioningInExtended(entry *filer_pb.Entry, enabled bool) error {
|
||||
if enabled {
|
||||
entry.Extended[s3_constants.ExtVersioningKey] = []byte(s3_constants.VersioningEnabled)
|
||||
} else {
|
||||
entry.Extended[s3_constants.ExtVersioningKey] = []byte(s3_constants.VersioningSuspended)
|
||||
// Don't set the header when versioning is not enabled
|
||||
delete(entry.Extended, s3_constants.ExtVersioningKey)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -49,6 +50,20 @@ func LoadVersioningFromExtended(entry *filer_pb.Entry) (bool, bool) {
|
||||
return false, false // not found
|
||||
}
|
||||
|
||||
// GetVersioningStatus returns the versioning status as a string: "", "Enabled", or "Suspended"
|
||||
// Empty string means versioning was never enabled
|
||||
func GetVersioningStatus(entry *filer_pb.Entry) string {
|
||||
if entry == nil || entry.Extended == nil {
|
||||
return "" // Never enabled
|
||||
}
|
||||
|
||||
if versioningBytes, exists := entry.Extended[s3_constants.ExtVersioningKey]; exists {
|
||||
return string(versioningBytes) // "Enabled" or "Suspended"
|
||||
}
|
||||
|
||||
return "" // Never enabled
|
||||
}
|
||||
|
||||
// CreateObjectLockConfiguration creates a new ObjectLockConfiguration with the specified parameters
|
||||
func CreateObjectLockConfiguration(enabled bool, mode string, days int, years int) *ObjectLockConfiguration {
|
||||
if !enabled {
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
|
||||
var (
|
||||
MAJOR_VERSION = int32(4)
|
||||
MINOR_VERSION = int32(03)
|
||||
MINOR_VERSION = int32(04)
|
||||
VERSION_NUMBER = fmt.Sprintf("%d.%02d", MAJOR_VERSION, MINOR_VERSION)
|
||||
VERSION = util.SizeLimit + " " + VERSION_NUMBER
|
||||
COMMIT = ""
|
||||
|
||||
+144
-29
@@ -74,6 +74,8 @@ type grpcState struct {
|
||||
lastWorkerInfo *types.WorkerData
|
||||
reconnectStop chan struct{}
|
||||
streamExit chan struct{}
|
||||
streamFailed chan struct{} // Signals when stream has failed
|
||||
regWait chan *worker_pb.RegistrationResponse
|
||||
}
|
||||
|
||||
// NewGrpcAdminClient creates a new gRPC admin client
|
||||
@@ -98,6 +100,25 @@ func NewGrpcAdminClient(adminAddress string, workerID string, dialOption grpc.Di
|
||||
return c
|
||||
}
|
||||
|
||||
// drainAndCloseRegWaitChannel drains any pending messages from the regWait channel
|
||||
// and then safely closes it. This prevents losing RegistrationResponse messages
|
||||
// that were sent before the channel is closed.
|
||||
func drainAndCloseRegWaitChannel(ch *chan *worker_pb.RegistrationResponse) {
|
||||
if ch == nil || *ch == nil {
|
||||
return
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-*ch:
|
||||
// continue draining until channel is empty
|
||||
default:
|
||||
close(*ch)
|
||||
*ch = nil
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// safeCloseChannel safely closes a channel and sets it to nil to prevent double-close panics.
|
||||
// NOTE: This function is NOT thread-safe. It is safe to use in this codebase because all calls
|
||||
// are serialized within the managerLoop goroutine. If this function is used in concurrent contexts
|
||||
@@ -140,7 +161,14 @@ out:
|
||||
req.Resp <- nil
|
||||
continue
|
||||
}
|
||||
err := c.sendRegistration(req.Worker)
|
||||
// Capture channel pointers to avoid race condition with reconnect
|
||||
streamFailedCh := state.streamFailed
|
||||
regWaitCh := state.regWait
|
||||
if streamFailedCh == nil || regWaitCh == nil {
|
||||
req.Resp <- fmt.Errorf("stream not ready for registration")
|
||||
continue
|
||||
}
|
||||
err := c.sendRegistration(req.Worker, streamFailedCh, regWaitCh)
|
||||
req.Resp <- err
|
||||
case ActionQueryConnected:
|
||||
respCh := cmd.data.(chan bool)
|
||||
@@ -225,14 +253,18 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
|
||||
// Start stream handlers BEFORE sending registration
|
||||
// This ensures handleIncoming is ready to receive the registration response
|
||||
s.streamExit = make(chan struct{})
|
||||
s.streamFailed = make(chan struct{})
|
||||
s.regWait = make(chan *worker_pb.RegistrationResponse, 1)
|
||||
go handleOutgoing(s.stream, s.streamExit, c.outgoing, c.cmds)
|
||||
go handleIncoming(c.workerID, s.stream, s.streamExit, c.incoming, c.cmds)
|
||||
go handleIncoming(c.workerID, s.stream, s.streamExit, c.incoming, c.cmds, s.streamFailed, s.regWait)
|
||||
|
||||
// Always check for worker info and send registration immediately as the very first message
|
||||
if s.lastWorkerInfo != nil {
|
||||
// Send registration via the normal outgoing channel and wait for response via incoming
|
||||
if err := c.sendRegistration(s.lastWorkerInfo); err != nil {
|
||||
if err := c.sendRegistration(s.lastWorkerInfo, s.streamFailed, s.regWait); err != nil {
|
||||
c.safeCloseChannel(&s.streamExit)
|
||||
c.safeCloseChannel(&s.streamFailed)
|
||||
drainAndCloseRegWaitChannel(&s.regWait)
|
||||
s.streamCancel()
|
||||
s.conn.Close()
|
||||
s.connected = false
|
||||
@@ -252,6 +284,8 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
|
||||
func (c *GrpcAdminClient) reconnect(s *grpcState) error {
|
||||
// Clean up existing connection completely
|
||||
c.safeCloseChannel(&s.streamExit)
|
||||
c.safeCloseChannel(&s.streamFailed)
|
||||
drainAndCloseRegWaitChannel(&s.regWait)
|
||||
if s.streamCancel != nil {
|
||||
s.streamCancel()
|
||||
}
|
||||
@@ -324,32 +358,70 @@ func handleOutgoing(
|
||||
streamExit <-chan struct{},
|
||||
outgoing <-chan *worker_pb.WorkerMessage,
|
||||
cmds chan<- grpcCommand) {
|
||||
|
||||
msgCh := make(chan *worker_pb.WorkerMessage)
|
||||
msgCh := make(chan *worker_pb.WorkerMessage, 1)
|
||||
errCh := make(chan error, 1) // Buffered to prevent blocking if the manager is busy
|
||||
// Goroutine to handle blocking stream.Recv() and simultaneously handle exit
|
||||
// signals
|
||||
|
||||
// Goroutine that reads from msgCh and performs the blocking stream.Send() calls.
|
||||
go func() {
|
||||
for msg := range msgCh {
|
||||
if err := stream.Send(msg); err != nil {
|
||||
errCh <- err
|
||||
return // Exit the receiver goroutine on error/EOF
|
||||
return
|
||||
}
|
||||
}
|
||||
close(errCh)
|
||||
}()
|
||||
|
||||
for msg := range outgoing {
|
||||
select {
|
||||
case msgCh <- msg:
|
||||
case err := <-errCh:
|
||||
// Helper function to handle stream errors and cleanup
|
||||
handleStreamError := func(err error) {
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to send message to admin: %v", err)
|
||||
cmds <- grpcCommand{action: ActionStreamError, data: err}
|
||||
return
|
||||
select {
|
||||
case cmds <- grpcCommand{action: ActionStreamError, data: err}:
|
||||
// Successfully queued
|
||||
default:
|
||||
// Manager busy, queue asynchronously to avoid blocking
|
||||
glog.V(2).Infof("Manager busy, queuing stream error asynchronously from outgoing handler: %v", err)
|
||||
go func(e error) {
|
||||
select {
|
||||
case cmds <- grpcCommand{action: ActionStreamError, data: e}:
|
||||
case <-time.After(2 * time.Second):
|
||||
glog.Warningf("Failed to send stream error to manager from outgoing handler, channel blocked: %v", e)
|
||||
}
|
||||
}(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Helper function to cleanup resources
|
||||
cleanup := func() {
|
||||
close(msgCh)
|
||||
<-errCh
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-streamExit:
|
||||
close(msgCh)
|
||||
<-errCh
|
||||
cleanup()
|
||||
return
|
||||
case err := <-errCh:
|
||||
handleStreamError(err)
|
||||
return
|
||||
case msg, ok := <-outgoing:
|
||||
if !ok {
|
||||
cleanup()
|
||||
return
|
||||
}
|
||||
select {
|
||||
case msgCh <- msg:
|
||||
// Message queued successfully
|
||||
case <-streamExit:
|
||||
cleanup()
|
||||
return
|
||||
case err := <-errCh:
|
||||
handleStreamError(err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -360,10 +432,15 @@ func handleIncoming(
|
||||
stream worker_pb.WorkerService_WorkerStreamClient,
|
||||
streamExit <-chan struct{},
|
||||
incoming chan<- *worker_pb.AdminMessage,
|
||||
cmds chan<- grpcCommand) {
|
||||
cmds chan<- grpcCommand,
|
||||
streamFailed chan<- struct{},
|
||||
regWait chan<- *worker_pb.RegistrationResponse) {
|
||||
glog.V(1).Infof("INCOMING HANDLER STARTED: Worker %s incoming message handler started", workerID)
|
||||
msgCh := make(chan *worker_pb.AdminMessage)
|
||||
errCh := make(chan error, 1) // Buffered to prevent blocking if the manager is busy
|
||||
// regWait is buffered with size 1 so that the registration response can be sent
|
||||
// even if the receiver goroutine has not yet started waiting on the channel.
|
||||
// This non-blocking send pattern avoids a race between sendRegistration and handleIncoming.
|
||||
// Goroutine to handle blocking stream.Recv() and simultaneously handle exit
|
||||
// signals
|
||||
go func() {
|
||||
@@ -385,7 +462,19 @@ func handleIncoming(
|
||||
// Message successfully received from the stream
|
||||
glog.V(4).Infof("MESSAGE RECEIVED: Worker %s received message from admin server: %T", workerID, msg.Message)
|
||||
|
||||
// Route message to waiting goroutines or general handler (original select logic)
|
||||
// If this is a registration response, also publish to the registration waiter.
|
||||
// regWait is buffered (size 1) so that the response can be sent even if sendRegistration
|
||||
// hasn't started waiting yet, preventing a race condition between the two goroutines.
|
||||
if rr := msg.GetRegistrationResponse(); rr != nil {
|
||||
select {
|
||||
case regWait <- rr:
|
||||
glog.V(3).Infof("REGISTRATION RESPONSE: Worker %s routed registration response to waiter", workerID)
|
||||
default:
|
||||
glog.V(2).Infof("REGISTRATION RESPONSE DROPPED: Worker %s registration response dropped (no waiter)", workerID)
|
||||
}
|
||||
}
|
||||
|
||||
// Route message to general handler.
|
||||
select {
|
||||
case incoming <- msg:
|
||||
glog.V(3).Infof("MESSAGE ROUTED: Worker %s successfully routed message to handler", workerID)
|
||||
@@ -401,8 +490,27 @@ func handleIncoming(
|
||||
glog.Errorf("RECEIVE ERROR: Worker %s failed to receive message from admin: %v", workerID, err)
|
||||
}
|
||||
|
||||
// Report the failure as a command to the managerLoop (blocking)
|
||||
cmds <- grpcCommand{action: ActionStreamError, data: err}
|
||||
// Signal that stream has failed (non-blocking)
|
||||
select {
|
||||
case streamFailed <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
|
||||
// Report the failure as a command to the managerLoop.
|
||||
// Try non-blocking first; if the manager is busy and the channel is full,
|
||||
// fall back to an asynchronous blocking send so the error is not lost.
|
||||
select {
|
||||
case cmds <- grpcCommand{action: ActionStreamError, data: err}:
|
||||
default:
|
||||
glog.V(2).Infof("Manager busy, queuing stream error asynchronously: %v", err)
|
||||
go func(e error) {
|
||||
select {
|
||||
case cmds <- grpcCommand{action: ActionStreamError, data: e}:
|
||||
case <-time.After(2 * time.Second):
|
||||
glog.Warningf("Failed to send stream error to manager, channel blocked: %v", e)
|
||||
}
|
||||
}(err)
|
||||
}
|
||||
|
||||
// Exit the main handler loop
|
||||
glog.V(1).Infof("INCOMING HANDLER STOPPED: Worker %s stopping incoming handler due to stream error", workerID)
|
||||
@@ -460,6 +568,8 @@ func (c *GrpcAdminClient) handleDisconnect(cmd grpcCommand, s *grpcState) {
|
||||
|
||||
// Send shutdown signal to stop handlers loop
|
||||
c.safeCloseChannel(&s.streamExit)
|
||||
c.safeCloseChannel(&s.streamFailed)
|
||||
drainAndCloseRegWaitChannel(&s.regWait)
|
||||
|
||||
// Cancel stream context
|
||||
if s.streamCancel != nil {
|
||||
@@ -495,7 +605,7 @@ func (c *GrpcAdminClient) RegisterWorker(worker *types.WorkerData) error {
|
||||
}
|
||||
|
||||
// sendRegistration sends the registration message and waits for response
|
||||
func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData) error {
|
||||
func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData, streamFailed <-chan struct{}, regWait <-chan *worker_pb.RegistrationResponse) error {
|
||||
capabilities := make([]string, len(worker.Capabilities))
|
||||
for i, cap := range worker.Capabilities {
|
||||
capabilities[i] = string(cap)
|
||||
@@ -519,6 +629,8 @@ func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData) error {
|
||||
case c.outgoing <- msg:
|
||||
case <-time.After(5 * time.Second):
|
||||
return fmt.Errorf("failed to send registration message: timeout")
|
||||
case <-streamFailed:
|
||||
return fmt.Errorf("stream failed before registration message could be sent")
|
||||
}
|
||||
|
||||
// Wait for registration response
|
||||
@@ -528,16 +640,19 @@ func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData) error {
|
||||
|
||||
for {
|
||||
select {
|
||||
case response := <-c.incoming:
|
||||
if regResp := response.GetRegistrationResponse(); regResp != nil {
|
||||
if regResp.Success {
|
||||
glog.Infof("Worker registered successfully: %s", regResp.Message)
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("registration failed: %s", regResp.Message)
|
||||
case regResp, ok := <-regWait:
|
||||
if !ok || regResp == nil {
|
||||
return fmt.Errorf("registration failed: channel closed unexpectedly")
|
||||
}
|
||||
if regResp.Success {
|
||||
glog.Infof("Worker registered successfully: %s", regResp.Message)
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("registration failed: %s", regResp.Message)
|
||||
case <-streamFailed:
|
||||
return fmt.Errorf("registration failed: stream closed by server")
|
||||
case <-timeout.C:
|
||||
return fmt.Errorf("registration timeout")
|
||||
return fmt.Errorf("registration failed: timeout waiting for response")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user