protect bpool from buffer pollution by invalid buffers (#20342)

Currently, the bucket events and replication targets are only reloaded
with buckets that failed to load during the first cluster startup,
which is wrong because if one bucket change was done in one node but
that node was not able to notify other nodes; the other nodes will
reload the bucket metadata config but fails to set the events and bucket
targets in the memory.

Dockerfile.hotfix

@@ -1,10 +1,10 @@
-FROM golang:1.21-alpine as build
+FROM golang:1.22-alpine as build
 
 ARG TARGETARCH
 ARG RELEASE
 
-ENV GOPATH /go
-ENV CGO_ENABLED 0
+ENV GOPATH=/go
+ENV CGO_ENABLED=0
 
 # Install curl and minisign
 RUN apk add -U --no-cache ca-certificates && \

Dockerfile.release

@@ -1,10 +1,10 @@
-FROM golang:1.21-alpine as build
+FROM golang:1.22-alpine AS build
 
 ARG TARGETARCH
 ARG RELEASE
 
-ENV GOPATH /go
-ENV CGO_ENABLED 0
+ENV GOPATH=/go
+ENV CGO_ENABLED=0
 
 # Install curl and minisign
 RUN apk add -U --no-cache ca-certificates && \

Dockerfile.release.fips

@@ -1,10 +1,10 @@
-FROM golang:1.21-alpine as build
+FROM golang:1.22-alpine AS build
 
 ARG TARGETARCH
 ARG RELEASE
 
-ENV GOPATH /go
-ENV CGO_ENABLED 0
+ENV GOPATH=/go
+ENV CGO_ENABLED=0
 
 # Install curl and minisign
 RUN apk add -U --no-cache ca-certificates && \

Dockerfile.release.old_cpu

@@ -1,10 +1,10 @@
-FROM golang:1.21-alpine as build
+FROM golang:1.22-alpine AS build
 
 ARG TARGETARCH
 ARG RELEASE
 
-ENV GOPATH /go
-ENV CGO_ENABLED 0
+ENV GOPATH=/go
+ENV CGO_ENABLED=0
 
 # Install curl and minisign
 RUN apk add -U --no-cache ca-certificates && \

cmd/admin-bucket-handlers.go

@@ -811,6 +811,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			}
 
 			bucketMap[bucket].NotificationConfigXML = configData
+			bucketMap[bucket].NotificationConfigUpdatedAt = updatedAt
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketPolicyConfig:
 			// Error out if Content-Length is beyond allowed size.

cmd/bucket-metadata-sys.go

@@ -124,6 +124,7 @@ func (sys *BucketMetadataSys) updateAndParse(ctx context.Context, bucket string,
 		meta.PolicyConfigUpdatedAt = updatedAt
 	case bucketNotificationConfig:
 		meta.NotificationConfigXML = configData
+		meta.NotificationConfigUpdatedAt = updatedAt
 	case bucketLifecycleConfig:
 		meta.LifecycleConfigXML = configData
 		meta.LifecycleConfigUpdatedAt = updatedAt
@@ -153,6 +154,8 @@ func (sys *BucketMetadataSys) updateAndParse(ctx context.Context, bucket string,
 		if err != nil {
 			return updatedAt, fmt.Errorf("Error encrypting bucket target metadata %w", err)
 		}
+		meta.BucketTargetsConfigUpdatedAt = updatedAt
+		meta.BucketTargetsConfigMetaUpdatedAt = updatedAt
 	default:
 		return updatedAt, fmt.Errorf("Unknown bucket %s metadata update requested %s", bucket, configFile)
 	}
@@ -504,7 +507,7 @@ func (sys *BucketMetadataSys) Init(ctx context.Context, buckets []string, objAPI
 }
 
 // concurrently load bucket metadata to speed up loading bucket metadata.
-func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []string, failedBuckets map[string]struct{}) {
+func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []string) {
 	g := errgroup.WithNErrs(len(buckets))
 	bucketMetas := make([]BucketMetadata, len(buckets))
 	for index := range buckets {
@@ -545,10 +548,6 @@ func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []stri
 
 	for i, meta := range bucketMetas {
 		if errs[i] != nil {
-			if failedBuckets == nil {
-				failedBuckets = make(map[string]struct{})
-			}
-			failedBuckets[buckets[i]] = struct{}{}
 			continue
 		}
 		globalEventNotifier.set(buckets[i], meta)   // set notification targets
@@ -556,7 +555,7 @@ func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []stri
 	}
 }
 
-func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context, failedBuckets map[string]struct{}) {
+func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context) {
 	const bucketMetadataRefresh = 15 * time.Minute
 
 	sleeper := newDynamicSleeper(2, 150*time.Millisecond, false)
@@ -586,7 +585,10 @@ func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context, fa
 			for i := range buckets {
 				wait := sleeper.Timer(ctx)
 
-				meta, err := loadBucketMetadata(ctx, sys.objAPI, buckets[i].Name)
+				bucket := buckets[i].Name
+				updated := false
+
+				meta, err := loadBucketMetadata(ctx, sys.objAPI, bucket)
 				if err != nil {
 					internalLogIf(ctx, err, logger.WarningKind)
 					wait() // wait to proceed to next entry.
@@ -594,14 +596,16 @@ func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context, fa
 				}
 
 				sys.Lock()
-				sys.metadataMap[buckets[i].Name] = meta
+				// Update if the bucket metadata in the memory is older than on-disk one
+				if lu := sys.metadataMap[bucket].lastUpdate(); lu.Before(meta.lastUpdate()) {
+					updated = true
+					sys.metadataMap[bucket] = meta
+				}
 				sys.Unlock()
 
-				// Initialize the failed buckets
-				if _, ok := failedBuckets[buckets[i].Name]; ok {
-					globalEventNotifier.set(buckets[i].Name, meta)
-					globalBucketTargetSys.set(buckets[i].Name, meta)
-					delete(failedBuckets, buckets[i].Name)
+				if updated {
+					globalEventNotifier.set(bucket, meta)
+					globalBucketTargetSys.set(bucket, meta)
 				}
 
 				wait() // wait to proceed to next entry.
@@ -622,13 +626,12 @@ func (sys *BucketMetadataSys) Initialized() bool {
 // Loads bucket metadata for all buckets into BucketMetadataSys.
 func (sys *BucketMetadataSys) init(ctx context.Context, buckets []string) {
 	count := globalEndpoints.ESCount() * 10
-	failedBuckets := make(map[string]struct{})
 	for {
 		if len(buckets) < count {
-			sys.concurrentLoad(ctx, buckets, failedBuckets)
+			sys.concurrentLoad(ctx, buckets)
 			break
 		}
-		sys.concurrentLoad(ctx, buckets[:count], failedBuckets)
+		sys.concurrentLoad(ctx, buckets[:count])
 		buckets = buckets[count:]
 	}
 
@@ -637,7 +640,7 @@ func (sys *BucketMetadataSys) init(ctx context.Context, buckets []string) {
 	sys.Unlock()
 
 	if globalIsDistErasure {
-		go sys.refreshBucketsMetadataLoop(ctx, failedBuckets)
+		go sys.refreshBucketsMetadataLoop(ctx)
 	}
 }
 

cmd/bucket-metadata.go

@@ -81,14 +81,19 @@ type BucketMetadata struct {
 	ReplicationConfigXML        []byte
 	BucketTargetsConfigJSON     []byte
 	BucketTargetsConfigMetaJSON []byte
-	PolicyConfigUpdatedAt       time.Time
-	ObjectLockConfigUpdatedAt   time.Time
-	EncryptionConfigUpdatedAt   time.Time
-	TaggingConfigUpdatedAt      time.Time
-	QuotaConfigUpdatedAt        time.Time
-	ReplicationConfigUpdatedAt  time.Time
-	VersioningConfigUpdatedAt   time.Time
-	LifecycleConfigUpdatedAt    time.Time
+
+	PolicyConfigUpdatedAt            time.Time
+	ObjectLockConfigUpdatedAt        time.Time
+	EncryptionConfigUpdatedAt        time.Time
+	TaggingConfigUpdatedAt           time.Time
+	QuotaConfigUpdatedAt             time.Time
+	ReplicationConfigUpdatedAt       time.Time
+	VersioningConfigUpdatedAt        time.Time
+	LifecycleConfigUpdatedAt         time.Time
+	NotificationConfigUpdatedAt      time.Time
+	BucketTargetsConfigUpdatedAt     time.Time
+	BucketTargetsConfigMetaUpdatedAt time.Time
+	// Add a new UpdatedAt field and update lastUpdate function
 
 	// Unexported fields. Must be updated atomically.
 	policyConfig           *policy.BucketPolicy
@@ -120,6 +125,46 @@ func newBucketMetadata(name string) BucketMetadata {
 	}
 }
 
+// Return the last update of this bucket metadata, which
+// means, the last update of any policy document.
+func (b BucketMetadata) lastUpdate() (t time.Time) {
+	if b.PolicyConfigUpdatedAt.After(t) {
+		t = b.PolicyConfigUpdatedAt
+	}
+	if b.ObjectLockConfigUpdatedAt.After(t) {
+		t = b.ObjectLockConfigUpdatedAt
+	}
+	if b.EncryptionConfigUpdatedAt.After(t) {
+		t = b.EncryptionConfigUpdatedAt
+	}
+	if b.TaggingConfigUpdatedAt.After(t) {
+		t = b.TaggingConfigUpdatedAt
+	}
+	if b.QuotaConfigUpdatedAt.After(t) {
+		t = b.QuotaConfigUpdatedAt
+	}
+	if b.ReplicationConfigUpdatedAt.After(t) {
+		t = b.ReplicationConfigUpdatedAt
+	}
+	if b.VersioningConfigUpdatedAt.After(t) {
+		t = b.VersioningConfigUpdatedAt
+	}
+	if b.LifecycleConfigUpdatedAt.After(t) {
+		t = b.LifecycleConfigUpdatedAt
+	}
+	if b.NotificationConfigUpdatedAt.After(t) {
+		t = b.NotificationConfigUpdatedAt
+	}
+	if b.BucketTargetsConfigUpdatedAt.After(t) {
+		t = b.BucketTargetsConfigUpdatedAt
+	}
+	if b.BucketTargetsConfigMetaUpdatedAt.After(t) {
+		t = b.BucketTargetsConfigMetaUpdatedAt
+	}
+
+	return
+}
+
 // Versioning returns true if versioning is enabled
 func (b BucketMetadata) Versioning() bool {
 	return b.LockEnabled || (b.versioningConfig != nil && b.versioningConfig.Enabled()) || (b.objectLockConfig != nil && b.objectLockConfig.Enabled())
@@ -440,6 +485,18 @@ func (b *BucketMetadata) defaultTimestamps() {
 	if b.LifecycleConfigUpdatedAt.IsZero() {
 		b.LifecycleConfigUpdatedAt = b.Created
 	}
+
+	if b.NotificationConfigUpdatedAt.IsZero() {
+		b.NotificationConfigUpdatedAt = b.Created
+	}
+
+	if b.BucketTargetsConfigUpdatedAt.IsZero() {
+		b.BucketTargetsConfigUpdatedAt = b.Created
+	}
+
+	if b.BucketTargetsConfigMetaUpdatedAt.IsZero() {
+		b.BucketTargetsConfigMetaUpdatedAt = b.Created
+	}
 }
 
 // Save config to supplied ObjectLayer api.

cmd/bucket-metadata_gen.go

@@ -156,6 +156,24 @@ func (z *BucketMetadata) DecodeMsg(dc *msgp.Reader) (err error) {
 				err = msgp.WrapError(err, "LifecycleConfigUpdatedAt")
 				return
 			}
+		case "NotificationConfigUpdatedAt":
+			z.NotificationConfigUpdatedAt, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "NotificationConfigUpdatedAt")
+				return
+			}
+		case "BucketTargetsConfigUpdatedAt":
+			z.BucketTargetsConfigUpdatedAt, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "BucketTargetsConfigUpdatedAt")
+				return
+			}
+		case "BucketTargetsConfigMetaUpdatedAt":
+			z.BucketTargetsConfigMetaUpdatedAt, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "BucketTargetsConfigMetaUpdatedAt")
+				return
+			}
 		default:
 			err = dc.Skip()
 			if err != nil {
@@ -169,9 +187,9 @@ func (z *BucketMetadata) DecodeMsg(dc *msgp.Reader) (err error) {
 
 // EncodeMsg implements msgp.Encodable
 func (z *BucketMetadata) EncodeMsg(en *msgp.Writer) (err error) {
-	// map header, size 22
+	// map header, size 25
 	// write "Name"
-	err = en.Append(0xde, 0x0, 0x16, 0xa4, 0x4e, 0x61, 0x6d, 0x65)
+	err = en.Append(0xde, 0x0, 0x19, 0xa4, 0x4e, 0x61, 0x6d, 0x65)
 	if err != nil {
 		return
 	}
@@ -390,15 +408,45 @@ func (z *BucketMetadata) EncodeMsg(en *msgp.Writer) (err error) {
 		err = msgp.WrapError(err, "LifecycleConfigUpdatedAt")
 		return
 	}
+	// write "NotificationConfigUpdatedAt"
+	err = en.Append(0xbb, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.NotificationConfigUpdatedAt)
+	if err != nil {
+		err = msgp.WrapError(err, "NotificationConfigUpdatedAt")
+		return
+	}
+	// write "BucketTargetsConfigUpdatedAt"
+	err = en.Append(0xbc, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.BucketTargetsConfigUpdatedAt)
+	if err != nil {
+		err = msgp.WrapError(err, "BucketTargetsConfigUpdatedAt")
+		return
+	}
+	// write "BucketTargetsConfigMetaUpdatedAt"
+	err = en.Append(0xd9, 0x20, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x4d, 0x65, 0x74, 0x61, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.BucketTargetsConfigMetaUpdatedAt)
+	if err != nil {
+		err = msgp.WrapError(err, "BucketTargetsConfigMetaUpdatedAt")
+		return
+	}
 	return
 }
 
 // MarshalMsg implements msgp.Marshaler
 func (z *BucketMetadata) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 22
+	// map header, size 25
 	// string "Name"
-	o = append(o, 0xde, 0x0, 0x16, 0xa4, 0x4e, 0x61, 0x6d, 0x65)
+	o = append(o, 0xde, 0x0, 0x19, 0xa4, 0x4e, 0x61, 0x6d, 0x65)
 	o = msgp.AppendString(o, z.Name)
 	// string "Created"
 	o = append(o, 0xa7, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64)
@@ -463,6 +511,15 @@ func (z *BucketMetadata) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "LifecycleConfigUpdatedAt"
 	o = append(o, 0xb8, 0x4c, 0x69, 0x66, 0x65, 0x63, 0x79, 0x63, 0x6c, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
 	o = msgp.AppendTime(o, z.LifecycleConfigUpdatedAt)
+	// string "NotificationConfigUpdatedAt"
+	o = append(o, 0xbb, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	o = msgp.AppendTime(o, z.NotificationConfigUpdatedAt)
+	// string "BucketTargetsConfigUpdatedAt"
+	o = append(o, 0xbc, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	o = msgp.AppendTime(o, z.BucketTargetsConfigUpdatedAt)
+	// string "BucketTargetsConfigMetaUpdatedAt"
+	o = append(o, 0xd9, 0x20, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x4d, 0x65, 0x74, 0x61, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	o = msgp.AppendTime(o, z.BucketTargetsConfigMetaUpdatedAt)
 	return
 }
 
@@ -616,6 +673,24 @@ func (z *BucketMetadata) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "LifecycleConfigUpdatedAt")
 				return
 			}
+		case "NotificationConfigUpdatedAt":
+			z.NotificationConfigUpdatedAt, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "NotificationConfigUpdatedAt")
+				return
+			}
+		case "BucketTargetsConfigUpdatedAt":
+			z.BucketTargetsConfigUpdatedAt, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "BucketTargetsConfigUpdatedAt")
+				return
+			}
+		case "BucketTargetsConfigMetaUpdatedAt":
+			z.BucketTargetsConfigMetaUpdatedAt, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "BucketTargetsConfigMetaUpdatedAt")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -630,6 +705,6 @@ func (z *BucketMetadata) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *BucketMetadata) Msgsize() (s int) {
-	s = 3 + 5 + msgp.StringPrefixSize + len(z.Name) + 8 + msgp.TimeSize + 12 + msgp.BoolSize + 17 + msgp.BytesPrefixSize + len(z.PolicyConfigJSON) + 22 + msgp.BytesPrefixSize + len(z.NotificationConfigXML) + 19 + msgp.BytesPrefixSize + len(z.LifecycleConfigXML) + 20 + msgp.BytesPrefixSize + len(z.ObjectLockConfigXML) + 20 + msgp.BytesPrefixSize + len(z.VersioningConfigXML) + 20 + msgp.BytesPrefixSize + len(z.EncryptionConfigXML) + 17 + msgp.BytesPrefixSize + len(z.TaggingConfigXML) + 16 + msgp.BytesPrefixSize + len(z.QuotaConfigJSON) + 21 + msgp.BytesPrefixSize + len(z.ReplicationConfigXML) + 24 + msgp.BytesPrefixSize + len(z.BucketTargetsConfigJSON) + 28 + msgp.BytesPrefixSize + len(z.BucketTargetsConfigMetaJSON) + 22 + msgp.TimeSize + 26 + msgp.TimeSize + 26 + msgp.TimeSize + 23 + msgp.TimeSize + 21 + msgp.TimeSize + 27 + msgp.TimeSize + 26 + msgp.TimeSize + 25 + msgp.TimeSize
+	s = 3 + 5 + msgp.StringPrefixSize + len(z.Name) + 8 + msgp.TimeSize + 12 + msgp.BoolSize + 17 + msgp.BytesPrefixSize + len(z.PolicyConfigJSON) + 22 + msgp.BytesPrefixSize + len(z.NotificationConfigXML) + 19 + msgp.BytesPrefixSize + len(z.LifecycleConfigXML) + 20 + msgp.BytesPrefixSize + len(z.ObjectLockConfigXML) + 20 + msgp.BytesPrefixSize + len(z.VersioningConfigXML) + 20 + msgp.BytesPrefixSize + len(z.EncryptionConfigXML) + 17 + msgp.BytesPrefixSize + len(z.TaggingConfigXML) + 16 + msgp.BytesPrefixSize + len(z.QuotaConfigJSON) + 21 + msgp.BytesPrefixSize + len(z.ReplicationConfigXML) + 24 + msgp.BytesPrefixSize + len(z.BucketTargetsConfigJSON) + 28 + msgp.BytesPrefixSize + len(z.BucketTargetsConfigMetaJSON) + 22 + msgp.TimeSize + 26 + msgp.TimeSize + 26 + msgp.TimeSize + 23 + msgp.TimeSize + 21 + msgp.TimeSize + 27 + msgp.TimeSize + 26 + msgp.TimeSize + 25 + msgp.TimeSize + 28 + msgp.TimeSize + 29 + msgp.TimeSize + 34 + msgp.TimeSize
 	return
 }

cmd/erasure-decode.go

@@ -48,14 +48,14 @@ func newParallelReader(readers []io.ReaderAt, e Erasure, offset, totalLength int
 		r2b[i] = i
 	}
 	bufs := make([][]byte, len(readers))
-	// Fill buffers
-	b := globalBytePoolCap.Load().Get()
 	shardSize := int(e.ShardSize())
-	if cap(b) < len(readers)*shardSize {
-		// We should always have enough capacity, but older objects may be bigger.
-		globalBytePoolCap.Load().Put(b)
-		b = nil
-	} else {
+	var b []byte
+
+	// We should always have enough capacity, but older objects may be bigger
+	// we do not need stashbuffer for them.
+	if globalBytePoolCap.Load().WidthCap() >= len(readers)*shardSize {
+		// Fill buffers
+		b = globalBytePoolCap.Load().Get()
 		// Seed the buffers.
 		for i := range bufs {
 			bufs[i] = b[i*shardSize : (i+1)*shardSize]

cmd/iam.go

@@ -417,7 +417,7 @@ func (sys *IAMSys) periodicRoutines(ctx context.Context, baseInterval time.Durat
 				took := time.Since(refreshStart).Seconds()
 				if took > maxDurationSecondsForLog {
 					// Log if we took a lot of time to load.
-					logger.Info("IAM refresh took (duration: %s)", took)
+					logger.Info("IAM refresh took (duration: %.2fs)", took)
 				}
 			}
 

cmd/object-api-deleteobject_test.go

@@ -93,14 +93,18 @@ func testDeleteObject(obj ObjectLayer, instanceType string, t TestErrHandler) {
 			md5Bytes := md5.Sum([]byte(object.content))
 			oi, err := obj.PutObject(context.Background(), testCase.bucketName, object.name, mustGetPutObjReader(t, strings.NewReader(object.content),
 				int64(len(object.content)), hex.EncodeToString(md5Bytes[:]), ""), ObjectOptions{})
-			t.Log(oi)
 			if err != nil {
+				t.Log(oi)
 				t.Fatalf("%s : %s", instanceType, err.Error())
 			}
 		}
 
 		oi, err := obj.DeleteObject(context.Background(), testCase.bucketName, testCase.pathToDelete, ObjectOptions{})
-		t.Log(oi, err)
+		if err != nil && !isErrObjectNotFound(err) {
+			t.Log(oi)
+			t.Errorf("Test %d: %s:  Expected to pass, but failed with: <ERROR> %s", i+1, instanceType, err)
+			continue
+		}
 
 		result, err := obj.ListObjects(context.Background(), testCase.bucketName, "", "", "", 1000)
 		if err != nil {

cmd/xl-storage.go

@@ -118,7 +118,8 @@ type xlStorage struct {
 	major, minor uint32
 	fsType       string
 
-	immediatePurge chan string
+	immediatePurge       chan string
+	immediatePurgeCancel context.CancelFunc
 
 	// mutex to prevent concurrent read operations overloading walks.
 	rotational bool
@@ -216,17 +217,21 @@ func newXLStorage(ep Endpoint, cleanUp bool) (s *xlStorage, err error) {
 	if globalIsTesting || globalIsCICD {
 		immediatePurgeQueue = 1
 	}
+
+	ctx, cancel := context.WithCancel(GlobalContext)
+
 	s = &xlStorage{
-		drivePath:      ep.Path,
-		endpoint:       ep,
-		globalSync:     globalFSOSync,
-		diskInfoCache:  cachevalue.New[DiskInfo](),
-		immediatePurge: make(chan string, immediatePurgeQueue),
+		drivePath:            ep.Path,
+		endpoint:             ep,
+		globalSync:           globalFSOSync,
+		diskInfoCache:        cachevalue.New[DiskInfo](),
+		immediatePurge:       make(chan string, immediatePurgeQueue),
+		immediatePurgeCancel: cancel,
 	}
 
 	defer func() {
-		if err == nil {
-			go s.cleanupTrashImmediateCallers(GlobalContext)
+		if cleanUp && err == nil {
+			go s.cleanupTrashImmediateCallers(ctx)
 		}
 	}()
 
@@ -399,7 +404,8 @@ func (s *xlStorage) Endpoint() Endpoint {
 	return s.endpoint
 }
 
-func (*xlStorage) Close() error {
+func (s *xlStorage) Close() error {
+	s.immediatePurgeCancel()
 	return nil
 }
 
@@ -1202,7 +1208,12 @@ func (s *xlStorage) cleanupTrashImmediateCallers(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case entry := <-s.immediatePurge:
-			removeAll(entry)
+			// Add deadlines such that immediate purge is not
+			// perpetually hung here.
+			w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
+			w.Run(func() error {
+				return removeAll(entry)
+			})
 		}
 	}
 }

internal/bpool/bpool.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2015-2023 MinIO, Inc.
+// Copyright (c) 2015-2024 MinIO, Inc.
 //
 // This file is part of MinIO Object Storage stack
 //
@@ -17,7 +17,9 @@
 
 package bpool
 
-import "github.com/klauspost/reedsolomon"
+import (
+	"github.com/klauspost/reedsolomon"
+)
 
 // BytePoolCap implements a leaky pool of []byte in the form of a bounded channel.
 type BytePoolCap struct {
@@ -29,11 +31,14 @@ type BytePoolCap struct {
 // NewBytePoolCap creates a new BytePool bounded to the given maxSize, with new
 // byte arrays sized based on width.
 func NewBytePoolCap(maxSize uint64, width int, capwidth int) (bp *BytePoolCap) {
-	if capwidth > 0 && capwidth < 64 {
+	if capwidth <= 0 {
+		panic("total buffer capacity must be provided")
+	}
+	if capwidth < 64 {
 		panic("buffer capped with smaller than 64 bytes is not supported")
 	}
-	if capwidth > 0 && width > capwidth {
-		panic("buffer length cannot be > capacity of the buffer")
+	if width > capwidth {
+		panic("minimum buffer length cannot be > capacity of the buffer")
 	}
 	return &BytePoolCap{
 		c:    make(chan []byte, maxSize),
@@ -60,11 +65,7 @@ func (bp *BytePoolCap) Get() (b []byte) {
 		// reuse existing buffer
 	default:
 		// create new aligned buffer
-		if bp.wcap > 0 {
-			b = reedsolomon.AllocAligned(1, bp.wcap)[0][:bp.w]
-		} else {
-			b = reedsolomon.AllocAligned(1, bp.w)[0]
-		}
+		b = reedsolomon.AllocAligned(1, bp.wcap)[0][:bp.w]
 	}
 	return
 }
@@ -74,8 +75,17 @@ func (bp *BytePoolCap) Put(b []byte) {
 	if bp == nil {
 		return
 	}
+
+	if cap(b) != bp.wcap {
+		// someone tried to put back buffer which is not part of this buffer pool
+		// we simply don't put this back into pool, a modified buffer provided
+		// by this package is no more usable, callers make sure to not modify
+		// the capacity of the buffer.
+		return
+	}
+
 	select {
-	case bp.c <- b:
+	case bp.c <- b[:bp.w]:
 		// buffer went back into pool
 	default:
 		// buffer didn't go back into pool, just discard
@@ -97,3 +107,11 @@ func (bp *BytePoolCap) WidthCap() (n int) {
 	}
 	return bp.wcap
 }
+
+// CurrentSize returns current size of buffer pool
+func (bp *BytePoolCap) CurrentSize() int {
+	if bp == nil {
+		return 0
+	}
+	return len(bp.c) * bp.w
+}

internal/bpool/bpool_test.go

@@ -17,7 +17,9 @@
 
 package bpool
 
-import "testing"
+import (
+	"testing"
+)
 
 // Tests - bytePool functionality.
 func TestBytePool(t *testing.T) {
@@ -25,20 +27,20 @@ func TestBytePool(t *testing.T) {
 	width := 1024
 	capWidth := 2048
 
-	bufPool := NewBytePoolCap(size, width, capWidth)
+	bp := NewBytePoolCap(size, width, capWidth)
 
 	// Check the width
-	if bufPool.Width() != width {
-		t.Fatalf("bytepool width invalid: got %v want %v", bufPool.Width(), width)
+	if bp.Width() != width {
+		t.Fatalf("bytepool width invalid: got %v want %v", bp.Width(), width)
 	}
 
 	// Check with width cap
-	if bufPool.WidthCap() != capWidth {
-		t.Fatalf("bytepool capWidth invalid: got %v want %v", bufPool.WidthCap(), capWidth)
+	if bp.WidthCap() != capWidth {
+		t.Fatalf("bytepool capWidth invalid: got %v want %v", bp.WidthCap(), capWidth)
 	}
 
 	// Check that retrieved buffer are of the expected width
-	b := bufPool.Get()
+	b := bp.Get()
 	if len(b) != width {
 		t.Fatalf("bytepool length invalid: got %v want %v", len(b), width)
 	}
@@ -46,14 +48,14 @@ func TestBytePool(t *testing.T) {
 		t.Fatalf("bytepool cap invalid: got %v want %v", cap(b), capWidth)
 	}
 
-	bufPool.Put(b)
+	bp.Put(b)
 
 	// Fill the pool beyond the capped pool size.
 	for i := uint64(0); i < size*2; i++ {
-		bufPool.Put(make([]byte, bufPool.w))
+		bp.Put(make([]byte, bp.w, bp.wcap))
 	}
 
-	b = bufPool.Get()
+	b = bp.Get()
 	if len(b) != width {
 		t.Fatalf("bytepool length invalid: got %v want %v", len(b), width)
 	}
@@ -61,31 +63,37 @@ func TestBytePool(t *testing.T) {
 		t.Fatalf("bytepool length invalid: got %v want %v", cap(b), capWidth)
 	}
 
-	bufPool.Put(b)
-
-	// Close the channel so we can iterate over it.
-	close(bufPool.c)
+	bp.Put(b)
 
 	// Check the size of the pool.
-	if uint64(len(bufPool.c)) != size {
-		t.Fatalf("bytepool size invalid: got %v want %v", len(bufPool.c), size)
+	if uint64(len(bp.c)) != size {
+		t.Fatalf("bytepool size invalid: got %v want %v", len(bp.c), size)
 	}
 
-	bufPoolNoCap := NewBytePoolCap(size, width, 0)
-	// Check the width
-	if bufPoolNoCap.Width() != width {
-		t.Fatalf("bytepool width invalid: got %v want %v", bufPool.Width(), width)
+	// lets drain the buf channel first before we validate invalid buffers.
+	for i := uint64(0); i < size; i++ {
+		bp.Get() // discard
 	}
 
-	// Check with width cap
-	if bufPoolNoCap.WidthCap() != 0 {
-		t.Fatalf("bytepool capWidth invalid: got %v want %v", bufPool.WidthCap(), 0)
+	// Try putting some invalid buffers into pool
+	bp.Put(make([]byte, bp.w, bp.wcap-1)) // wrong capacity is rejected (less)
+	bp.Put(make([]byte, bp.w, bp.wcap+1)) // wrong capacity is rejected (more)
+	bp.Put(make([]byte, width))           // wrong capacity is rejected (very less)
+	if len(bp.c) > 0 {
+		t.Fatal("bytepool should have rejected invalid packets")
 	}
-	b = bufPoolNoCap.Get()
+
+	// Try putting a short slice into pool
+	bp.Put(make([]byte, bp.w, bp.wcap)[:2])
+	if len(bp.c) != 1 {
+		t.Fatal("bytepool should have accepted short slice with sufficient capacity")
+	}
+
+	b = bp.Get()
 	if len(b) != width {
 		t.Fatalf("bytepool length invalid: got %v want %v", len(b), width)
 	}
-	if cap(b) != width {
-		t.Fatalf("bytepool length invalid: got %v want %v", cap(b), width)
-	}
+
+	// Close the channel.
+	close(bp.c)
 }

internal/http/headers.go

@@ -259,5 +259,6 @@ const (
 // http headers sent to webhook targets
 const (
 	// Reports the version of MinIO server
-	MinIOVersion = "x-minio-version"
+	MinIOVersion             = "x-minio-version"
+	WebhookEventPayloadCount = "x-minio-webhook-payload-count"
 )

internal/logger/target/http/http.go

@@ -26,6 +26,8 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -42,7 +44,7 @@ import (
 
 const (
 	// Timeout for the webhook http call
-	webhookCallTimeout = 3 * time.Second
+	webhookCallTimeout = 5 * time.Second
 
 	// maxWorkers is the maximum number of concurrent http loggers
 	maxWorkers = 16
@@ -219,17 +221,23 @@ func (h *Target) initMemoryStore(ctx context.Context) (err error) {
 	return nil
 }
 
-func (h *Target) send(ctx context.Context, payload []byte, payloadType string, timeout time.Duration) (err error) {
+func (h *Target) send(ctx context.Context, payload []byte, payloadCount int, payloadType string, timeout time.Duration) (err error) {
+	h.failedMessages.Add(int64(payloadCount))
+
 	defer func() {
 		if err != nil {
-			h.status.Store(statusOffline)
+			if xnet.IsNetworkOrHostDown(err, false) {
+				h.status.Store(statusOffline)
+			}
 		} else {
+			h.failedMessages.Add(-int64(payloadCount))
 			h.status.Store(statusOnline)
 		}
 	}()
 
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
+
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
 		h.Endpoint(), bytes.NewReader(payload))
 	if err != nil {
@@ -238,6 +246,7 @@ func (h *Target) send(ctx context.Context, payload []byte, payloadType string, t
 	if payloadType != "" {
 		req.Header.Set(xhttp.ContentType, payloadType)
 	}
+	req.Header.Set(xhttp.WebhookEventPayloadCount, strconv.Itoa(payloadCount))
 	req.Header.Set(xhttp.MinIOVersion, xhttp.GlobalMinIOVersion)
 	req.Header.Set(xhttp.MinioDeploymentID, xhttp.GlobalDeploymentID)
 
@@ -353,6 +362,7 @@ func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
 			return
 		}
 
+		var count int
 		if !isTick {
 			h.totalMessages.Add(1)
 
@@ -366,13 +376,15 @@ func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
 					h.failedMessages.Add(1)
 					continue
 				}
+				count++
+			} else {
+				entries = append(entries, entry)
+				count++
 			}
-
-			entries = append(entries, entry)
 		}
 
-		if len(entries) != h.batchSize {
-			if len(h.logCh) > 0 || len(globalBuffer) > 0 || len(entries) == 0 {
+		if count != h.batchSize {
+			if len(h.logCh) > 0 || len(globalBuffer) > 0 || count == 0 {
 				continue
 			}
 
@@ -406,16 +418,15 @@ func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
 		}
 
 		if !isDirQueue {
-			err = h.send(ctx, buf.Bytes(), h.payloadType, webhookCallTimeout)
+			err = h.send(ctx, buf.Bytes(), count, h.payloadType, webhookCallTimeout)
 		} else {
 			err = h.store.PutMultiple(entries)
 		}
 
 		if err != nil {
-
 			h.config.LogOnceIf(
 				context.Background(),
-				fmt.Errorf("unable to send webhook log entry to '%s' err '%w'", name, err),
+				fmt.Errorf("unable to send webhook log entry(s) to '%s' err '%w': %d", name, err, count),
 				name,
 			)
 
@@ -428,7 +439,7 @@ func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
 		}
 
 		entries = make([]interface{}, 0)
-
+		count = 0
 		if !isDirQueue {
 			buf.Reset()
 		}
@@ -529,14 +540,16 @@ func (h *Target) SendFromStore(key store.Key) (err error) {
 		return err
 	}
 
-	h.failedMessages.Add(1)
-	defer func() {
-		if err == nil {
-			h.failedMessages.Add(-1)
+	count := 1
+	v := strings.Split(key.Name, ":")
+	if len(v) == 2 {
+		count, err = strconv.Atoi(v[0])
+		if err != nil {
+			return err
 		}
-	}()
+	}
 
-	if err := h.send(context.Background(), eventData, h.payloadType, webhookCallTimeout); err != nil {
+	if err := h.send(context.Background(), eventData, count, h.payloadType, webhookCallTimeout); err != nil {
 		return err
 	}
 

internal/store/queuestore.go

@@ -20,6 +20,7 @@ package store
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
 	"sort"
@@ -107,17 +108,18 @@ func (store *QueueStore[_]) Delete() error {
 
 // PutMultiple - puts an item to the store.
 func (store *QueueStore[I]) PutMultiple(item []I) error {
-	store.Lock()
-	defer store.Unlock()
-	if uint64(len(store.entries)) >= store.entryLimit {
-		return errLimitExceeded
-	}
 	// Generate a new UUID for the key.
 	key, err := uuid.NewRandom()
 	if err != nil {
 		return err
 	}
-	return store.multiWrite(key.String(), item)
+
+	store.Lock()
+	defer store.Unlock()
+	if uint64(len(store.entries)) >= store.entryLimit {
+		return errLimitExceeded
+	}
+	return store.multiWrite(fmt.Sprintf("%d:%s", len(item), key.String()), item)
 }
 
 // multiWrite - writes an item to the directory.