mirror of
				https://github.com/minio/minio.git
				synced 2025-11-01 00:31:22 +01:00 
			
		
		
		
	
		
			
				
	
	
		
			614 lines
		
	
	
		
			19 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			614 lines
		
	
	
		
			19 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
| // Copyright (c) 2015-2021 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"
 | |
| 	"encoding/gob"
 | |
| 	"encoding/json"
 | |
| 	"errors"
 | |
| 	"io"
 | |
| 	"net/http"
 | |
| 	"strings"
 | |
| 	"sync/atomic"
 | |
| 	"time"
 | |
| 
 | |
| 	"github.com/dustin/go-humanize"
 | |
| 	"github.com/minio/madmin-go/v3"
 | |
| 	xioutil "github.com/minio/minio/internal/ioutil"
 | |
| 	"github.com/minio/minio/internal/logger"
 | |
| 	"github.com/minio/mux"
 | |
| 	"github.com/minio/pkg/v2/policy"
 | |
| )
 | |
| 
 | |
| // SiteReplicationAdd - PUT /minio/admin/v3/site-replication/add
 | |
| func (a adminAPIHandlers) SiteReplicationAdd(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, cred := validateAdminReq(ctx, w, r, policy.SiteReplicationAddAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var sites []madmin.PeerSite
 | |
| 	if err := parseJSONBody(ctx, r.Body, &sites, cred.SecretKey); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	opts := getSRAddOptions(r)
 | |
| 	status, err := globalSiteReplicationSys.AddPeerClusters(ctx, sites, opts)
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	body, err := json.Marshal(status)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	writeSuccessResponseJSON(w, body)
 | |
| }
 | |
| 
 | |
| func getSRAddOptions(r *http.Request) (opts madmin.SRAddOptions) {
 | |
| 	opts.ReplicateILMExpiry = r.Form.Get("replicateILMExpiry") == "true"
 | |
| 	return
 | |
| }
 | |
| 
 | |
| // SRPeerJoin - PUT /minio/admin/v3/site-replication/join
 | |
| //
 | |
| // used internally to tell current cluster to enable SR with
 | |
| // the provided peer clusters and service account.
 | |
| func (a adminAPIHandlers) SRPeerJoin(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, cred := validateAdminReq(ctx, w, r, policy.SiteReplicationAddAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var joinArg madmin.SRPeerJoinReq
 | |
| 	if err := parseJSONBody(ctx, r.Body, &joinArg, cred.SecretKey); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err := globalSiteReplicationSys.PeerJoinReq(ctx, joinArg); err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SRPeerBucketOps - PUT /minio/admin/v3/site-replication/bucket-ops?bucket=x&operation=y
 | |
| func (a adminAPIHandlers) SRPeerBucketOps(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationOperationAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	vars := mux.Vars(r)
 | |
| 	bucket := vars["bucket"]
 | |
| 	operation := madmin.BktOp(vars["operation"])
 | |
| 
 | |
| 	var err error
 | |
| 	switch operation {
 | |
| 	default:
 | |
| 		err = errSRInvalidRequest(errInvalidArgument)
 | |
| 	case madmin.MakeWithVersioningBktOp:
 | |
| 		createdAt, cerr := time.Parse(time.RFC3339Nano, strings.TrimSpace(r.Form.Get("createdAt")))
 | |
| 		if cerr != nil {
 | |
| 			createdAt = timeSentinel
 | |
| 		}
 | |
| 
 | |
| 		opts := MakeBucketOptions{
 | |
| 			LockEnabled:       r.Form.Get("lockEnabled") == "true",
 | |
| 			VersioningEnabled: r.Form.Get("versioningEnabled") == "true",
 | |
| 			ForceCreate:       r.Form.Get("forceCreate") == "true",
 | |
| 			CreatedAt:         createdAt,
 | |
| 		}
 | |
| 		err = globalSiteReplicationSys.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)
 | |
| 	case madmin.ConfigureReplBktOp:
 | |
| 		err = globalSiteReplicationSys.PeerBucketConfigureReplHandler(ctx, bucket)
 | |
| 	case madmin.DeleteBucketBktOp, madmin.ForceDeleteBucketBktOp:
 | |
| 		err = globalSiteReplicationSys.PeerBucketDeleteHandler(ctx, bucket, DeleteBucketOptions{
 | |
| 			Force:      operation == madmin.ForceDeleteBucketBktOp,
 | |
| 			SRDeleteOp: getSRBucketDeleteOp(true),
 | |
| 		})
 | |
| 	case madmin.PurgeDeletedBucketOp:
 | |
| 		globalSiteReplicationSys.purgeDeletedBucket(ctx, objectAPI, bucket)
 | |
| 	}
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SRPeerReplicateIAMItem - PUT /minio/admin/v3/site-replication/iam-item
 | |
| func (a adminAPIHandlers) SRPeerReplicateIAMItem(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationOperationAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var item madmin.SRIAMItem
 | |
| 	if err := parseJSONBody(ctx, r.Body, &item, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var err error
 | |
| 	switch item.Type {
 | |
| 	default:
 | |
| 		err = errSRInvalidRequest(errInvalidArgument)
 | |
| 	case madmin.SRIAMItemPolicy:
 | |
| 		if item.Policy == nil {
 | |
| 			err = globalSiteReplicationSys.PeerAddPolicyHandler(ctx, item.Name, nil, item.UpdatedAt)
 | |
| 		} else {
 | |
| 			policy, perr := policy.ParseConfig(bytes.NewReader(item.Policy))
 | |
| 			if perr != nil {
 | |
| 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, perr), r.URL)
 | |
| 				return
 | |
| 			}
 | |
| 			if policy.IsEmpty() {
 | |
| 				err = globalSiteReplicationSys.PeerAddPolicyHandler(ctx, item.Name, nil, item.UpdatedAt)
 | |
| 			} else {
 | |
| 				err = globalSiteReplicationSys.PeerAddPolicyHandler(ctx, item.Name, policy, item.UpdatedAt)
 | |
| 			}
 | |
| 		}
 | |
| 	case madmin.SRIAMItemSvcAcc:
 | |
| 		err = globalSiteReplicationSys.PeerSvcAccChangeHandler(ctx, item.SvcAccChange, item.UpdatedAt)
 | |
| 	case madmin.SRIAMItemPolicyMapping:
 | |
| 		err = globalSiteReplicationSys.PeerPolicyMappingHandler(ctx, item.PolicyMapping, item.UpdatedAt)
 | |
| 	case madmin.SRIAMItemSTSAcc:
 | |
| 		err = globalSiteReplicationSys.PeerSTSAccHandler(ctx, item.STSCredential, item.UpdatedAt)
 | |
| 	case madmin.SRIAMItemIAMUser:
 | |
| 		err = globalSiteReplicationSys.PeerIAMUserChangeHandler(ctx, item.IAMUser, item.UpdatedAt)
 | |
| 	case madmin.SRIAMItemGroupInfo:
 | |
| 		err = globalSiteReplicationSys.PeerGroupInfoChangeHandler(ctx, item.GroupInfo, item.UpdatedAt)
 | |
| 	}
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SRPeerReplicateBucketItem - PUT /minio/admin/v3/site-replication/peer/bucket-meta
 | |
| func (a adminAPIHandlers) SRPeerReplicateBucketItem(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationOperationAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var item madmin.SRBucketMeta
 | |
| 	if err := parseJSONBody(ctx, r.Body, &item, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if item.Bucket == "" {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errSRInvalidRequest(errInvalidArgument)), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var err error
 | |
| 	switch item.Type {
 | |
| 	default:
 | |
| 		err = globalSiteReplicationSys.PeerBucketMetadataUpdateHandler(ctx, item)
 | |
| 	case madmin.SRBucketMetaTypePolicy:
 | |
| 		if item.Policy == nil {
 | |
| 			err = globalSiteReplicationSys.PeerBucketPolicyHandler(ctx, item.Bucket, nil, item.UpdatedAt)
 | |
| 		} else {
 | |
| 			bktPolicy, berr := policy.ParseBucketPolicyConfig(bytes.NewReader(item.Policy), item.Bucket)
 | |
| 			if berr != nil {
 | |
| 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, berr), r.URL)
 | |
| 				return
 | |
| 			}
 | |
| 			if bktPolicy.IsEmpty() {
 | |
| 				err = globalSiteReplicationSys.PeerBucketPolicyHandler(ctx, item.Bucket, nil, item.UpdatedAt)
 | |
| 			} else {
 | |
| 				err = globalSiteReplicationSys.PeerBucketPolicyHandler(ctx, item.Bucket, bktPolicy, item.UpdatedAt)
 | |
| 			}
 | |
| 		}
 | |
| 	case madmin.SRBucketMetaTypeQuotaConfig:
 | |
| 		if item.Quota == nil {
 | |
| 			err = globalSiteReplicationSys.PeerBucketQuotaConfigHandler(ctx, item.Bucket, nil, item.UpdatedAt)
 | |
| 		} else {
 | |
| 			quotaConfig, err := parseBucketQuota(item.Bucket, item.Quota)
 | |
| 			if err != nil {
 | |
| 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 				return
 | |
| 			}
 | |
| 			if err = globalSiteReplicationSys.PeerBucketQuotaConfigHandler(ctx, item.Bucket, quotaConfig, item.UpdatedAt); err != nil {
 | |
| 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 				return
 | |
| 			}
 | |
| 		}
 | |
| 	case madmin.SRBucketMetaTypeVersionConfig:
 | |
| 		err = globalSiteReplicationSys.PeerBucketVersioningHandler(ctx, item.Bucket, item.Versioning, item.UpdatedAt)
 | |
| 	case madmin.SRBucketMetaTypeTags:
 | |
| 		err = globalSiteReplicationSys.PeerBucketTaggingHandler(ctx, item.Bucket, item.Tags, item.UpdatedAt)
 | |
| 	case madmin.SRBucketMetaTypeObjectLockConfig:
 | |
| 		err = globalSiteReplicationSys.PeerBucketObjectLockConfigHandler(ctx, item.Bucket, item.ObjectLockConfig, item.UpdatedAt)
 | |
| 	case madmin.SRBucketMetaTypeSSEConfig:
 | |
| 		err = globalSiteReplicationSys.PeerBucketSSEConfigHandler(ctx, item.Bucket, item.SSEConfig, item.UpdatedAt)
 | |
| 	case madmin.SRBucketMetaLCConfig:
 | |
| 		err = globalSiteReplicationSys.PeerBucketLCConfigHandler(ctx, item.Bucket, item.ExpiryLCConfig, item.UpdatedAt)
 | |
| 	}
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SiteReplicationInfo - GET /minio/admin/v3/site-replication/info
 | |
| func (a adminAPIHandlers) SiteReplicationInfo(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationInfoAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	info, err := globalSiteReplicationSys.GetClusterInfo(ctx)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err = json.NewEncoder(w).Encode(info); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| func (a adminAPIHandlers) SRPeerGetIDPSettings(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationAddAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	idpSettings := globalSiteReplicationSys.GetIDPSettings(ctx)
 | |
| 	if err := json.NewEncoder(w).Encode(idpSettings); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| func parseJSONBody(ctx context.Context, body io.Reader, v interface{}, encryptionKey string) error {
 | |
| 	data, err := io.ReadAll(body)
 | |
| 	if err != nil {
 | |
| 		return SRError{
 | |
| 			Cause: err,
 | |
| 			Code:  ErrSiteReplicationInvalidRequest,
 | |
| 		}
 | |
| 	}
 | |
| 	if encryptionKey != "" {
 | |
| 		data, err = madmin.DecryptData(encryptionKey, bytes.NewReader(data))
 | |
| 		if err != nil {
 | |
| 			logger.LogIf(ctx, err)
 | |
| 			return SRError{
 | |
| 				Cause: err,
 | |
| 				Code:  ErrSiteReplicationInvalidRequest,
 | |
| 			}
 | |
| 		}
 | |
| 	}
 | |
| 	return json.Unmarshal(data, v)
 | |
| }
 | |
| 
 | |
| // SiteReplicationStatus - GET /minio/admin/v3/site-replication/status
 | |
| func (a adminAPIHandlers) SiteReplicationStatus(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationInfoAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 	opts := getSRStatusOptions(r)
 | |
| 	// default options to all if status options are unset for backward compatibility
 | |
| 	var dfltOpts madmin.SRStatusOptions
 | |
| 	if opts == dfltOpts {
 | |
| 		opts.Buckets = true
 | |
| 		opts.Users = true
 | |
| 		opts.Policies = true
 | |
| 		opts.Groups = true
 | |
| 		opts.ILMExpiryRules = true
 | |
| 	}
 | |
| 	info, err := globalSiteReplicationSys.SiteReplicationStatus(ctx, objectAPI, opts)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err = json.NewEncoder(w).Encode(info); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SiteReplicationMetaInfo - GET /minio/admin/v3/site-replication/metainfo
 | |
| func (a adminAPIHandlers) SiteReplicationMetaInfo(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationInfoAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	opts := getSRStatusOptions(r)
 | |
| 	info, err := globalSiteReplicationSys.SiteReplicationMetaInfo(ctx, objectAPI, opts)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err = json.NewEncoder(w).Encode(info); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SiteReplicationEdit - PUT /minio/admin/v3/site-replication/edit
 | |
| func (a adminAPIHandlers) SiteReplicationEdit(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, cred := validateAdminReq(ctx, w, r, policy.SiteReplicationAddAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 	var site madmin.PeerInfo
 | |
| 	err := parseJSONBody(ctx, r.Body, &site, cred.SecretKey)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	opts := getSREditOptions(r)
 | |
| 	status, err := globalSiteReplicationSys.EditPeerCluster(ctx, site, opts)
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	body, err := json.Marshal(status)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	writeSuccessResponseJSON(w, body)
 | |
| }
 | |
| 
 | |
| func getSREditOptions(r *http.Request) (opts madmin.SREditOptions) {
 | |
| 	opts.DisableILMExpiryReplication = r.Form.Get("disableILMExpiryReplication") == "true"
 | |
| 	opts.EnableILMExpiryReplication = r.Form.Get("enableILMExpiryReplication") == "true"
 | |
| 	return
 | |
| }
 | |
| 
 | |
| // SRPeerEdit - PUT /minio/admin/v3/site-replication/peer/edit
 | |
| //
 | |
| // used internally to tell current cluster to update endpoint for peer
 | |
| func (a adminAPIHandlers) SRPeerEdit(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationAddAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var pi madmin.PeerInfo
 | |
| 	if err := parseJSONBody(ctx, r.Body, &pi, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err := globalSiteReplicationSys.PeerEditReq(ctx, pi); err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SRStateEdit - PUT /minio/admin/v3/site-replication/state/edit
 | |
| //
 | |
| // used internally to tell current cluster to update site replication state
 | |
| func (a adminAPIHandlers) SRStateEdit(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationOperationAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var state madmin.SRStateEditReq
 | |
| 	if err := parseJSONBody(ctx, r.Body, &state, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	if err := globalSiteReplicationSys.PeerStateEditReq(ctx, state); err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| func getSRStatusOptions(r *http.Request) (opts madmin.SRStatusOptions) {
 | |
| 	q := r.Form
 | |
| 	opts.Buckets = q.Get("buckets") == "true"
 | |
| 	opts.Policies = q.Get("policies") == "true"
 | |
| 	opts.Groups = q.Get("groups") == "true"
 | |
| 	opts.Users = q.Get("users") == "true"
 | |
| 	opts.ILMExpiryRules = q.Get("ilm-expiry-rules") == "true"
 | |
| 	opts.PeerState = q.Get("peer-state") == "true"
 | |
| 	opts.Entity = madmin.GetSREntityType(q.Get("entity"))
 | |
| 	opts.EntityValue = q.Get("entityvalue")
 | |
| 	opts.ShowDeleted = q.Get("showDeleted") == "true"
 | |
| 	opts.Metrics = q.Get("metrics") == "true"
 | |
| 	return
 | |
| }
 | |
| 
 | |
| // SiteReplicationRemove - PUT /minio/admin/v3/site-replication/remove
 | |
| func (a adminAPIHandlers) SiteReplicationRemove(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationRemoveAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 	var rreq madmin.SRRemoveReq
 | |
| 	err := parseJSONBody(ctx, r.Body, &rreq, "")
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	status, err := globalSiteReplicationSys.RemovePeerCluster(ctx, objectAPI, rreq)
 | |
| 	if err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	body, err := json.Marshal(status)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	writeSuccessResponseJSON(w, body)
 | |
| }
 | |
| 
 | |
| // SRPeerRemove - PUT /minio/admin/v3/site-replication/peer/remove
 | |
| //
 | |
| // used internally to tell current cluster to update endpoint for peer
 | |
| func (a adminAPIHandlers) SRPeerRemove(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationRemoveAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var req madmin.SRRemoveReq
 | |
| 	if err := parseJSONBody(ctx, r.Body, &req, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if err := globalSiteReplicationSys.InternalRemoveReq(ctx, objectAPI, req); err != nil {
 | |
| 		logger.LogIf(ctx, err)
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SiteReplicationResyncOp - PUT /minio/admin/v3/site-replication/resync/op
 | |
| func (a adminAPIHandlers) SiteReplicationResyncOp(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	objectAPI, _ := validateAdminReq(ctx, w, r, policy.SiteReplicationResyncAction)
 | |
| 	if objectAPI == nil {
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	var peerSite madmin.PeerInfo
 | |
| 	if err := parseJSONBody(ctx, r.Body, &peerSite, ""); err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	vars := mux.Vars(r)
 | |
| 	op := madmin.SiteResyncOp(vars["operation"])
 | |
| 	var (
 | |
| 		status madmin.SRResyncOpStatus
 | |
| 		err    error
 | |
| 	)
 | |
| 	switch op {
 | |
| 	case madmin.SiteResyncStart:
 | |
| 		status, err = globalSiteReplicationSys.startResync(ctx, objectAPI, peerSite)
 | |
| 	case madmin.SiteResyncCancel:
 | |
| 		status, err = globalSiteReplicationSys.cancelResync(ctx, objectAPI, peerSite)
 | |
| 	default:
 | |
| 		err = errSRInvalidRequest(errInvalidArgument)
 | |
| 	}
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	body, err := json.Marshal(status)
 | |
| 	if err != nil {
 | |
| 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 | |
| 		return
 | |
| 	}
 | |
| 	writeSuccessResponseJSON(w, body)
 | |
| }
 | |
| 
 | |
| // SiteReplicationDevNull - everything goes to io.Discard
 | |
| // [POST] /minio/admin/v3/site-replication/devnull
 | |
| func (a adminAPIHandlers) SiteReplicationDevNull(w http.ResponseWriter, r *http.Request) {
 | |
| 	ctx := r.Context()
 | |
| 
 | |
| 	globalSiteNetPerfRX.Connect()
 | |
| 	defer globalSiteNetPerfRX.Disconnect()
 | |
| 
 | |
| 	connectTime := time.Now()
 | |
| 	for {
 | |
| 		n, err := io.CopyN(xioutil.Discard, r.Body, 128*humanize.KiByte)
 | |
| 		atomic.AddUint64(&globalSiteNetPerfRX.RX, uint64(n))
 | |
| 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 | |
| 			// If there is a disconnection before globalNetPerfMinDuration (we give a margin of error of 1 sec)
 | |
| 			// would mean the network is not stable. Logging here will help in debugging network issues.
 | |
| 			if time.Since(connectTime) < (globalNetPerfMinDuration - time.Second) {
 | |
| 				logger.LogIf(ctx, err)
 | |
| 			}
 | |
| 		}
 | |
| 		if err != nil {
 | |
| 			if errors.Is(err, io.EOF) {
 | |
| 				w.WriteHeader(http.StatusNoContent)
 | |
| 			} else {
 | |
| 				w.WriteHeader(http.StatusBadRequest)
 | |
| 			}
 | |
| 			break
 | |
| 		}
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // SiteReplicationNetPerf - everything goes to io.Discard
 | |
| // [POST] /minio/admin/v3/site-replication/netperf
 | |
| func (a adminAPIHandlers) SiteReplicationNetPerf(w http.ResponseWriter, r *http.Request) {
 | |
| 	durationStr := r.Form.Get(peerRESTDuration)
 | |
| 	duration, _ := time.ParseDuration(durationStr)
 | |
| 	if duration < globalNetPerfMinDuration {
 | |
| 		duration = globalNetPerfMinDuration
 | |
| 	}
 | |
| 	result := siteNetperf(r.Context(), duration)
 | |
| 	logger.LogIf(r.Context(), gob.NewEncoder(w).Encode(result))
 | |
| }
 |