Fix healing on multiple zones (#8555)

It is expected in zone healing underlying
callers should return appropriate errors

cmd/bucket-handlers.go

@@ -564,6 +564,7 @@ func (api objectAPIHandlers) PutBucketHandler(w http.ResponseWriter, r *http.Req
 			return
 		}
 		globalBucketObjectLockConfig.Set(bucket, Retention{})
+		globalNotificationSys.PutBucketObjectLockConfig(ctx, bucket, Retention{})
 	}
 
 	// Make sure to add Location information here only for bucket
@@ -896,12 +897,7 @@ func (api objectAPIHandlers) DeleteBucketHandler(w http.ResponseWriter, r *http.
 		}
 	}
 
-	globalBucketObjectLockConfig.Delete(bucket)
-	globalNotificationSys.RemoveNotification(bucket)
-	globalPolicySys.Remove(bucket)
 	globalNotificationSys.DeleteBucket(ctx, bucket)
-	globalLifecycleSys.Remove(bucket)
-	globalNotificationSys.RemoveBucketLifecycle(ctx, bucket)
 
 	// Write success response.
 	writeSuccessNoContent(w)
@@ -1026,8 +1022,8 @@ func (api objectAPIHandlers) PutBucketObjectLockConfigHandler(w http.ResponseWri
 		globalBucketObjectLockConfig.Set(bucket, retention)
 		globalNotificationSys.PutBucketObjectLockConfig(ctx, bucket, retention)
 	} else {
-		globalBucketObjectLockConfig.Set(bucket, Retention{})
+		globalBucketObjectLockConfig.Remove(bucket)
-		globalBucketObjectLockConfig.Delete(bucket)
+		globalNotificationSys.RemoveBucketObjectLockConfig(ctx, bucket)
 	}
 
 	// Write success response.

cmd/notification.go

@@ -525,6 +525,11 @@ func (sys *NotificationSys) SetBucketPolicy(ctx context.Context, bucketName stri
 
 // DeleteBucket - calls DeleteBucket RPC call on all peers.
 func (sys *NotificationSys) DeleteBucket(ctx context.Context, bucketName string) {
+	globalNotificationSys.RemoveNotification(bucketName)
+	globalBucketObjectLockConfig.Remove(bucketName)
+	globalPolicySys.Remove(bucketName)
+	globalLifecycleSys.Remove(bucketName)
+
 	go func() {
 		ng := WithNPeers(len(sys.peerClients))
 		for idx, client := range sys.peerClients {
@@ -557,6 +562,23 @@ func (sys *NotificationSys) RemoveBucketPolicy(ctx context.Context, bucketName s
 	}()
 }
 
+// RemoveBucketObjectLockConfig - calls RemoveBucketObjectLockConfig RPC call on all peers.
+func (sys *NotificationSys) RemoveBucketObjectLockConfig(ctx context.Context, bucketName string) {
+	go func() {
+		ng := WithNPeers(len(sys.peerClients))
+		for idx, client := range sys.peerClients {
+			if client == nil {
+				continue
+			}
+			client := client
+			ng.Go(ctx, func() error {
+				return client.RemoveBucketObjectLockConfig(bucketName)
+			}, idx, *client.host)
+		}
+		ng.Wait()
+	}()
+}
+
 // SetBucketLifecycle - calls SetBucketLifecycle on all peers.
 func (sys *NotificationSys) SetBucketLifecycle(ctx context.Context, bucketName string,
 	bucketLifecycle *lifecycle.Lifecycle) {
@@ -984,21 +1006,22 @@ func (sys *NotificationSys) Send(args eventArgs) []event.TargetIDErr {
 
 // PutBucketObjectLockConfig - put bucket object lock configuration to all peers.
 func (sys *NotificationSys) PutBucketObjectLockConfig(ctx context.Context, bucketName string, retention Retention) {
-	var wg sync.WaitGroup
+	g := errgroup.WithNErrs(len(sys.peerClients))
-	for _, client := range sys.peerClients {
+	for index, client := range sys.peerClients {
 		if client == nil {
 			continue
 		}
-		wg.Add(1)
+		index := index
-		go func(client *peerRESTClient) {
+		g.Go(func() error {
-			defer wg.Done()
+			return sys.peerClients[index].PutBucketObjectLockConfig(bucketName, retention)
-			if err := client.PutBucketObjectLockConfig(bucketName, retention); err != nil {
+		}, index)
-				logger.GetReqInfo(ctx).AppendTags("remotePeer", client.host.Name)
+	}
+	for i, err := range g.Wait() {
+		if err != nil {
+			logger.GetReqInfo(ctx).AppendTags("remotePeer", sys.peerClients[i].host.String())
 			logger.LogIf(ctx, err)
 		}
-		}(client)
 	}
-	wg.Wait()
 }
 
 // NetReadPerfInfo - Network read performance information.

cmd/object-lock.go

@@ -118,8 +118,8 @@ func (config *BucketObjectLockConfig) Get(bucketName string) (r Retention, ok bo
 	return r, ok
 }
 
-// Delete - delete retention configuration.
+// Remove - removes retention configuration.
-func (config *BucketObjectLockConfig) Delete(bucketName string) {
+func (config *BucketObjectLockConfig) Remove(bucketName string) {
 	config.Lock()
 	delete(config.retentionMap, bucketName)
 	config.Unlock()
@@ -414,6 +414,7 @@ func checkGovernanceBypassAllowed(ctx context.Context, r *http.Request, bucket,
 	if err != nil {
 		// ignore case where object no longer exists
 		if toAPIError(ctx, err).Code == "NoSuchKey" {
+			oi.UserDefined = map[string]string{}
 			return oi, ErrNone
 		}
 		return oi, toAPIErrorCode(ctx, err)

cmd/peer-rest-client.go

@@ -361,6 +361,18 @@ func (client *peerRESTClient) RemoveBucketPolicy(bucket string) error {
 	return nil
 }
 
+// RemoveBucketObjectLockConfig - Remove bucket object lock config on the peer node.
+func (client *peerRESTClient) RemoveBucketObjectLockConfig(bucket string) error {
+	values := make(url.Values)
+	values.Set(peerRESTBucket, bucket)
+	respBody, err := client.call(peerRESTMethodBucketObjectLockConfigRemove, values, nil, -1)
+	if err != nil {
+		return err
+	}
+	defer http.DrainBody(respBody)
+	return nil
+}
+
 // SetBucketPolicy - Set bucket policy on the peer node.
 func (client *peerRESTClient) SetBucketPolicy(bucket string, bucketPolicy *policy.Policy) error {
 	values := make(url.Values)

cmd/peer-rest-common.go

@@ -59,6 +59,7 @@ const (
 	peerRESTMethodHardwareCPUInfo              = "/cpuhardwareinfo"
 	peerRESTMethodHardwareNetworkInfo          = "/networkhardwareinfo"
 	peerRESTMethodPutBucketObjectLockConfig    = "/putbucketobjectlockconfig"
+	peerRESTMethodBucketObjectLockConfigRemove = "/removebucketobjectlockconfig"
 )
 
 const (

cmd/peer-rest-server.go

@@ -510,7 +510,8 @@ func (s *peerRESTServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Requ
 
 	globalNotificationSys.RemoveNotification(bucketName)
 	globalPolicySys.Remove(bucketName)
-	globalBucketObjectLockConfig.Delete(bucketName)
+	globalBucketObjectLockConfig.Remove(bucketName)
+	globalLifecycleSys.Remove(bucketName)
 
 	w.(http.Flusher).Flush()
 }
@@ -761,6 +762,24 @@ func (s *peerRESTServer) PutBucketNotificationHandler(w http.ResponseWriter, r *
 	w.(http.Flusher).Flush()
 }
 
+// RemoveBucketObjectLockConfigHandler - handles DELETE bucket object lock configuration.
+func (s *peerRESTServer) RemoveBucketObjectLockConfigHandler(w http.ResponseWriter, r *http.Request) {
+	if !s.IsValid(w, r) {
+		s.writeErrorResponse(w, errors.New("Invalid request"))
+		return
+	}
+
+	vars := mux.Vars(r)
+	bucketName := vars[peerRESTBucket]
+	if bucketName == "" {
+		s.writeErrorResponse(w, errors.New("Bucket name is missing"))
+		return
+	}
+
+	globalBucketObjectLockConfig.Remove(bucketName)
+	w.(http.Flusher).Flush()
+}
+
 // PutBucketObjectLockConfigHandler - handles PUT bucket object lock configuration.
 func (s *peerRESTServer) PutBucketObjectLockConfigHandler(w http.ResponseWriter, r *http.Request) {
 	if !s.IsValid(w, r) {
@@ -1027,7 +1046,6 @@ func (s *peerRESTServer) IsValid(w http.ResponseWriter, r *http.Request) bool {
 func registerPeerRESTHandlers(router *mux.Router) {
 	server := &peerRESTServer{}
 	subrouter := router.PathPrefix(peerRESTPrefix).Subrouter()
-	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodPutBucketObjectLockConfig).HandlerFunc(httpTraceHdrs(server.PutBucketObjectLockConfigHandler)).Queries(restQueries(peerRESTBucket)...)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodNetReadPerfInfo).HandlerFunc(httpTraceHdrs(server.NetReadPerfInfoHandler)).Queries(restQueries(peerRESTNetPerfSize)...)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodCollectNetPerfInfo).HandlerFunc(httpTraceHdrs(server.CollectNetPerfInfoHandler)).Queries(restQueries(peerRESTNetPerfSize)...)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodGetLocks).HandlerFunc(httpTraceHdrs(server.GetLocksHandler))
@@ -1069,6 +1087,9 @@ func registerPeerRESTHandlers(router *mux.Router) {
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodBackgroundHealStatus).HandlerFunc(server.BackgroundHealStatusHandler)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLog).HandlerFunc(server.ConsoleLogHandler)
 
+	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodPutBucketObjectLockConfig).HandlerFunc(httpTraceHdrs(server.PutBucketObjectLockConfigHandler)).Queries(restQueries(peerRESTBucket)...)
+	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodBucketObjectLockConfigRemove).HandlerFunc(httpTraceHdrs(server.RemoveBucketObjectLockConfigHandler)).Queries(restQueries(peerRESTBucket)...)
+
 	// If none of the routes match add default error handler routes
 	router.NotFoundHandler = http.HandlerFunc(httpTraceAll(errorResponseHandler))
 	router.MethodNotAllowedHandler = http.HandlerFunc(httpTraceAll(errorResponseHandler))

cmd/xl-v1-healing.go

@@ -697,9 +697,10 @@ func (xl xlObjects) HealObject(ctx context.Context, bucket, object string, dryRu
 			writeQuorum = len(xl.getDisks())/2 + 1
 		}
 		if !dryRun && remove {
-			err = xl.deleteObject(healCtx, bucket, object, writeQuorum, false)
+			xl.deleteObject(healCtx, bucket, object, writeQuorum, false)
 		}
-		return defaultHealResult(xlMetaV1{}, storageDisks, errs, bucket, object), err
+		err = reduceReadQuorumErrs(ctx, errs, nil, writeQuorum-1)
+		return defaultHealResult(xlMetaV1{}, storageDisks, errs, bucket, object), toObjectErr(err, bucket, object)
 	}
 
 	latestXLMeta, err := getLatestXLMeta(healCtx, partsMetadata, errs)
@@ -718,7 +719,7 @@ func (xl xlObjects) HealObject(ctx context.Context, bucket, object string, dryRu
 		// Only if we get errors from all the disks we return error. Else we need to
 		// continue to return filled madmin.HealResultItem struct which includes info
 		// on what disks the file is available etc.
-		if reducedErr := reduceReadQuorumErrs(ctx, errs, nil, latestXLMeta.Erasure.DataBlocks); reducedErr != nil {
+		if err = reduceReadQuorumErrs(ctx, errs, nil, latestXLMeta.Erasure.DataBlocks); err != nil {
 			if m, ok := isObjectDangling(partsMetadata, errs, []error{}); ok {
 				writeQuorum := m.Erasure.DataBlocks + 1
 				if m.Erasure.DataBlocks == 0 {
@@ -728,7 +729,7 @@ func (xl xlObjects) HealObject(ctx context.Context, bucket, object string, dryRu
 					xl.deleteObject(ctx, bucket, object, writeQuorum, false)
 				}
 			}
-			return defaultHealResult(latestXLMeta, storageDisks, errs, bucket, object), toObjectErr(reducedErr, bucket, object)
+			return defaultHealResult(latestXLMeta, storageDisks, errs, bucket, object), toObjectErr(err, bucket, object)
 		}
 	}
 

cmd/xl-v1-healing_test.go

@@ -185,10 +185,12 @@ func TestHealObjectCorrupted(t *testing.T) {
 		xl.getDisks()[i].DeleteFile(bucket, filepath.Join(object, xlMetaJSONFile))
 	}
 
-	// Try healing now, expect to receive errDiskNotFound.
+	// Try healing now, expect to receive errFileNotFound.
 	_, err = objLayer.HealObject(context.Background(), bucket, object, false, true, madmin.HealDeepScan)
 	if err != nil {
-		t.Errorf("Expected nil but received %v", err)
+		if _, ok := err.(ObjectNotFound); !ok {
+			t.Errorf("Expect %v but received %v", ObjectNotFound{Bucket: bucket, Object: object}, err)
+		}
 	}
 
 	// since majority of xl.jsons are not available, object should be successfully deleted.