MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-11-08 21:24:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: Change ListBucketTargets handler (#10217)

Browse Source 
to list all targets across a tenant.
Also fixing some validations.
This commit is contained in:
poornas
2020-08-06 17:10:21 -07:00
committed by GitHub
parent ce129efa09
commit adcaa6f9de
10 changed files with 180 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
cmd/admin-bucket-handlers.go
View File

@@ -26,7 +26,6 @@ import (
"github.com/gorilla/mux"
"github.com/minio/minio/cmd/config"
"github.com/minio/minio/cmd/logger"
"github.com/minio/minio/pkg/auth"
"github.com/minio/minio/pkg/env"
iampolicy "github.com/minio/minio/pkg/iam/policy"
"github.com/minio/minio/pkg/madmin"
@@ -122,8 +121,8 @@ func (a adminAPIHandlers) GetBucketQuotaConfigHandler(w http.ResponseWriter, r *
writeSuccessResponseJSON(w, configData)
}
// SetBucketTargetHandler - sets a remote target for bucket
func (a adminAPIHandlers) SetBucketTargetHandler(w http.ResponseWriter, r *http.Request) {
// SetRemoteTargetHandler - sets a remote target for bucket
func (a adminAPIHandlers) SetRemoteTargetHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "SetBucketTarget")
defer logger.AuditLog(w, r, "SetBucketTarget", mustGetClaimsFromToken(r))
@@ -174,6 +173,7 @@ func (a adminAPIHandlers) SetBucketTargetHandler(w http.ResponseWriter, r *http.
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrBucketRemoteIdenticalToSource), r.URL)
return
}
target.SourceBucket = bucket
target.Arn = globalBucketTargetSys.getRemoteARN(bucket, &target)
if target.Arn == "" {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
@@ -183,7 +183,7 @@ func (a adminAPIHandlers) SetBucketTargetHandler(w http.ResponseWriter, r *http.
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
return
}
targets, err := globalBucketTargetSys.ListTargets(ctx, bucket)
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return
@@ -209,9 +209,9 @@ func (a adminAPIHandlers) SetBucketTargetHandler(w http.ResponseWriter, r *http.
writeSuccessResponseJSON(w, data)
}
// ListBucketTargetsHandler - lists remote target(s) for a bucket or gets a target
// ListRemoteTargetsHandler - lists remote target(s) for a bucket or gets a target
// for a particular ARN type
func (a adminAPIHandlers) ListBucketTargetsHandler(w http.ResponseWriter, r *http.Request) {
func (a adminAPIHandlers) ListRemoteTargetsHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "ListBucketTargets")
defer logger.AuditLog(w, r, "ListBucketTargets", mustGetClaimsFromToken(r))
@@ -225,28 +225,13 @@ func (a adminAPIHandlers) ListBucketTargetsHandler(w http.ResponseWriter, r *htt
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
cfg, err := globalBucketMetadataSys.GetBucketTargetsConfig(bucket)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
var (
targets []madmin.BucketTarget
tgt, ct madmin.BucketTarget
creds auth.Credentials
)
if cfg != nil && !cfg.Empty() {
for idx, t := range cfg.Targets {
if string(t.Type) == arnType || arnType == "" {
ct = cfg.Targets[idx]
// remove secretKey from creds
creds.AccessKey = ct.Credentials.AccessKey
tgt = madmin.BucketTarget{Endpoint: ct.Endpoint, Secure: ct.Secure, TargetBucket: ct.TargetBucket, Credentials: &creds, Arn: ct.Arn, Type: ct.Type}
targets = append(targets, tgt)
}
if bucket != "" {
if _, err := globalBucketMetadataSys.GetBucketTargetsConfig(bucket); err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
}
targets := globalBucketTargetSys.ListTargets(ctx, bucket, arnType)
data, err := json.Marshal(targets)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
@@ -256,8 +241,8 @@ func (a adminAPIHandlers) ListBucketTargetsHandler(w http.ResponseWriter, r *htt
writeSuccessResponseJSON(w, data)
}
// RemoveBucketTargetHandler - removes a remote target for bucket with specified ARN
func (a adminAPIHandlers) RemoveBucketTargetHandler(w http.ResponseWriter, r *http.Request) {
// RemoveRemoteTargetHandler - removes a remote target for bucket with specified ARN
func (a adminAPIHandlers) RemoveRemoteTargetHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "RemoveBucketTarget")
defer logger.AuditLog(w, r, "RemoveBucketTarget", mustGetClaimsFromToken(r))
@@ -275,11 +260,18 @@ func (a adminAPIHandlers) RemoveBucketTargetHandler(w http.ResponseWriter, r *ht
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
return
}
// Check if bucket exists.
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err := globalBucketTargetSys.RemoveTarget(ctx, bucket, arn); err != nil {
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
return
}
targets, err := globalBucketTargetSys.ListTargets(ctx, bucket)
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return

16
cmd/admin-router.go
View File

@@ -183,14 +183,14 @@ func registerAdminRouter(router *mux.Router, enableConfigOps, enableIAMOps bool)
}
// Bucket replication operations
// GetBucketTargetHandler
adminRouter.Methods(http.MethodGet).Path(adminVersion+"/list-bucket-targets").HandlerFunc(
httpTraceHdrs(adminAPI.ListBucketTargetsHandler)).Queries("bucket", "{bucket:.*}", "type", "{type:.*}")
// SetBucketTargetHandler
adminRouter.Methods(http.MethodPut).Path(adminVersion+"/set-bucket-target").HandlerFunc(
httpTraceHdrs(adminAPI.SetBucketTargetHandler)).Queries("bucket", "{bucket:.*}")
// SetBucketTargetHandler
adminRouter.Methods(http.MethodDelete).Path(adminVersion+"/remove-bucket-target").HandlerFunc(
httpTraceHdrs(adminAPI.RemoveBucketTargetHandler)).Queries("bucket", "{bucket:.*}", "arn", "{arn:.*}")
adminRouter.Methods(http.MethodGet).Path(adminVersion+"/list-remote-targets").HandlerFunc(
httpTraceHdrs(adminAPI.ListRemoteTargetsHandler)).Queries("bucket", "{bucket:.*}", "type", "{type:.*}")
// SetRemoteTargetHandler
adminRouter.Methods(http.MethodPut).Path(adminVersion+"/set-remote-target").HandlerFunc(
httpTraceHdrs(adminAPI.SetRemoteTargetHandler)).Queries("bucket", "{bucket:.*}")
// SetRemoteTargetHandler
adminRouter.Methods(http.MethodDelete).Path(adminVersion+"/remove-remote-target").HandlerFunc(
httpTraceHdrs(adminAPI.RemoveRemoteTargetHandler)).Queries("bucket", "{bucket:.*}", "arn", "{arn:.*}")
}
// -- Top APIs --

8
cmd/api-errors.go
View File

@@ -115,6 +115,7 @@ const (
ErrBucketRemoteArnInvalid
ErrBucketRemoteRemoveDisallowed
ErrReplicationTargetNotVersionedError
ErrReplicationSourceNotVersionedError
ErrReplicationNeedsVersioningError
ErrReplicationBucketNeedsVersioningError
ErrBucketReplicationDisabledError
@@ -871,6 +872,11 @@ var errorCodes = errorCodeMap{
Description: "The replication target does not have versioning enabled",
HTTPStatusCode: http.StatusBadRequest,
},
ErrReplicationSourceNotVersionedError: {
Code: "ReplicationSourceNotVersionedError",
Description: "The replication source does not have versioning enabled",
HTTPStatusCode: http.StatusBadRequest,
},
ErrReplicationNeedsVersioningError: {
Code: "InvalidRequest",
Description: "Versioning must be 'Enabled' on the bucket to apply a replication configuration",
@@ -1929,6 +1935,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrBucketRemoteRemoveDisallowed
case BucketReplicationTargetNotVersioned:
apiErr = ErrReplicationTargetNotVersionedError
case BucketReplicationSourceNotVersioned:
apiErr = ErrReplicationSourceNotVersionedError
case BucketQuotaExceeded:
apiErr = ErrAdminBucketQuotaExceeded
case *event.ErrInvalidEventName:

6
cmd/bucket-replication.go
View File

@@ -49,7 +49,7 @@ func getReplicationConfig(ctx context.Context, bucketName string) (rc *replicati
// validateReplicationDestination returns error if replication destination bucket missing or not configured
// It also returns true if replication destination is same as this server.
func validateReplicationDestination(ctx context.Context, bucket string, rCfg *replication.Config) (bool, error) {
clnt := globalBucketTargetSys.GetReplicationTargetClient(ctx, rCfg.ReplicationArn)
clnt := globalBucketTargetSys.GetReplicationTargetClient(ctx, rCfg.RoleArn)
if clnt == nil {
return false, BucketRemoteTargetNotFound{Bucket: bucket}
}
@@ -65,7 +65,7 @@ func validateReplicationDestination(ctx context.Context, bucket string, rCfg *re
}
}
// validate replication ARN against target endpoint
c, ok := globalBucketTargetSys.arnRemotesMap[rCfg.ReplicationArn]
c, ok := globalBucketTargetSys.arnRemotesMap[rCfg.RoleArn]
if ok {
if c.EndpointURL().String() == clnt.EndpointURL().String() {
sameTarget, _ := isLocalHost(clnt.EndpointURL().Hostname(), clnt.EndpointURL().Port(), globalMinioPort)
@@ -159,7 +159,7 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
logger.LogIf(ctx, err)
return
}
tgt := globalBucketTargetSys.GetReplicationTargetClient(ctx, cfg.ReplicationArn)
tgt := globalBucketTargetSys.GetReplicationTargetClient(ctx, cfg.RoleArn)
if tgt == nil {
return
}

89
cmd/bucket-targets.go
View File

@@ -18,7 +18,6 @@ package cmd
import (
"context"
"fmt"
"net/http"
"sync"
@@ -36,16 +35,42 @@ type BucketTargetSys struct {
clientsCache map[string]*miniogo.Core
}
// ListTargets - gets list of bucket targets for this bucket.
func (sys *BucketTargetSys) ListTargets(ctx context.Context, bucket string) (*madmin.BucketTargets, error) {
if globalIsGateway {
return nil, nil
// ListTargets lists bucket targets across tenant or for individual bucket, and returns
// results filtered by arnType
func (sys *BucketTargetSys) ListTargets(ctx context.Context, bucket, arnType string) (targets []madmin.BucketTarget) {
if bucket != "" {
if ts, err := sys.ListBucketTargets(ctx, bucket); err == nil {
for _, t := range ts.Targets {
if string(t.Type) == arnType || arnType == "" {
targets = append(targets, t.Clone())
}
}
}
return targets
}
sys.RLock()
defer sys.RUnlock()
for _, tgts := range sys.targetsMap {
for _, t := range tgts {
if string(t.Type) == arnType || arnType == "" {
targets = append(targets, t.Clone())
}
}
}
return
}
// ListBucketTargets - gets list of bucket targets for this bucket.
func (sys *BucketTargetSys) ListBucketTargets(ctx context.Context, bucket string) (*madmin.BucketTargets, error) {
sys.RLock()
defer sys.RUnlock()
tgts, ok := sys.targetsMap[bucket]
if ok {
return &madmin.BucketTargets{Targets: tgts}, nil
}
return nil, fmt.Errorf("No remote targets exist for bucket %s", bucket)
return nil, BucketRemoteTargetNotFound{Bucket: bucket}
}
// SetTarget - sets a new minio-go client target for this bucket.
@@ -60,8 +85,10 @@ func (sys *BucketTargetSys) SetTarget(ctx context.Context, bucket string, tgt *m
if err != nil {
return BucketRemoteTargetNotFound{Bucket: tgt.TargetBucket}
}
if tgt.Type == madmin.ReplicationArn {
if tgt.Type == madmin.ReplicationService {
if !globalBucketVersioningSys.Enabled(bucket) {
return BucketReplicationSourceNotVersioned{Bucket: bucket}
}
vcfg, err := clnt.GetBucketVersioning(ctx, tgt.TargetBucket)
if err != nil || vcfg.Status != string(versioning.Enabled) {
if isErrBucketNotFound(err) {
@@ -112,10 +139,10 @@ func (sys *BucketTargetSys) RemoveTarget(ctx context.Context, bucket, arnStr str
if err != nil {
return BucketRemoteArnInvalid{Bucket: bucket}
}
if arn.Type == madmin.ReplicationArn {
if arn.Type == madmin.ReplicationService {
// reject removal of remote target if replication configuration is present
rcfg, err := getReplicationConfig(ctx, bucket)
if err == nil && rcfg.ReplicationArn == arnStr {
if err == nil && rcfg.RoleArn == arnStr {
if _, ok := sys.arnRemotesMap[arnStr]; ok {
return BucketRemoteRemoveDisallowed{Bucket: bucket}
}
@@ -125,11 +152,17 @@ func (sys *BucketTargetSys) RemoveTarget(ctx context.Context, bucket, arnStr str
sys.Lock()
defer sys.Unlock()
targets := make([]madmin.BucketTarget, 0)
found := false
tgts := sys.targetsMap[bucket]
for _, tgt := range tgts {
if tgt.Arn != arnStr {
targets = append(targets, tgt)
continue
}
found = true
}
if !found {
return BucketRemoteTargetNotFound{Bucket: bucket}
}
sys.targetsMap[bucket] = targets
delete(sys.arnRemotesMap, arnStr)
@@ -168,6 +201,40 @@ func (sys *BucketTargetSys) Init(ctx context.Context, buckets []BucketInfo, objA
return nil
}
// UpdateTarget updates target to reflect metadata updates
func (sys *BucketTargetSys) UpdateTarget(bucket string, cfg *madmin.BucketTargets) {
if sys == nil {
return
}
sys.Lock()
defer sys.Unlock()
if cfg == nil || cfg.Empty() {
// remove target and arn association
if tgts, ok := sys.targetsMap[bucket]; ok {
for _, t := range tgts {
delete(sys.arnRemotesMap, t.Arn)
}
}
delete(sys.targetsMap, bucket)
return
}
if len(cfg.Targets) > 0 {
sys.targetsMap[bucket] = cfg.Targets
}
for _, tgt := range cfg.Targets {
tgtClient, err := sys.getRemoteTargetClient(&tgt)
if err != nil {
continue
}
sys.arnRemotesMap[tgt.Arn] = tgtClient
if _, ok := sys.clientsCache[tgtClient.EndpointURL().String()]; !ok {
sys.clientsCache[tgtClient.EndpointURL().String()] = tgtClient
}
}
sys.targetsMap[bucket] = cfg.Targets
}
// create minio-go clients for buckets having remote targets
func (sys *BucketTargetSys) load(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) {
for _, bucket := range buckets {
@@ -229,7 +296,7 @@ func (sys *BucketTargetSys) getRemoteARN(bucket string, target *madmin.BucketTar
return tgt.Arn
}
}
if !madmin.ArnType(target.Type).IsValid() {
if !madmin.ServiceType(target.Type).IsValid() {
return ""
}
arn := madmin.ARN{

7
cmd/object-api-errors.go
View File

@@ -411,6 +411,13 @@ func (e BucketReplicationTargetNotVersioned) Error() string {
return "Replication target does not have versioning enabled: " + e.Bucket
}
// BucketReplicationSourceNotVersioned replication source does not have versioning enabled.
type BucketReplicationSourceNotVersioned GenericError
func (e BucketReplicationSourceNotVersioned) Error() string {
return "Replication source does not have versioning enabled: " + e.Bucket
}
/// Bucket related errors.
// BucketNameInvalid - bucketname provided is invalid.

4
cmd/peer-rest-server.go
View File

@@ -610,6 +610,10 @@ func (s *peerRESTServer) LoadBucketMetadataHandler(w http.ResponseWriter, r *htt
if meta.notificationConfig != nil {
globalNotificationSys.AddRulesMap(bucketName, meta.notificationConfig.ToRulesMap())
}
if meta.bucketTargetConfig != nil {
globalBucketTargetSys.UpdateTarget(bucketName, meta.bucketTargetConfig)
}
}
// ReloadFormatHandler - Reload Format.