mirror of https://github.com/minio/minio.git
				
				
				
			
		
			
				
	
	
		
			412 lines
		
	
	
		
			13 KiB
		
	
	
	
		
			Go
		
	
	
	
			
		
		
	
	
			412 lines
		
	
	
		
			13 KiB
		
	
	
	
		
			Go
		
	
	
	
| /*
 | |
|  * MinIO Cloud Storage, (C) 2016 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 (
 | |
| 	"context"
 | |
| 	"crypto/tls"
 | |
| 	"fmt"
 | |
| 	"net/http"
 | |
| 	"net/url"
 | |
| 	"os"
 | |
| 	"sync"
 | |
| 	"time"
 | |
| 
 | |
| 	"github.com/dustin/go-humanize"
 | |
| 	xhttp "github.com/minio/minio/cmd/http"
 | |
| 	"github.com/minio/minio/cmd/logger"
 | |
| 	"github.com/minio/minio/pkg/sync/errgroup"
 | |
| )
 | |
| 
 | |
| var printEndpointError = func() func(Endpoint, error, bool) {
 | |
| 	var mutex sync.Mutex
 | |
| 	printOnce := make(map[Endpoint]map[string]int)
 | |
| 
 | |
| 	return func(endpoint Endpoint, err error, once bool) {
 | |
| 		reqInfo := (&logger.ReqInfo{}).AppendTags("endpoint", endpoint.String())
 | |
| 		ctx := logger.SetReqInfo(GlobalContext, reqInfo)
 | |
| 		mutex.Lock()
 | |
| 		defer mutex.Unlock()
 | |
| 
 | |
| 		m, ok := printOnce[endpoint]
 | |
| 		if !ok {
 | |
| 			m = make(map[string]int)
 | |
| 			m[err.Error()]++
 | |
| 			printOnce[endpoint] = m
 | |
| 			if once {
 | |
| 				logger.LogAlwaysIf(ctx, err)
 | |
| 				return
 | |
| 			}
 | |
| 		}
 | |
| 		// Once is set and we are here means error was already
 | |
| 		// printed once.
 | |
| 		if once {
 | |
| 			return
 | |
| 		}
 | |
| 		// once not set, check if same error occurred 3 times in
 | |
| 		// a row, then make sure we print it to call attention.
 | |
| 		if m[err.Error()] > 2 {
 | |
| 			logger.LogAlwaysIf(ctx, fmt.Errorf("Following error has been printed %d times.. %w", m[err.Error()], err))
 | |
| 			// Reduce the count to introduce further delay in printing
 | |
| 			// but let it again print after the 2th attempt
 | |
| 			m[err.Error()]--
 | |
| 			m[err.Error()]--
 | |
| 		}
 | |
| 		m[err.Error()]++
 | |
| 	}
 | |
| }()
 | |
| 
 | |
| // Migrates backend format of local disks.
 | |
| func formatErasureMigrateLocalEndpoints(endpoints Endpoints) error {
 | |
| 	g := errgroup.WithNErrs(len(endpoints))
 | |
| 	for index, endpoint := range endpoints {
 | |
| 		if !endpoint.IsLocal {
 | |
| 			continue
 | |
| 		}
 | |
| 		index := index
 | |
| 		g.Go(func() error {
 | |
| 			epPath := endpoints[index].Path
 | |
| 			formatPath := pathJoin(epPath, minioMetaBucket, formatConfigFile)
 | |
| 			if _, err := os.Stat(formatPath); err != nil {
 | |
| 				if os.IsNotExist(err) {
 | |
| 					return nil
 | |
| 				}
 | |
| 				return fmt.Errorf("unable to access (%s) %w", formatPath, err)
 | |
| 			}
 | |
| 			return formatErasureMigrate(epPath)
 | |
| 		}, index)
 | |
| 	}
 | |
| 	for _, err := range g.Wait() {
 | |
| 		if err != nil {
 | |
| 			return err
 | |
| 		}
 | |
| 	}
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| // Cleans up tmp directory of local disks.
 | |
| func formatErasureCleanupTmpLocalEndpoints(endpoints Endpoints) error {
 | |
| 	g := errgroup.WithNErrs(len(endpoints))
 | |
| 	for index, endpoint := range endpoints {
 | |
| 		if !endpoint.IsLocal {
 | |
| 			continue
 | |
| 		}
 | |
| 		index := index
 | |
| 		g.Go(func() error {
 | |
| 			epPath := endpoints[index].Path
 | |
| 			// If disk is not formatted there is nothing to be cleaned up.
 | |
| 			formatPath := pathJoin(epPath, minioMetaBucket, formatConfigFile)
 | |
| 			if _, err := os.Stat(formatPath); err != nil {
 | |
| 				if os.IsNotExist(err) {
 | |
| 					return nil
 | |
| 				}
 | |
| 				return fmt.Errorf("unable to access (%s) %w", formatPath, err)
 | |
| 			}
 | |
| 			if _, err := os.Stat(pathJoin(epPath, minioMetaTmpBucket+"-old")); err != nil {
 | |
| 				if !os.IsNotExist(err) {
 | |
| 					return fmt.Errorf("unable to access (%s) %w",
 | |
| 						pathJoin(epPath, minioMetaTmpBucket+"-old"),
 | |
| 						err)
 | |
| 				}
 | |
| 			}
 | |
| 
 | |
| 			// Need to move temporary objects left behind from previous run of minio
 | |
| 			// server to a unique directory under `minioMetaTmpBucket-old` to clean
 | |
| 			// up `minioMetaTmpBucket` for the current run.
 | |
| 			//
 | |
| 			// /disk1/.minio.sys/tmp-old/
 | |
| 			//  |__ 33a58b40-aecc-4c9f-a22f-ff17bfa33b62
 | |
| 			//  |__ e870a2c1-d09c-450c-a69c-6eaa54a89b3e
 | |
| 			//
 | |
| 			// In this example, `33a58b40-aecc-4c9f-a22f-ff17bfa33b62` directory contains
 | |
| 			// temporary objects from one of the previous runs of minio server.
 | |
| 			tmpOld := pathJoin(epPath, minioMetaTmpBucket+"-old", mustGetUUID())
 | |
| 			if err := renameAll(pathJoin(epPath, minioMetaTmpBucket),
 | |
| 				tmpOld); err != nil && err != errFileNotFound {
 | |
| 				return fmt.Errorf("unable to rename (%s -> %s) %w",
 | |
| 					pathJoin(epPath, minioMetaTmpBucket),
 | |
| 					tmpOld,
 | |
| 					osErrToFileErr(err))
 | |
| 			}
 | |
| 
 | |
| 			// Removal of tmp-old folder is backgrounded completely.
 | |
| 			go removeAll(pathJoin(epPath, minioMetaTmpBucket+"-old"))
 | |
| 
 | |
| 			if err := mkdirAll(pathJoin(epPath, minioMetaTmpBucket), 0777); err != nil {
 | |
| 				return fmt.Errorf("unable to create (%s) %w",
 | |
| 					pathJoin(epPath, minioMetaTmpBucket),
 | |
| 					err)
 | |
| 			}
 | |
| 			return nil
 | |
| 		}, index)
 | |
| 	}
 | |
| 	for _, err := range g.Wait() {
 | |
| 		if err != nil {
 | |
| 			return err
 | |
| 		}
 | |
| 	}
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| // Following error message is added to fix a regression in release
 | |
| // RELEASE.2018-03-16T22-52-12Z after migrating v1 to v2 to v3. This
 | |
| // migration failed to capture '.This' field properly which indicates
 | |
| // the disk UUID association. Below error message is returned when
 | |
| // we see this situation in format.json, for more info refer
 | |
| // https://github.com/minio/minio/issues/5667
 | |
| var errErasureV3ThisEmpty = fmt.Errorf("Erasure format version 3 has This field empty")
 | |
| 
 | |
| // IsServerResolvable - checks if the endpoint is resolvable
 | |
| // by sending a naked HTTP request with liveness checks.
 | |
| func IsServerResolvable(endpoint Endpoint) error {
 | |
| 	serverURL := &url.URL{
 | |
| 		Scheme: endpoint.Scheme,
 | |
| 		Host:   endpoint.Host,
 | |
| 		Path:   pathJoin(healthCheckPathPrefix, healthCheckLivenessPath),
 | |
| 	}
 | |
| 
 | |
| 	var tlsConfig *tls.Config
 | |
| 	if globalIsSSL {
 | |
| 		tlsConfig = &tls.Config{
 | |
| 			ServerName: endpoint.Hostname(),
 | |
| 			RootCAs:    globalRootCAs,
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	httpClient := &http.Client{
 | |
| 		Transport:
 | |
| 		// For more details about various values used here refer
 | |
| 		// https://golang.org/pkg/net/http/#Transport documentation
 | |
| 		&http.Transport{
 | |
| 			Proxy:                 http.ProxyFromEnvironment,
 | |
| 			DialContext:           xhttp.NewCustomDialContext(3 * time.Second),
 | |
| 			ResponseHeaderTimeout: 3 * time.Second,
 | |
| 			TLSHandshakeTimeout:   3 * time.Second,
 | |
| 			ExpectContinueTimeout: 3 * time.Second,
 | |
| 			TLSClientConfig:       tlsConfig,
 | |
| 			// Go net/http automatically unzip if content-type is
 | |
| 			// gzip disable this feature, as we are always interested
 | |
| 			// in raw stream.
 | |
| 			DisableCompression: true,
 | |
| 		},
 | |
| 	}
 | |
| 	defer httpClient.CloseIdleConnections()
 | |
| 
 | |
| 	ctx, cancel := context.WithTimeout(GlobalContext, 3*time.Second)
 | |
| 
 | |
| 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, serverURL.String(), nil)
 | |
| 	if err != nil {
 | |
| 		cancel()
 | |
| 		return err
 | |
| 	}
 | |
| 
 | |
| 	resp, err := httpClient.Do(req)
 | |
| 	cancel()
 | |
| 	if err != nil {
 | |
| 		return err
 | |
| 	}
 | |
| 	xhttp.DrainBody(resp.Body)
 | |
| 
 | |
| 	if resp.StatusCode != http.StatusOK {
 | |
| 		return StorageErr(resp.Status)
 | |
| 	}
 | |
| 
 | |
| 	if resp.Header.Get(xhttp.MinIOServerStatus) == unavailable {
 | |
| 		return StorageErr(unavailable)
 | |
| 	}
 | |
| 
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| // connect to list of endpoints and load all Erasure disk formats, validate the formats are correct
 | |
| // and are in quorum, if no formats are found attempt to initialize all of them for the first
 | |
| // time. additionally make sure to close all the disks used in this attempt.
 | |
| func connectLoadInitFormats(retryCount int, firstDisk bool, endpoints Endpoints, zoneCount, setCount, setDriveCount int, deploymentID string) (storageDisks []StorageAPI, format *formatErasureV3, err error) {
 | |
| 	// Initialize all storage disks
 | |
| 	storageDisks, errs := initStorageDisksWithErrors(endpoints)
 | |
| 
 | |
| 	defer func(storageDisks []StorageAPI) {
 | |
| 		if err != nil {
 | |
| 			closeStorageDisks(storageDisks)
 | |
| 		}
 | |
| 	}(storageDisks)
 | |
| 
 | |
| 	for i, err := range errs {
 | |
| 		if err != nil {
 | |
| 			if err != errDiskNotFound {
 | |
| 				return nil, nil, fmt.Errorf("Disk %s: %w", endpoints[i], err)
 | |
| 			}
 | |
| 			if retryCount >= 5 {
 | |
| 				logger.Info("Unable to connect to %s: %v\n", endpoints[i], IsServerResolvable(endpoints[i]))
 | |
| 			}
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	// Attempt to load all `format.json` from all disks.
 | |
| 	formatConfigs, sErrs := loadFormatErasureAll(storageDisks, false)
 | |
| 	// Check if we have
 | |
| 	for i, sErr := range sErrs {
 | |
| 		// print the error, nonetheless, which is perhaps unhandled
 | |
| 		if sErr != errUnformattedDisk && sErr != errDiskNotFound && retryCount >= 5 {
 | |
| 			if sErr != nil {
 | |
| 				logger.Info("Unable to read 'format.json' from %s: %v\n", endpoints[i], sErr)
 | |
| 			}
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	// Pre-emptively check if one of the formatted disks
 | |
| 	// is invalid. This function returns success for the
 | |
| 	// most part unless one of the formats is not consistent
 | |
| 	// with expected Erasure format. For example if a user is
 | |
| 	// trying to pool FS backend into an Erasure set.
 | |
| 	if err = checkFormatErasureValues(formatConfigs, setDriveCount); err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	// All disks report unformatted we should initialized everyone.
 | |
| 	if shouldInitErasureDisks(sErrs) && firstDisk {
 | |
| 		logger.Info("Formatting %s zone, %v set(s), %v drives per set.",
 | |
| 			humanize.Ordinal(zoneCount), setCount, setDriveCount)
 | |
| 
 | |
| 		// Initialize erasure code format on disks
 | |
| 		format, err = initFormatErasure(GlobalContext, storageDisks, setCount, setDriveCount, "", deploymentID, sErrs)
 | |
| 		if err != nil {
 | |
| 			return nil, nil, err
 | |
| 		}
 | |
| 
 | |
| 		// Assign globalDeploymentID on first run for the
 | |
| 		// minio server managing the first disk
 | |
| 		globalDeploymentID = format.ID
 | |
| 		return storageDisks, format, nil
 | |
| 	}
 | |
| 
 | |
| 	// Return error when quorum unformatted disks - indicating we are
 | |
| 	// waiting for first server to be online.
 | |
| 	if quorumUnformattedDisks(sErrs) && !firstDisk {
 | |
| 		return nil, nil, errNotFirstDisk
 | |
| 	}
 | |
| 
 | |
| 	// Return error when quorum unformatted disks but waiting for rest
 | |
| 	// of the servers to be online.
 | |
| 	if quorumUnformattedDisks(sErrs) && firstDisk {
 | |
| 		return nil, nil, errFirstDiskWait
 | |
| 	}
 | |
| 
 | |
| 	// Mark all root disks down
 | |
| 	markRootDisksAsDown(storageDisks, sErrs)
 | |
| 
 | |
| 	// Following function is added to fix a regressions which was introduced
 | |
| 	// in release RELEASE.2018-03-16T22-52-12Z after migrating v1 to v2 to v3.
 | |
| 	// This migration failed to capture '.This' field properly which indicates
 | |
| 	// the disk UUID association. Below function is called to handle and fix
 | |
| 	// this regression, for more info refer https://github.com/minio/minio/issues/5667
 | |
| 	if err = fixFormatErasureV3(storageDisks, endpoints, formatConfigs); err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	// If any of the .This field is still empty, we return error.
 | |
| 	if formatErasureV3ThisEmpty(formatConfigs) {
 | |
| 		return nil, nil, errErasureV3ThisEmpty
 | |
| 	}
 | |
| 
 | |
| 	format, err = getFormatErasureInQuorum(formatConfigs)
 | |
| 	if err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	if format.ID == "" {
 | |
| 		// Not a first disk, wait until first disk fixes deploymentID
 | |
| 		if !firstDisk {
 | |
| 			return nil, nil, errNotFirstDisk
 | |
| 		}
 | |
| 		if err = formatErasureFixDeploymentID(endpoints, storageDisks, format); err != nil {
 | |
| 			return nil, nil, err
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	globalDeploymentID = format.ID
 | |
| 
 | |
| 	if err = formatErasureFixLocalDeploymentID(endpoints, storageDisks, format); err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	// The will always recreate some directories inside .minio.sys of
 | |
| 	// the local disk such as tmp, multipart and background-ops
 | |
| 	initErasureMetaVolumesInLocalDisks(storageDisks, formatConfigs)
 | |
| 
 | |
| 	return storageDisks, format, nil
 | |
| }
 | |
| 
 | |
| // Format disks before initialization of object layer.
 | |
| func waitForFormatErasure(firstDisk bool, endpoints Endpoints, zoneCount, setCount, setDriveCount int, deploymentID string) ([]StorageAPI, *formatErasureV3, error) {
 | |
| 	if len(endpoints) == 0 || setCount == 0 || setDriveCount == 0 {
 | |
| 		return nil, nil, errInvalidArgument
 | |
| 	}
 | |
| 
 | |
| 	if err := formatErasureMigrateLocalEndpoints(endpoints); err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	if err := formatErasureCleanupTmpLocalEndpoints(endpoints); err != nil {
 | |
| 		return nil, nil, err
 | |
| 	}
 | |
| 
 | |
| 	// prepare getElapsedTime() to calculate elapsed time since we started trying formatting disks.
 | |
| 	// All times are rounded to avoid showing milli, micro and nano seconds
 | |
| 	formatStartTime := time.Now().Round(time.Second)
 | |
| 	getElapsedTime := func() string {
 | |
| 		return time.Now().Round(time.Second).Sub(formatStartTime).String()
 | |
| 	}
 | |
| 
 | |
| 	// Wait on each try for an update.
 | |
| 	ticker := time.NewTicker(500 * time.Millisecond)
 | |
| 	defer ticker.Stop()
 | |
| 	var tries int
 | |
| 	for {
 | |
| 		select {
 | |
| 		case <-ticker.C:
 | |
| 			storageDisks, format, err := connectLoadInitFormats(tries, firstDisk, endpoints, zoneCount, setCount, setDriveCount, deploymentID)
 | |
| 			if err != nil {
 | |
| 				tries++
 | |
| 				switch err {
 | |
| 				case errNotFirstDisk:
 | |
| 					// Fresh setup, wait for first server to be up.
 | |
| 					logger.Info("Waiting for the first server to format the disks.")
 | |
| 					continue
 | |
| 				case errFirstDiskWait:
 | |
| 					// Fresh setup, wait for other servers to come up.
 | |
| 					logger.Info("Waiting for all other servers to be online to format the disks.")
 | |
| 					continue
 | |
| 				case errErasureReadQuorum:
 | |
| 					// no quorum available continue to wait for minimum number of servers.
 | |
| 					logger.Info("Waiting for a minimum of %d disks to come online (elapsed %s)\n", len(endpoints)/2, getElapsedTime())
 | |
| 					continue
 | |
| 				case errErasureV3ThisEmpty:
 | |
| 					// need to wait for this error to be healed, so continue.
 | |
| 					continue
 | |
| 				default:
 | |
| 					// For all other unhandled errors we exit and fail.
 | |
| 					return nil, nil, err
 | |
| 				}
 | |
| 			}
 | |
| 			return storageDisks, format, nil
 | |
| 		case <-globalOSSignalCh:
 | |
| 			return nil, nil, fmt.Errorf("Initializing data volumes gracefully stopped")
 | |
| 		}
 | |
| 	}
 | |
| }
 |