MinIO大杂烩

{
  "Versions": [
    {
      "Header": {
        "EcM": 1,
        "EcN": 0,
        "Flags": 2,
        "ModTime": "2025-01-23T15:27:45.311572+08:00",
        "Signature": "d0c2b58b",
        "Type": 1,
        "VersionID": "00000000000000000000000000000000"
      },
      "Idx": 0,
      "Metadata": {
        "Type": 1,
        "V2Obj": {
          "CSumAlgo": 1,
          "DDir": "74hQxU7FTrq56ShK8pjqAA==",
          "EcAlgo": 1,
          "EcBSize": 1048576,
          "EcDist": [1],
          "EcIndex": 1,
          "EcM": 1,
          "EcN": 0,
          "ID": "AAAAAAAAAAAAAAAAAAAAAA==",
          "MTime": 1737617265311572000,
          "MetaSys": {},
          "MetaUsr": {
            "content-type": "text/plain",
            "etag": "90a1a2b65a4e40d55d758f2a59fe33b4"
          },
          "PartASizes": [2097152],
          "PartETags": null,
          "PartNums": [1],
          "PartSizes": [2097152],
          "Size": 2097152
        },
        "v": 1734527744
      }
    }
  ]
}

.
├── a.txt
│   ├── ef8850c5-4ec5-4eba-b9e9-284af298ea00
│   │   └── part.1
│   └── xl.meta
└── b.txt
    └── xl.meta

// Run the app - exit on error.
if err := newApp(appName).Run(args); err != nil {
  os.Exit(1) //nolint:gocritic
}

var serverCmd = cli.Command{
	Name:   "server",
	Usage:  "start object storage server",
	Flags:  append(ServerFlags, GlobalFlags...),
	Action: serverMain,

server
http://127.0.0.1:/Users/hanjing/mnt/minio0{1...3}
http://127.0.0.1:/Users/hanjing/mnt/minio0{4...6}

signal.Notify(globalOSSignalCh, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
go handleSignals()

	// Handle all server command args and build the disks layout
	bootstrapTrace("serverHandleCmdArgs", func() {
    // 这里确定了erasure set size的大小
		err := buildServerCtxt(ctx, &globalServerCtxt)
		logger.FatalIf(err, "Unable to prepare the list of endpoints")

		serverHandleCmdArgs(globalServerCtxt)
	})

type Resolver struct {
	// Timeout defines the maximum allowed time allowed for a lookup.
	Timeout time.Duration

	// Resolver is used to perform actual DNS lookup. If nil,
	// net.DefaultResolver is used instead.
	Resolver DNSResolver

	once  sync.Once
	mu    sync.RWMutex
	cache map[string]*cacheEntry
}

globalDNSCache = &dnscache.Resolver{
  Timeout: 5 * time.Second,
}

func runDNSCache(ctx *cli.Context) {
	dnsTTL := ctx.Duration("dns-cache-ttl")
	// Check if we have configured a custom DNS cache TTL.
	if dnsTTL <= 0 {
		if orchestrated {
			dnsTTL = 30 * time.Second
		} else {
			dnsTTL = 10 * time.Minute
		}
	}

	// Call to refresh will refresh names in cache.
	go func() {
		// Baremetal setups set DNS refresh window up to dnsTTL duration.
		t := time.NewTicker(dnsTTL)
		defer t.Stop()
		for {
			select {
			case <-t.C:
				globalDNSCache.Refresh()

			case <-GlobalContext.Done():
				return
			}
		}
	}()
}

// serverCtxt保存了磁盘布局
type disksLayout struct {
  // 是否使用拓展表达式
	legacy bool
  // server pool的集合
	pools  []poolDisksLayout
}
type poolDisksLayout struct {
  // server pool对应的命令行命令
	cmdline string
  // layout的第一位表示不同的erasure set，第二维表示同一个erasure set中不同的磁盘路径
	layout  [][]string
}

// mergeDisksLayoutFromArgs supports with and without ellipses transparently.
// 构造网络拓扑
func mergeDisksLayoutFromArgs(args []string, ctxt *serverCtxt) (err error) {
	if len(args) == 0 {
		return errInvalidArgument
	}

	ok := true
	// ok 表示是否使用拓展表达式，true表示不使用拓展表达式
	// 只要在其中一个arg中使用拓展表达式，结果均为false
	for _, arg := range args {
		ok = ok && !ellipses.HasEllipses(arg)
	}

	var setArgs [][]string

	// 通过环境变量得到erasure set的大小，默认为0
	v, err := env.GetInt(EnvErasureSetDriveCount, 0)
	if err != nil {
		return err
	}
	setDriveCount := uint64(v)

	// None of the args have ellipses use the old style.
	if ok {
		setArgs, err = GetAllSets(setDriveCount, args...)
		if err != nil {
			return err
		}
		// 所有的参数组成一个server pool
		ctxt.Layout = disksLayout{
			legacy: true,
			pools:  []poolDisksLayout{{layout: setArgs, cmdline: strings.Join(args, " ")}},
		}
		return
	}

	for _, arg := range args {
		if !ellipses.HasEllipses(arg) && len(args) > 1 {
			// TODO: support SNSD deployments to be decommissioned in future
			return fmt.Errorf("all args must have ellipses for pool expansion (%w) args: %s", errInvalidArgument, args)
		}
		setArgs, err = GetAllSets(setDriveCount, arg)
		if err != nil {
			return err
		}
		ctxt.Layout.pools = append(ctxt.Layout.pools, poolDisksLayout{cmdline: arg, layout: setArgs})
	}
	return
}

// Endpoint set represents parsed ellipses values, also provides
// methods to get the sets of endpoints.
type endpointSet struct {
  // 解析终端字符串得到的arg pattern，如果有多个ellipses，对应多个`Pattern`
	argPatterns []ellipses.ArgPattern
	endpoints   []string   // Endpoints saved from previous GetEndpoints().
  // 对于ellipses-style的参数
  // setIndexes对应一行，记录了server pool size /setSize 个 setSize值
	setIndexes  [][]uint64 // All the sets.
}
type ArgPattern []Pattern
// Pattern - ellipses pattern, describes the range and also the
// associated prefix and suffixes.
type Pattern struct {
	Prefix string
	Suffix string
	Seq    []string
}

func (s endpointSet) Get() (sets [][]string) {
	k := uint64(0)
	endpoints := s.getEndpoints()
	for i := range s.setIndexes {
		for j := range s.setIndexes[i] {
			sets = append(sets, endpoints[k:s.setIndexes[i][j]+k])
			k = s.setIndexes[i][j] + k
		}
	}

	return sets
}

	globalEndpoints, setupType, err = createServerEndpoints(globalMinioAddr, ctxt.Layout.pools, ctxt.Layout.legacy)
	logger.FatalIf(err, "Invalid command line arguments")
	globalNodes = globalEndpoints.GetNodes()

	globalIsErasure = (setupType == ErasureSetupType)
	globalIsDistErasure = (setupType == DistErasureSetupType)
	if globalIsDistErasure {
		globalIsErasure = true
	}
	globalIsErasureSD = (setupType == ErasureSDSetupType)
	if globalDynamicAPIPort && globalIsDistErasure {
		logger.FatalIf(errInvalidArgument, "Invalid --address=\"%s\", port '0' is not allowed in a distributed erasure coded setup", ctxt.Addr)
	}

	globalLocalNodeName = GetLocalPeer(globalEndpoints, globalMinioHost, globalMinioPort)
	nodeNameSum := sha256.Sum256([]byte(globalLocalNodeName))
	globalLocalNodeNameHex = hex.EncodeToString(nodeNameSum[:])

	// Initialize, see which NIC the service is running on, and save it as global value
	setGlobalInternodeInterface(ctxt.Interface)

// EndpointServerPools是 PoolEndpoints的集合，实际上描述整个部署的拓扑结构
type EndpointServerPools []PoolEndpoints
// PoolEndpoints represent endpoints in a given pool
// along with its setCount and setDriveCount.
// PoolEndpoints表示一个server pool的结构
type PoolEndpoints struct {
	// indicates if endpoints are provided in non-ellipses style
  // legacy 表示 是否使用遗留的方法表示终端，而不使用省略号表达式
	Legacy       bool
  // SetCount表示 server pool中的 erasure set的数量
	SetCount     int
  // DrivesPerSet 表示一个erasure set中的磁盘数量
	DrivesPerSet int
  // type Endpoints []Endpoint
  // 表示一个server pool中的所有disk
	Endpoints    Endpoints
  // server pool对应的命令行指令
	CmdLine      string
  // 操作系统信息
	Platform     string
}

type Endpoint struct {
	*url.URL
  // 如果是单个目录的输入，则 IsLocal为true
  // 如果输入参数ip是本地ip，IsLocal也为true
  // 其他情况下为false
	IsLocal bool

	PoolIdx, SetIdx, DiskIdx int
}

// SetupType - enum for setup type.
type SetupType int

const (
	// UnknownSetupType - starts with unknown setup type.
	UnknownSetupType SetupType = iota

	// FSSetupType - FS setup type enum.
	FSSetupType

	// ErasureSDSetupType - Erasure single drive setup enum.
	ErasureSDSetupType

	// ErasureSetupType - Erasure setup type enum.
	ErasureSetupType

	// DistErasureSetupType - Distributed Erasure setup type enum.
	DistErasureSetupType
)

// Returns the mode in which MinIO is running
func getMinioMode() string {
	switch {
	case globalIsDistErasure:
		return globalMinioModeDistErasure
	case globalIsErasure:
		return globalMinioModeErasure
	case globalIsErasureSD:
		return globalMinioModeErasureSD
	default:
		return globalMinioModeFS
	}
}

	// Configure server.
	bootstrapTrace("configureServer", func() {
		handler, err := configureServerHandler(globalEndpoints)
		if err != nil {
			logger.Fatal(config.ErrUnexpectedError(err), "Unable to configure one of server's RPC services")
		}
		// Allow grid to start after registering all services.
		close(globalGridStart)
		close(globalLockGridStart)

		httpServer := xhttp.NewServer(getServerListenAddrs()).
			UseHandler(setCriticalErrorHandler(corsHandler(handler))).
			UseTLSConfig(newTLSConfig(getCert)).
			UseIdleTimeout(globalServerCtxt.IdleTimeout).
			UseReadTimeout(globalServerCtxt.IdleTimeout).
			UseWriteTimeout(globalServerCtxt.IdleTimeout).
			UseReadHeaderTimeout(globalServerCtxt.ReadHeaderTimeout).
			UseBaseContext(GlobalContext).
			UseCustomLogger(log.New(io.Discard, "", 0)). // Turn-off random logging by Go stdlib
			UseTCPOptions(globalTCPOptions)

		httpServer.TCPOptions.Trace = bootstrapTraceMsg
		go func() {
			serveFn, err := httpServer.Init(GlobalContext, func(listenAddr string, err error) {
				bootLogIf(GlobalContext, fmt.Errorf("Unable to listen on `%s`: %v", listenAddr, err))
			})
			if err != nil {
				globalHTTPServerErrorCh <- err
				return
			}
			globalHTTPServerErrorCh <- serveFn()
		}()

		setHTTPServer(httpServer)
	})

// configureServer handler returns final handler for the http server.
func configureServerHandler(endpointServerPools EndpointServerPools) (http.Handler, error) {
	// Initialize router. `SkipClean(true)` stops minio/mux from
	// normalizing URL path minio/minio#3256
	router := mux.NewRouter().SkipClean(true).UseEncodedPath()

	// Initialize distributed NS lock.
	if globalIsDistErasure {
		registerDistErasureRouters(router, endpointServerPools)
	}

	// Add Admin router, all APIs are enabled in server mode.
	registerAdminRouter(router, true)

	// Add healthCheck router
	registerHealthCheckRouter(router)

	// Add server metrics router
	registerMetricsRouter(router)

	// Add STS router always.
	registerSTSRouter(router)

	// Add KMS router
	registerKMSRouter(router)

	// Add API router
	registerAPIRouter(router)

	router.Use(globalMiddlewares...)

	return router, nil
}

// GetObject
router.Methods(http.MethodGet).Path("/{object:.+}").
  HandlerFunc(s3APIMiddleware(api.GetObjectHandler, traceHdrsS3HFlag))

// Initialize API.
api := objectAPIHandlers{
  ObjectAPI: newObjectLayerFn,
}
// objectAPIHandlers implements and provides http handlers for S3 API.
type objectAPIHandlers struct {
	ObjectAPI func() ObjectLayer
}
func newObjectLayerFn() ObjectLayer {
	globalObjLayerMutex.RLock()
	defer globalObjLayerMutex.RUnlock()
	return globalObjectAPI
}

	var newObject ObjectLayer
	bootstrapTrace("newObjectLayer", func() {
		var err error
		newObject, err = newObjectLayer(GlobalContext, globalEndpoints)
		if err != nil {
			logFatalErrs(err, Endpoint{}, true)
		}
	})

// DefaultParityBlocks returns default parity blocks for 'drive' count
func DefaultParityBlocks(drive int) int {
	switch drive {
	case 1:
		return 0
	case 3, 2:
		return 1
	case 4, 5:
		return 2
	case 6, 7:
		return 3
	default:
		return 4
	}
}

// erasureServerPools
// minio 服务可以由多个server pool 组成，用来水平拓展
// erasureServerPools是server pool的集合
type erasureServerPools struct {
	poolMetaMutex sync.RWMutex
	poolMeta      poolMeta

	rebalMu   sync.RWMutex
	rebalMeta *rebalanceMeta

	deploymentID     [16]byte
	distributionAlgo string

	// server pool 由多个erasure set 组成
	// 这里的erasureSets结构实际上指单个 server pool
	serverPools []*erasureSets

	// Active decommission canceler
	decommissionCancelers []context.CancelFunc

	s3Peer *S3PeerSys

	mpCache *xsync.MapOf[string, MultipartInfo]
}

// erasureSets implements ObjectLayer combining a static list of erasure coded
// object sets. NOTE: There is no dynamic scaling allowed or intended in
// current design.
// server pool 由多个erasure set 组成
// 这里的erasureSets结构实际上指单个 server pool
// 上面这段话的意思是不能动态扩展server pool，初始指定后就不能再修改了
type erasureSets struct {
	sets []*erasureObjects

	// Reference format.
	format *formatErasureV3

	// erasureDisks mutex to lock erasureDisks.
	erasureDisksMu sync.RWMutex

	// Re-ordered list of disks per set.
	erasureDisks [][]StorageAPI

	// Distributed locker clients.
	erasureLockers setsDsyncLockers

	// Distributed lock owner (constant per running instance).
	erasureLockOwner string

	// List of endpoints provided on the command line.
	endpoints PoolEndpoints

	// String version of all the endpoints, an optimization
	// to avoid url.String() conversion taking CPU on
	// large disk setups.
	endpointStrings []string

	// Total number of sets and the number of disks per set.
	setCount, setDriveCount int
	defaultParityCount      int

	poolIndex int

	// Distribution algorithm of choice.
	distributionAlgo string
	deploymentID     [16]byte

	lastConnectDisksOpTime time.Time
}

// erasureObjects - Implements ER object layer.
// 表示一个erasure Set
type erasureObjects struct {
	setDriveCount      int
	defaultParityCount int

	setIndex  int
	poolIndex int

	// getDisks returns list of storageAPIs.
	getDisks func() []StorageAPI

	// getLockers returns list of remote and local lockers.
	getLockers func() ([]dsync.NetLocker, string)

	// getEndpoints returns list of endpoint belonging this set.
	// some may be local and some remote.
	getEndpoints func() []Endpoint

	// getEndpoints returns list of endpoint strings belonging this set.
	// some may be local and some remote.
	getEndpointStrings func() []string

	// Locker mutex map.
	nsMutex *nsLockMap
}

// Depending on the disk type network or local, initialize storage API.
func newStorageAPI(endpoint Endpoint, opts storageOpts) (storage StorageAPI, err error) {
	if endpoint.IsLocal {
		storage, err := newXLStorage(endpoint, opts.cleanUp)
		if err != nil {
			return nil, err
		}
		return newXLStorageDiskIDCheck(storage, opts.healthCheck), nil
	}

	return newStorageRESTClient(endpoint, opts.healthCheck, globalGrid.Load())
}

	// Return an error if ODirect is not supported. Single disk will have
	// oDirect off.
	// 在类似unix的平台上 disk.ODirectPlatform应该为true
	if globalIsErasureSD || !disk.ODirectPlatform {
		s.oDirect = false
	} else if err := s.checkODirectDiskSupport(info.FSType); err == nil {
		s.oDirect = true
	} else {
		return s, err
	}

// getDiskInfo returns given disk information.
func getDiskInfo(drivePath string) (di disk.Info, rootDrive bool, err error) {
	if err = checkPathLength(drivePath); err == nil {
		di, err = disk.GetInfo(drivePath, false)

		if !globalIsCICD && !globalIsErasureSD {
			if globalRootDiskThreshold > 0 {
				// Use MINIO_ROOTDISK_THRESHOLD_SIZE to figure out if
				// this disk is a root disk. treat those disks with
				// size less than or equal to the threshold as rootDrives.
				rootDrive = di.Total <= globalRootDiskThreshold
			} else {
				rootDrive, err = disk.IsRootDisk(drivePath, SlashSeparator)
			}
		}
	}

const (
	// Represents Erasure backend.
	formatBackendErasure = "xl"

	// Represents Erasure backend - single drive
	formatBackendErasureSingle = "xl-single"

	// formatErasureV1.Erasure.Version - version '1'.
	formatErasureVersionV1 = "1"

	// formatErasureV2.Erasure.Version - version '2'.
	formatErasureVersionV2 = "2"

	// formatErasureV3.Erasure.Version - version '3'.
	formatErasureVersionV3 = "3"

	// Distribution algorithm used, legacy
	formatErasureVersionV2DistributionAlgoV1 = "CRCMOD"

	// Distributed algorithm used, with N/2 default parity
	formatErasureVersionV3DistributionAlgoV2 = "SIPMOD"

	// Distributed algorithm used, with EC:4 default parity
	formatErasureVersionV3DistributionAlgoV3 = "SIPMOD+PARITY"
)

// PutObject
// http处理函数
router.Methods(http.MethodPut).Path("/{object:.+}").
  HandlerFunc(s3APIMiddleware(api.PutObjectHandler, traceHdrsS3HFlag))
// objectLayer层接口
putObject = objectAPI.PutObject

// Validate storage class metadata if present
// x-amz-storage-class minio 只支持 REDUCED_REDUNDANCY 和 Standard
if sc := r.Header.Get(xhttp.AmzStorageClass); sc != "" {
    if !storageclass.IsValid(sc) {
       writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidStorageClass), r.URL)
       return
    }
}

// maximum Upload size for objects in a single operation
// 5TiB
if isMaxObjectSize(size) {
    writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrEntityTooLarge), r.URL)
    return
}

	objInfo, err := putObject(ctx, bucket, object, pReader, opts)

// Returns always a same erasure coded set for a given input.
func (s *erasureSets) getHashedSetIndex(input string) int {
	// 通过hash得到对应erasure set，以下要素可能影响hash结果
	return hashKey(s.distributionAlgo, input, len(s.sets), s.deploymentID)
}

// PutObject - creates an object upon reading from the input stream
// until EOF, erasure codes the data across all disk and additionally
// writes `xl.meta` which carries the necessary metadata for future
// object operations.
func (er erasureObjects) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) {
	return er.putObject(ctx, bucket, object, data, opts)
}

// PutObject - writes an object to least used erasure pool.
func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
	// Validate put object input args.
	if err := checkPutObjectArgs(ctx, bucket, object); err != nil {
		return ObjectInfo{}, err
	}

	object = encodeDirObject(object)
	if z.SinglePool() {
		return z.serverPools[0].PutObject(ctx, bucket, object, data, opts)
	}

	idx, err := z.getPoolIdx(ctx, bucket, object, data.Size())
	if err != nil {
		return ObjectInfo{}, err
	}

	if opts.DataMovement && idx == opts.SrcPoolIdx {
		return ObjectInfo{}, DataMovementOverwriteErr{
			Bucket:    bucket,
			Object:    object,
			VersionID: opts.VersionID,
			Err:       errDataMovementSrcDstPoolSame,
		}
	}

	return z.serverPools[idx].PutObject(ctx, bucket, object, data, opts)
}

// getPoolIdx returns the found previous object and its corresponding pool idx,
// if none are found falls back to most available space pool, this function is
// designed to be only used by PutObject, CopyObject (newObject creation) and NewMultipartUpload.
func (z *erasureServerPools) getPoolIdx(ctx context.Context, bucket, object string, size int64) (idx int, err error) {
	idx, err = z.getPoolIdxExistingWithOpts(ctx, bucket, object, ObjectOptions{
		SkipDecommissioned: true,
		SkipRebalancing:    true,
	})
	if err != nil && !isErrObjectNotFound(err) {
		return idx, err
	}

	if isErrObjectNotFound(err) {
		idx = z.getAvailablePoolIdx(ctx, bucket, object, size)
		if idx < 0 {
			return -1, toObjectErr(errDiskFull)
		}
	}

	return idx, nil
}

// GetObjectInfo - reads object metadata from the hashedSet based on the object name.
func (s *erasureSets) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
	set := s.getHashedSet(object)
	return set.GetObjectInfo(ctx, bucket, object, opts)
}
// Returns always a same erasure coded set for a given input.
func (s *erasureSets) getHashedSet(input string) (set *erasureObjects) {
	return s.sets[s.getHashedSetIndex(input)]
}
func hashKey(algo string, key string, cardinality int, id [16]byte) int {
	switch algo {
	case formatErasureVersionV2DistributionAlgoV1:
		return crcHashMod(key, cardinality)
	case formatErasureVersionV3DistributionAlgoV2, formatErasureVersionV3DistributionAlgoV3:
		return sipHashMod(key, cardinality, id)
	default:
		// Unknown algorithm returns -1, also if cardinality is lesser than 0.
		return -1
	}
}

// func (er erasureObjects) putObject
if !opts.NoLock {
		lk := er.NewNSLock(bucket, object)
		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
		if err != nil {
			return ObjectInfo{}, err
		}
		ctx = lkctx.Context()
		defer lk.Unlock(lkctx)
	}

type erasureObjects struct {
	// getLockers returns list of remote and local lockers.
	getLockers func() ([]dsync.NetLocker, string)

	// Locker mutex map.
	nsMutex *nsLockMap
}

type erasureSets struct {
	sets []*erasureObjects

	// Distributed locker clients.
	erasureLockers setsDsyncLockers

  // Distributed lock owner (constant per running instance).
	erasureLockOwner string
  // setsDsyncLockers is encapsulated type for Close()
	type setsDsyncLockers [][]dsync.NetLocker

func (s *erasureSets) GetLockers(setIndex int) func() ([]dsync.NetLocker, string) {
	return func() ([]dsync.NetLocker, string) {
		lockers := make([]dsync.NetLocker, len(s.erasureLockers[setIndex]))
		copy(lockers, s.erasureLockers[setIndex])
    // erasureLockerOwner实际上是globalLocalNodeName
    // The name of this local node, fetched from arguments
	  // globalLocalNodeName    string
		return lockers, s.erasureLockOwner
	}
}

// nsLockMap - namespace lock map, provides primitives to Lock,
// Unlock, RLock and RUnlock.
type nsLockMap struct {
	// Indicates if namespace is part of a distributed setup.
	isDistErasure bool
	lockMap       map[string]*nsLock
	lockMapMutex  sync.Mutex
}
// newNSLock - return a new name space lock map.
func newNSLock(isDistErasure bool) *nsLockMap {
	nsMutex := nsLockMap{
		isDistErasure: isDistErasure,
	}
	if isDistErasure {
		return &nsMutex
	}
	nsMutex.lockMap = make(map[string]*nsLock)
	return &nsMutex
}

// GetObject
router.Methods(http.MethodGet).Path("/{object:.+}").
  HandlerFunc(s3APIMiddleware(api.GetObjectHandler, traceHdrsS3HFlag))