// This is not implemented/needed anymore, look for xl-sets.HealObjects()
188883
func (xl xlObjects) HealObjects(ctx context.Context, bucket, prefix string, healFn func(string, string) error) (e error) {
	return nil
}
1364361
func (h *healSequence) healMinioSysMeta(metaPrefix string) func() error {
	return func() error {
		// Get current object layer instance.
		objectAPI := newObjectLayerFn()
		if objectAPI == nil {
			return errServerNotInitialized
		}

		// NOTE: Healing on meta is run regardless
		// of any bucket being selected, this is to ensure that
		// meta are always upto date and correct.
		return objectAPI.HealObjects(h.ctx, minioMetaBucket, metaPrefix, func(bucket string, object string) error {
			if h.isQuitting() {
				return errHealStopSignalled
			}
			res, herr := objectAPI.HealObject(h.ctx, bucket, object, h.settings.DryRun, h.settings.Remove, h.settings.ScanMode)
			// Object might have been deleted, by the time heal
			// was attempted we ignore this object an move on.
			if isErrObjectNotFound(herr) {
				return nil
			}
			if herr != nil {
				return herr
			}
			res.Type = madmin.HealItemBucketMetadata
			return h.pushHealResultItem(res)
		})
	}
}
811673
func (xl xlObjects) HealObject(ctx context.Context, bucket, object string, dryRun bool, remove bool, scanMode madmin.HealScanMode) (hr madmin.HealResultItem, err error) {
	// Create context that also contains information about the object and bucket.
	// The top level handler might not have this information.
	reqInfo := logger.GetReqInfo(ctx)
	var newReqInfo *logger.ReqInfo
	if reqInfo != nil {
		newReqInfo = logger.NewReqInfo(reqInfo.RemoteHost, reqInfo.UserAgent, reqInfo.DeploymentID, reqInfo.RequestID, reqInfo.API, bucket, object)
	} else {
		newReqInfo = logger.NewReqInfo("", "", globalDeploymentID, "", "Heal", bucket, object)
	}
	healCtx := logger.SetReqInfo(context.Background(), newReqInfo)

	// Healing directories handle it separately.
	if hasSuffix(object, slashSeparator) {
		return xl.healObjectDir(healCtx, bucket, object, dryRun)
	}

	storageDisks := xl.getDisks()

	// Read metadata files from all the disks
	partsMetadata, errs := readAllXLMetadata(healCtx, storageDisks, bucket, object)

	// Check if the object is dangling, if yes and user requested
	// remove we simply delete it from namespace.
	if m, ok := isObjectDangling(partsMetadata, errs, []error{}); ok {
		writeQuorum := m.Erasure.DataBlocks + 1
		if m.Erasure.DataBlocks == 0 {
			writeQuorum = len(xl.getDisks())/2 + 1
		}
		if !dryRun && remove {
			err = xl.deleteObject(healCtx, bucket, object, writeQuorum, false)
		}
		return defaultHealResult(xlMetaV1{}, storageDisks, errs, bucket, object), err
	}

	latestXLMeta, err := getLatestXLMeta(healCtx, partsMetadata, errs)
	if err != nil {
		return defaultHealResult(xlMetaV1{}, storageDisks, errs, bucket, object), toObjectErr(err, bucket, object)
	}

	// Lock the object before healing.
	objectLock := xl.nsMutex.NewNSLock(bucket, object)
	if lerr := objectLock.GetRLock(globalHealingTimeout); lerr != nil {
		return defaultHealResult(latestXLMeta, storageDisks, errs, bucket, object), lerr
	}
	defer objectLock.RUnlock()

	errCount := 0
	for _, err := range errs {
		if err != nil {
			errCount++
		}
	}

	if errCount == len(errs) {
		// Only if we get errors from all the disks we return error. Else we need to
		// continue to return filled madmin.HealResultItem struct which includes info
		// on what disks the file is available etc.
		if reducedErr := reduceReadQuorumErrs(ctx, errs, nil, latestXLMeta.Erasure.DataBlocks); reducedErr != nil {
			if m, ok := isObjectDangling(partsMetadata, errs, []error{}); ok {
				writeQuorum := m.Erasure.DataBlocks + 1
				if m.Erasure.DataBlocks == 0 {
					writeQuorum = len(storageDisks)/2 + 1
				}
				if !dryRun && remove {
					err = xl.deleteObject(ctx, bucket, object, writeQuorum, false)
				}
			}
			return defaultHealResult(latestXLMeta, storageDisks, errs, bucket, object), toObjectErr(reducedErr, bucket, object)
		}
	}

	// Heal the object.
	return xl.healObject(healCtx, bucket, object, partsMetadata, errs, latestXLMeta, dryRun, remove, scanMode)
}
1350739
func (xl xlObjects) parentDirIsObject(ctx context.Context, bucket, parent string) bool {
	var isParentDirObject func(string) bool
	isParentDirObject = func(p string) bool {
		if p == "." || p == "/" {
			return false
		}
		if xl.isObject(bucket, p) {
			// If there is already a file at prefix "p", return true.
			return true
		}
		// Check if there is a file as one of the parent paths.
		return isParentDirObject(path.Dir(p))
	}
	return isParentDirObject(parent)
}
742621
func (s *xlSets) HealObject(ctx context.Context, bucket, object string, dryRun, remove bool, scanMode madmin.HealScanMode) (madmin.HealResultItem, error) {
	return s.getHashedSet(object).HealObject(ctx, bucket, object, dryRun, remove, scanMode)
}
1266520
func flattenObjects(node ast.Node) {
	ast.Walk(node, func(n ast.Node) (ast.Node, bool) {
		// We only care about lists, because this is what we modify
		list, ok := n.(*ast.ObjectList)
		if !ok {
			return n, true
		}

		// Rebuild the item list
		items := make([]*ast.ObjectItem, 0, len(list.Items))
		frontier := make([]*ast.ObjectItem, len(list.Items))
		copy(frontier, list.Items)
		for len(frontier) > 0 {
			// Pop the current item
			n := len(frontier)
			item := frontier[n-1]
			frontier = frontier[:n-1]

			switch v := item.Val.(type) {
			case *ast.ObjectType:
				items, frontier = flattenObjectType(v, item, items, frontier)
			case *ast.ListType:
				items, frontier = flattenListType(v, item, items, frontier)
			default:
				items = append(items, item)
			}
		}

		// Reverse the list since the frontier model runs things backwards
		for i := len(items)/2 - 1; i >= 0; i-- {
			opp := len(items) - 1 - i
			items[i], items[opp] = items[opp], items[i]
		}

		// Done! Set the original items
		list.Items = items
		return n, true
	})
}
156847
func (xl xlObjects) DeleteObject(ctx context.Context, bucket, object string) (err error) {
	// Acquire a write lock before deleting the object.
	objectLock := xl.nsMutex.NewNSLock(bucket, object)
	if perr := objectLock.GetLock(globalOperationTimeout); perr != nil {
		return perr
	}
	defer objectLock.Unlock()

	if err = checkDelObjArgs(ctx, bucket, object); err != nil {
		return err
	}

	var writeQuorum int
	var isObjectDir = hasSuffix(object, slashSeparator)

	if isObjectDir {
		_, err = xl.getObjectInfoDir(ctx, bucket, object)
		if err == errXLReadQuorum {
			if isObjectDirDangling(statAllDirs(ctx, xl.getDisks(), bucket, object)) {
				// If object is indeed dangling, purge it.
				err = nil
			}
		}
		if err != nil {
			return toObjectErr(err, bucket, object)
		}
	}

	if isObjectDir {
		writeQuorum = len(xl.getDisks())/2 + 1
	} else {
		// Read metadata associated with the object from all disks.
		partsMetadata, errs := readAllXLMetadata(ctx, xl.getDisks(), bucket, object)
		// get Quorum for this object
		_, writeQuorum, err = objectQuorumFromMeta(ctx, xl, partsMetadata, errs)
		if err != nil {
			return toObjectErr(err, bucket, object)
		}
	}

	// Delete the object on all disks.
	if err = xl.deleteObject(ctx, bucket, object, writeQuorum, isObjectDir); err != nil {
		return toObjectErr(err, bucket, object)
	}

	// Success.
	return nil
}
130697
func (xl xlObjects) ListBucketsHeal(ctx context.Context) ([]BucketInfo, error) {
	return nil, nil
}
230979
func newHealSequence(bucket, objPrefix, clientAddr string,
	numDisks int, hs madmin.HealOpts, forceStart bool) *healSequence {

	reqInfo := &logger.ReqInfo{RemoteHost: clientAddr, API: "Heal", BucketName: bucket}
	reqInfo.AppendTags("prefix", objPrefix)
	ctx := logger.SetReqInfo(context.Background(), reqInfo)

	return &healSequence{
		bucket:        bucket,
		objPrefix:     objPrefix,
		path:          pathJoin(bucket, objPrefix),
		startTime:     UTCNow(),
		clientToken:   mustGetUUID(),
		clientAddress: clientAddr,
		forceStarted:  forceStart,
		settings:      hs,
		currentStatus: healSequenceStatus{
			Summary:      healNotStartedStatus,
			HealSettings: hs,
			NumDisks:     numDisks,
			updateLock:   &sync.RWMutex{},
		},
		traverseAndHealDoneCh: make(chan error),
		stopSignalCh:          make(chan struct{}),
		ctx:                   ctx,
	}
}
469968
func (s *xlSets) CopyObjectPart(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, uploadID string, partID int,
	startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (partInfo PartInfo, err error) {
	destSet := s.getHashedSet(destObject)

	return destSet.PutObjectPart(ctx, destBucket, destObject, uploadID, partID, NewPutObjReader(srcInfo.Reader, nil, nil), dstOpts)
}
563168
func (i *ResourceInstance) HasObjects() bool {
	return i.Current != nil || len(i.Deposed) != 0
}
1001896
func (s *xlSets) ListObjectParts(ctx context.Context, bucket, object, uploadID string, partNumberMarker int, maxParts int, opts ObjectOptions) (result ListPartsInfo, err error) {
	return s.getHashedSet(object).ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
}
328631
func AddObjectLabels(obj runtime.Object, labels labels.Set) error {
	return AddObjectLabelsWithFlags(obj, labels, OverwriteExistingDstKey)
}
459437
func (s *xlSets) connectDisks() {
	for _, endpoint := range s.endpoints {
		if s.isConnected(endpoint) {
			continue
		}
		disk, format, err := connectEndpoint(endpoint)
		if err != nil {
			printEndpointError(endpoint, err)
			continue
		}
		i, j, err := findDiskIndex(s.format, format)
		if err != nil {
			// Close the internal connection to avoid connection leaks.
			disk.Close()
			printEndpointError(endpoint, err)
			continue
		}
		s.xlDisksMu.Lock()
		s.xlDisks[i][j] = disk
		s.xlDisksMu.Unlock()
	}
}
1021475
func (s *xlSets) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) {
	return s.getHashedSet(object).PutObject(ctx, bucket, object, data, opts)
}
905727
func (xl xlObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (oi ObjectInfo, e error) {
	cpSrcDstSame := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(dstBucket, dstObject))

	// Read metadata associated with the object from all disks.
	storageDisks := xl.getDisks()

	metaArr, errs := readAllXLMetadata(ctx, storageDisks, srcBucket, srcObject)

	// get Quorum for this object
	readQuorum, writeQuorum, err := objectQuorumFromMeta(ctx, xl, metaArr, errs)
	if err != nil {
		return oi, toObjectErr(err, srcBucket, srcObject)
	}

	if reducedErr := reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum); reducedErr != nil {
		return oi, toObjectErr(reducedErr, srcBucket, srcObject)
	}

	// List all online disks.
	_, modTime := listOnlineDisks(storageDisks, metaArr, errs)

	// Pick latest valid metadata.
	xlMeta, err := pickValidXLMeta(ctx, metaArr, modTime, readQuorum)
	if err != nil {
		return oi, toObjectErr(err, srcBucket, srcObject)
	}

	// Length of the file to read.
	length := xlMeta.Stat.Size

	// Check if this request is only metadata update.
	if cpSrcDstSame {
		// Update `xl.json` content on each disks.
		for index := range metaArr {
			metaArr[index].Meta = srcInfo.UserDefined
			metaArr[index].Meta["etag"] = srcInfo.ETag
		}

		var onlineDisks []StorageAPI

		tempObj := mustGetUUID()

		// Cleanup in case of xl.json writing failure
		defer xl.deleteObject(ctx, minioMetaTmpBucket, tempObj, writeQuorum, false)

		// Write unique `xl.json` for each disk.
		if onlineDisks, err = writeUniqueXLMetadata(ctx, storageDisks, minioMetaTmpBucket, tempObj, metaArr, writeQuorum); err != nil {
			return oi, toObjectErr(err, srcBucket, srcObject)
		}
		// Rename atomically `xl.json` from tmp location to destination for each disk.
		if _, err = renameXLMetadata(ctx, onlineDisks, minioMetaTmpBucket, tempObj, srcBucket, srcObject, writeQuorum); err != nil {
			return oi, toObjectErr(err, srcBucket, srcObject)
		}
		return xlMeta.ToObjectInfo(srcBucket, srcObject), nil
	}

	// Initialize pipe.
	pipeReader, pipeWriter := io.Pipe()

	go func() {
		var startOffset int64 // Read the whole file.
		if gerr := xl.getObject(ctx, srcBucket, srcObject, startOffset, length, pipeWriter, srcInfo.ETag, srcOpts); gerr != nil {
			pipeWriter.CloseWithError(toObjectErr(gerr, srcBucket, srcObject))
			return
		}
		pipeWriter.Close() // Close writer explicitly signaling we wrote all data.
	}()

	hashReader, err := hash.NewReader(pipeReader, length, "", "", length)
	if err != nil {
		logger.LogIf(ctx, err)
		return oi, toObjectErr(err, dstBucket, dstObject)
	}
	putOpts := ObjectOptions{UserDefined: srcInfo.UserDefined, ServerSideEncryption: dstOpts.ServerSideEncryption}
	objInfo, err := xl.putObject(ctx, dstBucket, dstObject, NewPutObjReader(hashReader, nil, nil), putOpts)
	if err != nil {
		return oi, toObjectErr(err, dstBucket, dstObject)
	}

	// Explicitly close the reader.
	pipeReader.Close()

	return objInfo, nil
}
47001
func GetAttrs(obj runtime.Object) (labels.Set, fields.Set, error) {
	persistentvolumeclaimObj, ok := obj.(*api.PersistentVolumeClaim)
	if !ok {
		return nil, nil, fmt.Errorf("not a persistentvolumeclaim")
	}
	return labels.Set(persistentvolumeclaimObj.Labels), PersistentVolumeClaimToSelectableFields(persistentvolumeclaimObj), nil
}