// 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 import (	"bytes"	"context"	crand "crypto/rand"	"crypto/subtle"	"encoding/json"	"errors"	"fmt"	"hash/crc32"	"io"	"io/ioutil"	"math"	"math/rand"	"net/http"	"net/url"	"os"	"path"	"regexp"	"runtime"	"sort"	"strconv"	"strings"	"time" 	"github.com/dustin/go-humanize"	"github.com/gorilla/mux"	"github.com/klauspost/compress/zip"	"github.com/minio/madmin-go"	"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"	"github.com/minio/minio/internal/pubsub"	iampolicy "github.com/minio/pkg/iam/policy"	xnet "github.com/minio/pkg/net"	"github.com/secure-io/sio-go") const (	maxEConfigJSONSize        = 262272	kubernetesVersionEndpoint = "https://kubernetes.default.svc/version") // Only valid query params for mgmt admin APIs.const (	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 {		return	} 	if globalInplaceUpdateDisabled {		// if MINIO_UPDATE=off - inplace update is disabled, mostly in containers.		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL)		return	} 	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)		return	} 	content, err := downloadReleaseURL(u, updateTimeout, mode)	if err != nil {		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	sha256Sum, lrTime, releaseInfo, err := parseReleaseData(content)	if err != nil {		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	u.Path = path.Dir(u.Path) + SlashSeparator + releaseInfo	crTime, err := GetCurrentReleaseTime()	if err != nil {		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	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)			return		} 		writeSuccessResponseJSON(w, jsonBytes)		return	} 	// 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)		return	} 	// 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)			return		}	} 	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)		return	} 	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)			return		}	} 	err = commitBinary()	if err != nil {		logger.LogIf(ctx, fmt.Errorf("server update failed with %w", err))		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	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)		return	} 	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	default:		logger.LogIf(ctx, fmt.Errorf("Unrecognized service action %s requested", action), logger.Application)		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrMalformedPOSTRequest), r.URL)		return	} 	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 {		return	} 	// 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 servertype 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 informationfunc (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 {		return	} 	// 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)		return	} 	// Reply with storage information (across nodes in a	// distributed setup) as json.	writeSuccessResponseJSON(w, jsonBytes)} // MetricsHandler - GET /minio/admin/v3/metrics// ----------// Get realtime server metricsfunc (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 {		return	}	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)	} else {		types = madmin.MetricsAll	}	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{}{}			}		}	}	done := ctx.Done()	ticker := time.NewTicker(interval)	defer ticker.Stop()	w.Header().Set(xhttp.ContentType, string(mimeJSON)) 	for n > 0 {		var m madmin.RealtimeMetrics		mLocal := collectLocalMetrics(types, hostMap)		m.Merge(&mLocal) 		// Allow half the interval for collecting remote...		cctx, cancel := context.WithTimeout(ctx, interval/2)		mRemote := collectRemoteMetrics(cctx, types, hostMap)		cancel()		m.Merge(&mRemote)		if !byhost {			m.ByHost = nil		} 		m.Final = n <= 1		// Marshal API response		jsonBytes, err := json.Marshal(m)		if err != nil {			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)			return		}		_, err = w.Write(jsonBytes)		if err != nil {			n = 0		} 		n--		if n <= 0 {			break		} 		// Flush before waiting for next...		w.(http.Flusher).Flush() 		select {		case <-ticker.C:		case <-done:			return		}	}} // DataUsageInfoHandler - GET /minio/admin/v3/datausage// ----------// Get server/cluster data usage infofunc (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 {		return	} 	dataUsageInfo, err := loadDataUsageFromBackend(ctx, objectAPI)	if err != nil {		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	dataUsageInfoJSON, err := json.Marshal(dataUsageInfo)	if err != nil {		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)		return	} 	writeSuccessResponseJSON(w, dataUsageInfoJSON)} func lriToLockEntry(l lockRequesterInfo, resource, server string) *madmin.LockEntry {	entry := &madmin.LockEntry{		Timestamp:  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"	} else {		entry.Type = "READ"

Lines 1–500 of 2985. Load more