sys.hMutex.RLock()
	defer sys.hMutex.RUnlock()
	if h, ok := sys.hc[ep.Host]; ok {
		return !h.Online
	}
	go sys.initHC(ep)
	return false
}

// markOffline sets endpoint to offline if network i/o timeout seen.
func (sys *BucketTargetSys) markOffline(ep *url.URL) {
	sys.hMutex.Lock()
	defer sys.hMutex.Unlock()
	if h, ok := sys.hc[ep.Host]; ok {
		h.Online = false
		sys.hc[ep.Host] = h
	}
}

	if objectAPI == nil {
		return
	}

	if GlobalKMS == nil {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrKMSNotConfigured), r.URL)
		return
	}

	if _, ok := GlobalKMS.(kms.KeyManager); !ok {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
		return
	}

	metrics, err := GlobalKMS.Metrics(ctx)
	if err != nil {

}

// Verify if request has AWS PreSign Version '4'.
func isRequestPresignedSignatureV4(r *http.Request) bool {
	_, ok := r.Form[xhttp.AmzCredential]
	return ok
}

// Verify request has AWS PreSign Version '2'.
func isRequestPresignedSignatureV2(r *http.Request) bool {
	_, ok := r.Form[xhttp.AmzAccessKeyID]
	return ok
}

// Verify if request has AWS Post policy Signature Version '4'.

	gatherers := prometheus.Gatherers{
		prometheus.DefaultGatherer,
		registry,
	}

	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		tc, ok := r.Context().Value(mcontext.ContextTraceKey).(*mcontext.TraceCtxt)
		if ok {
			tc.FuncName = "handler.MetricsLegacy"
			tc.ResponseRecorder.LogErrBody = true
		}

		mfs, err := gatherers.Gather()
		if err != nil {
			if len(mfs) == 0 {

		apiErr := InsufficientReadQuorum{}
		if len(params) >= 1 {
			apiErr.Bucket = params[0]
		}
		if len(params) >= 2 {
			apiErr.Object = decodeDirObject(params[1])
		}
		if v, ok := oerr.(InsufficientReadQuorum); ok {
			apiErr.Type = v.Type
		}
		return apiErr
	case errErasureWriteQuorum.Error():
		apiErr := InsufficientWriteQuorum{}
		if len(params) >= 1 {
			apiErr.Bucket = params[0]
		}

		rs, parseRangeErr = parseCopyPartRangeSpec(rangeHeader)
	} else {
		// This check is to see if client specified a header but the value
		// is empty for 'x-amz-copy-source-range'
		_, ok := r.Header[xhttp.AmzCopySourceRange]
		if ok {
			parseRangeErr = errInvalidRange
		}
	}

	checkCopyPartPrecondFn := func(o ObjectInfo) bool {
		if _, err := DecryptObjectInfo(&o, r); err != nil {

// that is not io.EOF.
func diskHealthCheckOK(ctx context.Context, err error) bool {
	// Check if context has a disk health check.
	tracker, ok := ctx.Value(healthDiskCtxKey{}).(*healthDiskCtxValue)
	if !ok {
		// No tracker, return
		return err == nil || errors.Is(err, io.EOF)
	}
	if err == nil || errors.Is(err, io.EOF) {
		tracker.logSuccess()
		return true
	}

func (z *erasureServerPools) updateRebalanceStats(ctx context.Context) error {
	var ok bool
	for i := range z.serverPools {
		if z.findIndex(i) == -1 {
			// Also ensure to initialize rebalanceStats to indicate
			// its a new pool that can receive rebalanced data.
			z.rebalMeta.PoolStats = append(z.rebalMeta.PoolStats, &rebalanceStats{})
			ok = true
		}
	}
	if ok {
		lock := z.serverPools[0].NewNSLock(minioMetaBucket, rebalMetaName)

}

func (target *KafkaTarget) addToBatch(key store.Key) error {
	if target.batch.IsFull() {
		if err := target.commitBatch(); err != nil {
			return err
		}
	}
	if _, ok := target.batch.GetByKey(key.Name); !ok {
		eventData, err := target.store.Get(key.Name)
		if err != nil {
			if os.IsNotExist(err) {
				return nil
			}
			return err
		}
		msg, err := target.toProducerMessage(eventData)

}

func lookupLoggerWebhookConfig(scfg config.Config, cfg Config) (Config, error) {
	for k, kv := range config.Merge(scfg[config.LoggerWebhookSubSys], EnvLoggerWebhookEnable, DefaultLoggerWebhookKVS) {
		if v, ok := cfg.HTTP[k]; ok && v.Enabled {
			// This target is already enabled using the
			// legacy environment variables, ignore.
			continue
		}
		subSysTarget := config.LoggerWebhookSubSys
		if k != config.Default {