Implement heal-upload admin API (#3914)

This API is meant for administrative tools like mc-admin to heal an
ongoing multipart upload on a Minio server.  N B This set of admin
APIs apply only for Minio servers.

`github.com/minio/minio/pkg/madmin` provides a go SDK for this (and
other admin) operations.  Specifically,

  func HealUpload(bucket, object, uploadID string, dryRun bool) error

Sample admin API request:
POST
/?heal&bucket=mybucket&object=myobject&upload-id=myuploadID&dry-run
- Header(s): ["x-minio-operation"] = "upload"

Notes:
- bucket, object and upload-id are mandatory query parameters
- if dry-run is set, API returns success if all parameters passed are
  valid.
This commit is contained in:
Krishnan Parthasarathi
2017-03-17 21:55:49 +05:30
committed by Harshavardhana
parent d4eea224d4
commit c192e5c9b2
8 changed files with 460 additions and 90 deletions

View File

@@ -17,7 +17,9 @@
package cmd
import (
"bytes"
"fmt"
"path/filepath"
"testing"
)
@@ -486,3 +488,73 @@ func TestListBucketsHeal(t *testing.T) {
t.Fatalf("Name of missing bucket is incorrect, expected: %s, found: %s", corruptedBucketName, buckets[0].Name)
}
}
// Tests healing of object.
func TestHealObjectXL(t *testing.T) {
root, err := newTestConfig(globalMinioDefaultRegion)
if err != nil {
t.Fatal(err)
}
defer removeAll(root)
nDisks := 16
fsDirs, err := getRandomDisks(nDisks)
if err != nil {
t.Fatal(err)
}
defer removeRoots(fsDirs)
endpoints, err := parseStorageEndpoints(fsDirs)
if err != nil {
t.Fatal(err)
}
// Everything is fine, should return nil
obj, _, err := initObjectLayer(endpoints)
if err != nil {
t.Fatal(err)
}
bucket := "bucket"
object := "object"
data := []byte("hello")
err = obj.MakeBucket(bucket)
if err != nil {
t.Fatalf("Failed to make a bucket - %v", err)
}
_, err = obj.PutObject(bucket, object, int64(len(data)), bytes.NewReader(data), nil, "")
if err != nil {
t.Fatalf("Failed to put an object - %v", err)
}
// Remove the object backend files from the first disk.
xl := obj.(*xlObjects)
firstDisk := xl.storageDisks[0]
err = firstDisk.DeleteFile(bucket, filepath.Join(object, xlMetaJSONFile))
if err != nil {
t.Fatalf("Failed to delete a file - %v", err)
}
err = obj.HealObject(bucket, object)
if err != nil {
t.Fatalf("Failed to heal object - %v", err)
}
_, err = firstDisk.StatFile(bucket, filepath.Join(object, xlMetaJSONFile))
if err != nil {
t.Errorf("Expected xl.json file to be present but stat failed - %v", err)
}
// Nil more than half the disks, to remove write quorum.
for i := 0; i <= len(xl.storageDisks)/2; i++ {
xl.storageDisks[i] = nil
}
// Try healing now, expect to receive errDiskNotFound.
err = obj.HealObject(bucket, object)
if errorCause(err) != errDiskNotFound {
t.Errorf("Expected %v but received %v", errDiskNotFound, err)
}
}