minio/cmd/admin-handlers.go at master · bood/minio

Repository navigation

// Copyright (c) 2015-2022 MinIO, Inc.
// This file is part of MinIO Object Storage stack
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
package cmd
	crand "crypto/rand"
	"crypto/rsa"
	"crypto/subtle"
	"crypto/x509"
	"encoding/base64"
	"encoding/json"
	"encoding/pem"
	"hash/crc32"
	"math/rand"
	"net/http"
	"github.com/dustin/go-humanize"
	"github.com/gorilla/mux"
	"github.com/klauspost/compress/zip"
	"github.com/minio/madmin-go"
	"github.com/minio/madmin-go/estream"
	"github.com/minio/minio/internal/dsync"
	"github.com/minio/minio/internal/handlers"
	xhttp "github.com/minio/minio/internal/http"
	"github.com/minio/minio/internal/kms"
	"github.com/minio/minio/internal/logger"
	"github.com/minio/minio/internal/logger/message/log"
	iampolicy "github.com/minio/pkg/iam/policy"
	xnet "github.com/minio/pkg/net"
	"github.com/secure-io/sio-go"
	maxEConfigJSONSize        = 262272
	kubernetesVersionEndpoint = "https://kubernetes.default.svc/version"
// Only valid query params for mgmt admin APIs.
	mgmtBucket      = "bucket"
	mgmtPrefix      = "prefix"
	mgmtClientToken = "clientToken"
	mgmtForceStart  = "forceStart"
	mgmtForceStop   = "forceStop"
// ServerUpdateHandler - POST /minio/admin/v3/update?updateURL={updateURL}
// ----------
// updates all minio servers and restarts them gracefully.
func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "ServerUpdate")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.ServerUpdateAdminAction)
	if objectAPI == nil {
	if globalInplaceUpdateDisabled {
		// if MINIO_UPDATE=off - inplace update is disabled, mostly in containers.
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL)
	vars := mux.Vars(r)
	updateURL := vars["updateURL"]
	mode := getMinioMode()
	if updateURL == "" {
		updateURL = minioReleaseInfoURL
		if runtime.GOOS == globalWindowsOSName {
			updateURL = minioReleaseWindowsInfoURL
	u, err := url.Parse(updateURL)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	content, err := downloadReleaseURL(u, updateTimeout, mode)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	sha256Sum, lrTime, releaseInfo, err := parseReleaseData(content)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	u.Path = path.Dir(u.Path) + SlashSeparator + releaseInfo
	crTime, err := GetCurrentReleaseTime()
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	if lrTime.Sub(crTime) <= 0 {
		updateStatus := madmin.ServerUpdateStatus{
			CurrentVersion: Version,
			UpdatedVersion: Version,
		// Marshal API response
		jsonBytes, err := json.Marshal(updateStatus)
		if err != nil {
			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
		writeSuccessResponseJSON(w, jsonBytes)
	// Download Binary Once
	reader, err := downloadBinary(u, mode)
	if err != nil {
		logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	// Push binary to other servers
	for _, nerr := range globalNotificationSys.VerifyBinary(ctx, u, sha256Sum, releaseInfo, reader) {
		if nerr.Err != nil {
			err := AdminError{
				Code:       AdminUpdateApplyFailure,
				Message:    nerr.Err.Error(),
				StatusCode: http.StatusInternalServerError,
			logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
			logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))
			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	err = verifyBinary(u, sha256Sum, releaseInfo, mode, reader)
	if err != nil {
		logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	for _, nerr := range globalNotificationSys.CommitBinary(ctx) {
		if nerr.Err != nil {
			err := AdminError{
				Code:       AdminUpdateApplyFailure,
				Message:    nerr.Err.Error(),
				StatusCode: http.StatusInternalServerError,
			logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
			logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))
			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	err = commitBinary()
	if err != nil {
		logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	updateStatus := madmin.ServerUpdateStatus{
		CurrentVersion: Version,
		UpdatedVersion: lrTime.Format(minioReleaseTagTimeLayout),
	// Marshal API response
	jsonBytes, err := json.Marshal(updateStatus)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	writeSuccessResponseJSON(w, jsonBytes)
	// Notify all other MinIO peers signal service.
	for _, nerr := range globalNotificationSys.SignalService(serviceRestart) {
		if nerr.Err != nil {
			logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
			logger.LogIf(ctx, nerr.Err)
	globalServiceSignalCh <- serviceRestart
// ServiceHandler - POST /minio/admin/v3/service?action={action}
// ----------
// Supports following actions:
// - restart (restarts all the MinIO instances in a setup)
// - stop (stops all the MinIO instances in a setup)
// - freeze (freezes all incoming S3 API calls)
// - unfreeze (unfreezes previously frozen S3 API calls)
func (a adminAPIHandlers) ServiceHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "Service")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	vars := mux.Vars(r)
	action := vars["action"]
	var serviceSig serviceSignal
	switch madmin.ServiceAction(action) {
	case madmin.ServiceActionRestart:
		serviceSig = serviceRestart
	case madmin.ServiceActionStop:
		serviceSig = serviceStop
	case madmin.ServiceActionFreeze:
		serviceSig = serviceFreeze
	case madmin.ServiceActionUnfreeze:
		serviceSig = serviceUnFreeze
		logger.LogIf(ctx, fmt.Errorf("Unrecognized service action %s requested", action), logger.Application)
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrMalformedPOSTRequest), r.URL)
	var objectAPI ObjectLayer
	switch serviceSig {
	case serviceRestart:
		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceRestartAdminAction)
	case serviceStop:
		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceStopAdminAction)
	case serviceFreeze, serviceUnFreeze:
		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceFreezeAdminAction)
	if objectAPI == nil {
	// Notify all other MinIO peers signal service.
	for _, nerr := range globalNotificationSys.SignalService(serviceSig) {
		if nerr.Err != nil {
			logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
			logger.LogIf(ctx, nerr.Err)
	// Reply to the client before restarting, stopping MinIO server.
	writeSuccessResponseHeadersOnly(w)
	switch serviceSig {
	case serviceFreeze:
		freezeServices()
	case serviceUnFreeze:
		unfreezeServices()
	case serviceRestart, serviceStop:
		globalServiceSignalCh <- serviceSig
// ServerProperties holds some server information such as, version, region
// uptime, etc..
type ServerProperties struct {
	Uptime       int64    `json:"uptime"`
	Version      string   `json:"version"`
	CommitID     string   `json:"commitID"`
	DeploymentID string   `json:"deploymentID"`
	Region       string   `json:"region"`
	SQSARN       []string `json:"sqsARN"`
// ServerConnStats holds transferred bytes from/to the server
type ServerConnStats struct {
	TotalInputBytes  uint64 `json:"transferred"`
	TotalOutputBytes uint64 `json:"received"`
	Throughput       uint64 `json:"throughput,omitempty"`
	S3InputBytes     uint64 `json:"transferredS3"`
	S3OutputBytes    uint64 `json:"receivedS3"`
	AdminInputBytes  uint64 `json:"transferredAdmin"`
	AdminOutputBytes uint64 `json:"receivedAdmin"`
// ServerHTTPAPIStats holds total number of HTTP operations from/to the server,
// including the average duration the call was spent.
type ServerHTTPAPIStats struct {
	APIStats map[string]int `json:"apiStats"`
// ServerHTTPStats holds all type of http operations performed to/from the server
// including their average execution time.
type ServerHTTPStats struct {
	S3RequestsInQueue      int32              `json:"s3RequestsInQueue"`
	S3RequestsIncoming     uint64             `json:"s3RequestsIncoming"`
	CurrentS3Requests      ServerHTTPAPIStats `json:"currentS3Requests"`
	TotalS3Requests        ServerHTTPAPIStats `json:"totalS3Requests"`
	TotalS3Errors          ServerHTTPAPIStats `json:"totalS3Errors"`
	TotalS35xxErrors       ServerHTTPAPIStats `json:"totalS35xxErrors"`
	TotalS34xxErrors       ServerHTTPAPIStats `json:"totalS34xxErrors"`
	TotalS3Canceled        ServerHTTPAPIStats `json:"totalS3Canceled"`
	TotalS3RejectedAuth    uint64             `json:"totalS3RejectedAuth"`
	TotalS3RejectedTime    uint64             `json:"totalS3RejectedTime"`
	TotalS3RejectedHeader  uint64             `json:"totalS3RejectedHeader"`
	TotalS3RejectedInvalid uint64             `json:"totalS3RejectedInvalid"`
// StorageInfoHandler - GET /minio/admin/v3/storageinfo
// ----------
// Get server information
func (a adminAPIHandlers) StorageInfoHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "StorageInfo")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.StorageInfoAdminAction)
	if objectAPI == nil {
	// ignores any errors here.
	storageInfo, _ := objectAPI.StorageInfo(ctx)
	// Collect any disk healing.
	healing, _ := getAggregatedBackgroundHealState(ctx, nil)
	healDisks := make(map[string]struct{}, len(healing.HealDisks))
	for _, disk := range healing.HealDisks {
		healDisks[disk] = struct{}{}
	// find all disks which belong to each respective endpoints
	for i, disk := range storageInfo.Disks {
		if _, ok := healDisks[disk.Endpoint]; ok {
			storageInfo.Disks[i].Healing = true
	// Marshal API response
	jsonBytes, err := json.Marshal(storageInfo)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	// Reply with storage information (across nodes in a
	// distributed setup) as json.
	writeSuccessResponseJSON(w, jsonBytes)
// MetricsHandler - GET /minio/admin/v3/metrics
// ----------
// Get realtime server metrics
func (a adminAPIHandlers) MetricsHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "Metrics")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.ServerInfoAdminAction)
	if objectAPI == nil {
	const defaultMetricsInterval = time.Second
	interval, err := time.ParseDuration(r.Form.Get("interval"))
	if err != nil || interval < time.Second {
		interval = defaultMetricsInterval
	n, err := strconv.Atoi(r.Form.Get("n"))
	if err != nil || n <= 0 {
		n = math.MaxInt32
	var types madmin.MetricType
	if t, _ := strconv.ParseUint(r.Form.Get("types"), 10, 64); t != 0 {
		types = madmin.MetricType(t)
		types = madmin.MetricsAll
	disks := strings.Split(r.Form.Get("disks"), ",")
	byDisk := strings.EqualFold(r.Form.Get("by-disk"), "true")
	var diskMap map[string]struct{}
	if len(disks) > 0 && disks[0] != "" {
		diskMap = make(map[string]struct{}, len(disks))
		for _, k := range disks {
			if k != "" {
				diskMap[k] = struct{}{}
	jobID := r.Form.Get("by-jobID")
	hosts := strings.Split(r.Form.Get("hosts"), ",")
	byHost := strings.EqualFold(r.Form.Get("by-host"), "true")
	var hostMap map[string]struct{}
	if len(hosts) > 0 && hosts[0] != "" {
		hostMap = make(map[string]struct{}, len(hosts))
		for _, k := range hosts {
			if k != "" {
				hostMap[k] = struct{}{}
	dID := r.Form.Get("by-depID")
	done := ctx.Done()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	w.Header().Set(xhttp.ContentType, string(mimeJSON))
	enc := json.NewEncoder(w)
	for n > 0 {
		var m madmin.RealtimeMetrics
		mLocal := collectLocalMetrics(types, collectMetricsOpts{
			hosts: hostMap,
			disks: diskMap,
			jobID: jobID,
			depID: dID,
		m.Merge(&mLocal)
		// Allow half the interval for collecting remote...
		cctx, cancel := context.WithTimeout(ctx, interval/2)
		mRemote := collectRemoteMetrics(cctx, types, collectMetricsOpts{
			hosts: hostMap,
			disks: diskMap,
			jobID: jobID,
			depID: dID,
		m.Merge(&mRemote)
		if !byHost {
			m.ByHost = nil
		if !byDisk {
			m.ByDisk = nil
		m.Final = n <= 1
		// Marshal API reesponse
		if err := enc.Encode(&m); err != nil {
		if n <= 0 {
		// Flush before waiting for next...
		w.(http.Flusher).Flush()
		case <-ticker.C:
		case <-done:
// DataUsageInfoHandler - GET /minio/admin/v3/datausage
// ----------
// Get server/cluster data usage info
func (a adminAPIHandlers) DataUsageInfoHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "DataUsageInfo")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.DataUsageInfoAdminAction)
	if objectAPI == nil {
	dataUsageInfo, err := loadDataUsageFromBackend(ctx, objectAPI)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	dataUsageInfoJSON, err := json.Marshal(dataUsageInfo)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	writeSuccessResponseJSON(w, dataUsageInfoJSON)
func lriToLockEntry(l lockRequesterInfo, now time.Time, resource, server string) *madmin.LockEntry {
	entry := &madmin.LockEntry{
		Timestamp:  l.Timestamp,
		Elapsed:    now.Sub(l.Timestamp),
		Resource:   resource,
		ServerList: []string{server},
		Source:     l.Source,
		Owner:      l.Owner,
		ID:         l.UID,
		Quorum:     l.Quorum,
	if l.Writer {
		entry.Type = "WRITE"
		entry.Type = "READ"
	return entry
func topLockEntries(peerLocks []*PeerLocks, stale bool) madmin.LockEntries {
	now := time.Now().UTC()
	entryMap := make(map[string]*madmin.LockEntry)
	for _, peerLock := range peerLocks {
		if peerLock == nil {
			continue
		for k, v := range peerLock.Locks {
			for _, lockReqInfo := range v {
				if val, ok := entryMap[lockReqInfo.Name]; ok {
					val.ServerList = append(val.ServerList, peerLock.Addr)
				} else {
					entryMap[lockReqInfo.Name] = lriToLockEntry(lockReqInfo, now, k, peerLock.Addr)
	var lockEntries madmin.LockEntries
	for _, v := range entryMap {
		if stale {
			lockEntries = append(lockEntries, *v)
			continue
		if len(v.ServerList) >= v.Quorum {
			lockEntries = append(lockEntries, *v)
	sort.Sort(lockEntries)
	return lockEntries
// PeerLocks holds server information result of one node
type PeerLocks struct {
	Addr  string
	Locks map[string][]lockRequesterInfo
// ForceUnlockHandler force unlocks requested resource
func (a adminAPIHandlers) ForceUnlockHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "ForceUnlock")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.ForceUnlockAdminAction)
	if objectAPI == nil {
	z, ok := objectAPI.(*erasureServerPools)
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
	vars := mux.Vars(r)
	var args dsync.LockArgs
	var lockers []dsync.NetLocker
	for _, path := range strings.Split(vars["paths"], ",") {
		if path == "" {
			continue
		args.Resources = append(args.Resources, path)
	for _, lks := range z.serverPools[0].erasureLockers {
		lockers = append(lockers, lks...)
	for _, locker := range lockers {
		locker.ForceUnlock(ctx, args)
// TopLocksHandler Get list of locks in use
func (a adminAPIHandlers) TopLocksHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "TopLocks")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.TopLocksAdminAction)
	if objectAPI == nil {
	count := 10 // by default list only top 10 entries
	if countStr := r.Form.Get("count"); countStr != "" {
		var err error
		count, err = strconv.Atoi(countStr)
		if err != nil {
			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	stale := r.Form.Get("stale") == "true" // list also stale locks
	peerLocks := globalNotificationSys.GetLocks(ctx, r)
	topLocks := topLockEntries(peerLocks, stale)
	// Marshal API response upto requested count.
	if len(topLocks) > count && count > 0 {
		topLocks = topLocks[:count]
	jsonBytes, err := json.Marshal(topLocks)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	// Reply with storage information (across nodes in a
	// distributed setup) as json.
	writeSuccessResponseJSON(w, jsonBytes)
// StartProfilingResult contains the status of the starting
// profiling action in a given server - deprecated API
type StartProfilingResult struct {
	NodeName string `json:"nodeName"`
	Success  bool   `json:"success"`
	Error    string `json:"error"`
// StartProfilingHandler - POST /minio/admin/v3/profiling/start?profilerType={profilerType}
// ----------
// Enable server profiling
func (a adminAPIHandlers) StartProfilingHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "StartProfiling")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	// Validate request signature.
	_, adminAPIErr := checkAdminRequestAuth(ctx, r, iampolicy.ProfilingAdminAction, "")
	if adminAPIErr != ErrNone {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
	if globalNotificationSys == nil {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
	vars := mux.Vars(r)
	profiles := strings.Split(vars["profilerType"], ",")
	thisAddr, err := xnet.ParseHost(globalLocalNodeName)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	globalProfilerMu.Lock()
	defer globalProfilerMu.Unlock()
	if globalProfiler == nil {
		globalProfiler = make(map[string]minioProfiler, 10)
	// Stop profiler of all types if already running
	for k, v := range globalProfiler {
		for _, p := range profiles {
			if p == k {
				v.Stop()
				delete(globalProfiler, k)
	// Start profiling on remote servers.
	var hostErrs []NotificationPeerErr
	for _, profiler := range profiles {
		hostErrs = append(hostErrs, globalNotificationSys.StartProfiling(profiler)...)
		// Start profiling locally as well.
		prof, err := startProfiler(profiler)
		if err != nil {
			hostErrs = append(hostErrs, NotificationPeerErr{
				Host: *thisAddr,
				Err:  err,
			globalProfiler[profiler] = prof
			hostErrs = append(hostErrs, NotificationPeerErr{
				Host: *thisAddr,
	var startProfilingResult []StartProfilingResult
	for _, nerr := range hostErrs {
		result := StartProfilingResult{NodeName: nerr.Host.String()}
		if nerr.Err != nil {
			result.Error = nerr.Err.Error()
			result.Success = true
		startProfilingResult = append(startProfilingResult, result)
	// Create JSON result and send it to the client
	startProfilingResultInBytes, err := json.Marshal(startProfilingResult)
	if err != nil {
		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
	writeSuccessResponseJSON(w, startProfilingResultInBytes)
// ProfileHandler - POST /minio/admin/v3/profile/?profilerType={profilerType}
// ----------
// Enable server profiling
func (a adminAPIHandlers) ProfileHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "Profile")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	// Validate request signature.
	_, adminAPIErr := checkAdminRequestAuth(ctx, r, iampolicy.ProfilingAdminAction, "")
	if adminAPIErr != ErrNone {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
	if globalNotificationSys == nil {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
	profileStr := r.Form.Get("profilerType")
	profiles := strings.Split(profileStr, ",")
	duration := time.Minute
	if dstr := r.Form.Get("duration"); dstr != "" {
		var err error
		duration, err = time.ParseDuration(dstr)
		if err != nil {
			writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
	// read request body
	io.CopyN(io.Discard, r.Body, 1)
	globalProfilerMu.Lock()
	if globalProfiler == nil {
		globalProfiler = make(map[string]minioProfiler, 10)
	// Stop profiler of all types if already running
	for k, v := range globalProfiler {
		delete(globalProfiler, k)
	// Start profiling on remote servers.
	for _, profiler := range profiles {
		globalNotificationSys.StartProfiling(profiler)
		// Start profiling locally as well.
		prof, err := startProfiler(profiler)
		if err == nil {
			globalProfiler[profiler] = prof
	globalProfilerMu.Unlock()
	timer := time.NewTimer(duration)
	defer timer.Stop()
		case <-ctx.Done():
			for k, v := range globalProfiler {
				v.Stop()
				delete(globalProfiler, k)
		case <-timer.C:
			if !globalNotificationSys.DownloadProfilingData(ctx, w) {
				writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminProfilerNotEnabled), r.URL)
// dummyFileInfo represents a dummy representation of a profile data file
// present only in memory, it helps to generate the zip stream.
type dummyFileInfo struct {
	name    string
	size    int64
	mode    os.FileMode
	modTime time.Time
	isDir   bool
	sys     interface{}
func (f dummyFileInfo) Name() string       { return f.name }
func (f dummyFileInfo) Size() int64        { return f.size }
func (f dummyFileInfo) Mode() os.FileMode  { return f.mode }
func (f dummyFileInfo) ModTime() time.Time { return f.modTime }
func (f dummyFileInfo) IsDir() bool        { return f.isDir }
func (f dummyFileInfo) Sys() interface{}   { return f.sys }
// DownloadProfilingHandler - POST /minio/admin/v3/profiling/download
// ----------
// Download profiling information of all nodes in a zip format - deprecated API
func (a adminAPIHandlers) DownloadProfilingHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "DownloadProfiling")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	// Validate request signature.
	_, adminAPIErr := checkAdminRequestAuth(ctx, r, iampolicy.ProfilingAdminAction, "")
	if adminAPIErr != ErrNone {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
	if globalNotificationSys == nil {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
	if !globalNotificationSys.DownloadProfilingData(ctx, w) {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminProfilerNotEnabled), r.URL)
type healInitParams struct {
	bucket, objPrefix     string
	hs                    madmin.HealOpts
	clientToken           string
	forceStart, forceStop bool
// extractHealInitParams - Validates params for heal init API.
func extractHealInitParams(vars map[string]string, qParms url.Values, r io.Reader) (hip healInitParams, err APIErrorCode) {
	hip.bucket = vars[mgmtBucket]
	hip.objPrefix = vars[mgmtPrefix]
	if hip.bucket == "" {
		if hip.objPrefix != "" {
			// Bucket is required if object-prefix is given
			err = ErrHealMissingBucket
	} else if isReservedOrInvalidBucket(hip.bucket, false) {
		err = ErrInvalidBucketName
	// empty prefix is valid.
	if !IsValidObjectPrefix(hip.objPrefix) {
		err = ErrInvalidObjectName
	if len(qParms[mgmtClientToken]) > 0 {
		hip.clientToken = qParms[mgmtClientToken][0]
	if _, ok := qParms[mgmtForceStart]; ok {
		hip.forceStart = true
	if _, ok := qParms[mgmtForceStop]; ok {
		hip.forceStop = true
	// Invalid request conditions:
	//   Cannot have both forceStart and forceStop in the same
	//   request; If clientToken is provided, request can only be
	//   to continue receiving logs, so it cannot be start or
	//   stop;
	if (hip.forceStart && hip.forceStop) ||
		(hip.clientToken != "" && (hip.forceStart || hip.forceStop)) {
		err = ErrInvalidRequest
	// ignore body if clientToken is provided
	if hip.clientToken == "" {
		jerr := json.NewDecoder(r).Decode(&hip.hs)
		if jerr != nil {
			logger.LogIf(GlobalContext, jerr, logger.Application)
			err = ErrRequestBodyParse
	err = ErrNone
// HealHandler - POST /minio/admin/v3/heal/
// -----------
// Start heal processing and return heal status items.
// On a successful heal sequence start, a unique client token is
// returned. Subsequent requests to this endpoint providing the client
// token will receive heal status records from the running heal
// sequence.
// If no client token is provided, and a heal sequence is in progress
// an error is returned with information about the running heal
// sequence. However, if the force-start flag is provided, the server
// aborts the running heal sequence and starts a new one.
func (a adminAPIHandlers) HealHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "Heal")
	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.HealAdminAction)
	if objectAPI == nil {
	hip, errCode := extractHealInitParams(mux.Vars(r), r.Form, r.Body)
	if errCode != ErrNone {
		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(errCode), r.URL)
	// Analyze the heal token and route the request accordingly
	token, success := proxyRequestByToken(ctx, w, r, hip.clientToken)
	if success {
	hip.clientToken = token
	// if request was not successful, try this server locally if token
	// is not found the call will fail anyways. if token is empty
	// try this server to generate a new token.
	type healResp struct {
		respBytes []byte
		apiErr    APIError
		errBody   string
	// Define a closure to start sending whitespace to client
	// after 10s unless a response item comes in
	keepConnLive := func(w http.ResponseWriter, r *http.Request, respCh chan healResp) {
		ticker := time.NewTicker(time.Second * 10)
		defer ticker.Stop()
		started := false
			select {
			case <-r.Context().Done():
			case <-ticker.C:
				if !started {
					// Start writing response to client
					started = true
					setCommonHeaders(w)
					setEventStreamHeaders(w)
					// Set 200 OK status
					w.WriteHeader(200)
				// Send whitespace and keep connection open
				if _, err := w.Write([]byte(" ")); err != nil {
					return
				w.(http.Flusher).Flush()
View remainder of file in raw view
FilesExpand file tree

admin-handlers.go

Latest commit

History

admin-handlers.go

File metadata and controls
