Disregard healing disks in crawling (#10349)

When crawling never use a disk we know is healing.

Most of the change involves keeping track of the original endpoint on xlStorage
and this also fixes DiskInfo.Endpoint never being populated.

Heal master will print `data-crawl: Disk "http://localhost:9001/data/mindev/data2/xl1" is 
Healing, skipping` once on a cycle (no more often than every 5m).

cmd/bitrot_test.go

@@ -34,7 +34,7 @@ func testBitrotReaderWriterAlgo(t *testing.T, bitrotAlgo BitrotAlgorithm) {
 	volume := "testvol"
 	filePath := "testfile"
 
-	disk, err := newXLStorage(tmpDir, "")
+	disk, err := newLocalXLStorage(tmpDir)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/erasure.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/bpool"
+	"github.com/minio/minio/pkg/color"
 	"github.com/minio/minio/pkg/dsync"
 	"github.com/minio/minio/pkg/madmin"
 	"github.com/minio/minio/pkg/sync/errgroup"
@@ -255,21 +256,45 @@ func (er erasureObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilt
 // CrawlAndGetDataUsage will start crawling buckets and send updated totals as they are traversed.
 // Updates are sent on a regular basis and the caller *must* consume them.
 func (er erasureObjects) crawlAndGetDataUsage(ctx context.Context, buckets []BucketInfo, bf *bloomFilter, updates chan<- dataUsageCache) error {
-	var disks []StorageAPI
+	if len(buckets) == 0 {
+		return nil
+	}
 
+	// Collect any disk healing.
+	healing, err := getAggregatedBackgroundHealState(ctx, true)
+	if err != nil {
+		return err
+	}
+	healDisks := make(map[string]struct{}, len(healing.HealDisks))
+	for _, disk := range healing.HealDisks {
+		healDisks[disk] = struct{}{}
+	}
+
+	// Collect disks we can use.
+	var disks []StorageAPI
 	for _, d := range er.getLoadBalancedDisks() {
 		if d == nil || !d.IsOnline() {
 			continue
 		}
+		di, err := d.DiskInfo()
+		if err != nil {
+			logger.LogIf(ctx, err)
+			continue
+		}
+		if _, ok := healDisks[di.Endpoint]; ok {
+			logger.Info(color.Green("data-crawl:")+" Disk %q is Healing, skipping disk.", di.Endpoint)
+			continue
+		}
 		disks = append(disks, d)
 	}
-	if len(disks) == 0 || len(buckets) == 0 {
+	if len(disks) == 0 {
+		logger.Info(color.Green("data-crawl:") + " No disks found, skipping crawl")
 		return nil
 	}
 
 	// Load bucket totals
 	oldCache := dataUsageCache{}
-	err := oldCache.load(ctx, er, dataUsageCacheName)
+	err = oldCache.load(ctx, er, dataUsageCacheName)
 	if err != nil {
 		return err
 	}

cmd/object-api-common.go

@@ -82,7 +82,7 @@ func dirObjectInfo(bucket, object string, size int64, metadata map[string]string
 // Depending on the disk type network or local, initialize storage API.
 func newStorageAPI(endpoint Endpoint) (storage StorageAPI, err error) {
 	if endpoint.IsLocal {
-		storage, err := newXLStorage(endpoint.Path, endpoint.Host)
+		storage, err := newXLStorage(endpoint)
 		if err != nil {
 			return nil, err
 		}

cmd/storage-rest-server.go

@@ -826,7 +826,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpointZones EndpointZones
 			if !endpoint.IsLocal {
 				continue
 			}
-			storage, err := newXLStorage(endpoint.Path, endpoint.Host)
+			storage, err := newXLStorage(endpoint)
 			if err != nil {
 				if err == errMinDiskSize {
 					logger.Fatal(config.ErrUnableToWriteInBackend(err).Hint(err.Error()), "Unable to initialize backend")

cmd/xl-storage.go

@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net/url"
 	"os"
 	slashpath "path"
 	"path/filepath"
@@ -91,6 +92,7 @@ type xlStorage struct {
 
 	diskPath string
 	hostname string
+	endpoint string
 
 	pool sync.Pool
 
@@ -233,7 +235,18 @@ func isDirEmpty(dirname string) bool {
 }
 
 // Initialize a new storage disk.
-func newXLStorage(path string, hostname string) (*xlStorage, error) {
+func newLocalXLStorage(path string) (*xlStorage, error) {
+	u := url.URL{Path: path}
+	return newXLStorage(Endpoint{
+		URL:     &u,
+		IsLocal: true,
+	})
+}
+
+// Initialize a new storage disk.
+func newXLStorage(ep Endpoint) (*xlStorage, error) {
+	path := ep.Path
+	hostname := ep.Host
 	var err error
 	if path, err = getValidPath(path, true); err != nil {
 		return nil, err
@@ -247,6 +260,7 @@ func newXLStorage(path string, hostname string) (*xlStorage, error) {
 	p := &xlStorage{
 		diskPath: path,
 		hostname: hostname,
+		endpoint: ep.String(),
 		pool: sync.Pool{
 			New: func() interface{} {
 				b := disk.AlignedBlock(readBlockSize)
@@ -422,6 +436,7 @@ func (s *xlStorage) DiskInfo() (info DiskInfo, err error) {
 		Used:      di.Total - di.Free,
 		RootDisk:  s.rootDisk,
 		MountPath: s.diskPath,
+		Endpoint:  s.endpoint,
 	}
 
 	diskID, err := s.GetDiskID()

cmd/xl-storage_test.go

@@ -122,7 +122,7 @@ func newXLStorageTestSetup() (*xlStorageDiskIDCheck, string, error) {
 	}
 
 	// Initialize a new xlStorage layer.
-	storage, err := newXLStorage(diskPath, "")
+	storage, err := newLocalXLStorage(diskPath)
 	if err != nil {
 		return nil, "", err
 	}
@@ -377,7 +377,7 @@ func TestNewXLStorage(t *testing.T) {
 	// Validate all test cases.
 	for i, testCase := range testCases {
 		// Initialize a new xlStorage layer.
-		_, err := newXLStorage(testCase.name, "")
+		_, err := newLocalXLStorage(testCase.name)
 		if err != testCase.err {
 			t.Fatalf("TestXLStorage %d failed wanted: %s, got: %s", i+1, err, testCase.err)
 		}
@@ -451,7 +451,7 @@ func TestXLStorageMakeVol(t *testing.T) {
 		}
 
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -460,7 +460,7 @@ func TestXLStorageMakeVol(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStorageNew, err := newXLStorage(permDeniedDir, "")
+		xlStorageNew, err := newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -550,7 +550,7 @@ func TestXLStorageDeleteVol(t *testing.T) {
 		}
 
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -559,7 +559,7 @@ func TestXLStorageDeleteVol(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStorageNew, err := newXLStorage(permDeniedDir, "")
+		xlStorageNew, err := newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -802,7 +802,7 @@ func TestXLStorageXlStorageListDir(t *testing.T) {
 		defer removePermDeniedFile(permDeniedDir)
 
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -811,7 +811,7 @@ func TestXLStorageXlStorageListDir(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStorageNew, err := newXLStorage(permDeniedDir, "")
+		xlStorageNew, err := newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -926,7 +926,7 @@ func TestXLStorageDeleteFile(t *testing.T) {
 		defer removePermDeniedFile(permDeniedDir)
 
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -935,7 +935,7 @@ func TestXLStorageDeleteFile(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStorageNew, err := newXLStorage(permDeniedDir, "")
+		xlStorageNew, err := newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -1124,7 +1124,7 @@ func TestXLStorageReadFile(t *testing.T) {
 		defer removePermDeniedFile(permDeniedDir)
 
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -1133,7 +1133,7 @@ func TestXLStorageReadFile(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStoragePermStorage, err := newXLStorage(permDeniedDir, "")
+		xlStoragePermStorage, err := newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -1294,7 +1294,7 @@ func TestXLStorageAppendFile(t *testing.T) {
 
 		var xlStoragePermStorage StorageAPI
 		// Initialize xlStorage storage layer for permission denied error.
-		_, err = newXLStorage(permDeniedDir, "")
+		_, err = newLocalXLStorage(permDeniedDir)
 		if err != nil && !os.IsPermission(err) {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}
@@ -1303,7 +1303,7 @@ func TestXLStorageAppendFile(t *testing.T) {
 			t.Fatalf("Unable to change permission to temporary directory %v. %v", permDeniedDir, err)
 		}
 
-		xlStoragePermStorage, err = newXLStorage(permDeniedDir, "")
+		xlStoragePermStorage, err = newLocalXLStorage(permDeniedDir)
 		if err != nil {
 			t.Fatalf("Unable to initialize xlStorage, %s", err)
 		}

cmd/xl-storage_unix_test.go

@@ -49,7 +49,7 @@ func TestIsValidUmaskVol(t *testing.T) {
 	testCase := testCases[0]
 
 	// Initialize a new xlStorage layer.
-	disk, err := newXLStorage(tmpPath, "")
+	disk, err := newLocalXLStorage(tmpPath)
 	if err != nil {
 		t.Fatalf("Initializing xlStorage failed with %s.", err)
 	}
@@ -91,7 +91,7 @@ func TestIsValidUmaskFile(t *testing.T) {
 	testCase := testCases[0]
 
 	// Initialize a new xlStorage layer.
-	disk, err := newXLStorage(tmpPath, "")
+	disk, err := newLocalXLStorage(tmpPath)
 	if err != nil {
 		t.Fatalf("Initializing xlStorage failed with %s.", err)
 	}

cmd/xl-storage_windows_test.go

@@ -48,7 +48,7 @@ func TestUNCPaths(t *testing.T) {
 
 	// Instantiate posix object to manage a disk
 	var fs StorageAPI
-	fs, err = newXLStorage(dir, "")
+	fs, err = newLocalXLStorage(dir)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -83,7 +83,7 @@ func TestUNCPathENOTDIR(t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	var fs StorageAPI
-	fs, err = newXLStorage(dir, "")
+	fs, err = newLocalXLStorage(dir)
 	if err != nil {
 		t.Fatal(err)
 	}