Compare commits

...

30 Commits

Author SHA1 Message Date
Klaus Post 216069d0da Remove 'null' version ID from directory object response (#17495)
Fixes #17494

Regression from #17132
2023-06-23 13:26:00 -07:00
Harshavardhana eefa047974 fix: keep decommission in a go-routine (#17496)
This was removed by mistake in #17491
2023-06-23 12:29:32 -07:00
Anis Eleuch d8dad5c9ea s3: Make/Delete buckets to use error quorum per pool (#17467) 2023-06-23 11:48:23 -07:00
Klaus Post bf8a68879c fix: Time ILM Actions for scanner info (#17493)
ILM Actions were not timed fix it.
2023-06-23 07:48:36 -07:00
Aditya Manthramurthy f3248a4b37 Redact all secrets from config viewing APIs (#17380)
This change adds a `Secret` property to `HelpKV` to identify secrets
like passwords and auth tokens that should not be revealed by the server
in its configuration fetching APIs. Configuration reporting APIs now do
not return secrets.
2023-06-23 07:45:27 -07:00
Harshavardhana d315d012a4 decom: during multiple pool decom preserve current pool status (#17491)
removal of completed pools must retain pool status of other
pools in draining, to resume any remaining draining operations.
2023-06-23 07:44:18 -07:00
Harshavardhana bd9bf3693f lambda: negative duration for presigned URL default to 1H (#17489)
fixes a bug where users created with Expiration as
timeSentinel is not rejected while generating the
presigned URL for lambda processing.
2023-06-23 00:17:24 -07:00
Klaus Post 15daa2e74a tooling: Add xlmeta --combine switch that will combine inline data (#17488)
Will combine or write partial data of each version found in the inspect data.

Example:

```
> xl-meta -export -combine inspect-data.1228fb52.zip

(... metadata json...)
}
Attempting to combine version "994f1113-da94-4be1-8551-9dbc54b204bc".
Read shard 1 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-01-of-13.data)
Read shard 2 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-02-of-13.data)
Read shard 3 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-03-of-13.data)
Read shard 4 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-04-of-13.data)
Read shard 6 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-06-of-13.data)
Read shard 7 Data shards 9 Parity 4 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-07-of-13.data)
Read shard 8 Data shards 8 Parity 5 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-08-of-13.data)
Read shard 9 Data shards 8 Parity 5 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-09-of-13.data)
Read shard 10 Data shards 8 Parity 5 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-10-of-13.data)
Read shard 11 Data shards 8 Parity 5 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-11-of-13.data)
Read shard 13 Data shards 8 Parity 5 (994f1113-da94-4be1-8551-9dbc54b204bc/shard-13-of-13.data)
Attempting to reconstruct using parity sets:
* Setup: Data shards: 9 - Parity blocks: 6
Have 6 complete remapped data shards and 6 complete parity shards. Could NOT reconstruct: too few shards given
* Setup: Data shards: 8 - Parity blocks: 5
Have 5 complete remapped data shards and 5 complete parity shards. Could reconstruct completely
0 bytes missing. Truncating 0 from the end.
Wrote output to 994f1113-da94-4be1-8551-9dbc54b204bc.complete
```

So far only inline data, but no real reason that external data can't also be included with some handling of blocks.

Supports only unencrypted data.
2023-06-22 12:41:24 -07:00
Harshavardhana 74759b05a5 make sure to set relevant config entries correctly (#17485)
Bonus: also allow skipping keys properly.
2023-06-22 10:04:02 -07:00
Aditya Manthramurthy 82ce78a17c Fix locking in policy attach API (#17426)
For policy attach/detach API to work correctly the server should hold a
lock before reading existing policy mapping and until after writing the
updated policy mapping. This is fixed in this change.

A site replication bug, where LDAP policy attach/detach were not
correctly propagated is also fixed in this change.

Bonus: Additionally, the server responds with the actual (or net)
changes performed in the attach/detach API call. For e.g. if a user
already has policy A applied, and a call to attach policies A and B is
performed, the server will respond that B was attached successfully.
2023-06-21 22:44:50 -07:00
Harshavardhana 9af6c6ceef under rebalance look for expired versions v/s remaining versions (#17482)
A continuation of PR #17479 for rebalance behavior must
also match the decommission behavior.

Fixes bug where rebalance would ignore rebalancing object
versions after one of the version returned "ObjectNotFound"
2023-06-21 13:23:20 -07:00
Harshavardhana 021372cc4c update helm release v5.0.11
Signed-off-by: Harshavardhana <harsha@minio.io>
2023-06-21 12:29:09 -07:00
Praveen raj Mani b94ab07c2f Honor global root CAs for kafka audit tls (#17481)
honor global root CAs for kafka audit tls
2023-06-21 10:50:40 -07:00
Harshavardhana 7605d07bb2 add support for bucket level request count per API (#17468)
New metrics added to calculate API request count
per bucket, per API.  Captures errors, including
4xx, 5xx HTTP status codes separately.
2023-06-21 09:41:59 -07:00
Harshavardhana ccc5801112 always look for expired versions v/s remaining versions (#17479)
while decommissioning it can so happen that the non-current
versions are all expired but there is a DEL marker as the
latest version.

For such objects, we should not decommission them instead
calculate the remaining versions and if the remaining versions
is one and that version is a DEL marker consider such
an object not to be scheduled for decommissioning.
2023-06-21 08:49:28 -07:00
Praveen raj Mani 7c72b25ef0 Add an option to make bucket notifications synchronous (#17406)
With the current asynchronous behaviour in sending notification events
to the targets, we can't provide guaranteed delivery as the systems
might go for restarts.

For such event-driven use-cases, we can provide an option to enable
synchronous events where the APIs wait until the event is successfully
sent or persisted.

This commit adds 'MINIO_API_SYNC_EVENTS' env which when set to 'on'
will enable sending/persisting events to targets synchronously.
2023-06-20 17:38:59 -07:00
Harshavardhana 02c2ec3027 skip onlineDisks with parity mismatch (#17478) 2023-06-20 13:18:24 -07:00
Harshavardhana 65c31fab12 fix: do not crash rebalance code instead set the object layer (#17465)
fixes #17421
2023-06-20 09:28:23 -07:00
jiuker b6b68be052 fix: replication check for duplicate endpoints detection with wrong route (#17474) 2023-06-20 09:27:54 -07:00
Harshavardhana 15911c85f6 safely ignore out of band deletions while decommissioning (#17473) 2023-06-20 08:31:42 -07:00
Aditya Manthramurthy 5a1612fe32 Bump up madmin-go and pkg deps (#17469) 2023-06-19 17:53:08 -07:00
Minio Trusted bbb7ae156c Update yaml files to latest version RELEASE.2023-06-19T19-52-50Z 2023-06-19 22:53:03 +00:00
Harshavardhana f9b8d1c699 fix: sio-error test to fail if commands fail (#17466) 2023-06-19 12:52:50 -07:00
Harshavardhana 1443b5927a allow quorum fileInfo to pick same parityBlocks (#17454)
Bonus: allow replication to proceed for 503 errors such as
with error code SlowDownRead
2023-06-18 18:20:15 -07:00
Anis Eleuch 35ef35b5c1 fix a integer divide by zero crash during rebalance (#17455)
A state is updated with a delete marker, which does not have parity or
data blocks defined, which can cause the integer divide by zero panics.

This commit fixes to avoid panics.
2023-06-18 11:14:53 -07:00
Harshavardhana 6806537eb3 event args list for fanOut notification must be sized same (#17450)
without this fan-out API can crash if client cancels
the on-going request.
2023-06-18 07:09:20 -07:00
Harshavardhana 64de61d15d fallback on etags if they match when mtime is not same (#17424)
on "unversioned" buckets there are situations
when successive concurrent I/O can lead to
an inconsistent state() with mtime while the
etag might be the same for the object on disk.

in such a scenario it is possible for us to
allow reading of the object since etag matches
and if etag matches we are guaranteed that we
have enough copies the object will be readable
and same.

This PR allows fallback in such scenarios.
2023-06-17 19:18:20 -07:00
Harshavardhana 22b7c8cd8a upgrade pkg and dperf to latest packages (#17448)
- dperf improvements in benchmarking read and write tests
- upgrade mimedb to use latest content-types
2023-06-17 07:31:36 -07:00
Poorna c4d0c49a5f ensure metadata updates go to same pool where version exists (#17451)
This PR also returns the replication status in 
proxy calls and defers replication attempt if 
HEAD on object version returned a error different
from NoSuchKey
2023-06-17 07:30:53 -07:00
Minio Trusted 142a5b0dcd Update yaml files to latest version RELEASE.2023-06-16T02-41-06Z 2023-06-16 05:36:19 +00:00
157 changed files with 1991 additions and 1202 deletions
+1 -1
View File
@@ -9,7 +9,7 @@ function _init() {
export CGO_ENABLED=0
## List of architectures and OS to test coss compilation.
SUPPORTED_OSARCH="linux/ppc64le linux/mips64 linux/amd64 linux/arm64 linux/s390x darwin/arm64 darwin/amd64 freebsd/amd64 windows/amd64 linux/arm linux/386 netbsd/amd64 linux/mips openbsd/amd64"
SUPPORTED_OSARCH="linux/ppc64le linux/mips64 linux/amd64 linux/arm64 linux/s390x darwin/arm64 darwin/amd64 freebsd/amd64 windows/amd64 linux/arm linux/386 netbsd/amd64 linux/mips"
}
function _build() {
+1 -1
View File
@@ -27,7 +27,7 @@ import (
"os"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
func main() {
+1 -1
View File
@@ -32,7 +32,7 @@ import (
jsoniter "github.com/json-iterator/go"
"github.com/klauspost/compress/zip"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/bucket/lifecycle"
objectlock "github.com/minio/minio/internal/bucket/object/lock"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"net/http"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/config"
iampolicy "github.com/minio/pkg/iam/policy"
+8 -4
View File
@@ -26,7 +26,7 @@ import (
"strconv"
"strings"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/cache"
"github.com/minio/minio/internal/config/etcd"
@@ -241,6 +241,8 @@ func setConfigKV(ctx context.Context, objectAPI ObjectLayer, kvBytes []byte) (re
// 1. `subsys:target` -> request for config of a single subsystem and target pair.
// 2. `subsys:` -> request for config of a single subsystem and the default target.
// 3. `subsys` -> request for config of all targets for the given subsystem.
//
// This is a reporting API and config secrets are redacted in the response.
func (a adminAPIHandlers) GetConfigKVHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "GetConfigKV")
@@ -268,7 +270,7 @@ func (a adminAPIHandlers) GetConfigKVHandler(w http.ResponseWriter, r *http.Requ
}
}
subSysConfigs, err := cfg.GetSubsysInfo(subSys, target)
subSysConfigs, err := cfg.GetSubsysInfo(subSys, target, true)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
@@ -490,7 +492,9 @@ func (a adminAPIHandlers) SetConfigHandler(w http.ResponseWriter, r *http.Reques
}
// GetConfigHandler - GET /minio/admin/v3/config
// Get config.json of this minio setup.
//
// This endpoint is mainly for exporting and backing up the configuration.
// Secrets are not redacted.
func (a adminAPIHandlers) GetConfigHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "GetConfig")
@@ -507,7 +511,7 @@ func (a adminAPIHandlers) GetConfigHandler(w http.ResponseWriter, r *http.Reques
hkvs := config.HelpSubSysMap[""]
for _, hkv := range hkvs {
// We ignore the error below, as we cannot get one.
cfgSubsysItems, _ := cfg.GetSubsysInfo(hkv.Key, "")
cfgSubsysItems, _ := cfg.GetSubsysInfo(hkv.Key, "", false)
for _, item := range cfgSubsysItems {
off := item.Config.Get(config.Enable) == config.EnableOff
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"net/http"
"strings"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/config"
cfgldap "github.com/minio/minio/internal/config/identity/ldap"
+2 -2
View File
@@ -22,7 +22,7 @@ import (
"io"
"net/http"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
iampolicy "github.com/minio/pkg/iam/policy"
@@ -150,7 +150,7 @@ func (a adminAPIHandlers) AttachDetachPolicyLDAP(w http.ResponseWriter, r *http.
}
// Call IAM subsystem
updatedAt, addedOrRemoved, err := globalIAMSys.PolicyDBUpdateLDAP(ctx, isAttach, par)
updatedAt, addedOrRemoved, _, err := globalIAMSys.PolicyDBUpdateLDAP(ctx, isAttach, par)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"strings"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/mux"
"github.com/minio/minio/internal/logger"
+1 -1
View File
@@ -30,7 +30,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
minio "github.com/minio/minio-go/v7"
"github.com/minio/pkg/sync/errgroup"
)
+35 -213
View File
@@ -26,11 +26,10 @@ import (
"net/http"
"os"
"sort"
"strings"
"time"
"github.com/klauspost/compress/zip"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/config/dns"
"github.com/minio/minio/internal/logger"
@@ -1771,24 +1770,40 @@ func (a adminAPIHandlers) ListPolicyMappingEntities(w http.ResponseWriter, r *ht
writeSuccessResponseJSON(w, econfigData)
}
// AttachPolicyBuiltin - POST /minio/admin/v3/idp/builtin/policy/attach
func (a adminAPIHandlers) AttachPolicyBuiltin(w http.ResponseWriter, r *http.Request) {
// AttachDetachPolicyBuiltin - POST /minio/admin/v3/idp/builtin/policy/{operation}
func (a adminAPIHandlers) AttachDetachPolicyBuiltin(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "AttachPolicyBuiltin")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.AttachPolicyAdminAction)
objectAPI, cred := validateAdminReq(ctx, w, r, iampolicy.UpdatePolicyAssociationAction,
iampolicy.AttachPolicyAdminAction)
if objectAPI == nil {
return
}
cred, _, s3Err := validateAdminSignature(ctx, r, "")
if s3Err != ErrNone {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
if r.ContentLength > maxEConfigJSONSize || r.ContentLength == -1 {
// More than maxConfigSize bytes were available
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminConfigTooLarge), r.URL)
return
}
password := cred.SecretKey
// Ensure body content type is opaque to ensure that request body has not
// been interpreted as form data.
contentType := r.Header.Get("Content-Type")
if contentType != "application/octet-stream" {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrBadRequest), r.URL)
return
}
operation := mux.Vars(r)["operation"]
if operation != "attach" && operation != "detach" {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminInvalidArgument), r.URL)
return
}
isAttach := operation == "attach"
password := cred.SecretKey
reqBytes, err := madmin.DecryptData(password, io.LimitReader(r.Body, r.ContentLength))
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
@@ -1806,235 +1821,42 @@ func (a adminAPIHandlers) AttachPolicyBuiltin(w http.ResponseWriter, r *http.Req
return
}
userOrGroup := par.User
isGroup := false
if userOrGroup == "" {
userOrGroup = par.Group
isGroup = true
}
if isGroup {
_, err := globalIAMSys.GetGroupDescription(userOrGroup)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
} else {
ok, _, err := globalIAMSys.IsTempUser(userOrGroup)
if err != nil && err != errNoSuchUser {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
if ok {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errIAMActionNotAllowed), r.URL)
return
}
// When the user is root credential you are not allowed to
// add policies for root user.
if userOrGroup == globalActiveCred.AccessKey {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errIAMActionNotAllowed), r.URL)
return
}
// Validate that user exists.
_, ok = globalIAMSys.GetUser(ctx, userOrGroup)
if !ok {
updatedAt, addedOrRemoved, _, err := globalIAMSys.PolicyDBUpdateBuiltin(ctx, isAttach, par)
if err != nil {
if err == errNoSuchUser || err == errNoSuchGroup {
if globalIAMSys.LDAPConfig.Enabled() {
// When LDAP is enabled, warn user that they are using the wrong
// API.
// API. FIXME: error can be no such group as well - fix errNoSuchUserLDAPWarn
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errNoSuchUserLDAPWarn), r.URL)
return
}
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errNoSuchUser), r.URL)
return
}
}
var existingPolicies []string
if isGroup {
existingPolicies, err = globalIAMSys.PolicyDBGet(userOrGroup, true)
} else {
existingPolicies, err = globalIAMSys.GetUserPolicies(userOrGroup)
}
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
policyMap := make(map[string]bool)
for _, p := range existingPolicies {
policyMap[p] = true
}
policiesToAttach := par.Policies
// Check if policy is already attached to user.
for _, p := range policiesToAttach {
if _, ok := policyMap[p]; ok {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrPolicyAlreadyAttached), r.URL)
return
}
}
existingPolicies = append(existingPolicies, policiesToAttach...)
newPolicies := strings.Join(existingPolicies, ",")
userType := regUser
updatedAt, err := globalIAMSys.PolicyDBSet(ctx, userOrGroup, newPolicies, userType, isGroup)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
logger.LogIf(ctx, globalSiteReplicationSys.IAMChangeHook(ctx, madmin.SRIAMItem{
Type: madmin.SRIAMItemPolicyMapping,
PolicyMapping: &madmin.SRPolicyMapping{
UserOrGroup: userOrGroup,
UserType: int(userType),
IsGroup: isGroup,
Policy: strings.Join(policiesToAttach, ","),
},
respBody := madmin.PolicyAssociationResp{
UpdatedAt: updatedAt,
}))
writeResponse(w, http.StatusCreated, nil, mimeNone)
}
// DetachPolicyBuiltin - POST /minio/admin/v3/idp/builtin/policy/detach
func (a adminAPIHandlers) DetachPolicyBuiltin(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "DetachPolicyBuiltin")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.AttachPolicyAdminAction)
if objectAPI == nil {
return
}
cred, _, s3Err := validateAdminSignature(ctx, r, "")
if s3Err != ErrNone {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
return
}
password := cred.SecretKey
reqBytes, err := madmin.DecryptData(password, io.LimitReader(r.Body, r.ContentLength))
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
var par madmin.PolicyAssociationReq
if err = json.Unmarshal(reqBytes, &par); err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
if err = par.IsValid(); err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
userOrGroup := par.User
isGroup := false
if userOrGroup == "" {
userOrGroup = par.Group
isGroup = true
}
if isGroup {
_, err := globalIAMSys.GetGroupDescription(userOrGroup)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
if isAttach {
respBody.PoliciesAttached = addedOrRemoved
} else {
ok, _, err := globalIAMSys.IsTempUser(userOrGroup)
if err != nil && err != errNoSuchUser {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
if ok {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errIAMActionNotAllowed), r.URL)
return
}
// Validate that user exists.
_, ok = globalIAMSys.GetUser(ctx, userOrGroup)
if !ok {
if globalIAMSys.LDAPConfig.Enabled() {
// When LDAP is enabled, warn user that they are using the wrong
// API.
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errNoSuchUserLDAPWarn), r.URL)
return
}
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, errNoSuchUser), r.URL)
return
}
respBody.PoliciesDetached = addedOrRemoved
}
userType := regUser
if globalIAMSys.GetUsersSysType() == LDAPUsersSysType {
userType = stsUser
}
var existingPolicies []string
if isGroup {
existingPolicies, err = globalIAMSys.PolicyDBGet(userOrGroup, true)
} else {
existingPolicies, err = globalIAMSys.GetUserPolicies(userOrGroup)
}
data, err := json.Marshal(respBody)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
policyMap := make(map[string]bool)
for _, p := range existingPolicies {
policyMap[p] = true
}
policiesToDetach := par.Policies
// Check if policy is already attached to user.
for _, p := range policiesToDetach {
if _, ok := policyMap[p]; ok {
delete(policyMap, p)
} else {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrPolicyNotAttached), r.URL)
return
}
}
newPoliciesSl := []string{}
for p := range policyMap {
newPoliciesSl = append(newPoliciesSl, p)
}
newPolicies := strings.Join(newPoliciesSl, ",")
updatedAt, err := globalIAMSys.PolicyDBSet(ctx, userOrGroup, newPolicies, userType, isGroup)
encryptedData, err := madmin.EncryptData(password, data)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
logger.LogIf(ctx, globalSiteReplicationSys.IAMChangeHook(ctx, madmin.SRIAMItem{
Type: madmin.SRIAMItemPolicyMapping,
PolicyMapping: &madmin.SRPolicyMapping{
UserOrGroup: userOrGroup,
UserType: int(userType),
IsGroup: isGroup,
Policy: newPolicies,
},
UpdatedAt: updatedAt,
}))
// Return successful JSON response
writeSuccessNoContent(w)
writeSuccessResponseJSON(w, encryptedData)
}
const (
+1 -1
View File
@@ -33,7 +33,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
cr "github.com/minio/minio-go/v7/pkg/credentials"
+4 -4
View File
@@ -45,8 +45,8 @@ import (
"github.com/dustin/go-humanize"
"github.com/klauspost/compress/zip"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v2/estream"
"github.com/minio/madmin-go/v3"
"github.com/minio/madmin-go/v3/estream"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/dsync"
"github.com/minio/minio/internal/handlers"
@@ -1525,7 +1525,7 @@ func (a adminAPIHandlers) TraceHandler(w http.ResponseWriter, r *http.Request) {
return shouldTrace(entry, traceOpts)
})
if err != nil {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
@@ -1605,7 +1605,7 @@ func (a adminAPIHandlers) ConsoleLogHandler(w http.ResponseWriter, r *http.Reque
err = globalConsoleSys.Subscribe(logCh, ctx.Done(), node, limitLines, logKind, nil)
if err != nil {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
+1 -1
View File
@@ -31,7 +31,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
"github.com/minio/mux"
)
+1 -1
View File
@@ -27,7 +27,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
)
+3 -6
View File
@@ -22,7 +22,7 @@ import (
"github.com/klauspost/compress/gzhttp"
"github.com/klauspost/compress/gzip"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
)
@@ -162,11 +162,8 @@ func registerAdminRouter(router *mux.Router, enableConfigOps bool) {
HandlerFunc(gz(httpTraceHdrs(adminAPI.SetPolicyForUserOrGroup))).
Queries("policyName", "{policyName:.*}", "userOrGroup", "{userOrGroup:.*}", "isGroup", "{isGroup:true|false}")
// Attach policies to user or group
adminRouter.Methods(http.MethodPost).Path(adminVersion + "/idp/builtin/policy/attach").HandlerFunc(gz(httpTraceHdrs(adminAPI.AttachPolicyBuiltin)))
// Detach policies from user or group
adminRouter.Methods(http.MethodPost).Path(adminVersion + "/idp/builtin/policy/detach").HandlerFunc(gz(httpTraceHdrs(adminAPI.DetachPolicyBuiltin)))
// Attach/Detach policies to/from user or group
adminRouter.Methods(http.MethodPost).Path(adminVersion + "/idp/builtin/policy/{operation}").HandlerFunc(gz(httpTraceHdrs(adminAPI.AttachDetachPolicyBuiltin)))
// Remove user IAM
adminRouter.Methods(http.MethodDelete).Path(adminVersion+"/remove-user").HandlerFunc(gz(httpTraceHdrs(adminAPI.RemoveUser))).Queries("accessKey", "{accessKey:.*}")
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"strings"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
+32 -20
View File
@@ -31,7 +31,7 @@ import (
"github.com/minio/minio/internal/ioutil"
"google.golang.org/api/googleapi"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/auth"
@@ -184,8 +184,10 @@ const (
ErrBucketAlreadyExists
ErrMetadataTooLarge
ErrUnsupportedMetadata
ErrUnsupportedHostHeader
ErrMaximumExpires
ErrSlowDown
ErrSlowDownRead
ErrSlowDownWrite
ErrInvalidPrefixMarker
ErrBadRequest
ErrKeyTooLongError
@@ -257,9 +259,9 @@ const (
ErrInvalidResourceName
ErrInvalidLifecycleQueryParameter
ErrServerNotInitialized
ErrOperationTimedOut
ErrRequestTimedout
ErrClientDisconnected
ErrOperationMaxedOut
ErrTooManyRequests
ErrInvalidRequest
ErrTransitionStorageClassNotFoundError
// MinIO storage class error codes
@@ -843,11 +845,16 @@ var errorCodes = errorCodeMap{
Description: "Request is not valid yet",
HTTPStatusCode: http.StatusForbidden,
},
ErrSlowDown: {
Code: "SlowDown",
ErrSlowDownRead: {
Code: "SlowDownRead",
Description: "Resource requested is unreadable, please reduce your request rate",
HTTPStatusCode: http.StatusServiceUnavailable,
},
ErrSlowDownWrite: {
Code: "SlowDownWrite",
Description: "Resource requested is unwritable, please reduce your request rate",
HTTPStatusCode: http.StatusServiceUnavailable,
},
ErrInvalidPrefixMarker: {
Code: "InvalidPrefixMarker",
Description: "Invalid marker prefix combination",
@@ -1416,7 +1423,7 @@ var errorCodes = errorCodeMap{
Description: "Cannot respond to plain-text request from TLS-encrypted server",
HTTPStatusCode: http.StatusBadRequest,
},
ErrOperationTimedOut: {
ErrRequestTimedout: {
Code: "RequestTimeout",
Description: "A timeout occurred while trying to lock a resource, please reduce your request rate",
HTTPStatusCode: http.StatusServiceUnavailable,
@@ -1426,9 +1433,9 @@ var errorCodes = errorCodeMap{
Description: "Client disconnected before response was ready",
HTTPStatusCode: 499, // No official code, use nginx value.
},
ErrOperationMaxedOut: {
Code: "SlowDown",
Description: "A timeout exceeded while waiting to proceed with the request, please reduce your request rate",
ErrTooManyRequests: {
Code: "TooManyRequests",
Description: "Deadline exceeded while waiting in incoming queue, please reduce your request rate",
HTTPStatusCode: http.StatusServiceUnavailable,
},
ErrUnsupportedMetadata: {
@@ -1436,6 +1443,11 @@ var errorCodes = errorCodeMap{
Description: "Your metadata headers are not supported.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrUnsupportedHostHeader: {
Code: "InvalidArgument",
Description: "Your Host header is malformed.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrObjectTampered: {
Code: "XMinioObjectTampered",
Description: errObjectTampered.Error(),
@@ -2039,7 +2051,7 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
}
// Only return ErrClientDisconnected if the provided context is actually canceled.
// This way downstream context.Canceled will still report ErrOperationTimedOut
// This way downstream context.Canceled will still report ErrRequestTimedout
if contextCanceled(ctx) && errors.Is(ctx.Err(), context.Canceled) {
return ErrClientDisconnected
}
@@ -2083,9 +2095,9 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
case errInvalidStorageClass:
apiErr = ErrInvalidStorageClass
case errErasureReadQuorum:
apiErr = ErrSlowDown
apiErr = ErrSlowDownRead
case errErasureWriteQuorum:
apiErr = ErrSlowDown
apiErr = ErrSlowDownWrite
// SSE errors
case errInvalidEncryptionParameters:
apiErr = ErrInvalidEncryptionParameters
@@ -2119,10 +2131,10 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrKMSKeyNotFoundException
case errKMSDefaultKeyAlreadyConfigured:
apiErr = ErrKMSDefaultKeyAlreadyConfigured
case context.Canceled, context.DeadlineExceeded:
apiErr = ErrOperationTimedOut
case errDiskNotFound:
apiErr = ErrSlowDown
case context.Canceled:
apiErr = ErrClientDisconnected
case context.DeadlineExceeded:
apiErr = ErrRequestTimedout
case objectlock.ErrInvalidRetentionDate:
apiErr = ErrInvalidRetentionDate
case objectlock.ErrPastObjectLockRetainDate:
@@ -2201,9 +2213,9 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
case InvalidPart:
apiErr = ErrInvalidPart
case InsufficientWriteQuorum:
apiErr = ErrSlowDown
apiErr = ErrSlowDownWrite
case InsufficientReadQuorum:
apiErr = ErrSlowDown
apiErr = ErrSlowDownRead
case InvalidMarkerPrefixCombination:
apiErr = ErrNotImplemented
case InvalidUploadIDKeyCombination:
@@ -2297,7 +2309,7 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
case *event.ErrUnsupportedConfiguration:
apiErr = ErrUnsupportedNotification
case OperationTimedOut:
apiErr = ErrOperationTimedOut
apiErr = ErrRequestTimedout
case BackendDown:
apiErr = ErrBackendDown
case ObjectNameTooLong:
+2 -2
View File
@@ -40,8 +40,8 @@ var toAPIErrorTests = []struct {
{err: ObjectNameInvalid{}, errCode: ErrInvalidObjectName},
{err: InvalidUploadID{}, errCode: ErrNoSuchUpload},
{err: InvalidPart{}, errCode: ErrInvalidPart},
{err: InsufficientReadQuorum{}, errCode: ErrSlowDown},
{err: InsufficientWriteQuorum{}, errCode: ErrSlowDown},
{err: InsufficientReadQuorum{}, errCode: ErrSlowDownRead},
{err: InsufficientWriteQuorum{}, errCode: ErrSlowDownWrite},
{err: InvalidMarkerPrefixCombination{}, errCode: ErrNotImplemented},
{err: InvalidUploadIDKeyCombination{}, errCode: ErrNotImplemented},
{err: MalformedUploadID{}, errCode: ErrNoSuchUpload},
+2
View File
@@ -945,6 +945,8 @@ func writeErrorResponse(ctx context.Context, w http.ResponseWriter, err APIError
}
func writeErrorResponseHeadersOnly(w http.ResponseWriter, err APIError) {
w.Header().Set(xMinIOErrCodeHeader, err.Code)
w.Header().Set(xMinIOErrDescHeader, "\""+err.Description+"\"")
writeResponse(w, err.HTTPStatusCode, nil, mimeNone)
}
+211 -209
View File
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"strconv"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/env"
)
+6 -2
View File
@@ -30,9 +30,11 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/env"
)
const (
@@ -345,7 +347,9 @@ func initAutoHeal(ctx context.Context, objAPI ObjectLayer) {
globalBackgroundHealState.pushHealLocalDisks(getLocalDisksToHeal()...)
go monitorLocalDisksAndHeal(ctx, z)
if env.Get("_MINIO_AUTO_DISK_HEALING", config.EnableOn) == config.EnableOn {
go monitorLocalDisksAndHeal(ctx, z)
}
}
func getLocalDisksToHeal() (disksToHeal Endpoints) {
+1 -1
View File
@@ -37,7 +37,7 @@ import (
"github.com/dustin/go-humanize"
"github.com/lithammer/shortuuid/v4"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
miniogo "github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/pubsub"
)
+1 -1
View File
@@ -26,7 +26,7 @@ import (
"net/http"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
+22 -5
View File
@@ -41,7 +41,7 @@ import (
"github.com/minio/mux"
"github.com/valyala/bytebufferpool"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio-go/v7/pkg/tags"
@@ -654,6 +654,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
if deleteResult.errInfo.Code != "" {
deleteErrors = append(deleteErrors, deleteResult.errInfo)
} else {
// All deletes on directory objects was with `nullVersionID`.
// Remove it from response.
if isDirObject(deleteResult.delInfo.ObjectName) && deleteResult.delInfo.VersionID == nullVersionID {
deleteResult.delInfo.VersionID = ""
}
deletedObjects = append(deletedObjects, deleteResult.delInfo)
}
}
@@ -669,6 +674,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
}
if replicateDeletes && (dobj.DeleteMarkerReplicationStatus() == replication.Pending || dobj.VersionPurgeStatus() == Pending) {
// copy so we can re-add null ID.
dobj := dobj
if isDirObject(dobj.ObjectName) && dobj.VersionID == "" {
dobj.VersionID = nullVersionID
}
dv := DeletedObjectReplicationInfo{
DeletedObject: dobj,
Bucket: bucket,
@@ -1315,6 +1325,16 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
Key: objInfo.Name,
Error: errs[i].Error(),
})
eventArgsList = append(eventArgsList, eventArgs{
EventName: event.ObjectCreatedPost,
BucketName: objInfo.Bucket,
Object: ObjectInfo{Name: objInfo.Name},
ReqParams: extractReqParams(r),
RespElements: extractRespElements(w),
UserAgent: fmt.Sprintf("%s MinIO-Fan-Out (failed: %v)", r.UserAgent(), errs[i]),
Host: handlers.GetSourceIP(r),
})
continue
}
@@ -1522,10 +1542,7 @@ func (api objectAPIHandlers) HeadBucketHandler(w http.ResponseWriter, r *http.Re
}
if s3Error := checkRequestAuthType(ctx, r, policy.ListBucketAction, bucket, ""); s3Error != ErrNone {
errCode := errorCodes.ToAPIErr(s3Error)
w.Header().Set(xMinIOErrCodeHeader, errCode.Code)
w.Header().Set(xMinIOErrDescHeader, "\""+errCode.Description+"\"")
writeErrorResponseHeadersOnly(w, errCode)
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(s3Error))
return
}
+2 -2
View File
@@ -749,7 +749,7 @@ func testAPIDeleteMultipleObjectsHandler(obj ObjectLayer, instanceType, bucketNa
for i := range requestList[0].Objects {
var vid string
if isDirObject(requestList[0].Objects[i].ObjectName) {
vid = nullVersionID
vid = ""
}
deletedObjects[i] = DeletedObject{
ObjectName: requestList[0].Objects[i].ObjectName,
@@ -766,7 +766,7 @@ func testAPIDeleteMultipleObjectsHandler(obj ObjectLayer, instanceType, bucketNa
for i := range requestList[1].Objects {
var vid string
if isDirObject(requestList[0].Objects[i].ObjectName) {
vid = nullVersionID
vid = ""
}
deletedObjects[i] = DeletedObject{
ObjectName: requestList[1].Objects[i].ObjectName,
+1 -1
View File
@@ -32,7 +32,7 @@ import (
"time"
"github.com/google/uuid"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/amztime"
sse "github.com/minio/minio/internal/bucket/encryption"
+1 -1
View File
@@ -25,7 +25,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio-go/v7/pkg/tags"
bucketsse "github.com/minio/minio/internal/bucket/encryption"
+1 -1
View File
@@ -29,7 +29,7 @@ import (
"path"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/tags"
bucketsse "github.com/minio/minio/internal/bucket/encryption"
"github.com/minio/minio/internal/bucket/lifecycle"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"net/http"
humanize "github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
"github.com/minio/pkg/bucket/policy"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"fmt"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
)
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/replication"
xhttp "github.com/minio/minio/internal/http"
)
+22 -2
View File
@@ -35,7 +35,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/encrypt"
"github.com/minio/minio-go/v7/pkg/tags"
@@ -170,7 +170,7 @@ func checkRemoteEndpoint(ctx context.Context, epURL *url.URL) error {
reqURL := &url.URL{
Scheme: epURL.Scheme,
Host: epURL.Host,
Path: healthCheckReadinessPath,
Path: healthCheckPathPrefix + healthCheckReadinessPath,
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
@@ -1384,6 +1384,25 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
return
}
}
// if target returns error other than NoSuchKey, defer replication attempt
if cerr != nil {
errResp := minio.ToErrorResponse(cerr)
switch errResp.Code {
case "NoSuchKey", "NoSuchVersion", "SlowDownRead":
rAction = replicateAll
default:
logger.LogIf(ctx, fmt.Errorf("unable to replicate %s/%s (%s). Target (%s) returned %s error on HEAD",
bucket, object, objInfo.VersionID, tgt.EndpointURL(), cerr))
sendEvent(eventArgs{
EventName: event.ObjectReplicationNotTracked,
BucketName: bucket,
Object: objInfo,
UserAgent: "Internal: [Replication]",
Host: globalLocalNodeName,
})
return
}
}
rinfo.ReplicationStatus = replication.Completed
rinfo.Size = size
rinfo.ReplicationAction = rAction
@@ -2094,6 +2113,7 @@ func proxyHeadToRepTarget(ctx context.Context, bucket, object string, rs *HTTPRa
StorageClass: objInfo.StorageClass,
ReplicationStatusInternal: objInfo.ReplicationStatus,
UserTags: tags.String(),
ReplicationStatus: replication.StatusType(objInfo.ReplicationStatus),
}
oi.UserDefined = make(map[string]string, len(objInfo.Metadata))
for k, v := range objInfo.Metadata {
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/replication"
xhttp "github.com/minio/minio/internal/http"
)
+1 -1
View File
@@ -25,7 +25,7 @@ import (
"time"
jsoniter "github.com/json-iterator/go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio/internal/bucket/replication"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"net/http"
humanize "github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/versioning"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"net/url"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
)
+1 -1
View File
@@ -52,7 +52,7 @@ import (
"github.com/minio/console/restapi"
"github.com/minio/console/restapi/operations"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio-go/v7/pkg/set"
+4 -1
View File
@@ -24,7 +24,7 @@ import (
"strings"
"sync"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/api"
"github.com/minio/minio/internal/config/cache"
@@ -616,6 +616,9 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
}
for n, l := range loggerCfg.AuditKafka {
if l.Enabled {
if l.TLS.Enable {
l.TLS.RootCAs = globalRootCAs
}
l.LogOnce = logger.LogOnceIf
loggerCfg.AuditKafka[n] = l
}
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"bytes"
"testing"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
)
+1 -1
View File
@@ -27,7 +27,7 @@ import (
"strings"
jsoniter "github.com/json-iterator/go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/kms"
)
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"sync"
"sync/atomic"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/minio/internal/logger/target/console"
"github.com/minio/minio/internal/logger/target/types"
+1 -1
View File
@@ -7,7 +7,7 @@ import (
"time"
"unsafe"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/lifecycle"
)
+2 -2
View File
@@ -34,7 +34,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/replication"
@@ -941,7 +941,7 @@ func (i *scannerItem) applyLifecycle(ctx context.Context, o ObjectLayer, oi Obje
console.Debugf(applyActionsLogPrefix+" lifecycle: %q Initial scan: %v\n", i.objectPath(), lcEvt.Action)
}
}
defer globalScannerMetrics.timeILM(lcEvt.Action)
defer globalScannerMetrics.timeILM(lcEvt.Action)()
switch lcEvt.Action {
case lifecycle.DeleteAction, lifecycle.DeleteVersionAction, lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
+1 -1
View File
@@ -33,7 +33,7 @@ import (
"github.com/cespare/xxhash/v2"
"github.com/klauspost/compress/zstd"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/logger"
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"sort"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
// BucketTargetUsageInfo - bucket target usage info provides
+3
View File
@@ -359,6 +359,9 @@ func createServerEndpoints(serverAddr string, args ...string) (
if err != nil {
return nil, -1, err
}
for i := range endpointList {
endpointList[i].SetPool(0)
}
endpointServerPools = append(endpointServerPools, PoolEndpoints{
Legacy: true,
SetCount: len(setArgs),
+52 -3
View File
@@ -38,6 +38,7 @@ import (
"github.com/minio/minio/internal/mountinfo"
"github.com/minio/pkg/env"
xnet "github.com/minio/pkg/net"
"golang.org/x/exp/slices"
)
// EndpointType - enum for endpoint type.
@@ -58,9 +59,17 @@ type ProxyEndpoint struct {
Transport http.RoundTripper
}
// Node holds information about a node in this cluster
type Node struct {
*url.URL
Pools []int
IsLocal bool
}
// Endpoint - any type of endpoint.
type Endpoint struct {
*url.URL
Pool int
IsLocal bool
}
@@ -97,6 +106,11 @@ func (endpoint *Endpoint) UpdateIsLocal() (err error) {
return nil
}
// SetPool sets a specific pool number to this node
func (endpoint *Endpoint) SetPool(i int) {
endpoint.Pool = i
}
// NewEndpoint - returns new endpoint based on given arguments.
func NewEndpoint(arg string) (ep Endpoint, e error) {
// isEmptyPath - check whether given path is not empty.
@@ -209,6 +223,35 @@ type PoolEndpoints struct {
// EndpointServerPools - list of list of endpoints
type EndpointServerPools []PoolEndpoints
// GetNodes returns a sorted list of nodes in this cluster
func (l EndpointServerPools) GetNodes() (nodes []Node) {
nodesMap := make(map[string]Node)
for _, pool := range l {
for _, ep := range pool.Endpoints {
node, ok := nodesMap[ep.Host]
if !ok {
node.IsLocal = ep.IsLocal
node.URL = &url.URL{
Scheme: ep.Scheme,
Host: ep.Host,
}
}
if !slices.Contains(node.Pools, ep.Pool) {
node.Pools = append(node.Pools, ep.Pool)
}
nodesMap[ep.Host] = node
}
}
nodes = make([]Node, 0, len(nodesMap))
for _, v := range nodesMap {
nodes = append(nodes, v)
}
sort.Slice(nodes, func(i, j int) bool {
return nodes[i].Host < nodes[j].Host
})
return
}
// GetPoolIdx return pool index
func (l EndpointServerPools) GetPoolIdx(pool string) int {
for id, ep := range globalEndpoints {
@@ -768,6 +811,8 @@ func CreatePoolEndpoints(serverAddr string, poolArgs ...[][]string) ([]Endpoints
return nil, setupType, config.ErrInvalidEndpoint(nil).Msg("use path style endpoint for single node setup")
}
endpoint.SetPool(0)
var endpoints Endpoints
endpoints = append(endpoints, endpoint)
setupType = ErasureSDSetupType
@@ -781,7 +826,7 @@ func CreatePoolEndpoints(serverAddr string, poolArgs ...[][]string) ([]Endpoints
return poolEndpoints, setupType, nil
}
for i, args := range poolArgs {
for poolIdx, args := range poolArgs {
var endpoints Endpoints
for _, iargs := range args {
// Convert args to endpoints
@@ -795,6 +840,10 @@ func CreatePoolEndpoints(serverAddr string, poolArgs ...[][]string) ([]Endpoints
return nil, setupType, config.ErrInvalidErasureEndpoints(nil).Msg(err.Error())
}
for i := range eps {
eps[i].SetPool(poolIdx)
}
endpoints = append(endpoints, eps...)
}
@@ -802,7 +851,7 @@ func CreatePoolEndpoints(serverAddr string, poolArgs ...[][]string) ([]Endpoints
return nil, setupType, config.ErrInvalidErasureEndpoints(nil).Msg("invalid number of endpoints")
}
poolEndpoints[i] = endpoints
poolEndpoints[poolIdx] = endpoints
}
for _, endpoints := range poolEndpoints {
@@ -960,7 +1009,7 @@ func CreateEndpoints(serverAddr string, args ...[]string) (Endpoints, SetupType,
_, serverAddrPort := mustSplitHostPort(serverAddr)
// For single arg, return FS setup.
// For single arg, return single drive setup.
if len(args) == 1 && len(args[0]) == 1 {
var endpoint Endpoint
endpoint, err = NewEndpoint(args[0][0])
+86 -3
View File
@@ -22,9 +22,41 @@ import (
"context"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
func commonETags(etags []string) (etag string, maxima int) {
etagOccurrenceMap := make(map[string]int, len(etags))
// Ignore the uuid sentinel and count the rest.
for _, etag := range etags {
if etag == "" {
continue
}
etagOccurrenceMap[etag]++
}
maxima = 0 // Counter for remembering max occurrence of elements.
latest := ""
// Find the common cardinality from previously collected
// occurrences of elements.
for etag, count := range etagOccurrenceMap {
if count < maxima {
continue
}
// We are at or above maxima
if count > maxima {
maxima = count
latest = etag
}
}
// Return the collected common max time, with maxima
return latest, maxima
}
// commonTime returns a maximally occurring time from a list of time.
func commonTimeAndOccurence(times []time.Time, group time.Duration) (maxTime time.Time, maxima int) {
timeOccurenceMap := make(map[int64]int, len(times))
@@ -86,6 +118,13 @@ func commonTime(modTimes []time.Time, quorum int) time.Time {
return timeSentinel
}
func commonETag(etags []string, quorum int) string {
if etag, count := commonETags(etags); count >= quorum {
return etag
}
return ""
}
// Beginning of unix time is treated as sentinel value here.
var (
timeSentinel = time.Unix(0, 0).UTC()
@@ -102,6 +141,33 @@ func bootModtimes(diskCount int) []time.Time {
return modTimes
}
func listObjectETags(partsMetadata []FileInfo, errs []error, quorum int) (etags []string) {
etags = make([]string, len(partsMetadata))
vidMap := map[string]int{}
for index, metadata := range partsMetadata {
if errs[index] != nil {
continue
}
vid := metadata.VersionID
if metadata.VersionID == "" {
vid = nullVersionID
}
vidMap[vid]++
etags[index] = metadata.Metadata["etag"]
}
for _, count := range vidMap {
// do we have enough common versions
// that have enough quorum to satisfy
// the etag.
if count >= quorum {
return etags
}
}
return make([]string, len(partsMetadata))
}
// Extracts list of times from FileInfo slice and returns, skips
// slice elements which have errors.
func listObjectModtimes(partsMetadata []FileInfo, errs []error) (modTimes []time.Time) {
@@ -162,7 +228,7 @@ func listObjectDiskMtimes(partsMetadata []FileInfo) (diskMTimes []time.Time) {
// - a slice of disks where disk having 'older' xl.meta (or nothing)
// are set to nil.
// - latest (in time) of the maximally occurring modTime(s), which has at least quorum occurrences.
func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error, quorum int) (onlineDisks []StorageAPI, modTime time.Time) {
func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error, quorum int) (onlineDisks []StorageAPI, modTime time.Time, etag string) {
onlineDisks = make([]StorageAPI, len(disks))
// List all the file commit ids from parts metadata.
@@ -171,6 +237,23 @@ func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error,
// Reduce list of UUIDs to a single common value.
modTime = commonTime(modTimes, quorum)
if modTime.IsZero() || modTime.Equal(timeSentinel) {
etags := listObjectETags(partsMetadata, errs, quorum)
etag = commonETag(etags, quorum)
if etag != "" { // allow this fallback only if a non-empty etag is found.
for index, e := range etags {
if partsMetadata[index].IsValid() && e == etag {
onlineDisks[index] = disks[index]
} else {
onlineDisks[index] = nil
}
}
return onlineDisks, modTime, etag
}
}
// Create a new online disks slice, which have common uuid.
for index, t := range modTimes {
if partsMetadata[index].IsValid() && t.Equal(modTime) {
@@ -180,7 +263,7 @@ func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error,
}
}
return onlineDisks, modTime
return onlineDisks, modTime, ""
}
// disksWithAllParts - This function needs to be called with
+4 -4
View File
@@ -27,7 +27,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
// Returns the latest updated FileInfo files and error in case of failure.
@@ -299,7 +299,7 @@ func TestListOnlineDisks(t *testing.T) {
}
rQuorum := len(errs) - z.serverPools[0].sets[0].defaultParityCount
onlineDisks, modTime := listOnlineDisks(erasureDisks, partsMetadata, test.errs, rQuorum)
onlineDisks, modTime, _ := listOnlineDisks(erasureDisks, partsMetadata, test.errs, rQuorum)
if !modTime.Equal(test.expectedTime) {
t.Fatalf("Expected modTime to be equal to %v but was found to be %v",
test.expectedTime, modTime)
@@ -481,7 +481,7 @@ func TestListOnlineDisksSmallObjects(t *testing.T) {
}
rQuorum := len(errs) - z.serverPools[0].sets[0].defaultParityCount
onlineDisks, modTime := listOnlineDisks(erasureDisks, partsMetadata, test.errs, rQuorum)
onlineDisks, modTime, _ := listOnlineDisks(erasureDisks, partsMetadata, test.errs, rQuorum)
if !modTime.Equal(test.expectedTime) {
t.Fatalf("Expected modTime to be equal to %v but was found to be %v",
test.expectedTime, modTime)
@@ -548,7 +548,7 @@ func TestDisksWithAllParts(t *testing.T) {
t.Fatalf("Failed to get quorum consistent fileInfo %v", err)
}
erasureDisks, _ = listOnlineDisks(erasureDisks, partsMetadata, errs, readQuorum)
erasureDisks, _, _ = listOnlineDisks(erasureDisks, partsMetadata, errs, readQuorum)
filteredDisks, errs, _ := disksWithAllParts(ctx, erasureDisks, partsMetadata,
errs, fi, bucket, object, madmin.HealDeepScan)
+8 -7
View File
@@ -26,7 +26,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/sync/errgroup"
)
@@ -386,11 +386,11 @@ func (er *erasureObjects) healObject(ctx context.Context, bucket string, object
// List of disks having latest version of the object xl.meta
// (by modtime).
onlineDisks, modTime := listOnlineDisks(storageDisks, partsMetadata, errs, readQuorum)
onlineDisks, modTime, etag := listOnlineDisks(storageDisks, partsMetadata, errs, readQuorum)
// Latest FileInfo for reference. If a valid metadata is not
// present, it is as good as object not found.
latestMeta, err := pickValidFileInfo(ctx, partsMetadata, modTime, readQuorum)
latestMeta, err := pickValidFileInfo(ctx, partsMetadata, modTime, etag, readQuorum)
if err != nil {
return result, err
}
@@ -503,7 +503,7 @@ func (er *erasureObjects) healObject(ctx context.Context, bucket string, object
if !latestMeta.XLV1 && !latestMeta.Deleted && !recreate && disksToHealCount > latestMeta.Erasure.ParityBlocks {
// When disk to heal count is greater than parity blocks we should simply error out.
err := fmt.Errorf("more drives are expected to heal than parity, returned errors: %v (dataErrs %v) -> %s/%s(%s)", errs, dataErrs, bucket, object, versionID)
err := fmt.Errorf("(%d > %d) more drives are expected to heal than parity, returned errors: %v (dataErrs %v) -> %s/%s(%s)", disksToHealCount, latestMeta.Erasure.ParityBlocks, errs, dataErrs, bucket, object, versionID)
logger.LogIf(ctx, err)
return er.defaultHealResult(latestMeta, storageDisks, storageEndpoints, errs,
bucket, object, versionID), err
@@ -1012,9 +1012,6 @@ func isObjectDangling(metaArr []FileInfo, errs []error, dataErrs []error) (valid
notFoundMetaErrs, corruptedMetaErrs, driveNotFoundMetaErrs := danglingErrsCount(errs)
notFoundPartsErrs, corruptedPartsErrs, driveNotFoundPartsErrs := danglingErrsCount(ndataErrs)
if driveNotFoundMetaErrs > 0 || driveNotFoundPartsErrs > 0 {
return validMeta, false
}
for _, m := range metaArr {
if m.IsValid() {
validMeta = m
@@ -1027,6 +1024,10 @@ func isObjectDangling(metaArr []FileInfo, errs []error, dataErrs []error) (valid
return validMeta, false
}
if driveNotFoundMetaErrs > 0 || driveNotFoundPartsErrs > 0 {
return validMeta, false
}
if validMeta.Deleted {
// notFoundPartsErrs is ignored since
// - delete marker does not have any parts
+1 -1
View File
@@ -32,7 +32,7 @@ import (
"github.com/dustin/go-humanize"
uuid2 "github.com/google/uuid"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config/storageclass"
)
+14 -5
View File
@@ -328,7 +328,7 @@ func (fi FileInfo) ObjectToPartOffset(ctx context.Context, offset int64) (partIn
return 0, 0, InvalidRange{}
}
func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.Time, quorum int) (FileInfo, error) {
func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.Time, etag string, quorum int) (FileInfo, error) {
// with less quorum return error.
if quorum < 1 {
return FileInfo{}, errErasureReadQuorum
@@ -336,12 +336,21 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.
metaHashes := make([]string, len(metaArr))
h := sha256.New()
for i, meta := range metaArr {
if meta.IsValid() && meta.ModTime.Equal(modTime) {
if !meta.IsValid() {
continue
}
etagOnly := modTime.Equal(timeSentinel) && (etag != "" && etag == meta.Metadata["etag"])
mtimeValid := meta.ModTime.Equal(modTime)
if mtimeValid || etagOnly {
fmt.Fprintf(h, "%v", meta.XLV1)
fmt.Fprintf(h, "%v", meta.GetDataDir())
if !etagOnly {
// Verify dataDir is same only when mtime is valid and etag is not considered.
fmt.Fprintf(h, "%v", meta.GetDataDir())
}
for _, part := range meta.Parts {
fmt.Fprintf(h, "part.%d", part.Number)
}
fmt.Fprintf(h, "%v+%v", meta.Erasure.DataBlocks, meta.Erasure.ParityBlocks)
fmt.Fprintf(h, "%v", meta.Erasure.Distribution)
// ILM transition fields
@@ -408,8 +417,8 @@ func pickValidDiskTimeWithQuorum(metaArr []FileInfo, quorum int) time.Time {
// pickValidFileInfo - picks one valid FileInfo content and returns from a
// slice of FileInfo.
func pickValidFileInfo(ctx context.Context, metaArr []FileInfo, modTime time.Time, quorum int) (FileInfo, error) {
return findFileInfoInQuorum(ctx, metaArr, modTime, quorum)
func pickValidFileInfo(ctx context.Context, metaArr []FileInfo, modTime time.Time, etag string, quorum int) (FileInfo, error) {
return findFileInfoInQuorum(ctx, metaArr, modTime, etag, quorum)
}
// writeUniqueFileInfo - writes unique `xl.meta` content for each disk concurrently.
+1 -1
View File
@@ -204,7 +204,7 @@ func TestFindFileInfoInQuorum(t *testing.T) {
for _, test := range tests {
test := test
t.Run("", func(t *testing.T) {
_, err := findFileInfoInQuorum(context.Background(), test.fis, test.modTime, test.expectedQuorum)
_, err := findFileInfoInQuorum(context.Background(), test.fis, test.modTime, "", test.expectedQuorum)
if err != test.expectedErr {
t.Errorf("Expected %s, got %s", test.expectedErr, err)
}
+3 -2
View File
@@ -83,8 +83,9 @@ func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object
if write {
quorum = writeQuorum
}
// List all online disks.
_, modTime := listOnlineDisks(storageDisks, partsMetadata, errs, quorum)
_, modTime, etag := listOnlineDisks(storageDisks, partsMetadata, errs, quorum)
if write {
reducedErr := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
@@ -98,7 +99,7 @@ func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object
}
// Pick one from the first valid metadata.
fi, err = pickValidFileInfo(ctx, partsMetadata, modTime, quorum)
fi, err = pickValidFileInfo(ctx, partsMetadata, modTime, etag, quorum)
return fi, partsMetadata, err
}
+42 -22
View File
@@ -32,7 +32,7 @@ import (
"time"
"github.com/klauspost/readahead"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/bucket/object/lock"
@@ -108,10 +108,10 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
}
// List all online disks.
onlineDisks, modTime := listOnlineDisks(storageDisks, metaArr, errs, readQuorum)
onlineDisks, modTime, etag := listOnlineDisks(storageDisks, metaArr, errs, readQuorum)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, etag, readQuorum)
if err != nil {
return oi, toObjectErr(err, srcBucket, srcObject)
}
@@ -347,11 +347,6 @@ func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, obje
var healOnce sync.Once
// once we have obtained a common FileInfo i.e latest, we should stick
// to single dataDir to read the content to avoid reading from some other
// dataDir that has stale FileInfo{} to ensure that we fail appropriately
// during reads and expect the same dataDir everywhere.
dataDir := fi.DataDir
for ; partIndex <= lastPartIndex; partIndex++ {
if length == totalBytesRead {
break
@@ -379,8 +374,11 @@ func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, obje
if !metaArr[index].IsValid() {
continue
}
if !metaArr[index].Erasure.Equal(fi.Erasure) {
continue
}
checksumInfo := metaArr[index].Erasure.GetChecksumInfo(partNumber)
partPath := pathJoin(object, dataDir, fmt.Sprintf("part.%d", partNumber))
partPath := pathJoin(object, metaArr[index].DataDir, fmt.Sprintf("part.%d", partNumber))
readers[index] = newBitrotReader(disk, metaArr[index].Data, bucket, partPath, tillOffset,
checksumInfo.Algorithm, checksumInfo.Hash, erasure.ShardSize())
@@ -664,10 +662,10 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
}
// List all online disks.
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs, readQuorum)
onlineDisks, modTime, etag := listOnlineDisks(disks, metaArr, errs, readQuorum)
// Pick latest valid metadata.
fi, err = pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
fi, err = pickValidFileInfo(ctx, metaArr, modTime, etag, readQuorum)
if err != nil {
return fi, nil, nil, err
}
@@ -693,9 +691,20 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
missingBlocks++
continue
}
if metaArr[i].IsValid() && metaArr[i].ModTime.Equal(fi.ModTime) {
continue
}
// verify metadata is valid, it has similar erasure info
// as well as common modtime, if modtime is not possible
// verify if it has common "etag" atleast.
if metaArr[i].IsValid() && metaArr[i].Erasure.Equal(fi.Erasure) {
ok := metaArr[i].ModTime.Equal(modTime)
if modTime.IsZero() || modTime.Equal(timeSentinel) {
ok = etag != "" && etag == fi.Metadata["etag"]
}
if ok {
continue
}
} // in all other cases metadata is corrupt, do not read from it.
metaArr[i] = FileInfo{}
onlineDisks[i] = nil
missingBlocks++
@@ -1013,13 +1022,24 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
data := r.Reader
if opts.CheckPrecondFn != nil {
obj, err := er.getObjectInfo(ctx, bucket, object, opts)
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
return objInfo, err
if !opts.NoLock {
ns := er.NewNSLock(bucket, object)
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
if err != nil {
return ObjectInfo{}, err
}
ctx = lkctx.Context()
defer ns.Unlock(lkctx)
opts.NoLock = true
}
if opts.CheckPrecondFn(obj) {
obj, err := er.getObjectInfo(ctx, bucket, object, opts)
if err == nil && opts.CheckPrecondFn(obj) {
return objInfo, PreConditionFailed{}
}
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) && !isErrReadQuorum(err) {
return objInfo, err
}
}
// Validate input data size and it can never be less than -1.
@@ -1794,10 +1814,10 @@ func (er erasureObjects) PutObjectMetadata(ctx context.Context, bucket, object s
}
// List all online disks.
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs, readQuorum)
onlineDisks, modTime, etag := listOnlineDisks(disks, metaArr, errs, readQuorum)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, etag, readQuorum)
if err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
@@ -1867,10 +1887,10 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin
}
// List all online disks.
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs, readQuorum)
onlineDisks, modTime, etag := listOnlineDisks(disks, metaArr, errs, readQuorum)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, etag, readQuorum)
if err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
+40 -19
View File
@@ -31,7 +31,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/console"
@@ -419,7 +419,9 @@ func (p poolMeta) save(ctx context.Context, pools []*erasureSets) error {
// Saves on all pools to make sure decommissioning of first pool is allowed.
for i, eset := range pools {
if err = saveConfig(ctx, eset, poolMetaName, buf); err != nil {
logger.LogIf(ctx, fmt.Errorf("saving pool.bin for pool index %d failed with: %v", i, err))
if !errors.Is(err, context.Canceled) {
logger.LogIf(ctx, fmt.Errorf("saving pool.bin for pool index %d failed with: %v", i, err))
}
return err
}
}
@@ -459,22 +461,34 @@ func (z *erasureServerPools) Init(ctx context.Context) error {
if !update {
z.poolMeta = meta
} else {
meta = poolMeta{} // to update write poolMeta fresh.
newMeta := poolMeta{} // to update write poolMeta fresh.
// looks like new pool was added we need to update,
// or this is a fresh installation (or an existing
// installation with pool removed)
meta.Version = poolMetaVersion
newMeta.Version = poolMetaVersion
for idx, pool := range z.serverPools {
meta.Pools = append(meta.Pools, PoolStatus{
var skip bool
for _, currentPool := range meta.Pools {
// Preserve any current pool status.
if currentPool.CmdLine == pool.endpoints.CmdLine {
newMeta.Pools = append(newMeta.Pools, currentPool)
skip = true
break
}
}
if skip {
continue
}
newMeta.Pools = append(newMeta.Pools, PoolStatus{
CmdLine: pool.endpoints.CmdLine,
ID: idx,
LastUpdate: UTCNow(),
})
}
if err = meta.save(ctx, z.serverPools); err != nil {
if err = newMeta.save(ctx, z.serverPools); err != nil {
return err
}
z.poolMeta = meta
z.poolMeta = newMeta
}
pools := meta.returnResumablePools()
@@ -720,18 +734,21 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
// to create the appropriate stack.
versionsSorter(fivs.Versions).reverse()
var decommissionedCount int
var decommissioned, expired int
for _, version := range fivs.Versions {
// Apply lifecycle rules on the objects that are expired.
if filterLifecycle(bi.Name, version.Name, version) {
decommissionedCount++
expired++
decommissioned++
continue
}
// any object with only single DEL marker we don't need
// to decommission, just skip it.
if version.Deleted && len(fivs.Versions) == 1 {
decommissionedCount++
// to decommission, just skip it, this also includes
// any other versions that have already expired.
remainingVersions := len(fivs.Versions) - expired
if version.Deleted && remainingVersions == 1 {
decommissioned++
continue
}
@@ -756,7 +773,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
SkipDecommissioned: true, // make sure we skip the decommissioned pool
})
var failure bool
if err != nil {
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
logger.LogIf(ctx, err)
failure = true
}
@@ -765,7 +782,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
z.poolMetaMutex.Unlock()
if !failure {
// Success keep a count.
decommissionedCount++
decommissioned++
}
continue
}
@@ -840,11 +857,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
if failure {
break // break out on first error
}
decommissionedCount++
decommissioned++
}
// if all versions were decommissioned, then we can delete the object versions.
if decommissionedCount == len(fivs.Versions) {
if decommissioned == len(fivs.Versions) {
stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name)
_, err := set.DeleteObject(ctx,
bi.Name,
@@ -1097,9 +1114,13 @@ func (z *erasureServerPools) Decommission(ctx context.Context, indices ...int) e
return err
}
for _, idx := range indices {
go z.doDecommissionInRoutine(ctx, idx)
}
go func() {
for _, idx := range indices {
// decommission all pools serially one after
// the other.
z.doDecommissionInRoutine(ctx, idx)
}
}()
// Successfully started decommissioning.
return nil
+22 -11
View File
@@ -32,7 +32,7 @@ import (
"time"
"github.com/lithammer/shortuuid/v4"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/env"
@@ -63,7 +63,10 @@ func (rs *rebalanceStats) update(bucket string, fi FileInfo) {
}
rs.NumVersions++
onDiskSz := fi.Size * int64(fi.Erasure.DataBlocks+fi.Erasure.ParityBlocks) / int64(fi.Erasure.DataBlocks)
onDiskSz := int64(0)
if !fi.Deleted {
onDiskSz = fi.Size * int64(fi.Erasure.DataBlocks+fi.Erasure.ParityBlocks) / int64(fi.Erasure.DataBlocks)
}
rs.Bytes += uint64(onDiskSz)
rs.Bucket = bucket
rs.Object = fi.Name
@@ -497,7 +500,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
// to create the appropriate stack.
versionsSorter(fivs.Versions).reverse()
var rebalanced int
var rebalanced, expired int
for _, version := range fivs.Versions {
// Skip transitioned objects for now. TBD
if version.IsRemote() {
@@ -507,12 +510,15 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
// Apply lifecycle rules on the objects that are expired.
if filterLifecycle(bucket, version.Name, version) {
rebalanced++
expired++
continue
}
// Empty delete markers we can assume that they can be purged
// as there is no more data associated with them.
if version.Deleted && len(fivs.Versions) == 1 {
// any object with only single DEL marker we don't need
// to rebalance, just skip it, this also includes
// any other versions that have already expired.
remainingVersions := len(fivs.Versions) - expired
if version.Deleted && remainingVersions == 1 {
rebalanced++
continue
}
@@ -547,9 +553,10 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
continue
}
var failure bool
var failure, ignore bool
for try := 0; try < 3; try++ {
// GetObjectReader.Close is called by rebalanceObject
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
gr, err := set.GetObjectNInfo(ctx,
bucket,
encodeDirObject(version.Name),
@@ -562,19 +569,21 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
})
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
// object deleted by the application, nothing to do here we move on.
return
ignore = true
stopFn(nil)
break
}
if err != nil {
failure = true
logger.LogIf(ctx, err)
stopFn(err)
continue
}
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name)
if err = z.rebalanceObject(ctx, bucket, gr); err != nil {
stopFn(err)
failure = true
logger.LogIf(ctx, err)
stopFn(err)
continue
}
@@ -582,7 +591,9 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
failure = false
break
}
if ignore {
continue
}
if failure {
break // break out on first error
}
+15 -3
View File
@@ -33,7 +33,7 @@ import (
"github.com/dustin/go-humanize"
"github.com/google/uuid"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/s3utils"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio-go/v7/pkg/tags"
@@ -141,6 +141,10 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
}
z.decommissionCancelers = make([]context.CancelFunc, len(z.serverPools))
// initialize the object layer.
setObjectLayer(z)
r := rand.New(rand.NewSource(time.Now().UnixNano()))
for {
err := z.Init(ctx) // Initializes all pools.
@@ -159,6 +163,7 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
globalLocalDrivesMu.Lock()
globalLocalDrives = localDrives
defer globalLocalDrivesMu.Unlock()
return z, nil
}
@@ -411,7 +416,9 @@ func (z *erasureServerPools) getPoolInfoExistingWithOpts(ctx context.Context, bu
}
// do not remove this check as it can lead to inconsistencies
// for all callers of bucket replication.
opts.VersionID = ""
if !opts.MetadataChg {
opts.VersionID = ""
}
pinfo.ObjInfo, pinfo.Err = pool.GetObjectInfo(ctx, bucket, object, opts)
poolObjInfos[i] = pinfo
}(i, pool, poolOpts[i])
@@ -443,7 +450,7 @@ func (z *erasureServerPools) getPoolInfoExistingWithOpts(ctx context.Context, bu
// found a pool
return pinfo, nil
}
if isErrReadQuorum(pinfo.Err) {
if isErrReadQuorum(pinfo.Err) && !opts.MetadataChg {
// read quorum is returned when the object is visibly
// present but its unreadable, we simply ask the writes to
// schedule to this pool instead. If there is no quorum
@@ -2268,6 +2275,7 @@ func (z *erasureServerPools) PutObjectMetadata(ctx context.Context, bucket, obje
return z.serverPools[0].PutObjectMetadata(ctx, bucket, object, opts)
}
opts.MetadataChg = true
// We don't know the size here set 1GiB atleast.
idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
if err != nil {
@@ -2284,6 +2292,8 @@ func (z *erasureServerPools) PutObjectTags(ctx context.Context, bucket, object s
return z.serverPools[0].PutObjectTags(ctx, bucket, object, tags, opts)
}
opts.MetadataChg = true
// We don't know the size here set 1GiB atleast.
idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
if err != nil {
@@ -2300,6 +2310,8 @@ func (z *erasureServerPools) DeleteObjectTags(ctx context.Context, bucket, objec
return z.serverPools[0].DeleteObjectTags(ctx, bucket, object, opts)
}
opts.MetadataChg = true
idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
if err != nil {
return ObjectInfo{}, err
+1 -1
View File
@@ -33,7 +33,7 @@ import (
"github.com/dchest/siphash"
"github.com/dustin/go-humanize"
"github.com/google/uuid"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/bpool"
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bpool"
"github.com/minio/minio/internal/dsync"
"github.com/minio/minio/internal/logger"
+14 -30
View File
@@ -35,20 +35,18 @@ import (
// EventNotifier - notifies external systems about events in MinIO.
type EventNotifier struct {
sync.RWMutex
targetList *event.TargetList
targetResCh chan event.TargetIDResult
bucketRulesMap map[string]event.RulesMap
bucketRemoteTargetRulesMap map[string]map[event.TargetID]event.RulesMap
targetList *event.TargetList
targetResCh chan event.TargetIDResult
bucketRulesMap map[string]event.RulesMap
}
// NewEventNotifier - creates new event notification object.
func NewEventNotifier() *EventNotifier {
// targetList/bucketRulesMap/bucketRemoteTargetRulesMap are populated by NotificationSys.InitBucketTargets()
return &EventNotifier{
targetList: event.NewTargetList(),
targetResCh: make(chan event.TargetIDResult),
bucketRulesMap: make(map[string]event.RulesMap),
bucketRemoteTargetRulesMap: make(map[string]map[event.TargetID]event.RulesMap),
targetList: event.NewTargetList(),
targetResCh: make(chan event.TargetIDResult),
bucketRulesMap: make(map[string]event.RulesMap),
}
}
@@ -122,10 +120,6 @@ func (evnot *EventNotifier) AddRulesMap(bucketName string, rulesMap event.RulesM
rulesMap = rulesMap.Clone()
for _, targetRulesMap := range evnot.bucketRemoteTargetRulesMap[bucketName] {
rulesMap.Add(targetRulesMap)
}
// Do not add for an empty rulesMap.
if len(rulesMap) == 0 {
delete(evnot.bucketRulesMap, bucketName)
@@ -174,29 +168,18 @@ func (evnot *EventNotifier) RemoveNotification(bucketName string) {
defer evnot.Unlock()
delete(evnot.bucketRulesMap, bucketName)
targetIDSet := event.NewTargetIDSet()
for targetID := range evnot.bucketRemoteTargetRulesMap[bucketName] {
targetIDSet[targetID] = struct{}{}
delete(evnot.bucketRemoteTargetRulesMap[bucketName], targetID)
}
evnot.targetList.Remove(targetIDSet)
delete(evnot.bucketRemoteTargetRulesMap, bucketName)
}
// RemoveAllRemoteTargets - closes and removes all notification targets.
func (evnot *EventNotifier) RemoveAllRemoteTargets() {
// RemoveAllBucketTargets - closes and removes all notification targets.
func (evnot *EventNotifier) RemoveAllBucketTargets() {
evnot.Lock()
defer evnot.Unlock()
for _, targetMap := range evnot.bucketRemoteTargetRulesMap {
targetIDSet := event.NewTargetIDSet()
for k := range targetMap {
targetIDSet[k] = struct{}{}
}
evnot.targetList.Remove(targetIDSet)
targetIDSet := event.NewTargetIDSet()
for k := range evnot.targetList.TargetMap() {
targetIDSet[k] = struct{}{}
}
evnot.targetList.Remove(targetIDSet)
}
// Send - sends the event to all registered notification targets
@@ -209,7 +192,8 @@ func (evnot *EventNotifier) Send(args eventArgs) {
return
}
evnot.targetList.Send(args.ToEvent(true), targetIDSet, evnot.targetResCh)
// If MINIO_API_SYNC_EVENTS is set, send events synchronously.
evnot.targetList.Send(args.ToEvent(true), targetIDSet, evnot.targetResCh, globalAPIConfig.isSyncEventsEnabled())
}
type eventArgs struct {
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"strings"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio/internal/auth"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/config/storageclass"
"github.com/minio/minio/internal/logger"
+6 -4
View File
@@ -27,7 +27,7 @@ import (
"time"
"github.com/minio/console/restapi"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/bucket/bandwidth"
@@ -224,7 +224,10 @@ var (
// registered listeners
globalConsoleSys *HTTPConsoleLoggerSys
// All unique drives for this deployment
globalEndpoints EndpointServerPools
// All unique nodes for this deployment
globalNodes []Node
// The name of this local node, fetched from arguments
globalLocalNodeName string
@@ -237,16 +240,15 @@ var (
// The global callhome config
globalCallhomeConfig callhome.Config
globalRemoteEndpoints map[string]Endpoint
// Global server's network statistics
globalConnStats = newConnStats()
// Global HTTP request statisitics
globalHTTPStats = newHTTPStats()
// Global bucket network statistics
// Global bucket network and API statistics
globalBucketConnStats = newBucketConnStats()
globalBucketHTTPStats = newBucketHTTPStats()
// Time when the server is started
globalBootTime = UTCNow()
+10 -1
View File
@@ -52,6 +52,7 @@ type apiConfig struct {
disableODirect bool
gzipObjects bool
rootAccess bool
syncEvents bool
}
const cgroupLimitFile = "/sys/fs/cgroup/memory/memory.limit_in_bytes"
@@ -166,6 +167,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
t.disableODirect = cfg.DisableODirect
t.gzipObjects = cfg.GzipObjects
t.rootAccess = cfg.RootAccess
t.syncEvents = cfg.SyncEvents
}
func (t *apiConfig) isDisableODirect() bool {
@@ -317,7 +319,7 @@ func maxClients(f http.HandlerFunc) http.HandlerFunc {
case <-deadlineTimer.C:
// Send a http timeout message
writeErrorResponse(r.Context(), w,
errorCodes.ToAPIErr(ErrOperationMaxedOut),
errorCodes.ToAPIErr(ErrTooManyRequests),
r.URL)
globalHTTPStats.addRequestsInQueue(-1)
return
@@ -353,3 +355,10 @@ func (t *apiConfig) getTransitionWorkers() int {
return t.transitionWorkers
}
func (t *apiConfig) isSyncEventsEnabled() bool {
t.mu.RLock()
defer t.mu.RUnlock()
return t.syncEvents
}
+19 -8
View File
@@ -26,7 +26,7 @@ import (
"regexp"
"strings"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/handlers"
xhttp "github.com/minio/minio/internal/http"
@@ -269,9 +269,26 @@ func trimAwsChunkedContentEncoding(contentEnc string) (trimmedContentEnc string)
func collectAPIStats(api string, f http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
resource, err := getResource(r.URL.Path, r.Host, globalDomainNames)
if err != nil {
defer logger.AuditLog(r.Context(), w, r, mustGetClaimsFromToken(r))
apiErr := errorCodes.ToAPIErr(ErrUnsupportedHostHeader)
apiErr.Description = fmt.Sprintf("%s: %v", apiErr.Description, err)
writeErrorResponse(r.Context(), w, apiErr, r.URL)
return
}
bucket, _ := path2BucketObject(resource)
globalHTTPStats.currentS3Requests.Inc(api)
defer globalHTTPStats.currentS3Requests.Dec(api)
if bucket != "" && bucket != minioReservedBucket {
globalBucketHTTPStats.updateHTTPStats(bucket, api, nil)
}
f.ServeHTTP(w, r)
tc, ok := r.Context().Value(mcontext.ContextTraceKey).(*mcontext.TraceCtxt)
@@ -299,16 +316,10 @@ func collectAPIStats(api string, f http.HandlerFunc) http.HandlerFunc {
globalConnStats.incS3InputBytes(int64(tc.RequestRecorder.Size()))
globalConnStats.incS3OutputBytes(int64(tc.ResponseRecorder.Size()))
resource, err := getResource(r.URL.Path, r.Host, globalDomainNames)
if err != nil {
logger.LogIf(r.Context(), fmt.Errorf("Unable to get the actual resource in the incoming request: %v", err))
return
}
bucket, _ := path2BucketObject(resource)
if bucket != "" && bucket != minioReservedBucket {
globalBucketConnStats.incS3InputBytes(bucket, int64(tc.RequestRecorder.Size()))
globalBucketConnStats.incS3OutputBytes(bucket, int64(tc.ResponseRecorder.Size()))
globalBucketHTTPStats.updateHTTPStats(bucket, api, tc.ResponseRecorder)
}
}
}
+123 -2
View File
@@ -120,6 +120,105 @@ type bucketS3RXTX struct {
s3OutputBytes uint64
}
type bucketHTTPAPIStats struct {
currentS3Requests *HTTPAPIStats
totalS3Requests *HTTPAPIStats
totalS34xxErrors *HTTPAPIStats
totalS35xxErrors *HTTPAPIStats
totalS3Canceled *HTTPAPIStats
}
type bucketHTTPStats struct {
sync.RWMutex
httpStats map[string]bucketHTTPAPIStats
}
func newBucketHTTPStats() *bucketHTTPStats {
return &bucketHTTPStats{
httpStats: make(map[string]bucketHTTPAPIStats),
}
}
func (bh *bucketHTTPStats) delete(bucket string) {
bh.Lock()
defer bh.Unlock()
delete(bh.httpStats, bucket)
}
func (bh *bucketHTTPStats) updateHTTPStats(bucket, api string, w *xhttp.ResponseRecorder) {
if bh == nil {
return
}
bh.Lock()
defer bh.Unlock()
hstats, ok := bh.httpStats[bucket]
if !ok {
hstats = bucketHTTPAPIStats{
currentS3Requests: &HTTPAPIStats{},
totalS3Requests: &HTTPAPIStats{},
totalS3Canceled: &HTTPAPIStats{},
totalS34xxErrors: &HTTPAPIStats{},
totalS35xxErrors: &HTTPAPIStats{},
}
}
if w == nil { // when response recorder nil, this is an active request
hstats.currentS3Requests.Inc(api)
bh.httpStats[bucket] = hstats
return
} // else {
hstats.currentS3Requests.Dec(api) // decrement this once we have the response recorder.
hstats.totalS3Requests.Inc(api)
code := w.StatusCode
switch {
case code == 0:
case code == 499:
// 499 is a good error, shall be counted as canceled.
hstats.totalS3Canceled.Inc(api)
case code >= http.StatusBadRequest:
if code >= http.StatusInternalServerError {
hstats.totalS35xxErrors.Inc(api)
} else {
hstats.totalS34xxErrors.Inc(api)
}
}
bh.httpStats[bucket] = hstats
}
func (bh *bucketHTTPStats) load(bucket string) bucketHTTPAPIStats {
if bh == nil {
return bucketHTTPAPIStats{
currentS3Requests: &HTTPAPIStats{},
totalS3Requests: &HTTPAPIStats{},
totalS3Canceled: &HTTPAPIStats{},
totalS34xxErrors: &HTTPAPIStats{},
totalS35xxErrors: &HTTPAPIStats{},
}
}
bh.RLock()
defer bh.RUnlock()
val, ok := bh.httpStats[bucket]
if ok {
return val
}
return bucketHTTPAPIStats{
currentS3Requests: &HTTPAPIStats{},
totalS3Requests: &HTTPAPIStats{},
totalS3Canceled: &HTTPAPIStats{},
totalS34xxErrors: &HTTPAPIStats{},
totalS35xxErrors: &HTTPAPIStats{},
}
}
type bucketConnStats struct {
sync.RWMutex
stats map[string]*bucketS3RXTX
@@ -225,10 +324,32 @@ func (stats *HTTPAPIStats) Dec(api string) {
}
}
// Get returns the current counter on input API string
func (stats *HTTPAPIStats) Get(api string) int {
if stats == nil {
return 0
}
stats.RLock()
defer stats.RUnlock()
val, ok := stats.apiStats[api]
if ok {
return val
}
return 0
}
// Load returns the recorded stats.
func (stats *HTTPAPIStats) Load() map[string]int {
stats.Lock()
defer stats.Unlock()
if stats == nil {
return map[string]int{}
}
stats.RLock()
defer stats.RUnlock()
apiStats := make(map[string]int, len(stats.apiStats))
for k, v := range stats.apiStats {
apiStats[k] = v
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"strings"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/handlers"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/mcontext"
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"unicode/utf8"
jsoniter "github.com/json-iterator/go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
+22 -32
View File
@@ -28,7 +28,7 @@ import (
"time"
jsoniter "github.com/json-iterator/go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/config/identity/openid"
@@ -914,11 +914,12 @@ func (store *IAMStoreSys) listGroups(ctx context.Context) (res []string, err err
// PolicyDBUpdate - adds or removes given policies to/from the user or group's
// policy associations.
func (store *IAMStoreSys) PolicyDBUpdate(ctx context.Context, name string, isGroup bool,
userType IAMUserType, policies []string, isAttach bool) (updatedAt time.Time, addedOrRemoved []string,
err error,
userType IAMUserType, policies []string, isAttach bool) (updatedAt time.Time,
addedOrRemoved, effectivePolicies []string, err error,
) {
if name == "" {
return updatedAt, nil, errInvalidArgument
err = errInvalidArgument
return
}
cache := store.lock()
@@ -932,11 +933,13 @@ func (store *IAMStoreSys) PolicyDBUpdate(ctx context.Context, name string, isGro
if store.getUsersSysType() == MinIOUsersSysType {
g, ok := cache.iamGroupsMap[name]
if !ok {
return updatedAt, nil, errNoSuchGroup
err = errNoSuchGroup
return
}
if g.Status == statusDisabled {
return updatedAt, nil, errGroupDisabled
err = errGroupDisabled
return
}
}
mp = cache.iamGroupPolicyMap[name]
@@ -945,6 +948,7 @@ func (store *IAMStoreSys) PolicyDBUpdate(ctx context.Context, name string, isGro
// Compute net policy change effect and updated policy mapping
existingPolicySet := mp.policySet()
policiesToUpdate := set.CreateStringSet(policies...)
var newPolicySet set.StringSet
newPolicyMapping := mp
if isAttach {
// new policies to attach => inputPolicies - existing (set difference)
@@ -952,26 +956,29 @@ func (store *IAMStoreSys) PolicyDBUpdate(ctx context.Context, name string, isGro
// validate that new policies to add are defined.
for _, p := range policiesToUpdate.ToSlice() {
if _, found := cache.iamPolicyDocsMap[p]; !found {
return updatedAt, nil, errNoSuchPolicy
err = errNoSuchPolicy
return
}
}
newPolicyMapping.Policies = strings.Join(existingPolicySet.Union(policiesToUpdate).ToSlice(), ",")
newPolicySet = existingPolicySet.Union(policiesToUpdate)
} else {
// policies to detach => inputPolicies ∩ existing (intersection)
policiesToUpdate = policiesToUpdate.Intersection(existingPolicySet)
newPolicyMapping.Policies = strings.Join(existingPolicySet.Difference(policiesToUpdate).ToSlice(), ",")
newPolicySet = existingPolicySet.Difference(policiesToUpdate)
}
newPolicyMapping.UpdatedAt = UTCNow()
// We return an error if the requested policy update will have no effect.
if policiesToUpdate.IsEmpty() {
return updatedAt, nil, errNoPolicyToAttachOrDetach
err = errNoPolicyToAttachOrDetach
return
}
newPolicies := newPolicySet.ToSlice()
newPolicyMapping.Policies = strings.Join(newPolicies, ",")
newPolicyMapping.UpdatedAt = UTCNow()
addedOrRemoved = policiesToUpdate.ToSlice()
if err := store.saveMappedPolicy(ctx, name, userType, isGroup, newPolicyMapping); err != nil {
return updatedAt, addedOrRemoved, err
if err = store.saveMappedPolicy(ctx, name, userType, isGroup, newPolicyMapping); err != nil {
return
}
if !isGroup {
cache.iamUserPolicyMap[name] = newPolicyMapping
@@ -979,7 +986,7 @@ func (store *IAMStoreSys) PolicyDBUpdate(ctx context.Context, name string, isGro
cache.iamGroupPolicyMap[name] = newPolicyMapping
}
cache.updatedAt = UTCNow()
return cache.updatedAt, addedOrRemoved, nil
return cache.updatedAt, addedOrRemoved, newPolicies, nil
}
// PolicyDBSet - update the policy mapping for the given user or group in
@@ -1451,23 +1458,6 @@ func (store *IAMStoreSys) GetUserInfo(name string) (u madmin.UserInfo, err error
}, nil
}
// GetUserPolicies - returns the policies attached to a user.
func (store *IAMStoreSys) GetUserPolicies(name string) ([]string, error) {
if name == "" {
return nil, errInvalidArgument
}
cache := store.rlock()
defer store.runlock()
if cache.iamUserPolicyMap[name].Policies == "" {
return []string{}, nil
}
policies := cache.iamUserPolicyMap[name].toSlice()
return policies, nil
}
// PolicyMappingNotificationHandler - handles updating a policy mapping from storage.
func (store *IAMStoreSys) PolicyMappingNotificationHandler(ctx context.Context, userOrGroup string, isGroup bool, userType IAMUserType) error {
if userOrGroup == "" {
+106 -21
View File
@@ -32,7 +32,7 @@ import (
"sync/atomic"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/arn"
"github.com/minio/minio/internal/auth"
@@ -813,7 +813,7 @@ func (sys *IAMSys) QueryLDAPPolicyEntities(ctx context.Context, q madmin.PolicyE
}
}
// IsTempUser - returns if given key is a temporary user.
// IsTempUser - returns if given key is a temporary user and parent user.
func (sys *IAMSys) IsTempUser(name string) (bool, string, error) {
if !sys.Initialized() {
return false, "", errServerNotInitialized
@@ -889,15 +889,6 @@ func (sys *IAMSys) QueryPolicyEntities(ctx context.Context, q madmin.PolicyEntit
}
}
// GetUserPolicies - get policies attached to a user.
func (sys *IAMSys) GetUserPolicies(name string) (p []string, err error) {
if !sys.Initialized() {
return p, errServerNotInitialized
}
return sys.store.GetUserPolicies(name)
}
// SetUserStatus - sets current user status, supports disabled or enabled.
func (sys *IAMSys) SetUserStatus(ctx context.Context, accessKey string, status madmin.AccountStatus) (updatedAt time.Time, err error) {
if !sys.Initialized() {
@@ -1582,13 +1573,93 @@ func (sys *IAMSys) PolicyDBSet(ctx context.Context, name, policy string, userTyp
return updatedAt, nil
}
// PolicyDBUpdateBuiltin - adds or removes policies from a user or a group
// verified to be an internal IDP user.
func (sys *IAMSys) PolicyDBUpdateBuiltin(ctx context.Context, isAttach bool,
r madmin.PolicyAssociationReq,
) (updatedAt time.Time, addedOrRemoved, effectivePolicies []string, err error) {
if !sys.Initialized() {
err = errServerNotInitialized
return
}
userOrGroup := r.User
var isGroup bool
if userOrGroup == "" {
isGroup = true
userOrGroup = r.Group
}
if isGroup {
_, err = sys.GetGroupDescription(userOrGroup)
if err != nil {
return
}
} else {
var isTemp bool
isTemp, _, err = sys.IsTempUser(userOrGroup)
if err != nil && err != errNoSuchUser {
return
}
if isTemp {
err = errIAMActionNotAllowed
return
}
// When the user is root credential you are not allowed to
// add policies for root user.
if userOrGroup == globalActiveCred.AccessKey {
err = errIAMActionNotAllowed
return
}
// Validate that user exists.
var userExists bool
_, userExists = sys.GetUser(ctx, userOrGroup)
if !userExists {
err = errNoSuchUser
return
}
}
updatedAt, addedOrRemoved, effectivePolicies, err = sys.store.PolicyDBUpdate(ctx, userOrGroup, isGroup,
regUser, r.Policies, isAttach)
if err != nil {
return
}
// Notify all other MinIO peers to reload policy
if !sys.HasWatcher() {
for _, nerr := range globalNotificationSys.LoadPolicyMapping(userOrGroup, regUser, isGroup) {
if nerr.Err != nil {
logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
logger.LogIf(ctx, nerr.Err)
}
}
}
logger.LogIf(ctx, globalSiteReplicationSys.IAMChangeHook(ctx, madmin.SRIAMItem{
Type: madmin.SRIAMItemPolicyMapping,
PolicyMapping: &madmin.SRPolicyMapping{
UserOrGroup: userOrGroup,
UserType: int(regUser),
IsGroup: isGroup,
Policy: strings.Join(effectivePolicies, ","),
},
UpdatedAt: updatedAt,
}))
return
}
// PolicyDBUpdateLDAP - adds or removes policies from a user or a group verified
// to be in the LDAP directory.
func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
r madmin.PolicyAssociationReq,
) (updatedAt time.Time, addedOrRemoved []string, err error) {
) (updatedAt time.Time, addedOrRemoved, effectivePolicies []string, err error) {
if !sys.Initialized() {
return updatedAt, nil, errServerNotInitialized
err = errServerNotInitialized
return
}
var dn string
@@ -1597,28 +1668,31 @@ func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
dn, err = sys.LDAPConfig.DoesUsernameExist(r.User)
if err != nil {
logger.LogIf(ctx, err)
return updatedAt, nil, err
return
}
if dn == "" {
return updatedAt, nil, errNoSuchUser
err = errNoSuchUser
return
}
isGroup = false
} else {
if exists, err := sys.LDAPConfig.DoesGroupDNExist(r.Group); err != nil {
var exists bool
if exists, err = sys.LDAPConfig.DoesGroupDNExist(r.Group); err != nil {
logger.LogIf(ctx, err)
return updatedAt, nil, err
return
} else if !exists {
return updatedAt, nil, errNoSuchGroup
err = errNoSuchGroup
return
}
dn = r.Group
isGroup = true
}
userType := stsUser
updatedAt, addedOrRemoved, err = sys.store.PolicyDBUpdate(ctx, dn, isGroup,
updatedAt, addedOrRemoved, effectivePolicies, err = sys.store.PolicyDBUpdate(ctx, dn, isGroup,
userType, r.Policies, isAttach)
if err != nil {
return updatedAt, nil, err
return
}
// Notify all other MinIO peers to reload policy
@@ -1631,7 +1705,18 @@ func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
}
}
return updatedAt, addedOrRemoved, nil
logger.LogIf(ctx, globalSiteReplicationSys.IAMChangeHook(ctx, madmin.SRIAMItem{
Type: madmin.SRIAMItemPolicyMapping,
PolicyMapping: &madmin.SRPolicyMapping{
UserOrGroup: dn,
UserType: int(userType),
IsGroup: isGroup,
Policy: strings.Join(effectivePolicies, ","),
},
UpdatedAt: updatedAt,
}))
return
}
// PolicyDBGet - gets policy set on a user or group. If a list of groups is
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"time"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
iampolicy "github.com/minio/pkg/iam/policy"
+1 -1
View File
@@ -22,7 +22,7 @@ package cmd
import (
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
const (
+1 -1
View File
@@ -129,7 +129,7 @@ func (api objectAPIHandlers) ListenNotificationHandler(w http.ResponseWriter, r
return rulesMap.MatchSimple(ev.EventName, ev.S3.Object.Key)
})
if err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if bucketName != "" {
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"context"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/disk"
)
+106 -15
View File
@@ -28,7 +28,7 @@ import (
"time"
"github.com/minio/kes-go"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/logger"
"github.com/minio/minio/internal/mcontext"
@@ -483,7 +483,7 @@ func getUsageLastScanActivityMD() MetricDescription {
Namespace: minioMetricNamespace,
Subsystem: usageSubsystem,
Name: lastActivityTime,
Help: "Time elapsed (in nano seconds) since last scan activity. This is set to 0 until first scan cycle",
Help: "Time elapsed (in nano seconds) since last scan activity.",
Type: gaugeMetric,
}
}
@@ -703,7 +703,7 @@ func getS3RequestsInQueueMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: waitingTotal,
Help: "Number of S3 requests in the waiting queue",
Help: "Total number of S3 requests in the waiting queue",
Type: gaugeMetric,
}
}
@@ -713,7 +713,7 @@ func getIncomingS3RequestsMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: incomingTotal,
Help: "Volatile number of total incoming S3 requests",
Help: "Total number of incoming S3 requests",
Type: gaugeMetric,
}
}
@@ -723,7 +723,7 @@ func getS3RequestsTotalMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: total,
Help: "Total number S3 requests",
Help: "Total number of S3 requests",
Type: counterMetric,
}
}
@@ -733,7 +733,7 @@ func getS3RequestsErrorsMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: errorsTotal,
Help: "Total number S3 requests with (4xx and 5xx) errors",
Help: "Total number of S3 requests with (4xx and 5xx) errors",
Type: counterMetric,
}
}
@@ -743,7 +743,7 @@ func getS3Requests4xxErrorsMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: "4xx_" + errorsTotal,
Help: "Total number S3 requests with (4xx) errors",
Help: "Total number of S3 requests with (4xx) errors",
Type: counterMetric,
}
}
@@ -753,7 +753,7 @@ func getS3Requests5xxErrorsMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: "5xx_" + errorsTotal,
Help: "Total number S3 requests with (5xx) errors",
Help: "Total number of S3 requests with (5xx) errors",
Type: counterMetric,
}
}
@@ -763,7 +763,7 @@ func getS3RequestsCanceledMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsSubsystem,
Name: canceledTotal,
Help: "Total number S3 requests that were canceled from the client while processing",
Help: "Total number of S3 requests that were canceled by the client",
Type: counterMetric,
}
}
@@ -773,7 +773,7 @@ func getS3RejectedAuthRequestsTotalMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsRejectedSubsystem,
Name: authTotal,
Help: "Total number S3 requests rejected for auth failure",
Help: "Total number of S3 requests rejected for auth failure",
Type: counterMetric,
}
}
@@ -783,7 +783,7 @@ func getS3RejectedHeaderRequestsTotalMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsRejectedSubsystem,
Name: headerTotal,
Help: "Total number S3 requests rejected for invalid header",
Help: "Total number of S3 requests rejected for invalid header",
Type: counterMetric,
}
}
@@ -793,7 +793,7 @@ func getS3RejectedTimestampRequestsTotalMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsRejectedSubsystem,
Name: timestampTotal,
Help: "Total number S3 requests rejected for invalid timestamp",
Help: "Total number of S3 requests rejected for invalid timestamp",
Type: counterMetric,
}
}
@@ -803,7 +803,7 @@ func getS3RejectedInvalidRequestsTotalMD() MetricDescription {
Namespace: s3MetricNamespace,
Subsystem: requestsRejectedSubsystem,
Name: invalidTotal,
Help: "Total number S3 invalid requests",
Help: "Total number of invalid S3 requests",
Type: counterMetric,
}
}
@@ -913,7 +913,7 @@ func getHealLastActivityTimeMD() MetricDescription {
Namespace: healMetricNamespace,
Subsystem: timeSubsystem,
Name: lastActivityTime,
Help: "Time elapsed (in nano seconds) since last self healing activity. This is set to -1 until initial self heal activity",
Help: "Time elapsed (in nano seconds) since last self healing activity.",
Type: gaugeMetric,
}
}
@@ -1309,6 +1309,56 @@ func getExpiryPendingTasksMD() MetricDescription {
}
}
func getBucketS3RequestsInFlightMD() MetricDescription {
return MetricDescription{
Namespace: bucketMetricNamespace,
Subsystem: requestsSubsystem,
Name: inflightTotal,
Help: "Total number of S3 requests currently in flight on a bucket",
Type: gaugeMetric,
}
}
func getBucketS3RequestsTotalMD() MetricDescription {
return MetricDescription{
Namespace: bucketMetricNamespace,
Subsystem: requestsSubsystem,
Name: total,
Help: "Total number of S3 requests on a bucket",
Type: counterMetric,
}
}
func getBucketS3Requests4xxErrorsMD() MetricDescription {
return MetricDescription{
Namespace: bucketMetricNamespace,
Subsystem: requestsSubsystem,
Name: "4xx_" + errorsTotal,
Help: "Total number of S3 requests with (4xx) errors on a bucket",
Type: counterMetric,
}
}
func getBucketS3Requests5xxErrorsMD() MetricDescription {
return MetricDescription{
Namespace: bucketMetricNamespace,
Subsystem: requestsSubsystem,
Name: "5xx_" + errorsTotal,
Help: "Total number of S3 requests with (5xx) errors on a bucket",
Type: counterMetric,
}
}
func getBucketS3RequestsCanceledMD() MetricDescription {
return MetricDescription{
Namespace: bucketMetricNamespace,
Subsystem: requestsSubsystem,
Name: canceledTotal,
Help: "Total number of S3 requests that were canceled from the client while processing on a bucket",
Type: counterMetric,
}
}
func getILMNodeMetrics() *MetricsGroup {
mg := &MetricsGroup{
cacheInterval: 10 * time.Second,
@@ -1456,7 +1506,7 @@ func getIAMNodeMetrics() *MetricsGroup {
Namespace: nodeMetricNamespace,
Subsystem: iamSubsystem,
Name: "since_last_sync_millis",
Help: "Time (in milliseconds) since last successful IAM data sync. This is set to 0 until the first sync after server start.",
Help: "Time (in milliseconds) since last successful IAM data sync.",
Type: gaugeMetric,
},
Value: float64(sinceLastSyncMillis),
@@ -2072,6 +2122,47 @@ func getBucketUsageMetrics() *MetricsGroup {
})
}
httpStats := globalBucketHTTPStats.load(bucket)
for k, v := range httpStats.currentS3Requests.Load() {
metrics = append(metrics, Metric{
Description: getBucketS3RequestsInFlightMD(),
Value: float64(v),
VariableLabels: map[string]string{"bucket": bucket, "api": k},
})
}
for k, v := range httpStats.totalS3Requests.Load() {
metrics = append(metrics, Metric{
Description: getBucketS3RequestsTotalMD(),
Value: float64(v),
VariableLabels: map[string]string{"bucket": bucket, "api": k},
})
}
for k, v := range httpStats.totalS3Canceled.Load() {
metrics = append(metrics, Metric{
Description: getBucketS3RequestsCanceledMD(),
Value: float64(v),
VariableLabels: map[string]string{"bucket": bucket, "api": k},
})
}
for k, v := range httpStats.totalS34xxErrors.Load() {
metrics = append(metrics, Metric{
Description: getBucketS3Requests4xxErrorsMD(),
Value: float64(v),
VariableLabels: map[string]string{"bucket": bucket, "api": k},
})
}
for k, v := range httpStats.totalS35xxErrors.Load() {
metrics = append(metrics, Metric{
Description: getBucketS3Requests5xxErrorsMD(),
Value: float64(v),
VariableLabels: map[string]string{"bucket": bucket, "api": k},
})
}
if stats.hasReplicationUsage() {
for arn, stat := range stats.Stats {
metrics = append(metrics, Metric{
+1 -1
View File
@@ -22,7 +22,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
const (
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
// naughtyDisk wraps a POSIX disk and returns programmed errors
+1 -1
View File
@@ -18,7 +18,7 @@
package cmd
import (
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
// GetTotalCapacity gets the total capacity in the cluster.
+2 -2
View File
@@ -29,7 +29,7 @@ import (
"github.com/cespare/xxhash/v2"
"github.com/klauspost/compress/zip"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/bandwidth"
"github.com/minio/minio/internal/logger"
xnet "github.com/minio/pkg/net"
@@ -453,6 +453,7 @@ func (sys *NotificationSys) DeleteBucketMetadata(ctx context.Context, bucketName
globalBucketTargetSys.Delete(bucketName)
globalEventNotifier.RemoveNotification(bucketName)
globalBucketConnStats.delete(bucketName)
globalBucketHTTPStats.delete(bucketName)
if localMetacacheMgr != nil {
localMetacacheMgr.deleteBucketCache(bucketName)
}
@@ -1032,7 +1033,6 @@ func (sys *NotificationSys) GetPeerOnlineCount() (nodesOnline, nodesOffline int)
// NewNotificationSys - creates new notification system object.
func NewNotificationSys(endpoints EndpointServerPools) *NotificationSys {
// targetList/bucketRulesMap/bucketRemoteTargetRulesMap are populated by NotificationSys.Init()
remote, all := newPeerRestClients(endpoints)
return &NotificationSys{
peerClients: remote,
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/replication"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/logger"
+3 -1
View File
@@ -23,7 +23,7 @@ import (
"net/http"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7/pkg/encrypt"
"github.com/minio/minio-go/v7/pkg/tags"
"github.com/minio/minio/internal/hash"
@@ -99,6 +99,8 @@ type ObjectOptions struct {
IndexCB func() []byte
InclFreeVersions bool
MetadataChg bool // is true if it is a metadata update operation.
}
// ExpirationOptions represents object options for object expiration at objectLayer.
+2 -8
View File
@@ -662,10 +662,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
}
}
}
errCode := errorCodes.ToAPIErr(s3Error)
w.Header().Set(xMinIOErrCodeHeader, errCode.Code)
w.Header().Set(xMinIOErrDescHeader, "\""+errCode.Description+"\"")
writeErrorResponseHeadersOnly(w, errCode)
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(s3Error))
return
}
@@ -700,10 +697,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
}
if s3Error := authorizeRequest(ctx, r, policy.GetObjectAction); s3Error != ErrNone {
errCode := errorCodes.ToAPIErr(s3Error)
w.Header().Set(xMinIOErrCodeHeader, errCode.Code)
w.Header().Set(xMinIOErrDescHeader, "\""+errCode.Description+"\"")
writeErrorResponseHeadersOnly(w, errCode)
writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(s3Error))
return
}
+1 -1
View File
@@ -47,7 +47,7 @@ func getLambdaEventData(bucket, object string, cred auth.Credentials, r *http.Re
}
duration := time.Until(cred.Expiration)
if cred.Expiration.IsZero() || duration > time.Hour {
if duration > time.Hour || duration < time.Hour {
// Always limit to 1 hour.
duration = time.Hour
}
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"sync/atomic"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/disk"
ioutilx "github.com/minio/minio/internal/ioutil"
)
+1 -1
View File
@@ -29,7 +29,7 @@ import (
"strings"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/bucket/bandwidth"
"github.com/minio/minio/internal/event"
xhttp "github.com/minio/minio/internal/http"
+2 -1
View File
@@ -30,7 +30,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
b "github.com/minio/minio/internal/bucket/bandwidth"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/logger"
@@ -533,6 +533,7 @@ func (s *peerRESTServer) DeleteBucketMetadataHandler(w http.ResponseWriter, r *h
globalBucketTargetSys.Delete(bucketName)
globalEventNotifier.RemoveNotification(bucketName)
globalBucketConnStats.delete(bucketName)
globalBucketHTTPStats.delete(bucketName)
if localMetacacheMgr != nil {
localMetacacheMgr.deleteBucketCache(bucketName)
}
+106 -85
View File
@@ -21,37 +21,81 @@ import (
"context"
"encoding/gob"
"errors"
"fmt"
"io"
"net/url"
"strconv"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/logger"
"github.com/minio/minio/internal/rest"
xnet "github.com/minio/pkg/net"
"github.com/minio/pkg/sync/errgroup"
"golang.org/x/exp/slices"
)
var errPeerOffline = errors.New("peer is offline")
type peerS3Client interface {
ListBuckets(ctx context.Context, opts BucketOptions) ([]BucketInfo, error)
GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error)
MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error
DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error
GetHost() string
SetPools([]int)
GetPools() []int
}
type localPeerS3Client struct {
host string
pools []int
}
func (l *localPeerS3Client) GetHost() string {
return l.host
}
func (l *localPeerS3Client) SetPools(p []int) {
l.pools = make([]int, len(p))
copy(l.pools, p)
}
func (l localPeerS3Client) GetPools() []int {
return l.pools
}
func (l localPeerS3Client) ListBuckets(ctx context.Context, opts BucketOptions) ([]BucketInfo, error) {
return listBucketsLocal(ctx, opts)
}
func (l localPeerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
return getBucketInfoLocal(ctx, bucket, opts)
}
func (l localPeerS3Client) MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error {
return makeBucketLocal(ctx, bucket, opts)
}
func (l localPeerS3Client) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
return deleteBucketLocal(ctx, bucket, opts)
}
// client to talk to peer Nodes.
type peerS3Client struct {
host *xnet.Host
type remotePeerS3Client struct {
host string
pools []int
restClient *rest.Client
}
// Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
// permanently. The only way to restore the connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
// after verifying format.json
func (client *peerS3Client) call(method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
func (client *remotePeerS3Client) call(method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
return client.callWithContext(GlobalContext, method, values, body, length)
}
// Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
// permanently. The only way to restore the connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
// after verifying format.json
func (client *peerS3Client) callWithContext(ctx context.Context, method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
func (client *remotePeerS3Client) callWithContext(ctx context.Context, method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
if values == nil {
values = make(url.Values)
}
@@ -67,16 +111,15 @@ func (client *peerS3Client) callWithContext(ctx context.Context, method string,
// S3PeerSys - S3 peer call system.
type S3PeerSys struct {
peerClients []*peerS3Client // Excludes self
allPeerClients []*peerS3Client // Includes nil client for self
peerClients []peerS3Client // Excludes self
poolsCount int
}
// NewS3PeerSys - creates new S3 peer calls.
func NewS3PeerSys(endpoints EndpointServerPools) *S3PeerSys {
remote, all := newPeerS3Clients(endpoints)
return &S3PeerSys{
peerClients: remote,
allPeerClients: all,
peerClients: newPeerS3Clients(endpoints.GetNodes()),
poolsCount: len(endpoints),
}
}
@@ -84,14 +127,8 @@ func NewS3PeerSys(endpoints EndpointServerPools) *S3PeerSys {
func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) (result []BucketInfo, err error) {
g := errgroup.WithNErrs(len(sys.peerClients))
localBuckets, err := listBucketsLocal(ctx, opts)
if err != nil {
return nil, err
}
nodeBuckets := make([][]BucketInfo, len(sys.peerClients)+1)
nodeBuckets := make([][]BucketInfo, len(sys.peerClients))
errs := []error{nil}
nodeBuckets[0] = localBuckets
for idx, client := range sys.peerClients {
idx := idx
@@ -104,14 +141,14 @@ func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) (resu
if err != nil {
return err
}
nodeBuckets[idx+1] = localBuckets
nodeBuckets[idx] = localBuckets
return nil
}, idx)
}
errs = append(errs, g.Wait()...)
quorum := (len(sys.allPeerClients) / 2)
quorum := len(sys.peerClients)/2 + 1
if err = reduceReadQuorumErrs(ctx, errs, bucketOpIgnoredErrs, quorum); err != nil {
return nil, err
}
@@ -156,11 +193,7 @@ func (sys *S3PeerSys) GetBucketInfo(ctx context.Context, bucket string, opts Buc
errs := g.Wait()
bucketInfo, err := getBucketInfoLocal(ctx, bucket, opts)
errs = append(errs, err)
bucketInfos = append(bucketInfos, bucketInfo)
quorum := (len(sys.allPeerClients) / 2)
quorum := len(sys.peerClients)/2 + 1
if err = reduceReadQuorumErrs(ctx, errs, bucketOpIgnoredErrs, quorum); err != nil {
return BucketInfo{}, toObjectErr(err, bucket)
}
@@ -174,7 +207,7 @@ func (sys *S3PeerSys) GetBucketInfo(ctx context.Context, bucket string, opts Buc
return BucketInfo{}, toObjectErr(errVolumeNotFound, bucket)
}
func (client *peerS3Client) ListBuckets(ctx context.Context, opts BucketOptions) ([]BucketInfo, error) {
func (client *remotePeerS3Client) ListBuckets(ctx context.Context, opts BucketOptions) ([]BucketInfo, error) {
v := url.Values{}
v.Set(peerS3BucketDeleted, strconv.FormatBool(opts.Deleted))
@@ -190,7 +223,7 @@ func (client *peerS3Client) ListBuckets(ctx context.Context, opts BucketOptions)
}
// GetBucketInfo returns bucket stat info from a peer
func (client *peerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
func (client *remotePeerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
v := url.Values{}
v.Set(peerS3Bucket, bucket)
v.Set(peerS3BucketDeleted, strconv.FormatBool(opts.Deleted))
@@ -209,7 +242,6 @@ func (client *peerS3Client) GetBucketInfo(ctx context.Context, bucket string, op
// MakeBucket creates bucket across all peers
func (sys *S3PeerSys) MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error {
g := errgroup.WithNErrs(len(sys.peerClients))
for idx, client := range sys.peerClients {
client := client
g.Go(func() error {
@@ -219,30 +251,24 @@ func (sys *S3PeerSys) MakeBucket(ctx context.Context, bucket string, opts MakeBu
return client.MakeBucket(ctx, bucket, opts)
}, idx)
}
errs := g.Wait()
errs = append(errs, makeBucketLocal(ctx, bucket, opts))
quorum := (len(sys.allPeerClients) / 2) + 1
err := reduceWriteQuorumErrs(ctx, errs, bucketOpIgnoredErrs, quorum)
// Perform MRF on missing buckets for temporary errors.
for _, err := range errs {
if err == nil {
continue
for poolIdx := 0; poolIdx < sys.poolsCount; poolIdx++ {
perPoolErrs := make([]error, 0, len(sys.peerClients))
for i, client := range sys.peerClients {
if slices.Contains(client.GetPools(), poolIdx) {
perPoolErrs = append(perPoolErrs, errs[i])
}
}
if errors.Is(err, errPeerOffline) || errors.Is(err, errDiskNotFound) ||
isNetworkError(err) {
globalMRFState.addPartialOp(partialOperation{
bucket: bucket,
})
if poolErr := reduceReadQuorumErrs(ctx, errs, bucketOpIgnoredErrs, len(perPoolErrs)/2+1); poolErr != nil {
return toObjectErr(poolErr, bucket)
}
}
return toObjectErr(err, bucket)
return nil
}
// MakeBucket creates a bucket on a peer
func (client *peerS3Client) MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error {
func (client *remotePeerS3Client) MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error {
v := url.Values{}
v.Set(peerS3Bucket, bucket)
v.Set(peerS3BucketForceCreate, strconv.FormatBool(opts.ForceCreate))
@@ -259,7 +285,6 @@ func (client *peerS3Client) MakeBucket(ctx context.Context, bucket string, opts
// DeleteBucket deletes bucket across all peers
func (sys *S3PeerSys) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
g := errgroup.WithNErrs(len(sys.peerClients))
for idx, client := range sys.peerClients {
client := client
g.Go(func() error {
@@ -269,32 +294,26 @@ func (sys *S3PeerSys) DeleteBucket(ctx context.Context, bucket string, opts Dele
return client.DeleteBucket(ctx, bucket, opts)
}, idx)
}
errs := g.Wait()
errs = append(errs, deleteBucketLocal(ctx, bucket, opts))
var errReturn error
for _, err := range errs {
if errReturn == nil && err != nil {
// always return first error
errReturn = toObjectErr(err, bucket)
break
for poolIdx := 0; poolIdx < sys.poolsCount; poolIdx++ {
perPoolErrs := make([]error, 0, len(sys.peerClients))
for i, client := range sys.peerClients {
if slices.Contains(client.GetPools(), poolIdx) {
perPoolErrs = append(perPoolErrs, errs[i])
}
}
}
for _, err := range errs {
if err == nil && errReturn != nil {
if poolErr := reduceReadQuorumErrs(ctx, errs, bucketOpIgnoredErrs, len(perPoolErrs)/2+1); poolErr != nil {
// re-create successful deletes, since we are return an error.
sys.MakeBucket(ctx, bucket, MakeBucketOptions{})
break
return toObjectErr(poolErr, bucket)
}
}
return errReturn
return nil
}
// DeleteBucket deletes bucket on a peer
func (client *peerS3Client) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
func (client *remotePeerS3Client) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
v := url.Values{}
v.Set(peerS3Bucket, bucket)
v.Set(peerS3BucketForceDelete, strconv.FormatBool(opts.Force))
@@ -308,33 +327,35 @@ func (client *peerS3Client) DeleteBucket(ctx context.Context, bucket string, opt
return nil
}
func (client remotePeerS3Client) GetHost() string {
return client.host
}
func (client remotePeerS3Client) GetPools() []int {
return client.pools
}
func (client *remotePeerS3Client) SetPools(p []int) {
client.pools = make([]int, len(p))
copy(client.pools, p)
}
// newPeerS3Clients creates new peer clients.
// The two slices will point to the same clients,
// but 'all' will contain nil entry for local client.
// The 'all' slice will be in the same order across the cluster.
func newPeerS3Clients(endpoints EndpointServerPools) (remote, all []*peerS3Client) {
if !globalIsDistErasure {
// Only useful in distributed setups
return nil, nil
}
hosts := endpoints.hostsSorted()
remote = make([]*peerS3Client, 0, len(hosts))
all = make([]*peerS3Client, len(hosts))
for i, host := range hosts {
if host == nil {
continue
func newPeerS3Clients(nodes []Node) (peers []peerS3Client) {
peers = make([]peerS3Client, len(nodes))
for i, node := range nodes {
if node.IsLocal {
peers[i] = &localPeerS3Client{host: node.Host}
} else {
peers[i] = newPeerS3Client(node.Host)
}
all[i] = newPeerS3Client(host)
remote = append(remote, all[i])
peers[i].SetPools(node.Pools)
}
if len(all) != len(remote)+1 {
logger.LogIf(context.Background(), fmt.Errorf("WARNING: Expected number of all hosts (%v) to be remote +1 (%v)", len(all), len(remote)))
}
return remote, all
return
}
// Returns a peer S3 client.
func newPeerS3Client(peer *xnet.Host) *peerS3Client {
func newPeerS3Client(peer string) peerS3Client {
scheme := "http"
if globalIsTLS {
scheme = "https"
@@ -342,7 +363,7 @@ func newPeerS3Client(peer *xnet.Host) *peerS3Client {
serverURL := &url.URL{
Scheme: scheme,
Host: peer.String(),
Host: peer,
Path: peerS3Path,
}
@@ -360,5 +381,5 @@ func newPeerS3Client(peer *xnet.Host) *peerS3Client {
return !isNetworkError(err)
}
return &peerS3Client{host: peer, restClient: restClient}
return &remotePeerS3Client{host: peer, restClient: restClient}
}
+1 -1
View File
@@ -29,7 +29,7 @@ import (
"time"
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/pkg/randreader"
)
+35 -36
View File
@@ -36,7 +36,7 @@ import (
"github.com/coreos/go-systemd/v22/daemon"
"github.com/minio/cli"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio-go/v7/pkg/set"
@@ -246,23 +246,19 @@ func serverHandleCmdArgs(ctx *cli.Context) {
globalEndpoints, setupType, err = createServerEndpoints(globalMinioAddr, serverCmdArgs(ctx)...)
logger.FatalIf(err, "Invalid command line arguments")
globalNodes = globalEndpoints.GetNodes()
globalLocalNodeName = GetLocalPeer(globalEndpoints, globalMinioHost, globalMinioPort)
nodeNameSum := sha256.Sum256([]byte(globalLocalNodeName))
globalLocalNodeNameHex = hex.EncodeToString(nodeNameSum[:])
globalNodeNamesHex = make(map[string]struct{})
globalRemoteEndpoints = make(map[string]Endpoint)
for _, z := range globalEndpoints {
for _, ep := range z.Endpoints {
if ep.IsLocal {
globalRemoteEndpoints[globalLocalNodeName] = ep
} else {
globalRemoteEndpoints[ep.Host] = ep
}
nodeNameSum := sha256.Sum256([]byte(ep.Host))
globalNodeNamesHex[hex.EncodeToString(nodeNameSum[:])] = struct{}{}
for _, n := range globalNodes {
nodeName := n.Host
if n.IsLocal {
nodeName = globalLocalNodeName
}
nodeNameSum := sha256.Sum256([]byte(nodeName))
globalNodeNamesHex[hex.EncodeToString(nodeNameSum[:])] = struct{}{}
}
// allow transport to be HTTP/1.1 for proxying.
@@ -413,12 +409,9 @@ func bootstrapTrace(msg string) {
})
}
func initServer(ctx context.Context, newObject ObjectLayer) error {
func initServerConfig(ctx context.Context, newObject ObjectLayer) error {
t1 := time.Now()
// Once the config is fully loaded, initialize the new object layer.
setObjectLayer(newObject)
r := rand.New(rand.NewSource(time.Now().UnixNano()))
for {
@@ -633,6 +626,12 @@ func serverMain(ctx *cli.Context) {
}
}
if !globalDisableFreezeOnBoot {
// Freeze the services until the bucket notification subsystem gets initialized.
bootstrapTrace("freezeServices")
freezeServices()
}
bootstrapTrace("newObjectLayer")
newObject, err := newObjectLayer(GlobalContext, globalEndpoints)
if err != nil {
@@ -645,31 +644,22 @@ func serverMain(ctx *cli.Context) {
bootstrapTrace("newSharedLock")
globalLeaderLock = newSharedLock(GlobalContext, newObject, "leader.lock")
// Enable background operations for erasure coding
// Enable background operations on
//
// - Disk auto healing
// - MRF (most recently failed) healing
// - Background expiration routine for lifecycle policies
bootstrapTrace("initAutoHeal")
initAutoHeal(GlobalContext, newObject)
bootstrapTrace("initHealMRF")
initHealMRF(GlobalContext, newObject)
bootstrapTrace("initBackgroundExpiry")
initBackgroundExpiry(GlobalContext, newObject)
if !globalCLIContext.StrictS3Compat {
logger.Info(color.RedBold("WARNING: Strict AWS S3 compatible incoming PUT, POST content payload validation is turned off, caution is advised do not use in production"))
}
if globalActiveCred.Equal(auth.DefaultCredentials) {
msg := fmt.Sprintf("WARNING: Detected default credentials '%s', we recommend that you change these values with 'MINIO_ROOT_USER' and 'MINIO_ROOT_PASSWORD' environment variables",
globalActiveCred)
logger.Info(color.RedBold(msg))
}
if !globalDisableFreezeOnBoot {
// Freeze the services until the bucket notification subsystem gets initialized.
freezeServices()
}
bootstrapTrace("initializing the server")
if err = initServer(GlobalContext, newObject); err != nil {
bootstrapTrace("initServerConfig")
if err = initServerConfig(GlobalContext, newObject); err != nil {
var cerr config.Err
// For any config error, we don't need to drop into safe-mode
// instead its a user error and should be fixed by user.
@@ -685,12 +675,22 @@ func serverMain(ctx *cli.Context) {
logger.LogIf(GlobalContext, err)
}
if !globalCLIContext.StrictS3Compat {
logger.Info(color.RedBold("WARNING: Strict AWS S3 compatible incoming PUT, POST content payload validation is turned off, caution is advised do not use in production"))
}
if globalActiveCred.Equal(auth.DefaultCredentials) {
msg := fmt.Sprintf("WARNING: Detected default credentials '%s', we recommend that you change these values with 'MINIO_ROOT_USER' and 'MINIO_ROOT_PASSWORD' environment variables",
globalActiveCred)
logger.Info(color.RedBold(msg))
}
// Initialize users credentials and policies in background right after config has initialized.
go func() {
bootstrapTrace("globalIAMSys.Init")
globalIAMSys.Init(GlobalContext, newObject, globalEtcdClient, globalRefreshIAMInterval)
// Initialize
// Initialize Console UI
if globalBrowserEnabled {
bootstrapTrace("initConsoleServer")
srv, err := initConsoleServer()
@@ -721,7 +721,6 @@ func serverMain(ctx *cli.Context) {
go func() {
if !globalDisableFreezeOnBoot {
bootstrapTrace("freezeServices")
defer unfreezeServices()
defer bootstrapTrace("unfreezeServices")
t := time.AfterFunc(5*time.Minute, func() {
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"runtime"
"runtime/debug"
"github.com/minio/madmin-go/v2/kernel"
"github.com/minio/madmin-go/v3/kernel"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/sys"
)
+1 -1
View File
@@ -25,7 +25,7 @@ import (
"strings"
humanize "github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
color "github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/logger"
xnet "github.com/minio/pkg/net"
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"strings"
"testing"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
// Tests if we generate storage info.
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio/internal/auth"
+4 -4
View File
@@ -51,10 +51,6 @@ func handleSignals() {
// send signal to various go-routines that they need to quit.
cancelGlobalContext()
if globalEventNotifier != nil {
globalEventNotifier.RemoveAllRemoteTargets()
}
if httpServer := newHTTPServerFn(); httpServer != nil {
err = httpServer.Shutdown()
if !errors.Is(err, http.ErrServerClosed) {
@@ -71,6 +67,10 @@ func handleSignals() {
logger.LogIf(context.Background(), srv.Shutdown())
}
if globalEventNotifier != nil {
globalEventNotifier.RemoveAllBucketTargets()
}
return (err == nil && oerr == nil)
}
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"testing"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
xhttp "github.com/minio/minio/internal/http"
)
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
)
//go:generate msgp -file=$GOFILE
+1 -1
View File
@@ -34,7 +34,7 @@ import (
"sync"
"time"
"github.com/minio/madmin-go/v2"
"github.com/minio/madmin-go/v3"
minioClient "github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio-go/v7/pkg/replication"

Some files were not shown because too many files have changed in this diff Show More