fix: introduce isLeafDir in healing to fix the crash (#5920)

This PR also supports healing directories.

Fixes #5917

cmd/acl-handlers.go

@@ -0,0 +1,141 @@
+/*
+ * Minio Cloud Storage, (C) 2018 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"encoding/xml"
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"github.com/minio/minio/pkg/policy"
+)
+
+// Data types used for returning dummy access control
+// policy XML, these variables shouldn't be used elsewhere
+// they are only defined to be used in this file alone.
+type grantee struct {
+	XMLNS       string `xml:"xmlns:xsi,attr"`
+	XMLXSI      string `xml:"xsi:type,attr"`
+	Type        string `xml:"Type"`
+	ID          string `xml:"ID,omitempty"`
+	DisplayName string `xml:"DisplayName,omitempty"`
+	URI         string `xml:"URI,omitempty"`
+}
+
+type grant struct {
+	Grantee    grantee `xml:"Grantee"`
+	Permission string  `xml:"Permission"`
+}
+
+type accessControlPolicy struct {
+	XMLName           xml.Name `xml:"AccessControlPolicy"`
+	Owner             Owner    `xml:"Owner"`
+	AccessControlList struct {
+		Grants []grant `xml:"Grant"`
+	} `xml:"AccessControlList"`
+}
+
+// GetBucketACLHandler - GET Bucket ACL
+// -----------------
+// This operation uses the ACL
+// subresource to return the ACL of a specified bucket.
+func (api objectAPIHandlers) GetBucketACLHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, "GetBucketACL")
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+
+	objAPI := api.ObjectAPI()
+	if objAPI == nil {
+		writeErrorResponse(w, ErrServerNotInitialized, r.URL)
+		return
+	}
+
+	// Allow getBucketACL if policy action is set, since this is a dummy call
+	// we are simply re-purposing the bucketPolicyAction.
+	if s3Error := checkRequestAuthType(ctx, r, policy.GetBucketPolicyAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(w, s3Error, r.URL)
+		return
+	}
+
+	// Before proceeding validate if bucket exists.
+	_, err := objAPI.GetBucketInfo(ctx, bucket)
+	if err != nil {
+		writeErrorResponse(w, toAPIErrorCode(err), r.URL)
+		return
+	}
+
+	acl := &accessControlPolicy{}
+	acl.AccessControlList.Grants = append(acl.AccessControlList.Grants, grant{
+		Grantee: grantee{
+			Type: "CanonicalUser",
+		},
+		Permission: "FULL_CONTROL",
+	})
+	if err := xml.NewEncoder(w).Encode(acl); err != nil {
+		writeErrorResponse(w, toAPIErrorCode(err), r.URL)
+		return
+	}
+
+	w.(http.Flusher).Flush()
+}
+
+// GetObjectACLHandler - GET Object ACL
+// -----------------
+// This operation uses the ACL
+// subresource to return the ACL of a specified object.
+func (api objectAPIHandlers) GetObjectACLHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, "GetObjectACL")
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	object := vars["object"]
+
+	objAPI := api.ObjectAPI()
+	if objAPI == nil {
+		writeErrorResponse(w, ErrServerNotInitialized, r.URL)
+		return
+	}
+
+	// Allow getObjectACL if policy action is set, since this is a dummy call
+	// we are simply re-purposing the bucketPolicyAction.
+	if s3Error := checkRequestAuthType(ctx, r, policy.GetBucketPolicyAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(w, s3Error, r.URL)
+		return
+	}
+
+	// Before proceeding validate if object exists.
+	_, err := objAPI.GetObjectInfo(ctx, bucket, object)
+	if err != nil {
+		writeErrorResponse(w, toAPIErrorCode(err), r.URL)
+		return
+	}
+
+	acl := &accessControlPolicy{}
+	acl.AccessControlList.Grants = append(acl.AccessControlList.Grants, grant{
+		Grantee: grantee{
+			Type: "CanonicalUser",
+		},
+		Permission: "FULL_CONTROL",
+	})
+	if err := xml.NewEncoder(w).Encode(acl); err != nil {
+		writeErrorResponse(w, toAPIErrorCode(err), r.URL)
+		return
+	}
+
+	w.(http.Flusher).Flush()
+}

cmd/api-router.go

@@ -83,6 +83,10 @@ func registerAPIRouter(router *mux.Router) {
 		bucket.Methods("GET").HandlerFunc(httpTraceAll(api.GetBucketLocationHandler)).Queries("location", "")
 		// GetBucketPolicy
 		bucket.Methods("GET").HandlerFunc(httpTraceAll(api.GetBucketPolicyHandler)).Queries("policy", "")
+
+		// GetBucketACL -- this is a dummy call.
+		bucket.Methods("GET").HandlerFunc(httpTraceAll(api.GetBucketACLHandler)).Queries("acl", "")
+
 		// GetBucketNotification
 		bucket.Methods("GET").HandlerFunc(httpTraceAll(api.GetBucketNotificationHandler)).Queries("notification", "")
 		// ListenBucketNotification

cmd/generic-handlers.go

@@ -420,6 +420,11 @@ func setIgnoreResourcesHandler(h http.Handler) http.Handler {
 // Checks requests for not implemented Bucket resources
 func ignoreNotImplementedBucketResources(req *http.Request) bool {
 	for name := range req.URL.Query() {
+		// Enable GetBucketACL dummy call specifically.
+		if name == "acl" && req.Method == http.MethodGet {
+			return false
+		}
+
 		if notimplementedBucketResourceNames[name] {
 			return true
 		}
@@ -430,6 +435,10 @@ func ignoreNotImplementedBucketResources(req *http.Request) bool {
 // Checks requests for not implemented Object resources
 func ignoreNotImplementedObjectResources(req *http.Request) bool {
 	for name := range req.URL.Query() {
+		// Enable GetObjectACL dummy call specifically.
+		if name == "acl" && req.Method == http.MethodGet {
+			return false
+		}
 		if notimplementedObjectResourceNames[name] {
 			return true
 		}

cmd/tree-walk.go

@@ -153,6 +153,12 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 			return err
 		}
 	}
+
+	// When isleaf check is delayed, make sure that it is set correctly here.
+	if delayIsLeaf && isLeaf == nil {
+		return errInvalidArgument
+	}
+
 	// For an empty list return right here.
 	if len(entries) == 0 {
 		return nil
@@ -169,6 +175,7 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 	if len(entries) == 0 {
 		return nil
 	}
+
 	for i, entry := range entries {
 		var leaf, leafDir bool
 
@@ -187,7 +194,6 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 		}
 
 		isDir := !leafDir && !leaf
-
 		if i == 0 && markerDir == entry {
 			if !recursive {
 				// Skip as the marker would already be listed in the previous listing.

cmd/xl-sets.go

@@ -1360,13 +1360,17 @@ func (s *xlSets) listObjectsHeal(ctx context.Context, bucket, prefix, marker, de
 			return s.getHashedSet(entry).isObject(bucket, entry)
 		}
 
+		isLeafDir := func(bucket, entry string) bool {
+			return s.getHashedSet(entry).isObjectDir(bucket, entry)
+		}
+
 		var setDisks = make([][]StorageAPI, len(s.sets))
 		for _, set := range s.sets {
 			setDisks = append(setDisks, set.getLoadBalancedDisks())
 		}
 
 		listDir := listDirSetsHealFactory(isLeaf, setDisks...)
-		walkResultCh = startTreeWalk(ctx, bucket, prefix, marker, recursive, listDir, nil, nil, endWalkCh)
+		walkResultCh = startTreeWalk(ctx, bucket, prefix, marker, recursive, listDir, nil, isLeafDir, endWalkCh)
 	}
 
 	var objInfos []ObjectInfo
@@ -1403,20 +1407,7 @@ func (s *xlSets) listObjectsHeal(ctx context.Context, bucket, prefix, marker, de
 	result := ListObjectsInfo{IsTruncated: !eof}
 	for _, objInfo := range objInfos {
 		result.NextMarker = objInfo.Name
-		if objInfo.IsDir {
-			result.Prefixes = append(result.Prefixes, objInfo.Name)
-			continue
-		}
-
-		// Add each object seen to the result - objects are
-		// checked for healing later.
-		result.Objects = append(result.Objects, ObjectInfo{
-			Bucket:  bucket,
-			Name:    objInfo.Name,
-			ModTime: objInfo.ModTime,
-			Size:    objInfo.Size,
-			IsDir:   false,
-		})
+		result.Objects = append(result.Objects, objInfo)
 	}
 	return result, nil
 }

cmd/xl-v1-healing.go

@@ -539,12 +539,74 @@ func healObject(ctx context.Context, storageDisks []StorageAPI, bucket string, o
 	return result, nil
 }
 
+// healObjectDir - heals object directory specifically, this special call
+// is needed since we do not have a special backend format for directories.
+func (xl xlObjects) healObjectDir(ctx context.Context, bucket, object string, dryRun bool) (hr madmin.HealResultItem, err error) {
+	storageDisks := xl.getDisks()
+
+	// Initialize heal result object
+	hr = madmin.HealResultItem{
+		Type:         madmin.HealItemObject,
+		Bucket:       bucket,
+		Object:       object,
+		DiskCount:    len(storageDisks),
+		ParityBlocks: len(storageDisks) / 2,
+		DataBlocks:   len(storageDisks) / 2,
+		ObjectSize:   0,
+	}
+
+	// Prepare object creation in all disks
+	for _, disk := range storageDisks {
+		if disk == nil {
+			hr.Before.Drives = append(hr.Before.Drives, madmin.HealDriveInfo{
+				UUID:  "",
+				State: madmin.DriveStateOffline,
+			})
+			hr.After.Drives = append(hr.After.Drives, madmin.HealDriveInfo{
+				UUID:  "",
+				State: madmin.DriveStateMissing,
+			})
+			continue
+		}
+
+		drive := disk.String()
+		hr.Before.Drives = append(hr.Before.Drives, madmin.HealDriveInfo{
+			UUID:     "",
+			Endpoint: drive,
+			State:    madmin.DriveStateMissing,
+		})
+		hr.After.Drives = append(hr.After.Drives, madmin.HealDriveInfo{
+			UUID:     "",
+			Endpoint: drive,
+			State:    madmin.DriveStateMissing,
+		})
+
+		if !dryRun {
+			if err := disk.MakeVol(pathJoin(bucket, object)); err != nil && err != errVolumeExists {
+				return hr, toObjectErr(err, bucket, object)
+			}
+
+			for i, v := range hr.Before.Drives {
+				if v.Endpoint == drive {
+					hr.After.Drives[i].State = madmin.DriveStateOk
+				}
+			}
+		}
+	}
+	return hr, nil
+}
+
 // HealObject - heal the given object.
 //
 // FIXME: If an object object was deleted and one disk was down,
 // and later the disk comes back up again, heal on the object
 // should delete it.
 func (xl xlObjects) HealObject(ctx context.Context, bucket, object string, dryRun bool) (hr madmin.HealResultItem, err error) {
+	// Healing directories handle it separately.
+	if hasSuffix(object, slashSeparator) {
+		return xl.healObjectDir(ctx, bucket, object, dryRun)
+	}
+
 	// FIXME: Metadata is read again in the healObject() call below.
 	// Read metadata files from all the disks
 	partsMetadata, errs := readAllXLMetadata(ctx, xl.getDisks(), bucket, object)

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '2'
 # 9001 through 9004.
 services:
  minio1:
-  image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+  image: minio/minio:RELEASE.2018-05-10T00-00-42Z
   volumes:
    - data1:/data
   ports:
@@ -15,7 +15,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio2:
-  image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+  image: minio/minio:RELEASE.2018-05-10T00-00-42Z
   volumes:
    - data2:/data
   ports:
@@ -25,7 +25,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio3:
-  image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+  image: minio/minio:RELEASE.2018-05-10T00-00-42Z
   volumes:
    - data3:/data
   ports:
@@ -35,7 +35,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio4:
-  image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+  image: minio/minio:RELEASE.2018-05-10T00-00-42Z
   volumes:
    - data4:/data
   ports:

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.1'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio1-data:/export
     ports:
@@ -20,7 +20,7 @@ services:
       - access_key
 
   minio2:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio2-data:/export
     ports:
@@ -38,7 +38,7 @@ services:
       - access_key
 
   minio3:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio3-data:/export
     ports:
@@ -56,7 +56,7 @@ services:
       - access_key
 
   minio4:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio4-data:/export
     ports:

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio1-data:/export
     ports:
@@ -20,7 +20,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio2:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio2-data:/export
     ports:
@@ -38,7 +38,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio3:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio3-data:/export
     ports:
@@ -56,7 +56,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio4:
-    image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+    image: minio/minio:RELEASE.2018-05-10T00-00-42Z
     volumes:
       - minio4-data:/export
     ports:

docs/orchestration/kubernetes-yaml/README.md

@@ -124,7 +124,7 @@ spec:
         - name: data 
           mountPath: "/data"
         # Pulls the lastest Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - server
         - /data
@@ -303,7 +303,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - server
         - http://minio-0.minio.default.svc.cluster.local/data
@@ -514,7 +514,7 @@ spec:
       containers:
       - name: minio
         # Pulls the default Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - gateway
         - gcs

docs/orchestration/kubernetes-yaml/minio-distributed-statefulset.yaml

@@ -21,7 +21,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - server
         - http://minio-0.minio.default.svc.cluster.local/data

docs/orchestration/kubernetes-yaml/minio-gcs-gateway-deployment.yaml

@@ -21,7 +21,7 @@ spec:
       containers:
       - name: minio
         # Pulls the default Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - gateway
         - gcs

docs/orchestration/kubernetes-yaml/minio-standalone-deployment.yaml

@@ -29,7 +29,7 @@ spec:
         - name: data 
           mountPath: "/data"
         # Pulls the lastest Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         args:
         - server
         - /data

docs/orchestration/kubernetes/README.md

@@ -30,7 +30,7 @@ Above command deploys Minio on the Kubernetes cluster in the default configurati
 | Parameter                  | Description                         | Default                                                 |
 |----------------------------|-------------------------------------|---------------------------------------------------------|
 | `image`                    | Minio image name                    | `minio/minio`                                           |
-| `imageTag`                 | Minio image tag. Possible values listed [here](https://hub.docker.com/r/minio/minio/tags/).| `RELEASE.2018-05-04T23-13-12Z`|
+| `imageTag`                 | Minio image tag. Possible values listed [here](https://hub.docker.com/r/minio/minio/tags/).| `RELEASE.2018-05-10T00-00-42Z`|
 | `imagePullPolicy`          | Image pull policy                   | `Always`                                                |
 | `mode`                     | Minio server mode (`standalone`, `shared` or `distributed`)| `standalone`                     |
 | `numberOfNodes`            | Number of nodes (applicable only for Minio distributed mode). Should be 4 <= x <= 16 | `4`    |

docs/orchestration/minikube/statefulset.yaml

@@ -54,7 +54,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2018-05-04T23-13-12Z
+        image: minio/minio:RELEASE.2018-05-10T00-00-42Z
         imagePullPolicy: IfNotPresent
         args: ["server", "http://minio-0.minio.default.svc.cluster.local/data", "http://minio-1.minio.default.svc.cluster.local/data", "http://minio-2.minio.default.svc.cluster.local/data", "http://minio-3.minio.default.svc.cluster.local/data"]
         ports: