Compare commits

...

26 Commits

Author SHA1 Message Date
Olli Janatuinen b413ff9fdb Support user certificate based authentication on SFTP (#19650) 2024-05-06 23:41:25 -07:00
Harshavardhana 6a15580817 fix: collect quorum errors for deletePrefix() (#19685)
do not return error for single drive being offline.
2024-05-06 22:44:46 -07:00
Cesar N 39633a5581 Set Console Redirect URL env variable (#19683) 2024-05-06 19:47:59 -07:00
Alex 1e83f15e2f Update Console version to v1.4.0 (#19684) 2024-05-06 19:47:37 -07:00
Harshavardhana 888d2bb1d8 support ETag value to be '*' (#19682)
This supports '*' as per behavior to
comply with AWS S3 behavior for

- 'If-Match: *'
- 'If-None-Match: *'
2024-05-06 17:08:42 -07:00
Klaus Post 847ee5ac45 Make WalkDir return errors (#19677)
If used, 'opts.Marker` will cause many missed entries since results are returned 
unsorted, and pools are serialized.

Switch to fully concurrent listing and merging across pools to return sorted entries.
2024-05-06 13:27:52 -07:00
jiuker 9a9a49aa84 fix: Ignore AWSAccessKeyId check for SignV2 policy condition (#19673) 2024-05-06 03:52:41 -07:00
Harshavardhana a03ca80269 support 'mc support perf object' with root login disabled (#19672)
It is expected that whoever is using the credentials which has
the proper set of permissions must be able to run.

`mc support perf object`

While the root login is disabled.
2024-05-06 02:45:10 -07:00
Harshavardhana 523bd769f1 add support for specific error response for InvalidRange (#19668)
fixes #19648

AWS S3 returns the actual object size as part of XML
response for InvalidRange error, this is used apparently
by SDKs to retry the request without the range.
2024-05-05 09:56:21 -07:00
Harshavardhana 8ff70ea5a9 turn-off coloring if we have std{err,out} dumb terminals (#19667) 2024-05-03 17:17:57 -07:00
Harshavardhana da3e7747ca avoid using 10MiB EC buffers in maxAPI calculations (#19665)
max requests per node is more conservative in its value
causing premature serialization of the calls, avoid it
for newer deployments.
2024-05-03 13:08:20 -07:00
Klaus Post 4afb59e63f fix: walk missing entries with opts.Marker set (#19661)
'opts.Marker` is causing many missed entries if used since results are returned unsorted. Also since pools are serialized.

Switch to do fully concurrent listing and merging across pools to return sorted entries.

Returning errors on listings is impossible with the current API, so document that.

Return an error at once if no drives are found instead of just returning an empty listing and no error.
2024-05-03 10:26:51 -07:00
Harshavardhana 1526e7ece3 extend server config.yaml to support per pool set drive count (#19663)
This is to support deployments migrating from a multi-pooled
wider stripe to lower stripe. MINIO_STORAGE_CLASS_STANDARD
is still expected to be same for all pools. So you can satisfy
adding custom drive count based pools by adjusting the storage
class value.

```
version: v2
address: ':9000'
rootUser: 'minioadmin'
rootPassword: 'minioadmin'
console-address: ':9001'
pools: # Specify the nodes and drives with pools
  -
    args:
        - 'node{11...14}.example.net/data{1...4}'
  -
    args:
        - 'node{15...18}.example.net/data{1...4}'
  -
    args:
        - 'node{19...22}.example.net/data{1...4}'
  -
    args:
        - 'node{23...34}.example.net/data{1...10}'
    set-drive-count: 6
```
2024-05-03 08:54:03 -07:00
Krishnan Parthasarathi 6c07bfee8a With retention, skip actions expiring all versions (#19657)
ILM actions due to ExpiredObjectDeleteAllVersions and
DelMarkerExpiration are ignored when object locking is enabled on a
bucket.
Note: This applies to object versions which may not have retention
configured on them. This applies to all object versions in this bucket,
including those created before the retention config was applied.
2024-05-03 04:18:58 -07:00
Poorna 446c760820 replication: Avoid proxying if requested object is a deletemarker (#19656)
Fixes: #19654
2024-05-02 13:15:54 -07:00
Shireesh Anjal 04f92f1291 Change endpoint format for per-bucket metrics (#19655)
Per-bucket metrics endpoints always start with /bucket and the bucket
name is appended to the path. e.g. if the collector path is /bucket/api,
the endpoint for the bucket "mybucket" would be
/minio/metrics/v3/bucket/api/mybucket

Change the existing bucket api endpoint accordingly from /api/bucket to
/bucket/api
2024-05-02 10:37:57 -07:00
Klaus Post 4a60a7794d Use better gzip for log rotate (#19651)
Should be 2x faster with same usage.
2024-05-02 04:38:40 -07:00
Bala FA e5b16adb1c Add cluster IAM metrics in metrics-v3 (#19595)
Signed-off-by: Bala.FA <bala@minio.io>
2024-05-02 01:20:42 -07:00
Harshavardhana 402a3ac719 support compression after rotation of logs (#19647) 2024-05-01 15:38:07 -07:00
Aditya Manthramurthy f3d61c51fc fix: Filter out cust. AssumeRole Token for audit (#19646)
The `Token` parameter is a sensitive value that should not be output in the Audit log for STS AssumeRoleWithCustomToken API.

Bonus: Add a simple tool that echoes audit logs to the console.
2024-05-01 14:31:13 -07:00
Klaus Post 0cde17ae5d Return listing when exceeding min disk errors (#19644)
When listing, with drives returning `errFileNotFound,` `errVolumeNotFound`, or `errUnformattedDisk,`, 
we could get below `minDisks` drives being left.

This would result in a quorum never being reachable for any object. Therefore, the listing 
would continue, but no results would ever be produced.

Include `fnf` in the mindisk check since it is incremented on these errors. This will stop 
listing when minDisks are left.

Allow `opts.minDisks` to not return errVolumeNotFound or errFileNotFound and return that. 
That will allow for good results even if disks return something else.

We switch `errUnformattedDisk` to a regular error. If we have enough of those, we should just fail.
2024-05-01 10:59:08 -07:00
Harshavardhana 8c1bba681b add logrotate support for MinIO logs (#19641) 2024-05-01 10:57:52 -07:00
Klaus Post dbfb5e797b Wait one minute after startup to restart decommissioning (#19645)
Typically not all drives are connected, so we delay 3 minutes before resuming.
This greatly reduces risk of starting to list unconnected drives, or drives we risk being disconnected soon.

This delay is not applied when starting with an admin call.
2024-05-01 08:18:21 -07:00
Harshavardhana 08ff702434 enhance ListSVCs() API to return more info to avoid InfoSvc() (#19642)
ConsoleUI like applications rely on combination of

ListServiceAccounts() and InfoServiceAccount() to populate
UI elements, however individually these calls can be slow
causing the entire UI to load sluggishly.
2024-05-01 05:41:13 -07:00
Klaus Post 0e2148264a Fix --stfp "mac-algos=..." overwrites cipher algorithms (#19643)
Setting MAC algorithms overwrites cipher algorithms.

Followup to #19636
2024-05-01 04:07:40 -07:00
Minio Trusted a75f42344b Update yaml files to latest version RELEASE.2024-05-01T01-11-10Z 2024-05-01 02:45:52 +00:00
64 changed files with 1611 additions and 556 deletions
+7 -2
View File
@@ -1039,8 +1039,13 @@ func (a adminAPIHandlers) ListServiceAccounts(w http.ResponseWriter, r *http.Req
for _, svc := range serviceAccounts {
expiryTime := svc.Expiration
serviceAccountList = append(serviceAccountList, madmin.ServiceAccountInfo{
AccessKey: svc.AccessKey,
Expiration: &expiryTime,
Description: svc.Description,
ParentUser: svc.ParentUser,
Name: svc.Name,
AccountStatus: svc.Status,
AccessKey: svc.AccessKey,
ImpliedPolicy: svc.IsImpliedPolicy(),
Expiration: &expiryTime,
})
}
+59 -1
View File
@@ -50,6 +50,7 @@ import (
"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/auth"
"github.com/minio/minio/internal/dsync"
"github.com/minio/minio/internal/grid"
"github.com/minio/minio/internal/handlers"
@@ -1627,6 +1628,47 @@ func (a adminAPIHandlers) NetperfHandler(w http.ResponseWriter, r *http.Request)
}
}
func isAllowedRWAccess(r *http.Request, cred auth.Credentials, bucketName string) (rd, wr bool) {
owner := cred.AccessKey == globalActiveCred.AccessKey
// Set prefix value for "s3:prefix" policy conditionals.
r.Header.Set("prefix", "")
// Set delimiter value for "s3:delimiter" policy conditionals.
r.Header.Set("delimiter", SlashSeparator)
isAllowedAccess := func(bucketName string) (rd, wr bool) {
if globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.GetObjectAction,
BucketName: bucketName,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
ObjectName: "",
Claims: cred.Claims,
}) {
rd = true
}
if globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.PutObjectAction,
BucketName: bucketName,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
ObjectName: "",
Claims: cred.Claims,
}) {
wr = true
}
return rd, wr
}
return isAllowedAccess(bucketName)
}
// ObjectSpeedTestHandler - reports maximum speed of a cluster by performing PUT and
// GET operations on the server, supports auto tuning by default by automatically
// increasing concurrency and stopping when we have reached the limits on the
@@ -1635,11 +1677,24 @@ func (a adminAPIHandlers) ObjectSpeedTestHandler(w http.ResponseWriter, r *http.
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
objectAPI, _ := validateAdminReq(ctx, w, r, policy.HealthInfoAdminAction)
objectAPI, creds := validateAdminReq(ctx, w, r, policy.HealthInfoAdminAction)
if objectAPI == nil {
return
}
if !globalAPIConfig.permitRootAccess() {
rd, wr := isAllowedRWAccess(r, creds, globalObjectPerfBucket)
if !rd || !wr {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, AdminError{
Code: "XMinioSpeedtestInsufficientPermissions",
Message: fmt.Sprintf("%s does not have read and write access to '%s' bucket", creds.AccessKey,
globalObjectPerfBucket),
StatusCode: http.StatusForbidden,
}), r.URL)
return
}
}
sizeStr := r.Form.Get(peerRESTSize)
durationStr := r.Form.Get(peerRESTDuration)
concurrentStr := r.Form.Get(peerRESTConcurrent)
@@ -1648,6 +1703,7 @@ func (a adminAPIHandlers) ObjectSpeedTestHandler(w http.ResponseWriter, r *http.
autotune := r.Form.Get("autotune") == "true"
noClear := r.Form.Get("noclear") == "true"
enableSha256 := r.Form.Get("enableSha256") == "true"
enableMultipart := r.Form.Get("enableMultipart") == "true"
size, err := strconv.Atoi(sizeStr)
if err != nil {
@@ -1721,6 +1777,8 @@ func (a adminAPIHandlers) ObjectSpeedTestHandler(w http.ResponseWriter, r *http.
storageClass: storageClass,
bucketName: customBucket,
enableSha256: enableSha256,
enableMultipart: enableMultipart,
creds: creds,
})
var prevResult madmin.SpeedTestResult
for {
+37 -20
View File
@@ -56,19 +56,23 @@ type APIError struct {
Code string
Description string
HTTPStatusCode int
ObjectSize string
RangeRequested string
}
// APIErrorResponse - error response format
type APIErrorResponse struct {
XMLName xml.Name `xml:"Error" json:"-"`
Code string
Message string
Key string `xml:"Key,omitempty" json:"Key,omitempty"`
BucketName string `xml:"BucketName,omitempty" json:"BucketName,omitempty"`
Resource string
Region string `xml:"Region,omitempty" json:"Region,omitempty"`
RequestID string `xml:"RequestId" json:"RequestId"`
HostID string `xml:"HostId" json:"HostId"`
XMLName xml.Name `xml:"Error" json:"-"`
Code string
Message string
Key string `xml:"Key,omitempty" json:"Key,omitempty"`
BucketName string `xml:"BucketName,omitempty" json:"BucketName,omitempty"`
Resource string
Region string `xml:"Region,omitempty" json:"Region,omitempty"`
RequestID string `xml:"RequestId" json:"RequestId"`
HostID string `xml:"HostId" json:"HostId"`
ActualObjectSize string `xml:"ActualObjectSize,omitempty" json:"ActualObjectSize,omitempty"`
RangeRequested string `xml:"RangeRequested,omitempty" json:"RangeRequested,omitempty"`
}
// APIErrorCode type of error status.
@@ -2412,10 +2416,9 @@ func toAPIError(ctx context.Context, err error) APIError {
apiErr := errorCodes.ToAPIErr(toAPIErrorCode(ctx, err))
switch apiErr.Code {
case "NotImplemented":
desc := fmt.Sprintf("%s (%v)", apiErr.Description, err)
apiErr = APIError{
Code: apiErr.Code,
Description: desc,
Description: fmt.Sprintf("%s (%v)", apiErr.Description, err),
HTTPStatusCode: apiErr.HTTPStatusCode,
}
case "XMinioBackendDown":
@@ -2432,7 +2435,19 @@ func toAPIError(ctx context.Context, err error) APIError {
HTTPStatusCode: e.HTTPStatusCode,
}
case batchReplicationJobError:
apiErr = APIError(e)
apiErr = APIError{
Description: e.Description,
Code: e.Code,
HTTPStatusCode: e.HTTPStatusCode,
}
case InvalidRange:
apiErr = APIError{
Code: "InvalidRange",
Description: e.Error(),
HTTPStatusCode: errorCodes[ErrInvalidRange].HTTPStatusCode,
ObjectSize: strconv.FormatInt(e.ResourceSize, 10),
RangeRequested: fmt.Sprintf("%d-%d", e.OffsetBegin, e.OffsetEnd),
}
case InvalidArgument:
apiErr = APIError{
Code: "InvalidArgument",
@@ -2559,13 +2574,15 @@ func getAPIError(code APIErrorCode) APIError {
func getAPIErrorResponse(ctx context.Context, err APIError, resource, requestID, hostID string) APIErrorResponse {
reqInfo := logger.GetReqInfo(ctx)
return APIErrorResponse{
Code: err.Code,
Message: err.Description,
BucketName: reqInfo.BucketName,
Key: reqInfo.ObjectName,
Resource: resource,
Region: globalSite.Region,
RequestID: requestID,
HostID: hostID,
Code: err.Code,
Message: err.Description,
BucketName: reqInfo.BucketName,
Key: reqInfo.ObjectName,
Resource: resource,
Region: globalSite.Region,
RequestID: requestID,
HostID: hostID,
ActualObjectSize: err.ObjectSize,
RangeRequested: err.RangeRequested,
}
}
+22 -15
View File
@@ -472,17 +472,17 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo
stopFn(toDelCopy[i], err)
batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s versionID=%s due to %v (attempts=%d)", ri.Bucket, toDelCopy[i].ObjectName, toDelCopy[i].VersionID, err, attempts))
failed++
if attempts == retryAttempts { // all retry attempts failed, record failure
if oi, ok := oiCache.Get(toDelCopy[i]); ok {
ri.trackCurrentBucketObject(r.Bucket, *oi, false)
}
} else {
if oi, ok := oiCache.Get(toDelCopy[i]); ok {
ri.trackCurrentBucketObject(r.Bucket, *oi, false, attempts)
}
if attempts != retryAttempts {
// retry
toDel = append(toDel, toDelCopy[i])
}
} else {
stopFn(toDelCopy[i], nil)
if oi, ok := oiCache.Get(toDelCopy[i]); ok {
ri.trackCurrentBucketObject(r.Bucket, *oi, true)
ri.trackCurrentBucketObject(r.Bucket, *oi, true, attempts)
}
}
}
@@ -537,7 +537,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
ctx, cancel := context.WithCancel(ctx)
defer cancel()
results := make(chan ObjectInfo, workerSize)
results := make(chan itemOrErr[ObjectInfo], workerSize)
if err := api.Walk(ctx, r.Bucket, r.Prefix, results, WalkOptions{
Marker: lastObject,
LatestOnly: false, // we need to visit all versions of the object to implement purge: retainVersions
@@ -584,11 +584,18 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
versionsCount int
toDel []expireObjInfo
)
failed := true
for result := range results {
if result.Err != nil {
failed = true
batchLogIf(ctx, result.Err)
continue
}
// Apply filter to find the matching rule to apply expiry
// actions accordingly.
// nolint:gocritic
if result.IsLatest {
if result.Item.IsLatest {
// send down filtered entries to be deleted using
// DeleteObjects method
if len(toDel) > 10 { // batch up to 10 objects/versions to be expired simultaneously.
@@ -609,7 +616,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
var match BatchJobExpireFilter
var found bool
for _, rule := range r.Rules {
if rule.Matches(result, now) {
if rule.Matches(result.Item, now) {
match = rule
found = true
break
@@ -619,18 +626,18 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
continue
}
prevObj = result
prevObj = result.Item
matchedFilter = match
versionsCount = 1
// Include the latest version
if matchedFilter.Purge.RetainVersions == 0 {
toDel = append(toDel, expireObjInfo{
ObjectInfo: result,
ObjectInfo: result.Item,
ExpireAll: true,
})
continue
}
} else if prevObj.Name == result.Name {
} else if prevObj.Name == result.Item.Name {
if matchedFilter.Purge.RetainVersions == 0 {
continue // including latest version in toDel suffices, skipping other versions
}
@@ -643,7 +650,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
continue // retain versions
}
toDel = append(toDel, expireObjInfo{
ObjectInfo: result,
ObjectInfo: result.Item,
})
}
// Send any remaining objects downstream
@@ -658,8 +665,8 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
<-expireDoneCh // waits for the expire goroutine to complete
wk.Wait() // waits for all expire workers to retire
ri.Complete = ri.ObjectsFailed == 0
ri.Failed = ri.ObjectsFailed > 0
ri.Complete = !failed && ri.ObjectsFailed == 0
ri.Failed = failed || ri.ObjectsFailed > 0
globalBatchJobsMetrics.save(job.ID, ri)
// Close the saverQuitCh - this also triggers saving in-memory state
+47 -18
View File
@@ -447,7 +447,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
} else {
stopFn(oi, nil)
}
ri.trackCurrentBucketObject(r.Target.Bucket, oi, success)
ri.trackCurrentBucketObject(r.Target.Bucket, oi, success, attempts)
globalBatchJobsMetrics.save(job.ID, ri)
// persist in-memory state to disk after every 10secs.
batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
@@ -690,6 +690,7 @@ type batchJobInfo struct {
StartTime time.Time `json:"startTime" msg:"st"`
LastUpdate time.Time `json:"lastUpdate" msg:"lu"`
RetryAttempts int `json:"retryAttempts" msg:"ra"`
Attempts int `json:"attempts" msg:"at"`
Complete bool `json:"complete" msg:"cmp"`
Failed bool `json:"failed" msg:"fld"`
@@ -833,13 +834,15 @@ func (ri *batchJobInfo) clone() *batchJobInfo {
ObjectsFailed: ri.ObjectsFailed,
BytesTransferred: ri.BytesTransferred,
BytesFailed: ri.BytesFailed,
Attempts: ri.Attempts,
}
}
func (ri *batchJobInfo) countItem(size int64, dmarker, success bool) {
func (ri *batchJobInfo) countItem(size int64, dmarker, success bool, attempt int) {
if ri == nil {
return
}
ri.Attempts++
if success {
if dmarker {
ri.DeleteMarkers++
@@ -847,7 +850,19 @@ func (ri *batchJobInfo) countItem(size int64, dmarker, success bool) {
ri.Objects++
ri.BytesTransferred += size
}
if attempt > 1 {
if dmarker {
ri.DeleteMarkersFailed--
} else {
ri.ObjectsFailed--
ri.BytesFailed += size
}
}
} else {
if attempt > 1 {
// Only count first attempt
return
}
if dmarker {
ri.DeleteMarkersFailed++
} else {
@@ -921,7 +936,7 @@ func (ri *batchJobInfo) trackMultipleObjectVersions(bucket string, info ObjectIn
}
}
func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool) {
func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool, attempt int) {
if ri == nil {
return
}
@@ -931,7 +946,7 @@ func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo,
ri.Bucket = bucket
ri.Object = info.Name
ri.countItem(info.Size, info.DeleteMarker, success)
ri.countItem(info.Size, info.DeleteMarker, success, attempt)
}
func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInfo) {
@@ -945,7 +960,7 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf
ri.Bucket = bucket
for i := range batch {
ri.Object = batch[i].Name
ri.countItem(batch[i].Size, batch[i].DeleteMarker, true)
ri.countItem(batch[i].Size, batch[i].DeleteMarker, true, 1)
}
}
@@ -1057,8 +1072,8 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
c.SetAppInfo("minio-"+batchJobPrefix, r.APIVersion+" "+job.ID)
var (
walkCh = make(chan ObjectInfo, 100)
slowCh = make(chan ObjectInfo, 100)
walkCh = make(chan itemOrErr[ObjectInfo], 100)
slowCh = make(chan itemOrErr[ObjectInfo], 100)
)
if !*r.Source.Snowball.Disable && r.Source.Type.isMinio() && r.Target.Type.isMinio() {
@@ -1084,7 +1099,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
if err := r.writeAsArchive(ctx, api, cl, batch); err != nil {
batchLogIf(ctx, err)
for _, b := range batch {
slowCh <- b
slowCh <- itemOrErr[ObjectInfo]{Item: b}
}
} else {
ri.trackCurrentBucketBatch(r.Source.Bucket, batch)
@@ -1095,12 +1110,12 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
}
}
for obj := range walkCh {
if obj.DeleteMarker || !obj.VersionPurgeStatus.Empty() || obj.Size >= int64(smallerThan) {
if obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
slowCh <- obj
continue
}
batch = append(batch, obj)
batch = append(batch, obj.Item)
if len(batch) < *r.Source.Snowball.Batch {
continue
@@ -1153,8 +1168,13 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
prevObj := ""
skipReplicate := false
for result := range slowCh {
result := result
for res := range slowCh {
if res.Err != nil {
ri.Failed = true
batchLogIf(ctx, res.Err)
continue
}
result := res.Item
if result.Name != prevObj {
prevObj = result.Name
skipReplicate = result.DeleteMarker && s3Type
@@ -1183,7 +1203,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
} else {
stopFn(result, nil)
}
ri.trackCurrentBucketObject(r.Source.Bucket, result, success)
ri.trackCurrentBucketObject(r.Source.Bucket, result, success, attempts)
globalBatchJobsMetrics.save(job.ID, ri)
// persist in-memory state to disk after every 10secs.
batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
@@ -1231,6 +1251,7 @@ type batchReplicationJobError struct {
Code string
Description string
HTTPStatusCode int
ObjectSize int64
}
func (e batchReplicationJobError) Error() string {
@@ -1483,7 +1504,7 @@ func (a adminAPIHandlers) ListBatchJobs(w http.ResponseWriter, r *http.Request)
jobType = string(madmin.BatchJobReplicate)
}
resultCh := make(chan ObjectInfo)
resultCh := make(chan itemOrErr[ObjectInfo])
ctx, cancel := context.WithCancel(ctx)
defer cancel()
@@ -1495,8 +1516,12 @@ func (a adminAPIHandlers) ListBatchJobs(w http.ResponseWriter, r *http.Request)
listResult := madmin.ListBatchJobsResult{}
for result := range resultCh {
if result.Err != nil {
writeErrorResponseJSON(ctx, w, toAPIError(ctx, result.Err), r.URL)
return
}
req := &BatchJobRequest{}
if err := req.load(ctx, objectAPI, result.Name); err != nil {
if err := req.load(ctx, objectAPI, result.Item.Name); err != nil {
if !errors.Is(err, errNoSuchJob) {
batchLogIf(ctx, err)
}
@@ -1701,7 +1726,7 @@ func newBatchJobPool(ctx context.Context, o ObjectLayer, workers int) *BatchJobP
}
func (j *BatchJobPool) resume() {
results := make(chan ObjectInfo, 100)
results := make(chan itemOrErr[ObjectInfo], 100)
ctx, cancel := context.WithCancel(j.ctx)
defer cancel()
if err := j.objLayer.Walk(ctx, minioMetaBucket, batchJobPrefix, results, WalkOptions{}); err != nil {
@@ -1709,12 +1734,16 @@ func (j *BatchJobPool) resume() {
return
}
for result := range results {
if result.Err != nil {
batchLogIf(j.ctx, result.Err)
continue
}
// ignore batch-replicate.bin and batch-rotate.bin entries
if strings.HasSuffix(result.Name, slashSeparator) {
if strings.HasSuffix(result.Item.Name, slashSeparator) {
continue
}
req := &BatchJobRequest{}
if err := req.load(ctx, j.objLayer, result.Name); err != nil {
if err := req.load(ctx, j.objLayer, result.Item.Name); err != nil {
batchLogIf(ctx, err)
continue
}
+30 -5
View File
@@ -419,6 +419,12 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.WrapError(err, "RetryAttempts")
return
}
case "at":
z.Attempts, err = dc.ReadInt()
if err != nil {
err = msgp.WrapError(err, "Attempts")
return
}
case "cmp":
z.Complete, err = dc.ReadBool()
if err != nil {
@@ -492,9 +498,9 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) {
// EncodeMsg implements msgp.Encodable
func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
// map header, size 16
// map header, size 17
// write "v"
err = en.Append(0xde, 0x0, 0x10, 0xa1, 0x76)
err = en.Append(0xde, 0x0, 0x11, 0xa1, 0x76)
if err != nil {
return
}
@@ -553,6 +559,16 @@ func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
err = msgp.WrapError(err, "RetryAttempts")
return
}
// write "at"
err = en.Append(0xa2, 0x61, 0x74)
if err != nil {
return
}
err = en.WriteInt(z.Attempts)
if err != nil {
err = msgp.WrapError(err, "Attempts")
return
}
// write "cmp"
err = en.Append(0xa3, 0x63, 0x6d, 0x70)
if err != nil {
@@ -659,9 +675,9 @@ func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
// MarshalMsg implements msgp.Marshaler
func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) {
o = msgp.Require(b, z.Msgsize())
// map header, size 16
// map header, size 17
// string "v"
o = append(o, 0xde, 0x0, 0x10, 0xa1, 0x76)
o = append(o, 0xde, 0x0, 0x11, 0xa1, 0x76)
o = msgp.AppendInt(o, z.Version)
// string "jid"
o = append(o, 0xa3, 0x6a, 0x69, 0x64)
@@ -678,6 +694,9 @@ func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) {
// string "ra"
o = append(o, 0xa2, 0x72, 0x61)
o = msgp.AppendInt(o, z.RetryAttempts)
// string "at"
o = append(o, 0xa2, 0x61, 0x74)
o = msgp.AppendInt(o, z.Attempts)
// string "cmp"
o = append(o, 0xa3, 0x63, 0x6d, 0x70)
o = msgp.AppendBool(o, z.Complete)
@@ -765,6 +784,12 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
err = msgp.WrapError(err, "RetryAttempts")
return
}
case "at":
z.Attempts, bts, err = msgp.ReadIntBytes(bts)
if err != nil {
err = msgp.WrapError(err, "Attempts")
return
}
case "cmp":
z.Complete, bts, err = msgp.ReadBoolBytes(bts)
if err != nil {
@@ -839,6 +864,6 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
func (z *batchJobInfo) Msgsize() (s int) {
s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size
s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size
return
}
+12 -9
View File
@@ -356,7 +356,7 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
retryAttempts := ri.RetryAttempts
ctx, cancel := context.WithCancel(ctx)
results := make(chan ObjectInfo, 100)
results := make(chan itemOrErr[ObjectInfo], 100)
if err := api.Walk(ctx, r.Bucket, r.Prefix, results, WalkOptions{
Marker: lastObject,
Filter: selectObj,
@@ -365,9 +365,14 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
// Do not need to retry if we can't list objects on source.
return err
}
for result := range results {
result := result
failed := false
for res := range results {
if res.Err != nil {
failed = true
batchLogIf(ctx, res.Err)
break
}
result := res.Item
sseKMS := crypto.S3KMS.IsEncrypted(result.UserDefined)
sseS3 := crypto.S3.IsEncrypted(result.UserDefined)
if !sseKMS && !sseS3 { // neither sse-s3 nor sse-kms disallowed
@@ -377,7 +382,6 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
go func() {
defer wk.Give()
for attempts := 1; attempts <= retryAttempts; attempts++ {
attempts := attempts
stopFn := globalBatchJobsMetrics.trace(batchJobMetricKeyRotation, job.ID, attempts)
success := true
if err := r.KeyRotate(ctx, api, result); err != nil {
@@ -387,8 +391,7 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
} else {
stopFn(result, nil)
}
ri.trackCurrentBucketObject(r.Bucket, result, success)
ri.RetryAttempts = attempts
ri.trackCurrentBucketObject(r.Bucket, result, success, attempts)
globalBatchJobsMetrics.save(job.ID, ri)
// persist in-memory state to disk after every 10secs.
batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
@@ -407,8 +410,8 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
}
wk.Wait()
ri.Complete = ri.ObjectsFailed == 0
ri.Failed = ri.ObjectsFailed > 0
ri.Complete = !failed && ri.ObjectsFailed == 0
ri.Failed = failed || ri.ObjectsFailed > 0
globalBatchJobsMetrics.save(job.ID, ri)
// persist in-memory state to disk.
batchLogIf(ctx, ri.updateAfter(ctx, api, 0, job))
+17 -6
View File
@@ -2732,7 +2732,7 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
s.workerCh <- struct{}{}
}()
// Allocate new results channel to receive ObjectInfo.
objInfoCh := make(chan ObjectInfo)
objInfoCh := make(chan itemOrErr[ObjectInfo])
cfg, err := getReplicationConfig(ctx, opts.bucket)
if err != nil {
replLogIf(ctx, fmt.Errorf("replication resync of %s for arn %s failed with %w", opts.bucket, opts.arn, err))
@@ -2867,7 +2867,12 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
}
}(ctx, i)
}
for obj := range objInfoCh {
for res := range objInfoCh {
if res.Err != nil {
resyncStatus = ResyncFailed
replLogIf(ctx, res.Err)
return
}
select {
case <-s.resyncCancelCh:
resyncStatus = ResyncCanceled
@@ -2876,11 +2881,11 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
return
default:
}
if heal && lastCheckpoint != "" && lastCheckpoint != obj.Name {
if heal && lastCheckpoint != "" && lastCheckpoint != res.Item.Name {
continue
}
lastCheckpoint = ""
roi := getHealReplicateObjectInfo(obj, rcfg)
roi := getHealReplicateObjectInfo(res.Item, rcfg)
if !roi.ExistingObjResync.mustResync() {
continue
}
@@ -3140,7 +3145,7 @@ func getReplicationDiff(ctx context.Context, objAPI ObjectLayer, bucket string,
return nil, err
}
objInfoCh := make(chan ObjectInfo, 10)
objInfoCh := make(chan itemOrErr[ObjectInfo], 10)
if err := objAPI.Walk(ctx, bucket, opts.Prefix, objInfoCh, WalkOptions{}); err != nil {
replLogIf(ctx, err)
return nil, err
@@ -3152,11 +3157,17 @@ func getReplicationDiff(ctx context.Context, objAPI ObjectLayer, bucket string,
diffCh := make(chan madmin.DiffInfo, 4000)
go func() {
defer xioutil.SafeClose(diffCh)
for obj := range objInfoCh {
for res := range objInfoCh {
if res.Err != nil {
diffCh <- madmin.DiffInfo{Err: res.Err}
return
}
if contextCanceled(ctx) {
// Just consume input...
continue
}
obj := res.Item
// Ignore object prefixes which are excluded
// from versioning via the MinIO bucket versioning extension.
if globalBucketVersioningSys.PrefixSuspended(bucket, obj.Name) {
+1 -1
View File
@@ -65,5 +65,5 @@ var (
MinioBannerName = "MinIO Object Storage Server"
// MinioLicense - MinIO server license.
MinioLicense = "GNU AGPLv3 <https://www.gnu.org/licenses/agpl-3.0.html>"
MinioLicense = "GNU AGPLv3 - https://www.gnu.org/licenses/agpl-3.0.html"
)
+11
View File
@@ -63,6 +63,7 @@ import (
"github.com/minio/pkg/v2/ellipses"
"github.com/minio/pkg/v2/env"
xnet "github.com/minio/pkg/v2/net"
"golang.org/x/term"
)
// serverDebugLog will enable debug printing
@@ -73,6 +74,13 @@ var (
)
func init() {
if !term.IsTerminal(int(os.Stdout.Fd())) || !term.IsTerminal(int(os.Stderr.Fd())) {
color.TurnOff()
}
if env.Get("NO_COLOR", "") != "" || env.Get("TERM", "") == "dumb" {
color.TurnOff()
}
if runtime.GOOS == "windows" {
if mousetrap.StartedByExplorer() {
fmt.Printf("Don't double-click %s\n", os.Args[0])
@@ -131,6 +139,9 @@ func minioConfigToConsoleFeatures() {
os.Setenv("CONSOLE_LOG_QUERY_AUTH_TOKEN", value)
}
}
if value := env.Get(config.EnvBrowserRedirectURL, ""); value != "" {
os.Setenv("CONSOLE_BROWSER_REDIRECT_URL", value)
}
// pass the console subpath configuration
if globalBrowserRedirectURL != nil {
subPath := path.Clean(pathJoin(strings.TrimSpace(globalBrowserRedirectURL.Path), SlashSeparator))
+1 -1
View File
@@ -575,7 +575,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
configLogIf(ctx, fmt.Errorf("Invalid api configuration: %w", err))
}
globalAPIConfig.init(apiConfig, setDriveCounts)
globalAPIConfig.init(apiConfig, setDriveCounts, objAPI.Legacy())
autoGenerateRootCredentials() // Generate the KMS root credentials here since we don't know whether API root access is disabled until now.
setRemoteInstanceTransport(NewHTTPTransportWithTimeout(apiConfig.RemoteTransportDeadline))
case config.CompressionSubSys:
+4 -3
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2015-2021 MinIO, Inc.
// Copyright (c) 2015-2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
@@ -20,6 +20,7 @@ package cmd
import (
"container/ring"
"context"
"io"
"sync"
"sync/atomic"
@@ -49,10 +50,10 @@ type HTTPConsoleLoggerSys struct {
// NewConsoleLogger - creates new HTTPConsoleLoggerSys with all nodes subscribed to
// the console logging pub sub system
func NewConsoleLogger(ctx context.Context) *HTTPConsoleLoggerSys {
func NewConsoleLogger(ctx context.Context, w io.Writer) *HTTPConsoleLoggerSys {
return &HTTPConsoleLoggerSys{
pubsub: pubsub.New[log.Info, madmin.LogMask](8),
console: console.New(),
console: console.New(w),
logBuf: ring.New(defaultLogBufferCount),
}
}
+6 -8
View File
@@ -1199,17 +1199,15 @@ func evalActionFromLifecycle(ctx context.Context, lc lifecycle.Lifecycle, lr loc
console.Debugf(applyActionsLogPrefix+" lifecycle: Secondary scan: %v\n", event.Action)
}
if event.Action == lifecycle.NoneAction {
return event
}
if obj.IsLatest && event.Action == lifecycle.DeleteAllVersionsAction {
if lr.LockEnabled && enforceRetentionForDeletion(ctx, obj) {
switch event.Action {
case lifecycle.DeleteAllVersionsAction, lifecycle.DelMarkerDeleteAllVersionsAction:
// Skip if bucket has object locking enabled; To prevent the
// possibility of violating an object retention on one of the
// noncurrent versions of this object.
if lr.LockEnabled {
return lifecycle.Event{Action: lifecycle.NoneAction}
}
}
switch event.Action {
case lifecycle.DeleteVersionAction, lifecycle.DeleteRestoredVersionAction:
// Defensive code, should never happen
if obj.VersionID == "" {
+96
View File
@@ -20,12 +20,15 @@ package cmd
import (
"context"
"encoding/xml"
"fmt"
"strings"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/versioning"
)
@@ -141,3 +144,96 @@ func TestApplyNewerNoncurrentVersionsLimit(t *testing.T) {
}
}
}
func TestEvalActionFromLifecycle(t *testing.T) {
// Tests cover only ExpiredObjectDeleteAllVersions and DelMarkerExpiration actions
obj := ObjectInfo{
Name: "foo",
ModTime: time.Now().Add(-31 * 24 * time.Hour),
Size: 100 << 20,
VersionID: uuid.New().String(),
IsLatest: true,
NumVersions: 4,
}
delMarker := ObjectInfo{
Name: "foo-deleted",
ModTime: time.Now().Add(-61 * 24 * time.Hour),
Size: 0,
VersionID: uuid.New().String(),
IsLatest: true,
DeleteMarker: true,
NumVersions: 4,
}
deleteAllILM := `<LifecycleConfiguration>
<Rule>
<Expiration>
<Days>30</Days>
<ExpiredObjectAllVersions>true</ExpiredObjectAllVersions>
</Expiration>
<Filter></Filter>
<Status>Enabled</Status>
<ID>DeleteAllVersions</ID>
</Rule>
</LifecycleConfiguration>`
delMarkerILM := `<LifecycleConfiguration>
<Rule>
<ID>DelMarkerExpiration</ID>
<Filter></Filter>
<Status>Enabled</Status>
<DelMarkerExpiration>
<Days>60</Days>
</DelMarkerExpiration>
</Rule>
</LifecycleConfiguration>`
deleteAllLc, err := lifecycle.ParseLifecycleConfig(strings.NewReader(deleteAllILM))
if err != nil {
t.Fatalf("Failed to parse deleteAllILM test ILM policy %v", err)
}
delMarkerLc, err := lifecycle.ParseLifecycleConfig(strings.NewReader(delMarkerILM))
if err != nil {
t.Fatalf("Failed to parse delMarkerILM test ILM policy %v", err)
}
tests := []struct {
ilm lifecycle.Lifecycle
retention lock.Retention
obj ObjectInfo
want lifecycle.Action
}{
{
// with object locking
ilm: *deleteAllLc,
retention: lock.Retention{LockEnabled: true},
obj: obj,
want: lifecycle.NoneAction,
},
{
// without object locking
ilm: *deleteAllLc,
retention: lock.Retention{},
obj: obj,
want: lifecycle.DeleteAllVersionsAction,
},
{
// with object locking
ilm: *delMarkerLc,
retention: lock.Retention{LockEnabled: true},
obj: delMarker,
want: lifecycle.NoneAction,
},
{
// without object locking
ilm: *delMarkerLc,
retention: lock.Retention{},
obj: delMarker,
want: lifecycle.DelMarkerDeleteAllVersionsAction,
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("TestEvalAction-%d", i), func(t *testing.T) {
if got := evalActionFromLifecycle(context.TODO(), test.ilm, test.retention, nil, test.obj); got.Action != test.want {
t.Fatalf("Expected %v but got %v", test.want, got)
}
})
}
}
+26 -28
View File
@@ -23,7 +23,6 @@ import (
"net/url"
"runtime"
"sort"
"strconv"
"strings"
"github.com/cespare/xxhash/v2"
@@ -134,7 +133,7 @@ func possibleSetCountsWithSymmetry(setCounts []uint64, argPatterns []ellipses.Ar
// on each index, this function also determines the final set size
// The final set size has the affinity towards choosing smaller
// indexes (total sets)
func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint64, argPatterns []ellipses.ArgPattern) (setIndexes [][]uint64, err error) {
func getSetIndexes(args []string, totalSizes []uint64, setDriveCount uint64, argPatterns []ellipses.ArgPattern) (setIndexes [][]uint64, err error) {
if len(totalSizes) == 0 || len(args) == 0 {
return nil, errInvalidArgument
}
@@ -142,7 +141,7 @@ func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint6
setIndexes = make([][]uint64, len(totalSizes))
for _, totalSize := range totalSizes {
// Check if totalSize has minimum range upto setSize
if totalSize < setSizes[0] || totalSize < customSetDriveCount {
if totalSize < setSizes[0] || totalSize < setDriveCount {
msg := fmt.Sprintf("Incorrect number of endpoints provided %s", args)
return nil, config.ErrInvalidNumberOfErasureEndpoints(nil).Msg(msg)
}
@@ -167,11 +166,11 @@ func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint6
var setSize uint64
// Custom set drive count allows to override automatic distribution.
// only meant if you want to further optimize drive distribution.
if customSetDriveCount > 0 {
if setDriveCount > 0 {
msg := fmt.Sprintf("Invalid set drive count. Acceptable values for %d number drives are %d", commonSize, setCounts)
var found bool
for _, ss := range setCounts {
if ss == customSetDriveCount {
if ss == setDriveCount {
found = true
}
}
@@ -180,8 +179,7 @@ func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint6
}
// No automatic symmetry calculation expected, user is on their own
setSize = customSetDriveCount
globalCustomErasureDriveCount = true
setSize = setDriveCount
} else {
// Returns possible set counts with symmetry.
setCounts = possibleSetCountsWithSymmetry(setCounts, argPatterns)
@@ -256,7 +254,7 @@ func getTotalSizes(argPatterns []ellipses.ArgPattern) []uint64 {
// Parses all arguments and returns an endpointSet which is a collection
// of endpoints following the ellipses pattern, this is what is used
// by the object layer for initializing itself.
func parseEndpointSet(customSetDriveCount uint64, args ...string) (ep endpointSet, err error) {
func parseEndpointSet(setDriveCount uint64, args ...string) (ep endpointSet, err error) {
argPatterns := make([]ellipses.ArgPattern, len(args))
for i, arg := range args {
patterns, perr := ellipses.FindEllipsesPatterns(arg)
@@ -266,7 +264,7 @@ func parseEndpointSet(customSetDriveCount uint64, args ...string) (ep endpointSe
argPatterns[i] = patterns
}
ep.setIndexes, err = getSetIndexes(args, getTotalSizes(argPatterns), customSetDriveCount, argPatterns)
ep.setIndexes, err = getSetIndexes(args, getTotalSizes(argPatterns), setDriveCount, argPatterns)
if err != nil {
return endpointSet{}, config.ErrInvalidErasureEndpoints(nil).Msg(err.Error())
}
@@ -281,23 +279,14 @@ func parseEndpointSet(customSetDriveCount uint64, args ...string) (ep endpointSe
// specific set size.
// For example: {1...64} is divided into 4 sets each of size 16.
// This applies to even distributed setup syntax as well.
func GetAllSets(args ...string) ([][]string, error) {
var customSetDriveCount uint64
if v := env.Get(EnvErasureSetDriveCount, ""); v != "" {
driveCount, err := strconv.Atoi(v)
if err != nil {
return nil, config.ErrInvalidErasureSetSize(err)
}
customSetDriveCount = uint64(driveCount)
}
func GetAllSets(setDriveCount uint64, args ...string) ([][]string, error) {
var setArgs [][]string
if !ellipses.HasEllipses(args...) {
var setIndexes [][]uint64
// Check if we have more one args.
if len(args) > 1 {
var err error
setIndexes, err = getSetIndexes(args, []uint64{uint64(len(args))}, customSetDriveCount, nil)
setIndexes, err = getSetIndexes(args, []uint64{uint64(len(args))}, setDriveCount, nil)
if err != nil {
return nil, err
}
@@ -311,7 +300,7 @@ func GetAllSets(args ...string) ([][]string, error) {
}
setArgs = s.Get()
} else {
s, err := parseEndpointSet(customSetDriveCount, args...)
s, err := parseEndpointSet(setDriveCount, args...)
if err != nil {
return nil, err
}
@@ -336,8 +325,6 @@ const (
EnvErasureSetDriveCount = "MINIO_ERASURE_SET_DRIVE_COUNT"
)
var globalCustomErasureDriveCount = false
type node struct {
nodeName string
disks []string
@@ -366,8 +353,13 @@ func (el *endpointsList) add(arg string) error {
return nil
}
type poolArgs struct {
args []string
setDriveCount uint64
}
// buildDisksLayoutFromConfFile supports with and without ellipses transparently.
func buildDisksLayoutFromConfFile(pools [][]string) (layout disksLayout, err error) {
func buildDisksLayoutFromConfFile(pools []poolArgs) (layout disksLayout, err error) {
if len(pools) == 0 {
return layout, errInvalidArgument
}
@@ -375,7 +367,7 @@ func buildDisksLayoutFromConfFile(pools [][]string) (layout disksLayout, err err
for _, list := range pools {
var endpointsList endpointsList
for _, arg := range list {
for _, arg := range list.args {
switch {
case ellipses.HasList(arg):
patterns, err := ellipses.FindListPatterns(arg)
@@ -436,7 +428,7 @@ func buildDisksLayoutFromConfFile(pools [][]string) (layout disksLayout, err err
}
}
setArgs, err := GetAllSets(eps...)
setArgs, err := GetAllSets(list.setDriveCount, eps...)
if err != nil {
return layout, err
}
@@ -469,9 +461,15 @@ func mergeDisksLayoutFromArgs(args []string, ctxt *serverCtxt) (err error) {
var setArgs [][]string
v, err := env.GetInt(EnvErasureSetDriveCount, 0)
if err != nil {
return err
}
setDriveCount := uint64(v)
// None of the args have ellipses use the old style.
if ok {
setArgs, err = GetAllSets(args...)
setArgs, err = GetAllSets(setDriveCount, args...)
if err != nil {
return err
}
@@ -487,7 +485,7 @@ func mergeDisksLayoutFromArgs(args []string, ctxt *serverCtxt) (err error) {
// TODO: support SNSD deployments to be decommissioned in future
return fmt.Errorf("all args must have ellipses for pool expansion (%w) args: %s", errInvalidArgument, args)
}
setArgs, err = GetAllSets(arg)
setArgs, err = GetAllSets(setDriveCount, arg)
if err != nil {
return err
}
+23 -11
View File
@@ -251,6 +251,18 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
opts.NoDecryption = true
}
if objInfo.Size == 0 {
if _, _, err := rs.GetOffsetLength(objInfo.Size); err != nil {
// Make sure to return object info to provide extra information.
return &GetObjectReader{
ObjInfo: objInfo,
}, err
}
// Zero byte objects don't even need to further initialize pipes etc.
return NewGetObjectReaderFromReader(bytes.NewReader(nil), objInfo, opts)
}
if objInfo.IsRemote() {
gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, h, objInfo, opts)
if err != nil {
@@ -260,11 +272,6 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
return gr.WithCleanupFuncs(nsUnlocker), nil
}
if objInfo.Size == 0 {
// Zero byte objects don't even need to further initialize pipes etc.
return NewGetObjectReaderFromReader(bytes.NewReader(nil), objInfo, opts)
}
fn, off, length, err := NewGetObjectReader(rs, objInfo, opts)
if err != nil {
return nil, err
@@ -1801,6 +1808,14 @@ func (er erasureObjects) commitRenameDataDir(ctx context.Context, bucket, object
func (er erasureObjects) deletePrefix(ctx context.Context, bucket, prefix string) error {
disks := er.getDisks()
// Assume (N/2 + 1) quorum for Delete()
// this is a theoretical assumption such that
// for delete's we do not need to honor storage
// class for objects that have reduced quorum
// due to storage class - this only needs to be honored
// for Read() requests alone that we already do.
writeQuorum := len(disks)/2 + 1
g := errgroup.WithNErrs(len(disks))
for index := range disks {
index := index
@@ -1814,12 +1829,9 @@ func (er erasureObjects) deletePrefix(ctx context.Context, bucket, prefix string
})
}, index)
}
for _, err := range g.Wait() {
if err != nil {
return err
}
}
return nil
// return errors if any during deletion
return reduceWriteQuorumErrs(ctx, g.Wait(), objectOpIgnoredErrs, writeQuorum)
}
// DeleteObject - deletes an object, this call doesn't necessary reply
+4
View File
@@ -535,6 +535,10 @@ func (z *erasureServerPools) Init(ctx context.Context) error {
if len(poolIndices) > 0 && globalEndpoints[poolIndices[0]].Endpoints[0].IsLocal {
go func() {
// Resume decommissioning of pools, but wait 3 minutes for cluster to stabilize.
if err := sleepContext(ctx, 3*time.Minute); err != nil {
return
}
r := rand.New(rand.NewSource(time.Now().UnixNano()))
for {
if err := z.Decommission(ctx, poolIndices...); err != nil {
+183 -154
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2015-2021 MinIO, Inc.
// Copyright (c) 2015-2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
@@ -644,6 +644,15 @@ func (z *erasureServerPools) Shutdown(ctx context.Context) error {
return nil
}
// Legacy returns 'true' if distribution algo is CRCMOD
func (z *erasureServerPools) Legacy() (ok bool) {
ok = true
for _, set := range z.serverPools {
ok = ok && set.Legacy()
}
return ok
}
func (z *erasureServerPools) BackendInfo() (b madmin.BackendInfo) {
b.Type = madmin.Erasure
@@ -2034,178 +2043,198 @@ func (z *erasureServerPools) HealBucket(ctx context.Context, bucket string, opts
}
// Walk a bucket, optionally prefix recursively, until we have returned
// all the content to objectInfo channel, it is callers responsibility
// to allocate a receive channel for ObjectInfo, upon any unhandled
// error walker returns error. Optionally if context.Done() is received
// then Walk() stops the walker.
func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, results chan<- ObjectInfo, opts WalkOptions) error {
// all the contents of the provided bucket+prefix.
// TODO: Note that most errors will result in a truncated listing.
func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, results chan<- itemOrErr[ObjectInfo], opts WalkOptions) error {
if err := checkListObjsArgs(ctx, bucket, prefix, ""); err != nil {
// Upon error close the channel.
xioutil.SafeClose(results)
return err
}
parentCtx := ctx
ctx, cancelCause := context.WithCancelCause(ctx)
var entries []chan metaCacheEntry
vcfg, _ := globalBucketVersioningSys.Get(bucket)
ctx, cancel := context.WithCancel(ctx)
go func() {
defer cancel()
defer xioutil.SafeClose(results)
for _, erasureSet := range z.serverPools {
var wg sync.WaitGroup
for _, set := range erasureSet.sets {
set := set
wg.Add(1)
go func() {
defer wg.Done()
disks, infos, _ := set.getOnlineDisksWithHealingAndInfo(true)
if len(disks) == 0 {
cancel()
for poolIdx, erasureSet := range z.serverPools {
for setIdx, set := range erasureSet.sets {
set := set
listOut := make(chan metaCacheEntry, 1)
entries = append(entries, listOut)
disks, infos, _ := set.getOnlineDisksWithHealingAndInfo(true)
if len(disks) == 0 {
xioutil.SafeClose(results)
err := fmt.Errorf("Walk: no online disks found in pool %d, set %d", setIdx, poolIdx)
cancelCause(err)
return err
}
go func() {
defer xioutil.SafeClose(listOut)
send := func(e metaCacheEntry) {
if e.isDir() {
// Ignore directories.
return
}
select {
case listOut <- e:
case <-ctx.Done():
}
}
send := func(objInfo ObjectInfo) bool {
select {
case <-ctx.Done():
return false
case results <- objInfo:
return true
askDisks := getListQuorum(opts.AskDisks, set.setDriveCount)
if askDisks == -1 {
newDisks := getQuorumDisks(disks, infos, (len(disks)+1)/2)
if newDisks != nil {
// If we found disks signature in quorum, we proceed to list
// from a single drive, shuffling of the drives is subsequently.
disks = newDisks
askDisks = 1
} else {
// If we did not find suitable disks, perform strict quorum listing
// as no disk agrees on quorum anymore.
askDisks = getListQuorum("strict", set.setDriveCount)
}
}
// Special case: ask all disks if the drive count is 4
if set.setDriveCount == 4 || askDisks > len(disks) {
askDisks = len(disks) // use all available drives
}
var fallbackDisks []StorageAPI
if askDisks > 0 && len(disks) > askDisks {
rand.Shuffle(len(disks), func(i, j int) {
disks[i], disks[j] = disks[j], disks[i]
})
fallbackDisks = disks[askDisks:]
disks = disks[:askDisks]
}
requestedVersions := 0
if opts.LatestOnly {
requestedVersions = 1
}
// However many we ask, versions must exist on ~50%
listingQuorum := (askDisks + 1) / 2
// How to resolve partial results.
resolver := metadataResolutionParams{
dirQuorum: listingQuorum,
objQuorum: listingQuorum,
bucket: bucket,
requestedVersions: requestedVersions,
}
path := baseDirFromPrefix(prefix)
filterPrefix := strings.Trim(strings.TrimPrefix(prefix, path), slashSeparator)
if path == prefix {
filterPrefix = ""
}
lopts := listPathRawOptions{
disks: disks,
fallbackDisks: fallbackDisks,
bucket: bucket,
path: path,
filterPrefix: filterPrefix,
recursive: true,
forwardTo: opts.Marker,
minDisks: listingQuorum,
reportNotFound: false,
agreed: send,
partial: func(entries metaCacheEntries, _ []error) {
entry, ok := entries.resolve(&resolver)
if ok {
send(*entry)
}
}
},
finished: nil,
}
askDisks := getListQuorum(opts.AskDisks, set.setDriveCount)
if askDisks == -1 {
newDisks := getQuorumDisks(disks, infos, (len(disks)+1)/2)
if newDisks != nil {
// If we found disks signature in quorum, we proceed to list
// from a single drive, shuffling of the drives is subsequently.
disks = newDisks
askDisks = 1
} else {
// If we did not find suitable disks, perform strict quorum listing
// as no disk agrees on quorum anymore.
askDisks = getListQuorum("strict", set.setDriveCount)
}
}
if err := listPathRaw(ctx, lopts); err != nil {
cancelCause(fmt.Errorf("listPathRaw returned %w: opts(%#v)", err, lopts))
return
}
}()
}
}
// Special case: ask all disks if the drive count is 4
if set.setDriveCount == 4 || askDisks > len(disks) {
askDisks = len(disks) // use all available drives
}
var fallbackDisks []StorageAPI
if askDisks > 0 && len(disks) > askDisks {
rand.Shuffle(len(disks), func(i, j int) {
disks[i], disks[j] = disks[j], disks[i]
})
fallbackDisks = disks[askDisks:]
disks = disks[:askDisks]
}
requestedVersions := 0
if opts.LatestOnly {
requestedVersions = 1
}
loadEntry := func(entry metaCacheEntry) {
if entry.isDir() {
// Convert and filter merged entries.
merged := make(chan metaCacheEntry, 100)
vcfg, _ := globalBucketVersioningSys.Get(bucket)
go func() {
defer cancelCause(nil)
defer xioutil.SafeClose(results)
sentErr := false
sendErr := func(err error) {
if !sentErr {
select {
case results <- itemOrErr[ObjectInfo]{Err: err}:
sentErr = true
case <-parentCtx.Done():
}
}
}
send := func(oi ObjectInfo) bool {
select {
case results <- itemOrErr[ObjectInfo]{Item: oi}:
return true
case <-ctx.Done():
sendErr(context.Cause(ctx))
return false
}
}
for entry := range merged {
if opts.LatestOnly {
fi, err := entry.fileInfo(bucket)
if err != nil {
sendErr(err)
return
}
if opts.Filter != nil {
if opts.Filter(fi) {
if !send(fi.ToObjectInfo(bucket, fi.Name, vcfg != nil && vcfg.Versioned(fi.Name))) {
return
}
if opts.LatestOnly {
fi, err := entry.fileInfo(bucket)
if err != nil {
cancel()
return
}
if opts.Filter != nil {
if opts.Filter(fi) {
if !send(fi.ToObjectInfo(bucket, fi.Name, vcfg != nil && vcfg.Versioned(fi.Name))) {
return
}
}
} else {
if !send(fi.ToObjectInfo(bucket, fi.Name, vcfg != nil && vcfg.Versioned(fi.Name))) {
return
}
}
} else {
fivs, err := entry.fileInfoVersions(bucket)
if err != nil {
cancel()
return
}
// Note: entry.fileInfoVersions returns versions sorted in reverse chronological order based on ModTime
if opts.VersionsSort == WalkVersionsSortAsc {
versionsSorter(fivs.Versions).reverse()
}
for _, version := range fivs.Versions {
if opts.Filter != nil {
if opts.Filter(version) {
if !send(version.ToObjectInfo(bucket, version.Name, vcfg != nil && vcfg.Versioned(version.Name))) {
return
}
}
} else {
if !send(version.ToObjectInfo(bucket, version.Name, vcfg != nil && vcfg.Versioned(version.Name))) {
return
}
}
}
}
}
// However many we ask, versions must exist on ~50%
listingQuorum := (askDisks + 1) / 2
// How to resolve partial results.
resolver := metadataResolutionParams{
dirQuorum: listingQuorum,
objQuorum: listingQuorum,
bucket: bucket,
requestedVersions: requestedVersions,
}
path := baseDirFromPrefix(prefix)
filterPrefix := strings.Trim(strings.TrimPrefix(prefix, path), slashSeparator)
if path == prefix {
filterPrefix = ""
}
lopts := listPathRawOptions{
disks: disks,
fallbackDisks: fallbackDisks,
bucket: bucket,
path: path,
filterPrefix: filterPrefix,
recursive: true,
forwardTo: opts.Marker,
minDisks: 1,
reportNotFound: false,
agreed: loadEntry,
partial: func(entries metaCacheEntries, _ []error) {
entry, ok := entries.resolve(&resolver)
if ok {
loadEntry(*entry)
}
},
finished: nil,
}
if err := listPathRaw(ctx, lopts); err != nil {
storageLogIf(ctx, fmt.Errorf("listPathRaw returned %w: opts(%#v)", err, lopts))
cancel()
} else {
if !send(fi.ToObjectInfo(bucket, fi.Name, vcfg != nil && vcfg.Versioned(fi.Name))) {
return
}
}()
}
continue
}
fivs, err := entry.fileInfoVersions(bucket)
if err != nil {
sendErr(err)
return
}
// Note: entry.fileInfoVersions returns versions sorted in reverse chronological order based on ModTime
if opts.VersionsSort == WalkVersionsSortAsc {
versionsSorter(fivs.Versions).reverse()
}
for _, version := range fivs.Versions {
if opts.Filter != nil {
if opts.Filter(version) {
if !send(version.ToObjectInfo(bucket, version.Name, vcfg != nil && vcfg.Versioned(version.Name))) {
return
}
}
} else {
if !send(version.ToObjectInfo(bucket, version.Name, vcfg != nil && vcfg.Versioned(version.Name))) {
return
}
}
}
wg.Wait()
}
}()
go func() {
// Merge all entries from all disks.
// We leave quorum at 1, since entries are already resolved to have the desired quorum.
// mergeEntryChannels will close 'merged' channel upon completion or cancellation.
storageLogIf(ctx, mergeEntryChannels(ctx, entries, merged, 1))
}()
return nil
}
+5
View File
@@ -194,6 +194,11 @@ func findDiskIndex(refFormat, format *formatErasureV3) (int, int, error) {
return -1, -1, fmt.Errorf("DriveID: %s not found", format.Erasure.This)
}
// Legacy returns 'true' if distribution algo is CRCMOD
func (s *erasureSets) Legacy() (ok bool) {
return s.distributionAlgo == formatErasureVersionV2DistributionAlgoV1
}
// connectDisks - attempt to connect all the endpoints, loads format
// and re-arranges the disks in proper position.
func (s *erasureSets) connectDisks() {
+2 -5
View File
@@ -443,11 +443,8 @@ func checkFormatErasureValues(formats []*formatErasureV3, disks []StorageAPI, se
return fmt.Errorf("%s drive is already being used in another erasure deployment. (Number of drives specified: %d but the number of drives found in the %s drive's format.json: %d)",
disks[i], len(formats), humanize.Ordinal(i+1), len(formatErasure.Erasure.Sets)*len(formatErasure.Erasure.Sets[0]))
}
// Only if custom erasure drive count is set, verify if the
// set_drive_count was manually set - we need to honor what is
// present on the drives.
if globalCustomErasureDriveCount && len(formatErasure.Erasure.Sets[0]) != setDriveCount {
return fmt.Errorf("%s drive is already formatted with %d drives per erasure set. This cannot be changed to %d, please revert your MINIO_ERASURE_SET_DRIVE_COUNT setting", disks[i], len(formatErasure.Erasure.Sets[0]), setDriveCount)
if len(formatErasure.Erasure.Sets[0]) != setDriveCount {
return fmt.Errorf("%s drive is already formatted with %d drives per erasure set. This cannot be changed to %d", disks[i], len(formatErasure.Erasure.Sets[0]), setDriveCount)
}
}
return nil
+21 -21
View File
@@ -22,6 +22,7 @@ import (
"net/http"
"os"
"runtime"
"slices"
"strconv"
"strings"
"sync"
@@ -38,13 +39,11 @@ import (
type apiConfig struct {
mu sync.RWMutex
requestsDeadline time.Duration
requestsPool chan struct{}
clusterDeadline time.Duration
listQuorum string
corsAllowOrigins []string
// total drives per erasure set across pools.
totalDriveCount int
requestsDeadline time.Duration
requestsPool chan struct{}
clusterDeadline time.Duration
listQuorum string
corsAllowOrigins []string
replicationPriority string
replicationMaxWorkers int
transitionWorkers int
@@ -110,7 +109,7 @@ func availableMemory() (available uint64) {
return
}
func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) {
t.mu.Lock()
defer t.mu.Unlock()
@@ -125,27 +124,24 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
}
t.corsAllowOrigins = corsAllowOrigin
maxSetDrives := 0
for _, setDriveCount := range setDriveCounts {
t.totalDriveCount += setDriveCount
if setDriveCount > maxSetDrives {
maxSetDrives = setDriveCount
}
}
var apiRequestsMaxPerNode int
if cfg.RequestsMax <= 0 {
maxSetDrives := slices.Max(setDriveCounts)
// Returns 75% of max memory allowed
maxMem := availableMemory()
// max requests per node is calculated as
// total_ram / ram_per_request
// ram_per_request is (2MiB+128KiB) * driveCount \
// + 2 * 10MiB (default erasure block size v1) + 2 * 1MiB (default erasure block size v2)
blockSize := xioutil.LargeBlock + xioutil.SmallBlock
apiRequestsMaxPerNode = int(maxMem / uint64(maxSetDrives*blockSize+int(blockSizeV1*2+blockSizeV2*2)))
if globalIsDistErasure {
logger.Info("Automatically configured API requests per node based on available memory on the system: %d", apiRequestsMaxPerNode)
if legacy {
// ram_per_request is (1MiB+32KiB) * driveCount \
// + 2 * 10MiB (default erasure block size v1) + 2 * 1MiB (default erasure block size v2)
apiRequestsMaxPerNode = int(maxMem / uint64(maxSetDrives*blockSize+int(blockSizeV1*2+blockSizeV2*2)))
} else {
// ram_per_request is (1MiB+32KiB) * driveCount \
// + 2 * 1MiB (default erasure block size v2)
apiRequestsMaxPerNode = int(maxMem / uint64(maxSetDrives*blockSize+int(blockSizeV2*2)))
}
} else {
apiRequestsMaxPerNode = cfg.RequestsMax
@@ -154,6 +150,10 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
}
}
if globalIsDistErasure {
logger.Info("Configured max API requests per node based on available memory: %d", apiRequestsMaxPerNode)
}
if cap(t.requestsPool) != apiRequestsMaxPerNode {
// Only replace if needed.
// Existing requests will use the previous limit,
+5 -1
View File
@@ -60,7 +60,11 @@ func (h *HTTPRangeSpec) GetLength(resourceSize int64) (rangeLength int64, err er
}
case h.Start >= resourceSize:
return 0, errInvalidRange
return 0, InvalidRange{
OffsetBegin: h.Start,
OffsetEnd: h.End,
ResourceSize: resourceSize,
}
case h.End > -1:
end := h.End
+2 -2
View File
@@ -72,7 +72,7 @@ func TestHTTPRequestRangeSpec(t *testing.T) {
if err == nil {
t.Errorf("Case %d: Did not get an expected error - got %v", i, rs)
}
if err == errInvalidRange {
if isErrInvalidRange(err) {
t.Errorf("Case %d: Got invalid range error instead of a parse error", i)
}
if rs != nil {
@@ -95,7 +95,7 @@ func TestHTTPRequestRangeSpec(t *testing.T) {
if err1 == nil {
o, l, err2 = rs.GetOffsetLength(resourceSize)
}
if err1 == errInvalidRange || (err1 == nil && err2 == errInvalidRange) {
if isErrInvalidRange(err1) || (err1 == nil && isErrInvalidRange(err2)) {
continue
}
t.Errorf("Case %d: Expected errInvalidRange but: %v %v %d %d %v", i, rs, err1, o, l, err2)
+13 -12
View File
@@ -645,36 +645,37 @@ func (iamOS *IAMObjectStore) deleteGroupInfo(ctx context.Context, name string) e
return err
}
// helper type for listIAMConfigItems
type itemOrErr struct {
Item string
Err error
}
// Lists objects in the minioMetaBucket at the given path prefix. All returned
// items have the pathPrefix removed from their names.
func listIAMConfigItems(ctx context.Context, objAPI ObjectLayer, pathPrefix string) <-chan itemOrErr {
ch := make(chan itemOrErr)
func listIAMConfigItems(ctx context.Context, objAPI ObjectLayer, pathPrefix string) <-chan itemOrErr[string] {
ch := make(chan itemOrErr[string])
go func() {
defer xioutil.SafeClose(ch)
// Allocate new results channel to receive ObjectInfo.
objInfoCh := make(chan ObjectInfo)
objInfoCh := make(chan itemOrErr[ObjectInfo])
if err := objAPI.Walk(ctx, minioMetaBucket, pathPrefix, objInfoCh, WalkOptions{}); err != nil {
select {
case ch <- itemOrErr{Err: err}:
case ch <- itemOrErr[string]{Err: err}:
case <-ctx.Done():
}
return
}
for obj := range objInfoCh {
item := strings.TrimPrefix(obj.Name, pathPrefix)
if obj.Err != nil {
select {
case ch <- itemOrErr[string]{Err: obj.Err}:
case <-ctx.Done():
return
}
}
item := strings.TrimPrefix(obj.Item.Name, pathPrefix)
item = strings.TrimSuffix(item, SlashSeparator)
select {
case ch <- itemOrErr{Item: item}:
case ch <- itemOrErr[string]{Item: item}:
case <-ctx.Done():
return
}
+1 -2
View File
@@ -134,7 +134,6 @@ func newApp(name string) *cli.App {
// Register all commands.
registerCommand(serverCmd)
registerCommand(gatewayCmd) // hidden kept for guiding users.
// Set up app.
cli.HelpFlag = cli.BoolFlag{
@@ -181,7 +180,7 @@ func versionBanner(c *cli.Context) io.Reader {
banner := &strings.Builder{}
fmt.Fprintln(banner, color.Bold("%s version %s (commit-id=%s)", c.App.Name, c.App.Version, CommitID))
fmt.Fprintln(banner, color.Blue("Runtime:")+color.Bold(" %s %s/%s", runtime.Version(), runtime.GOOS, runtime.GOARCH))
fmt.Fprintln(banner, color.Blue("License:")+color.Bold(" GNU AGPLv3 <https://www.gnu.org/licenses/agpl-3.0.html>"))
fmt.Fprintln(banner, color.Blue("License:")+color.Bold(" GNU AGPLv3 - https://www.gnu.org/licenses/agpl-3.0.html"))
fmt.Fprintln(banner, color.Blue("Copyright:")+color.Bold(" 2015-%s MinIO, Inc.", CopyrightYear))
return strings.NewReader(banner.String())
}
+18 -20
View File
@@ -1010,8 +1010,7 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
// not a storage error.
return nil
}
askDisks := len(disks)
readers := make([]*metacacheReader, askDisks)
readers := make([]*metacacheReader, len(disks))
defer func() {
for _, r := range readers {
r.Close()
@@ -1093,16 +1092,14 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
case nil:
default:
switch err.Error() {
case errFileNotFound.Error(),
errVolumeNotFound.Error(),
errUnformattedDisk.Error():
case errFileNotFound.Error():
atEOF++
fnf++
// This is a special case, to handle bucket does
// not exist situations.
if errors.Is(err, errVolumeNotFound) {
vnf++
}
continue
case errVolumeNotFound.Error():
atEOF++
fnf++
vnf++
continue
}
hasErr++
@@ -1141,8 +1138,17 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
topEntries[i] = entry
}
// Stop if we exceed number of bad disks
if hasErr > len(disks)-opts.minDisks && hasErr > 0 {
// Since minDisks is set to quorum, we return if we have enough.
if vnf > 0 && vnf >= len(readers)-opts.minDisks {
return errVolumeNotFound
}
// Since minDisks is set to quorum, we return if we have enough.
if fnf > 0 && fnf >= len(readers)-opts.minDisks {
return errFileNotFound
}
// Stop if we exceed number of bad disks.
if hasErr > 0 && hasErr+fnf > len(disks)-opts.minDisks {
if opts.finished != nil {
opts.finished(errs)
}
@@ -1160,10 +1166,6 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
return errors.New(strings.Join(combinedErr, ", "))
}
if vnf == len(readers) {
return errVolumeNotFound
}
// Break if all at EOF or error.
if atEOF+hasErr == len(readers) {
if hasErr > 0 && opts.finished != nil {
@@ -1172,10 +1174,6 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
break
}
if fnf == len(readers) {
return errFileNotFound
}
if agree == len(readers) {
// Everybody agreed
for _, r := range readers {
+69
View File
@@ -0,0 +1,69 @@
// Copyright (c) 2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"context"
"sync/atomic"
"time"
)
const (
lastSyncDurationMillis = "last_sync_duration_millis"
pluginAuthnServiceFailedRequestsMinute = "plugin_authn_service_failed_requests_minute"
pluginAuthnServiceLastFailSeconds = "plugin_authn_service_last_fail_seconds"
pluginAuthnServiceLastSuccSeconds = "plugin_authn_service_last_succ_seconds"
pluginAuthnServiceSuccAvgRttMsMinute = "plugin_authn_service_succ_avg_rtt_ms_minute"
pluginAuthnServiceSuccMaxRttMsMinute = "plugin_authn_service_succ_max_rtt_ms_minute"
pluginAuthnServiceTotalRequestsMinute = "plugin_authn_service_total_requests_minute"
sinceLastSyncMillis = "since_last_sync_millis"
syncFailures = "sync_failures"
syncSuccesses = "sync_successes"
)
var (
lastSyncDurationMillisMD = NewCounterMD(lastSyncDurationMillis, "Last successful IAM data sync duration in milliseconds")
pluginAuthnServiceFailedRequestsMinuteMD = NewCounterMD(pluginAuthnServiceFailedRequestsMinute, "When plugin authentication is configured, returns failed requests count in the last full minute")
pluginAuthnServiceLastFailSecondsMD = NewCounterMD(pluginAuthnServiceLastFailSeconds, "When plugin authentication is configured, returns time (in seconds) since the last failed request to the service")
pluginAuthnServiceLastSuccSecondsMD = NewCounterMD(pluginAuthnServiceLastSuccSeconds, "When plugin authentication is configured, returns time (in seconds) since the last successful request to the service")
pluginAuthnServiceSuccAvgRttMsMinuteMD = NewCounterMD(pluginAuthnServiceSuccAvgRttMsMinute, "When plugin authentication is configured, returns average round-trip-time of successful requests in the last full minute")
pluginAuthnServiceSuccMaxRttMsMinuteMD = NewCounterMD(pluginAuthnServiceSuccMaxRttMsMinute, "When plugin authentication is configured, returns maximum round-trip-time of successful requests in the last full minute")
pluginAuthnServiceTotalRequestsMinuteMD = NewCounterMD(pluginAuthnServiceTotalRequestsMinute, "When plugin authentication is configured, returns total requests count in the last full minute")
sinceLastSyncMillisMD = NewCounterMD(sinceLastSyncMillis, "Time (in milliseconds) since last successful IAM data sync.")
syncFailuresMD = NewCounterMD(syncFailures, "Number of failed IAM data syncs since server start.")
syncSuccessesMD = NewCounterMD(syncSuccesses, "Number of successful IAM data syncs since server start.")
)
// loadClusterIAMMetrics - `MetricsLoaderFn` for cluster IAM metrics.
func loadClusterIAMMetrics(_ context.Context, m MetricValues, _ *metricsCache) error {
m.Set(lastSyncDurationMillis, float64(atomic.LoadUint64(&globalIAMSys.LastRefreshDurationMilliseconds)))
pluginAuthNMetrics := globalAuthNPlugin.Metrics()
m.Set(pluginAuthnServiceFailedRequestsMinute, float64(pluginAuthNMetrics.FailedRequests))
m.Set(pluginAuthnServiceLastFailSeconds, pluginAuthNMetrics.LastUnreachableSecs)
m.Set(pluginAuthnServiceLastSuccSeconds, pluginAuthNMetrics.LastReachableSecs)
m.Set(pluginAuthnServiceSuccAvgRttMsMinute, pluginAuthNMetrics.AvgSuccRTTMs)
m.Set(pluginAuthnServiceSuccMaxRttMsMinute, pluginAuthNMetrics.MaxSuccRTTMs)
m.Set(pluginAuthnServiceTotalRequestsMinute, float64(pluginAuthNMetrics.TotalRequests))
lastSyncTime := atomic.LoadUint64(&globalIAMSys.LastRefreshTimeUnixNano)
if lastSyncTime != 0 {
m.Set(sinceLastSyncMillis, float64((uint64(time.Now().UnixNano())-lastSyncTime)/uint64(time.Millisecond)))
}
m.Set(syncFailures, float64(atomic.LoadUint64(&globalIAMSys.TotalRefreshFailures)))
m.Set(syncSuccesses, float64(atomic.LoadUint64(&globalIAMSys.TotalRefreshSuccesses)))
return nil
}
+9 -12
View File
@@ -221,18 +221,15 @@ func (h *metricsV3Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
pathComponents := mux.Vars(r)["pathComps"]
isListingRequest := r.Form.Has("list")
// Parse optional buckets query parameter.
bucketsParam := r.Form["buckets"]
buckets := make([]string, 0, len(bucketsParam))
for _, bp := range bucketsParam {
bp = strings.TrimSpace(bp)
if bp == "" {
continue
}
splits := strings.Split(bp, ",")
for _, split := range splits {
buckets = append(buckets, strings.TrimSpace(split))
}
buckets := []string{}
if strings.HasPrefix(pathComponents, "/bucket/") {
// bucket specific metrics, extract the bucket name from the path.
// it's the last part of the path. e.g. /bucket/api/<bucket-name>
bucketIdx := strings.LastIndex(pathComponents, "/")
buckets = append(buckets, pathComponents[bucketIdx+1:])
// remove bucket from pathComponents as it is dyanamic and
// hence not included in the collector path.
pathComponents = pathComponents[:bucketIdx]
}
innerHandler := h.handle(pathComponents, isListingRequest, buckets)
+23 -1
View File
@@ -29,9 +29,13 @@ import (
//
// These are paths under the top-level /minio/metrics/v3 metrics endpoint. Each
// of these paths returns a set of V3 metrics.
//
// Per-bucket metrics endpoints always start with /bucket and the bucket name is
// appended to the path. e.g. if the collector path is /bucket/api, the endpoint
// for the bucket "mybucket" would be /minio/metrics/v3/bucket/api/mybucket
const (
apiRequestsCollectorPath collectorPath = "/api/requests"
apiBucketCollectorPath collectorPath = "/api/bucket"
apiBucketCollectorPath collectorPath = "/bucket/api"
systemNetworkInternodeCollectorPath collectorPath = "/system/network/internode"
systemDriveCollectorPath collectorPath = "/system/drive"
@@ -46,6 +50,7 @@ const (
clusterErasureSetCollectorPath collectorPath = "/cluster/erasure-set"
clusterAuditCollectorPath collectorPath = "/cluster/audit"
clusterNotificationCollectorPath collectorPath = "/cluster/notification"
clusterIAMCollectorPath collectorPath = "/cluster/iam"
)
const (
@@ -278,6 +283,22 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
loadClusterNotificationMetrics,
)
clusterIAMMG := NewMetricsGroup(clusterIAMCollectorPath,
[]MetricDescriptor{
lastSyncDurationMillisMD,
pluginAuthnServiceFailedRequestsMinuteMD,
pluginAuthnServiceLastFailSecondsMD,
pluginAuthnServiceLastSuccSecondsMD,
pluginAuthnServiceSuccAvgRttMsMinuteMD,
pluginAuthnServiceSuccMaxRttMsMinuteMD,
pluginAuthnServiceTotalRequestsMinuteMD,
sinceLastSyncMillisMD,
syncFailuresMD,
syncSuccessesMD,
},
loadClusterIAMMetrics,
)
allMetricGroups := []*MetricsGroup{
apiRequestsMG,
apiBucketMG,
@@ -294,6 +315,7 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
clusterErasureSetMG,
clusterAuditMG,
clusterNotificationMG,
clusterIAMMG,
}
// Bucket metrics are special, they always include the bucket label. These
+4 -1
View File
@@ -595,7 +595,7 @@ type InvalidRange struct {
}
func (e InvalidRange) Error() string {
return fmt.Sprintf("The requested range \"bytes %d -> %d of %d\" is not satisfiable.", e.OffsetBegin, e.OffsetEnd, e.ResourceSize)
return fmt.Sprintf("The requested range 'bytes=%d-%d' is not satisfiable", e.OffsetBegin, e.OffsetEnd)
}
// ObjectTooLarge error returned when the size of the object > max object size allowed (5G) per request.
@@ -758,6 +758,9 @@ func isErrMethodNotAllowed(err error) bool {
}
func isErrInvalidRange(err error) bool {
if errors.Is(err, errInvalidRange) {
return true
}
_, ok := err.(InvalidRange)
return ok
}
+2 -1
View File
@@ -244,6 +244,7 @@ type ObjectLayer interface {
Shutdown(context.Context) error
NSScanner(ctx context.Context, updates chan<- DataUsageInfo, wantCycle uint32, scanMode madmin.HealScanMode) error
BackendInfo() madmin.BackendInfo
Legacy() bool // Only returns true for deployments which use CRCMOD as its object distribution algorithm.
StorageInfo(ctx context.Context, metrics bool) StorageInfo
LocalStorageInfo(ctx context.Context, metrics bool) StorageInfo
@@ -256,7 +257,7 @@ type ObjectLayer interface {
ListObjectsV2(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int, fetchOwner bool, startAfter string) (result ListObjectsV2Info, err error)
ListObjectVersions(ctx context.Context, bucket, prefix, marker, versionMarker, delimiter string, maxKeys int) (result ListObjectVersionsInfo, err error)
// Walk lists all objects including versions, delete markers.
Walk(ctx context.Context, bucket, prefix string, results chan<- ObjectInfo, opts WalkOptions) error
Walk(ctx context.Context, bucket, prefix string, results chan<- itemOrErr[ObjectInfo], opts WalkOptions) error
// Object operations.
+4 -4
View File
@@ -878,7 +878,7 @@ func _testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler, v
// Marker being set to a value which is lesser than and all object names when sorted (37).
// Expected to send all the objects in the bucket in this case.
{"test-bucket-list-object", "", "Abc", "", 10, resultCases[14], nil, true},
// Marker is to a hierarhical value (38-39).
// Marker is to a hierarchical value (38-39).
{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", "", 10, resultCases[15], nil, true},
{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "", 10, resultCases[16], nil, true},
// Testing with marker and truncation, but no prefix (40-42).
@@ -909,7 +909,7 @@ func _testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler, v
{"test-bucket-list-object", "Asia", "", SlashSeparator, 10, resultCases[25], nil, true},
{"test-bucket-list-object", "new", "", SlashSeparator, 10, resultCases[26], nil, true},
{"test-bucket-list-object", "Asia/India/", "", SlashSeparator, 10, resultCases[27], nil, true},
// Test with marker set as hierarhical value and with delimiter. (58-59)
// Test with marker set as hierarchical value and with delimiter. (58-59)
{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", SlashSeparator, 10, resultCases[28], nil, true},
{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", SlashSeparator, 10, resultCases[29], nil, true},
// Test with prefix and delimiter set to '/'. (60)
@@ -1618,7 +1618,7 @@ func testListObjectVersions(obj ObjectLayer, instanceType string, t1 TestErrHand
// Marker being set to a value which is lesser than and all object names when sorted (35).
// Expected to send all the objects in the bucket in this case.
{"test-bucket-list-object", "", "Abc", "", 10, resultCases[14], nil, true},
// Marker is to a hierarhical value (36-37).
// Marker is to a hierarchical value (36-37).
{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", "", 10, resultCases[15], nil, true},
{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "", 10, resultCases[16], nil, true},
// Testing with marker and truncation, but no prefix (38-40).
@@ -1649,7 +1649,7 @@ func testListObjectVersions(obj ObjectLayer, instanceType string, t1 TestErrHand
{"test-bucket-list-object", "Asia", "", SlashSeparator, 10, resultCases[25], nil, true},
{"test-bucket-list-object", "new", "", SlashSeparator, 10, resultCases[26], nil, true},
{"test-bucket-list-object", "Asia/India/", "", SlashSeparator, 10, resultCases[27], nil, true},
// Test with marker set as hierarhical value and with delimiter. (56-57)
// Test with marker set as hierarchical value and with delimiter. (56-57)
{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", SlashSeparator, 10, resultCases[28], nil, true},
{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", SlashSeparator, 10, resultCases[29], nil, true},
// Test with prefix and delimiter set to '/'. (58)
+4
View File
@@ -23,6 +23,7 @@ import (
"net/http"
"regexp"
"strconv"
"strings"
"time"
"github.com/minio/minio/internal/amztime"
@@ -325,6 +326,9 @@ func canonicalizeETag(etag string) string {
// isETagEqual return true if the canonical representations of two ETag strings
// are equal, false otherwise
func isETagEqual(left, right string) bool {
if strings.TrimSpace(right) == "*" {
return true
}
return canonicalizeETag(left) == canonicalizeETag(right)
}
+18 -15
View File
@@ -488,21 +488,24 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
reader *GetObjectReader
perr error
)
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
globalReplicationStats.incProxy(bucket, getObjectAPI, false)
// proxy to replication target if active-active replication is in place.
reader, proxy, perr = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts, proxytgts)
if perr != nil {
globalReplicationStats.incProxy(bucket, getObjectAPI, true)
proxyGetErr := ErrorRespToObjectError(perr, bucket, object)
if !isErrBucketNotFound(proxyGetErr) && !isErrObjectNotFound(proxyGetErr) && !isErrVersionNotFound(proxyGetErr) &&
!isErrPreconditionFailed(proxyGetErr) && !isErrInvalidRange(proxyGetErr) {
replLogIf(ctx, fmt.Errorf("Proxying request (replication) failed for %s/%s(%s) - %w", bucket, object, opts.VersionID, perr))
// avoid proxying if version is a delete marker
if !isErrMethodNotAllowed(err) && !(gr != nil && gr.ObjInfo.DeleteMarker) {
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
globalReplicationStats.incProxy(bucket, getObjectAPI, false)
// proxy to replication target if active-active replication is in place.
reader, proxy, perr = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts, proxytgts)
if perr != nil {
globalReplicationStats.incProxy(bucket, getObjectAPI, true)
proxyGetErr := ErrorRespToObjectError(perr, bucket, object)
if !isErrBucketNotFound(proxyGetErr) && !isErrObjectNotFound(proxyGetErr) && !isErrVersionNotFound(proxyGetErr) &&
!isErrPreconditionFailed(proxyGetErr) && !isErrInvalidRange(proxyGetErr) {
replLogIf(ctx, fmt.Errorf("Proxying request (replication) failed for %s/%s(%s) - %w", bucket, object, opts.VersionID, perr))
}
}
if reader != nil && proxy.Proxy && perr == nil {
gr = reader
}
}
if reader != nil && proxy.Proxy && perr == nil {
gr = reader
}
}
if reader == nil || !proxy.Proxy {
@@ -1024,7 +1027,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
objInfo, err := getObjectInfo(ctx, bucket, object, opts)
var proxy proxyResult
if err != nil {
if err != nil && !objInfo.DeleteMarker && !isErrMethodNotAllowed(err) {
// proxy HEAD to replication target if active-active replication configured on bucket
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
+1
View File
@@ -715,6 +715,7 @@ func (client *peerRESTClient) SpeedTest(ctx context.Context, opts speedTestOpts)
values.Set(peerRESTStorageClass, opts.storageClass)
values.Set(peerRESTBucket, opts.bucketName)
values.Set(peerRESTEnableSha256, strconv.FormatBool(opts.enableSha256))
values.Set(peerRESTEnableMultipart, strconv.FormatBool(opts.enableMultipart))
respBody, err := client.callWithContext(context.Background(), peerRESTMethodSpeedTest, values, nil, -1)
if err != nil {
+28 -26
View File
@@ -18,7 +18,7 @@
package cmd
const (
peerRESTVersion = "v38" // Convert RPC calls
peerRESTVersion = "v39" // add more flags to speedtest API
peerRESTVersionPrefix = SlashSeparator + peerRESTVersion
peerRESTPrefix = minioReservedBucketPath + "/peer"
peerRESTPath = peerRESTPrefix + peerRESTVersionPrefix
@@ -38,31 +38,33 @@ const (
)
const (
peerRESTBucket = "bucket"
peerRESTBuckets = "buckets"
peerRESTUser = "user"
peerRESTGroup = "group"
peerRESTUserTemp = "user-temp"
peerRESTPolicy = "policy"
peerRESTUserOrGroup = "user-or-group"
peerRESTUserType = "user-type"
peerRESTIsGroup = "is-group"
peerRESTSignal = "signal"
peerRESTSubSys = "sub-sys"
peerRESTProfiler = "profiler"
peerRESTSize = "size"
peerRESTConcurrent = "concurrent"
peerRESTDuration = "duration"
peerRESTStorageClass = "storage-class"
peerRESTEnableSha256 = "enableSha256"
peerRESTMetricsTypes = "types"
peerRESTDisk = "disk"
peerRESTHost = "host"
peerRESTJobID = "job-id"
peerRESTDepID = "depID"
peerRESTStartRebalance = "start-rebalance"
peerRESTMetrics = "metrics"
peerRESTDryRun = "dry-run"
peerRESTBucket = "bucket"
peerRESTBuckets = "buckets"
peerRESTUser = "user"
peerRESTGroup = "group"
peerRESTUserTemp = "user-temp"
peerRESTPolicy = "policy"
peerRESTUserOrGroup = "user-or-group"
peerRESTUserType = "user-type"
peerRESTIsGroup = "is-group"
peerRESTSignal = "signal"
peerRESTSubSys = "sub-sys"
peerRESTProfiler = "profiler"
peerRESTSize = "size"
peerRESTConcurrent = "concurrent"
peerRESTDuration = "duration"
peerRESTStorageClass = "storage-class"
peerRESTEnableSha256 = "enableSha256"
peerRESTEnableMultipart = "enableMultipart"
peerRESTAccessKey = "access-key"
peerRESTMetricsTypes = "types"
peerRESTDisk = "disk"
peerRESTHost = "host"
peerRESTJobID = "job-id"
peerRESTDepID = "depID"
peerRESTStartRebalance = "start-rebalance"
peerRESTMetrics = "metrics"
peerRESTDryRun = "dry-run"
peerRESTURL = "url"
peerRESTSha256Sum = "sha256sum"
+15 -6
View File
@@ -1059,6 +1059,13 @@ func (s *peerRESTServer) SpeedTestHandler(w http.ResponseWriter, r *http.Request
storageClass := r.Form.Get(peerRESTStorageClass)
bucketName := r.Form.Get(peerRESTBucket)
enableSha256 := r.Form.Get(peerRESTEnableSha256) == "true"
enableMultipart := r.Form.Get(peerRESTEnableMultipart) == "true"
u, ok := globalIAMSys.GetUser(r.Context(), r.Form.Get(peerRESTAccessKey))
if !ok {
s.writeErrorResponse(w, errAuthentication)
return
}
size, err := strconv.Atoi(sizeStr)
if err != nil {
@@ -1078,12 +1085,14 @@ func (s *peerRESTServer) SpeedTestHandler(w http.ResponseWriter, r *http.Request
done := keepHTTPResponseAlive(w)
result, err := selfSpeedTest(r.Context(), speedTestOpts{
objectSize: size,
concurrency: concurrent,
duration: duration,
storageClass: storageClass,
bucketName: bucketName,
enableSha256: enableSha256,
objectSize: size,
concurrency: concurrent,
duration: duration,
storageClass: storageClass,
bucketName: bucketName,
enableSha256: enableSha256,
enableMultipart: enableMultipart,
creds: u.Credentials,
})
if err != nil {
result.Error = err.Error()
+22 -4
View File
@@ -33,6 +33,7 @@ import (
"github.com/dustin/go-humanize"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
xhttp "github.com/minio/minio/internal/http"
xioutil "github.com/minio/minio/internal/ioutil"
"github.com/minio/pkg/v2/randreader"
@@ -72,7 +73,7 @@ func (f *firstByteRecorder) Read(p []byte) (n int, err error) {
}
// Runs the speedtest on local MinIO process.
func selfSpeedTest(ctx context.Context, opts speedTestOpts) (SpeedTestResult, error) {
func selfSpeedTest(ctx context.Context, opts speedTestOpts) (res SpeedTestResult, err error) {
objAPI := newObjectLayerFn()
if objAPI == nil {
return SpeedTestResult{}, errServerNotInitialized
@@ -96,7 +97,24 @@ func selfSpeedTest(ctx context.Context, opts speedTestOpts) (SpeedTestResult, er
popts := minio.PutObjectOptions{
UserMetadata: userMetadata,
DisableContentSha256: !opts.enableSha256,
DisableMultipart: true,
DisableMultipart: !opts.enableMultipart,
}
clnt := globalMinioClient
if !globalAPIConfig.permitRootAccess() {
region := globalSite.Region
if region == "" {
region = "us-east-1"
}
clnt, err = minio.New(globalLocalNodeName, &minio.Options{
Creds: credentials.NewStaticV4(opts.creds.AccessKey, opts.creds.SecretKey, opts.creds.SessionToken),
Secure: globalIsTLS,
Transport: globalRemoteTargetTransport,
Region: region,
})
if err != nil {
return res, err
}
}
var mu sync.Mutex
@@ -109,7 +127,7 @@ func selfSpeedTest(ctx context.Context, opts speedTestOpts) (SpeedTestResult, er
t := time.Now()
reader := newRandomReader(opts.objectSize)
tmpObjName := pathJoin(objNamePrefix, fmt.Sprintf("%d/%d", i, objCountPerThread[i]))
info, err := globalMinioClient.PutObject(uploadsCtx, opts.bucketName, tmpObjName, reader, int64(opts.objectSize), popts)
info, err := clnt.PutObject(uploadsCtx, opts.bucketName, tmpObjName, reader, int64(opts.objectSize), popts)
if err != nil {
if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
errOnce.Do(func() {
@@ -150,7 +168,7 @@ func selfSpeedTest(ctx context.Context, opts speedTestOpts) (SpeedTestResult, er
var downloadTTFB madmin.TimeDurations
wg.Add(opts.concurrency)
c := minio.Core{Client: globalMinioClient}
c := minio.Core{Client: clnt}
for i := 0; i < opts.concurrency; i++ {
go func(i int) {
defer wg.Done()
+19 -12
View File
@@ -210,18 +210,23 @@ func testPostPolicyBucketHandler(obj ObjectLayer, instanceType string, t TestErr
// Test cases for signature-V2.
testCasesV2 := []struct {
expectedStatus int
accessKey string
secretKey string
formData map[string]string
}{
{http.StatusForbidden, "invalidaccesskey", credentials.SecretKey},
{http.StatusForbidden, credentials.AccessKey, "invalidsecretkey"},
{http.StatusNoContent, credentials.AccessKey, credentials.SecretKey},
{http.StatusForbidden, credentials.SecretKey, map[string]string{"AWSAccessKeyId": "invalidaccesskey"}},
{http.StatusForbidden, "invalidsecretkey", map[string]string{"AWSAccessKeyId": credentials.AccessKey}},
{http.StatusNoContent, credentials.SecretKey, map[string]string{"AWSAccessKeyId": credentials.AccessKey}},
{http.StatusForbidden, credentials.SecretKey, map[string]string{"Awsaccesskeyid": "invalidaccesskey"}},
{http.StatusForbidden, "invalidsecretkey", map[string]string{"Awsaccesskeyid": credentials.AccessKey}},
{http.StatusNoContent, credentials.SecretKey, map[string]string{"Awsaccesskeyid": credentials.AccessKey}},
// Forbidden with key not in policy.conditions for signed requests V2.
{http.StatusForbidden, credentials.SecretKey, map[string]string{"Awsaccesskeyid": credentials.AccessKey, "AnotherKey": "AnotherContent"}},
}
for i, test := range testCasesV2 {
// initialize HTTP NewRecorder, this records any mutations to response writer inside the handler.
rec := httptest.NewRecorder()
req, perr := newPostRequestV2("", bucketName, "testobject", test.accessKey, test.secretKey)
req, perr := newPostRequestV2("", bucketName, "testobject", test.secretKey, test.formData)
if perr != nil {
t.Fatalf("Test %d: %s: Failed to create HTTP request for PostPolicyHandler: <ERROR> %v", i+1, instanceType, perr)
}
@@ -593,7 +598,7 @@ func postPresignSignatureV4(policyBase64 string, t time.Time, secretAccessKey, l
return signature
}
func newPostRequestV2(endPoint, bucketName, objectName string, accessKey, secretKey string) (*http.Request, error) {
func newPostRequestV2(endPoint, bucketName, objectName string, secretKey string, formInputData map[string]string) (*http.Request, error) {
// Expire the request five minutes from now.
expirationTime := UTCNow().Add(time.Minute * 5)
// Create a new post policy.
@@ -605,12 +610,14 @@ func newPostRequestV2(endPoint, bucketName, objectName string, accessKey, secret
signature := calculateSignatureV2(encodedPolicy, secretKey)
formData := map[string]string{
"AWSAccessKeyId": accessKey,
"bucket": bucketName,
"key": objectName + "/${filename}",
"policy": encodedPolicy,
"signature": signature,
"X-Amz-Ignore-AWSAccessKeyId": "",
"bucket": bucketName,
"key": objectName + "/${filename}",
"policy": encodedPolicy,
"signature": signature,
}
for key, value := range formInputData {
formData[key] = value
}
// Create the multipart form.
+8 -2
View File
@@ -347,10 +347,16 @@ func checkPostPolicy(formValues http.Header, postPolicyForm PostPolicyForm) erro
}
delete(checkHeader, formCanonicalName)
}
// For SignV2 - Signature field will be ignored
// Policy is generated from Signature with other fields, so it should be ignored
// For SignV2 - Signature/AWSAccessKeyId field will be ignored.
if _, ok := formValues[xhttp.AmzSignatureV2]; ok {
delete(checkHeader, xhttp.AmzSignatureV2)
for k := range checkHeader {
// case-insensitivity for AWSAccessKeyId
if strings.EqualFold(k, xhttp.AmzAccessKeyID) {
delete(checkHeader, k)
break
}
}
}
if len(checkHeader) != 0 {
+127 -33
View File
@@ -18,6 +18,7 @@
package cmd
import (
"bytes"
"context"
"encoding/hex"
"errors"
@@ -28,6 +29,7 @@ import (
"net"
"os"
"os/signal"
"path/filepath"
"runtime"
"strings"
"syscall"
@@ -191,20 +193,25 @@ var ServerFlags = []cli.Flag{
EnvVar: "MINIO_RECV_BUF_SIZE",
Hidden: true,
},
}
var gatewayCmd = cli.Command{
Name: "gateway",
Usage: "start object storage gateway",
Hidden: true,
Flags: append(ServerFlags, GlobalFlags...),
HideHelpCommand: true,
Action: gatewayMain,
}
func gatewayMain(ctx *cli.Context) error {
logger.Fatal(errInvalidArgument, "Gateway is deprecated, To continue to use Gateway please use releases no later than 'RELEASE.2022-10-24T18-35-07Z'. We recommend all our users to migrate from gateway mode to server mode. Please read https://blog.min.io/deprecation-of-the-minio-gateway/")
return nil
cli.StringFlag{
Name: "log-dir",
Usage: "specify the directory to save the server log",
EnvVar: "MINIO_LOG_DIR",
Hidden: true,
},
cli.IntFlag{
Name: "log-size",
Usage: "specify the maximum server log file size in bytes before its rotated",
Value: 10 * humanize.MiByte,
EnvVar: "MINIO_LOG_SIZE",
Hidden: true,
},
cli.BoolFlag{
Name: "log-compress",
Usage: "specify if we want the rotated logs to be gzip compressed or not",
EnvVar: "MINIO_LOG_COMPRESS",
Hidden: true,
},
}
var serverCmd = cli.Command{
@@ -281,23 +288,7 @@ func serverCmdArgs(ctx *cli.Context) []string {
return strings.Fields(v)
}
func mergeServerCtxtFromConfigFile(configFile string, ctxt *serverCtxt) error {
rd, err := Open(configFile)
if err != nil {
return err
}
defer rd.Close()
cf := &config.ServerConfig{}
dec := yaml.NewDecoder(rd)
dec.SetStrict(true)
if err = dec.Decode(cf); err != nil {
return err
}
if cf.Version != "v1" {
return fmt.Errorf("unexpected version: %s", cf.Version)
}
func configCommonToSrvCtx(cf config.ServerConfigCommon, ctxt *serverCtxt) {
ctxt.RootUser = cf.RootUser
ctxt.RootPwd = cf.RootPwd
@@ -325,8 +316,75 @@ func mergeServerCtxtFromConfigFile(configFile string, ctxt *serverCtxt) error {
if cf.Options.SFTP.SSHPrivateKey != "" {
ctxt.SFTP = append(ctxt.SFTP, fmt.Sprintf("ssh-private-key=%s", cf.Options.SFTP.SSHPrivateKey))
}
}
ctxt.Layout, err = buildDisksLayoutFromConfFile(cf.Pools)
func mergeServerCtxtFromConfigFile(configFile string, ctxt *serverCtxt) error {
rd, err := xioutil.ReadFile(configFile)
if err != nil {
return err
}
cfReader := bytes.NewReader(rd)
cv := config.ServerConfigVersion{}
if err = yaml.Unmarshal(rd, &cv); err != nil {
return err
}
switch cv.Version {
case "v1", "v2":
default:
return fmt.Errorf("unexpected version: %s", cv.Version)
}
cfCommon := config.ServerConfigCommon{}
if err = yaml.Unmarshal(rd, &cfCommon); err != nil {
return err
}
configCommonToSrvCtx(cfCommon, ctxt)
v, err := env.GetInt(EnvErasureSetDriveCount, 0)
if err != nil {
return err
}
setDriveCount := uint64(v)
var pools []poolArgs
switch cv.Version {
case "v1":
cfV1 := config.ServerConfigV1{}
if err = yaml.Unmarshal(rd, &cfV1); err != nil {
return err
}
pools = make([]poolArgs, 0, len(cfV1.Pools))
for _, list := range cfV1.Pools {
pools = append(pools, poolArgs{
args: list,
setDriveCount: setDriveCount,
})
}
case "v2":
cf := config.ServerConfig{}
cfReader.Seek(0, io.SeekStart)
if err = yaml.Unmarshal(rd, &cf); err != nil {
return err
}
pools = make([]poolArgs, 0, len(cf.Pools))
for _, list := range cf.Pools {
driveCount := list.SetDriveCount
if setDriveCount > 0 {
driveCount = setDriveCount
}
pools = append(pools, poolArgs{
args: list.Args,
setDriveCount: driveCount,
})
}
}
ctxt.Layout, err = buildDisksLayoutFromConfFile(pools)
return err
}
@@ -667,6 +725,30 @@ func getServerListenAddrs() []string {
return addrs.ToSlice()
}
var globalLoggerOutput io.WriteCloser
func initializeLogRotate(ctx *cli.Context) (io.WriteCloser, error) {
lgDir := ctx.String("log-dir")
if lgDir == "" {
return os.Stderr, nil
}
lgDirAbs, err := filepath.Abs(lgDir)
if err != nil {
return nil, err
}
lgSize := ctx.Int("log-size")
output, err := logger.NewDir(logger.Options{
Directory: lgDirAbs,
MaximumFileSize: int64(lgSize),
Compress: ctx.Bool("log-compress"),
})
if err != nil {
return nil, err
}
logger.EnableJSON()
return output, nil
}
// serverMain handler called for 'minio server' command.
func serverMain(ctx *cli.Context) {
var warnings []string
@@ -679,11 +761,23 @@ func serverMain(ctx *cli.Context) {
// Initialize globalConsoleSys system
bootstrapTrace("newConsoleLogger", func() {
globalConsoleSys = NewConsoleLogger(GlobalContext)
output, err := initializeLogRotate(ctx)
if err == nil {
logger.Output = output
globalConsoleSys = NewConsoleLogger(GlobalContext, output)
globalLoggerOutput = output
} else {
logger.Output = os.Stderr
globalConsoleSys = NewConsoleLogger(GlobalContext, os.Stderr)
}
logger.AddSystemTarget(GlobalContext, globalConsoleSys)
// Set node name, only set for distributed setup.
globalConsoleSys.SetNodeName(globalLocalNodeName)
if err != nil {
// We can only log here since we need globalConsoleSys initialized
logger.Fatal(err, "invalid --logrorate-dir option")
}
})
// Always load ENV variables from files first.
+55 -1
View File
@@ -18,6 +18,7 @@
package cmd
import (
"bytes"
"context"
"crypto/subtle"
"errors"
@@ -162,6 +163,7 @@ func startSFTPServer(args []string) {
port int
publicIP string
sshPrivateKey string
userCaKeyFile string
)
allowPubKeys := supportedPubKeyAuthAlgos
allowKexAlgos := preferredKexAlgos
@@ -196,7 +198,9 @@ func startSFTPServer(args []string) {
case "cipher-algos":
allowCiphers = filterAlgos(arg, strings.Split(tokens[1], ","), supportedCiphers)
case "mac-algos":
allowCiphers = filterAlgos(arg, strings.Split(tokens[1], ","), supportedMACs)
allowMACs = filterAlgos(arg, strings.Split(tokens[1], ","), supportedMACs)
case "trusted-user-ca-key":
userCaKeyFile = tokens[1]
}
}
@@ -278,6 +282,56 @@ func startSFTPServer(args []string) {
},
}
if userCaKeyFile != "" {
keyBytes, err := os.ReadFile(userCaKeyFile)
if err != nil {
logger.Fatal(fmt.Errorf("invalid arguments passed, trusted user certificate authority public key file is not accessible: %v", err), "unable to start SFTP server")
}
caPublicKey, _, _, _, err := ssh.ParseAuthorizedKey(keyBytes)
if err != nil {
logger.Fatal(fmt.Errorf("invalid arguments passed, trusted user certificate authority public key file is not parseable: %v", err), "unable to start SFTP server")
}
sshConfig.PublicKeyCallback = func(c ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) {
_, ok := globalIAMSys.GetUser(context.Background(), c.User())
if !ok {
return nil, errNoSuchUser
}
// Verify that client provided certificate, not only public key.
cert, ok := key.(*ssh.Certificate)
if !ok {
return nil, errSftpPublicKeyWithoutCert
}
// ssh.CheckCert called by ssh.Authenticate accepts certificates
// with empty principles list so we block those in here.
if len(cert.ValidPrincipals) == 0 {
return nil, errSftpCertWithoutPrincipals
}
// Verify that certificate provided by user is issued by trusted CA,
// username in authentication request matches to identities in certificate
// and that certificate type is correct.
checker := ssh.CertChecker{}
checker.IsUserAuthority = func(k ssh.PublicKey) bool {
return bytes.Equal(k.Marshal(), caPublicKey.Marshal())
}
_, err = checker.Authenticate(c, key)
if err != nil {
return nil, err
}
return &ssh.Permissions{
CriticalOptions: map[string]string{
"accessKey": c.User(),
},
Extensions: make(map[string]string),
}, nil
}
}
sshConfig.AddHostKey(private)
handleSFTPSession := func(channel ssh.Channel, sconn *ssh.ServerConn) {
+4
View File
@@ -31,6 +31,10 @@ import (
func handleSignals() {
// Custom exit function
exit := func(success bool) {
if globalLoggerOutput != nil {
globalLoggerOutput.Close()
}
// If global profiler is set stop before we exit.
globalProfilerMu.Lock()
defer globalProfilerMu.Unlock()
+16 -9
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2015-2021 MinIO, Inc.
// Copyright (c) 2015-2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
@@ -27,6 +27,7 @@ import (
"github.com/minio/dperf/pkg/dperf"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
xioutil "github.com/minio/minio/internal/ioutil"
)
@@ -41,6 +42,8 @@ type speedTestOpts struct {
storageClass string
bucketName string
enableSha256 bool
enableMultipart bool
creds auth.Credentials
}
// Get the max throughput and iops numbers.
@@ -107,12 +110,14 @@ func objectSpeedTest(ctx context.Context, opts speedTestOpts) chan madmin.SpeedT
// if the default concurrency yields zero results, throw an error.
if throughputHighestResults[i].Downloads == 0 && opts.concurrencyStart == concurrency {
errStr = fmt.Sprintf("no results for downloads upon first attempt, concurrency %d and duration %s", opts.concurrencyStart, opts.duration)
errStr = fmt.Sprintf("no results for downloads upon first attempt, concurrency %d and duration %s",
opts.concurrencyStart, opts.duration)
}
// if the default concurrency yields zero results, throw an error.
if throughputHighestResults[i].Uploads == 0 && opts.concurrencyStart == concurrency {
errStr = fmt.Sprintf("no results for uploads upon first attempt, concurrency %d and duration %s", opts.concurrencyStart, opts.duration)
errStr = fmt.Sprintf("no results for uploads upon first attempt, concurrency %d and duration %s",
opts.concurrencyStart, opts.duration)
}
result.PUTStats.Servers = append(result.PUTStats.Servers, madmin.SpeedTestStatServer{
@@ -160,12 +165,14 @@ func objectSpeedTest(ctx context.Context, opts speedTestOpts) chan madmin.SpeedT
}
sopts := speedTestOpts{
objectSize: opts.objectSize,
concurrency: concurrency,
duration: opts.duration,
storageClass: opts.storageClass,
bucketName: opts.bucketName,
enableSha256: opts.enableSha256,
objectSize: opts.objectSize,
concurrency: concurrency,
duration: opts.duration,
storageClass: opts.storageClass,
bucketName: opts.bucketName,
enableSha256: opts.enableSha256,
enableMultipart: opts.enableMultipart,
creds: opts.creds,
}
results := globalNotificationSys.SpeedTest(ctx, sopts)
+3 -1
View File
@@ -929,7 +929,9 @@ func (sts *stsAPIHandlers) AssumeRoleWithCustomToken(w http.ResponseWriter, r *h
ctx := newContext(r, w, "AssumeRoleWithCustomToken")
claims := make(map[string]interface{})
defer logger.AuditLog(ctx, w, r, claims)
auditLogFilterKeys := []string{stsToken}
defer logger.AuditLog(ctx, w, r, claims, auditLogFilterKeys...)
if !globalIAMSys.Initialized() {
writeSTSErrorResponse(ctx, w, ErrSTSIAMNotInitialized, errIAMNotInitialized)
+22 -1
View File
@@ -109,7 +109,7 @@ func TestMain(m *testing.M) {
setMaxResources(nil)
// Initialize globalConsoleSys system
globalConsoleSys = NewConsoleLogger(context.Background())
globalConsoleSys = NewConsoleLogger(context.Background(), io.Discard)
globalInternodeTransport = NewInternodeHTTPTransport(0)()
@@ -217,6 +217,27 @@ func prepareErasure(ctx context.Context, nDisks int) (ObjectLayer, []string, err
return nil, nil, err
}
// Wait up to 10 seconds for disks to come online.
pools := obj.(*erasureServerPools)
t := time.Now()
for _, pool := range pools.serverPools {
for _, sets := range pool.erasureDisks {
for _, s := range sets {
if !s.IsLocal() {
for {
if s.IsOnline() {
break
}
time.Sleep(100 * time.Millisecond)
if time.Since(t) > 10*time.Second {
return nil, nil, errors.New("timeout waiting for disk to come online")
}
}
}
}
}
}
return obj, fsDirs, nil
}
+6
View File
@@ -119,3 +119,9 @@ var errInvalidMaxParts = errors.New("Part number is greater than the maximum all
// error returned for session policies > 2048
var errSessionPolicyTooLarge = errors.New("Session policy should not exceed 2048 characters")
// error returned in SFTP when user used public key without certificate
var errSftpPublicKeyWithoutCert = errors.New("public key authentication without certificate is not accepted")
// error returned in SFTP when user used certificate which does not contain principal(s)
var errSftpCertWithoutPrincipals = errors.New("certificates without principal(s) are not accepted")
+11 -9
View File
@@ -1127,16 +1127,18 @@ func ptr[T any](a T) *T {
return &a
}
func max(a, b int) int {
if a > b {
return a
// sleepContext sleeps for d duration or until ctx is done.
func sleepContext(ctx context.Context, d time.Duration) error {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(d):
}
return b
return nil
}
func min(a, b int) int {
if a < b {
return a
}
return b
// helper type to return either item or error.
type itemOrErr[V any] struct {
Item V
Err error
}
+62
View File
@@ -0,0 +1,62 @@
//go:build ignore
// +build ignore
// Copyright (c) 2015-2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package main
import (
"bytes"
"encoding/json"
"flag"
"fmt"
"io"
"log"
"net/http"
)
var port int
func init() {
flag.IntVar(&port, "port", 8080, "Port to listen on")
}
func mainHandler(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(r.Body)
defer r.Body.Close()
if err != nil {
log.Printf("Error reading request body: %v", err)
w.WriteHeader(http.StatusBadRequest)
return
}
log.Printf(">>> %s %s\n", r.Method, r.URL.Path)
var out bytes.Buffer
json.Indent(&out, body, "", " ")
log.Printf("%s\n", out.String())
w.WriteHeader(http.StatusOK)
}
func main() {
flag.Parse()
http.HandleFunc("/", mainHandler)
log.Printf("Listening on :%d\n", port)
log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", port), nil))
}
+17
View File
@@ -0,0 +1,17 @@
# `auditlog-echo`: A tool to view MinIO Audit logs on the console
1. Run the tool with:
```
go run docs/auditlog/auditlog-echo.go
```
The listen port has a default value (8080), but can be set with the `-port` flag.
2. Configure audit logging in MinIO with for example:
```
mc admin config set myminio audit_webhook enable=on endpoint=http://localhost:8080
```
3. Make any requests to MinIO and see audit logs printed to the tool's console.
+46 -23
View File
@@ -15,34 +15,57 @@ minio server --config config.yaml
Lets you start MinIO server with all inputs to start MinIO server provided via this configuration file, once the configuration file is provided all other pre-existing values on disk for configuration are overridden by the new values set in this configuration file.
Following is an example YAML configuration structure.
```
version: v1
address: ':9000'
rootUser: 'minioadmin'
rootPassword: 'pBU94AGAY85e'
console-address: ':9001'
certs-dir: '/home/user/.minio/certs/'
```yaml
version: v2
address: ":9000"
rootUser: "minioadmin"
rootPassword: "minioadmin"
console-address: ":9001"
certs-dir: "/home/user/.minio/certs/"
pools: # Specify the nodes and drives with pools
-
- 'https://server-example-pool1:9000/mnt/disk{1...4}/'
- 'https://server{1...2}-pool1:9000/mnt/disk{1...4}/'
- 'https://server3-pool1:9000/mnt/disk{1...4}/'
- 'https://server4-pool1:9000/mnt/disk{1...4}/'
-
- 'https://server-example-pool2:9000/mnt/disk{1...4}/'
- 'https://server{1...2}-pool2:9000/mnt/disk{1...4}/'
- 'https://server3-pool2:9000/mnt/disk{1...4}/'
- 'https://server4-pool2:9000/mnt/disk{1...4}/'
...
- args:
- "https://server-example-pool1:9000/mnt/disk{1...4}/"
- "https://server{1...2}-pool1:9000/mnt/disk{1...4}/"
- "https://server3-pool1:9000/mnt/disk{1...4}/"
- "https://server4-pool1:9000/mnt/disk{1...4}/"
- args:
- "https://server-example-pool2:9000/mnt/disk{1...4}/"
- "https://server{1...2}-pool2:9000/mnt/disk{1...4}/"
- "https://server3-pool2:9000/mnt/disk{1...4}/"
- "https://server4-pool2:9000/mnt/disk{1...4}/"
# more args
options:
ftp: # settings for MinIO to act as an ftp server
address: ':8021'
passive-port-range: '30000-40000'
address: ":8021"
passive-port-range: "30000-40000"
sftp: # settings for MinIO to act as an sftp server
address: ':8022'
ssh-private-key: '/home/user/.ssh/id_rsa'
address: ":8022"
ssh-private-key: "/home/user/.ssh/id_rsa"
```
If you are using the config `v1` YAML you should migrate your `pools:` field values to the following format
`v1` format
```yaml
pools: # Specify the nodes and drives with pools
-
- "https://server-example-pool1:9000/mnt/disk{1...4}/"
- "https://server{1...2}-pool1:9000/mnt/disk{1...4}/"
- "https://server3-pool1:9000/mnt/disk{1...4}/"
- "https://server4-pool1:9000/mnt/disk{1...4}/"
```
to `v2` format
```yaml
pools:
- args:
- "https://server-example-pool1:9000/mnt/disk{1...4}/"
- "https://server{1...2}-pool1:9000/mnt/disk{1...4}/"
- "https://server3-pool1:9000/mnt/disk{1...4}/"
- "https://server4-pool1:9000/mnt/disk{1...4}/"
set-drive-count: 4 # Advanced option, must be used under guidance from MinIO team.
```
### Things to know
+13
View File
@@ -242,3 +242,16 @@ hmac-sha1
hmac-sha1-96
```
### Certificate-based authentication
`--sftp=trusted-user-ca-key=...` specifies a file containing public key of certificate authority that is trusted
to sign user certificates for authentication.
Implementation is identical with "TrustedUserCAKeys" setting in OpenSSH server with exception that only one CA
key can be defined.
If a certificate is presented for authentication and has its signing CA key is in this file, then it may be
used for authentication for any user listed in the certificate's principals list.
Note that certificates that lack a list of principals will not be permitted for authentication using trusted-user-ca-key.
For more details on certificates, see the CERTIFICATES section in ssh-keygen(1).
+1 -1
View File
@@ -81,6 +81,6 @@ func mainHandler(w http.ResponseWriter, r *http.Request) {
func main() {
http.HandleFunc("/", mainHandler)
log.Print("Listing on :8081")
log.Print("Listening on :8081")
log.Fatal(http.ListenAndServe(":8081", nil))
}
+24 -9
View File
@@ -86,18 +86,18 @@ The standard metrics group for GoCollector is not shown below.
| `minio_api_requests_traffic_sent_bytes` | `counter` | Total number of bytes sent | `type,pool_index,server` |
| `minio_api_requests_traffic_received_bytes` | `counter` | Total number of bytes received | `type,pool_index,server` |
### `/api/bucket`
### `/bucket/api`
| Name | Type | Help | Labels |
|----------------------------------------------|-----------|------------------------------------------------------------------|-----------------------------------------|
| `minio_api_bucket_traffic_received_bytes` | `counter` | Total number of bytes sent for a bucket | `bucket,type,server,pool_index` |
| `minio_api_bucket_traffic_sent_bytes` | `counter` | Total number of bytes received for a bucket | `bucket,type,server,pool_index` |
| `minio_api_bucket_inflight_total` | `gauge` | Total number of requests currently in flight for a bucket | `bucket,name,type,server,pool_index` |
| `minio_api_bucket_total` | `counter` | Total number of requests for a bucket | `bucket,name,type,server,pool_index` |
| `minio_api_bucket_canceled_total` | `counter` | Total number of requests canceled by the client for a bucket | `bucket,name,type,server,pool_index` |
| `minio_api_bucket_4xx_errors_total` | `counter` | Total number of requests with 4xx errors for a bucket | `bucket,name,type,server,pool_index` |
| `minio_api_bucket_5xx_errors_total` | `counter` | Total number of requests with 5xx errors for a bucket | `bucket,name,type,server,pool_index` |
| `minio_api_bucket_ttfb_seconds_distribution` | `counter` | Distribution of time to first byte across API calls for a bucket | `bucket,name,le,type,server,pool_index` |
| `minio_bucket_api_traffic_received_bytes` | `counter` | Total number of bytes sent for a bucket | `bucket,type,server,pool_index` |
| `minio_bucket_api_traffic_sent_bytes` | `counter` | Total number of bytes received for a bucket | `bucket,type,server,pool_index` |
| `minio_bucket_api_inflight_total` | `gauge` | Total number of requests currently in flight for a bucket | `bucket,name,type,server,pool_index` |
| `minio_bucket_api_total` | `counter` | Total number of requests for a bucket | `bucket,name,type,server,pool_index` |
| `minio_bucket_api_canceled_total` | `counter` | Total number of requests canceled by the client for a bucket | `bucket,name,type,server,pool_index` |
| `minio_bucket_api_4xx_errors_total` | `counter` | Total number of requests with 4xx errors for a bucket | `bucket,name,type,server,pool_index` |
| `minio_bucket_api_5xx_errors_total` | `counter` | Total number of requests with 5xx errors for a bucket | `bucket,name,type,server,pool_index` |
| `minio_bucket_api_ttfb_seconds_distribution` | `counter` | Distribution of time to first byte across API calls for a bucket | `bucket,name,le,type,server,pool_index` |
### `/system/drive`
@@ -253,3 +253,18 @@ The standard metrics group for GoCollector is not shown below.
| `minio_cluster_notification_events_errors_total` | `counter` | Events that were failed to be sent to the targets | |
| `minio_cluster_notification_events_sent_total` | `counter` | Total number of events sent to the targets | |
| `minio_cluster_notification_events_skipped_total` | `counter` | Events that were skipped to be sent to the targets due to the in-memory queue being full | |
### `/cluster/iam`
| Name | Type | Help | Labels |
|-----------------------------------------------------------------|-----------|--------------------------------------------------------------------------------------------------------------------------|--------|
| `minio_cluster_iam_last_sync_duration_millis` | `counter` | Last successful IAM data sync duration in milliseconds | |
| `minio_cluster_iam_plugin_authn_service_failed_requests_minute` | `counter` | When plugin authentication is configured, returns failed requests count in the last full minute | |
| `minio_cluster_iam_plugin_authn_service_last_fail_seconds` | `counter` | When plugin authentication is configured, returns time (in seconds) since the last failed request to the service | |
| `minio_cluster_iam_plugin_authn_service_last_succ_seconds` | `counter` | When plugin authentication is configured, returns time (in seconds) since the last successful request to the service | |
| `minio_cluster_iam_plugin_authn_service_succ_avg_rtt_ms_minute` | `counter` | When plugin authentication is configured, returns average round-trip-time of successful requests in the last full minute | |
| `minio_cluster_iam_plugin_authn_service_succ_max_rtt_ms_minute` | `counter` | When plugin authentication is configured, returns maximum round-trip-time of successful requests in the last full minute | |
| `minio_cluster_iam_plugin_authn_service_total_requests_minute` | `counter` | When plugin authentication is configured, returns total requests count in the last full minute | |
| `minio_cluster_iam_since_last_sync_millis` | `counter` | Time (in milliseconds) since last successful IAM data sync | |
| `minio_cluster_iam_sync_failures` | `counter` | Number of failed IAM data syncs since server start | |
| `minio_cluster_iam_sync_successes` | `counter` | Number of successful IAM data syncs since server start | |
@@ -2,7 +2,7 @@ version: '3.7'
# Settings and configurations that are common for all containers
x-minio-common: &minio-common
image: quay.io/minio/minio:RELEASE.2024-04-28T17-53-50Z
image: quay.io/minio/minio:RELEASE.2024-05-01T01-11-10Z
command: server --console-address ":9001" http://minio{1...4}/data{1...2}
expose:
- "9000"
+4 -4
View File
@@ -45,13 +45,13 @@ require (
github.com/lithammer/shortuuid/v4 v4.0.0
github.com/miekg/dns v1.1.58
github.com/minio/cli v1.24.2
github.com/minio/console v1.3.0
github.com/minio/console v1.4.0
github.com/minio/csvparser v1.0.0
github.com/minio/dnscache v0.1.1
github.com/minio/dperf v0.5.3
github.com/minio/highwayhash v1.0.2
github.com/minio/kms-go/kes v0.3.0
github.com/minio/madmin-go/v3 v3.0.50
github.com/minio/madmin-go/v3 v3.0.51
github.com/minio/minio-go/v7 v7.0.70
github.com/minio/mux v1.9.0
github.com/minio/pkg/v2 v2.0.17
@@ -95,6 +95,7 @@ require (
golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f
golang.org/x/oauth2 v0.19.0
golang.org/x/sys v0.19.0
golang.org/x/term v0.19.0
golang.org/x/time v0.5.0
google.golang.org/api v0.172.0
gopkg.in/yaml.v2 v2.4.0
@@ -194,7 +195,7 @@ require (
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/minio/colorjson v1.0.6 // indirect
github.com/minio/filepath v1.0.0 // indirect
github.com/minio/mc v0.0.0-20240425223512-5dfaa31d67be // indirect
github.com/minio/mc v0.0.0-20240430174448-dcb911bed9d5 // indirect
github.com/minio/md5-simd v1.1.2 // indirect
github.com/minio/websocket v1.6.0 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
@@ -241,7 +242,6 @@ require (
golang.org/x/mod v0.17.0 // indirect
golang.org/x/net v0.24.0 // indirect
golang.org/x/sync v0.7.0 // indirect
golang.org/x/term v0.19.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/tools v0.20.0 // indirect
google.golang.org/genproto v0.0.0-20240415180920-8c6c420018be // indirect
+6 -6
View File
@@ -424,8 +424,8 @@ github.com/minio/cli v1.24.2 h1:J+fCUh9mhPLjN3Lj/YhklXvxj8mnyE/D6FpFduXJ2jg=
github.com/minio/cli v1.24.2/go.mod h1:bYxnK0uS629N3Bq+AOZZ+6lwF77Sodk4+UL9vNuXhOY=
github.com/minio/colorjson v1.0.6 h1:m7TUvpvt0u7FBmVIEQNIa0T4NBQlxrcMBp4wJKsg2Ik=
github.com/minio/colorjson v1.0.6/go.mod h1:LUXwS5ZGNb6Eh9f+t+3uJiowD3XsIWtsvTriUBeqgYs=
github.com/minio/console v1.3.0 h1:iQyCHz6B42gIb/vDWKaFDYeBpaB9nJU8kqf06QAxYDY=
github.com/minio/console v1.3.0/go.mod h1:y5QbdoppH6KqOY2KlV/DOo5BthDu1yOLhwIYcmhAIag=
github.com/minio/console v1.4.0 h1:WqZMFWQRnUtpumgdgXLpPXenz4hHB1E3SZydiMbv7jY=
github.com/minio/console v1.4.0/go.mod h1:keSmHo7VWg6NxdJSh1/S0+GdBxtQkqxWgs/rSqL8i8w=
github.com/minio/csvparser v1.0.0 h1:xJEHcYK8ZAjeW4hNV9Zu30u+/2o4UyPnYgyjWp8b7ZU=
github.com/minio/csvparser v1.0.0/go.mod h1:lKXskSLzPgC5WQyzP7maKH7Sl1cqvANXo9YCto8zbtM=
github.com/minio/dnscache v0.1.1 h1:AMYLqomzskpORiUA1ciN9k7bZT1oB3YZN4cEIi88W5o=
@@ -438,10 +438,10 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
github.com/minio/kms-go/kes v0.3.0 h1:SU8VGVM/Hk9w1OiSby3OatkcojooUqIdDHl6dtM6NkY=
github.com/minio/kms-go/kes v0.3.0/go.mod h1:w6DeVT878qEOU3nUrYVy1WOT5H1Ig9hbDIh698NYJKY=
github.com/minio/madmin-go/v3 v3.0.50 h1:+RQMetVFvPQmAOEDN/xmLhwk9+xOzu3rqwnlZEskgvg=
github.com/minio/madmin-go/v3 v3.0.50/go.mod h1:ZDF7kf5fhmxLhbGTqyq5efs4ao0v4eWf7nOuef/ljJs=
github.com/minio/mc v0.0.0-20240425223512-5dfaa31d67be h1:HobtnPBvp53b57oT+yV8VkrRBVICMX2UFhGG/Ch4KTw=
github.com/minio/mc v0.0.0-20240425223512-5dfaa31d67be/go.mod h1:aOiBeSNmpfJn1yyz+EujrTM+XmUwkXiM69zSXg12VDM=
github.com/minio/madmin-go/v3 v3.0.51 h1:brGOvDP8KvoHb/bdzCHUPFCbTtrN8o507uPHZpyuinM=
github.com/minio/madmin-go/v3 v3.0.51/go.mod h1:IFAwr0XMrdsLovxAdCcuq/eoL4nRuMVQQv0iubJANQw=
github.com/minio/mc v0.0.0-20240430174448-dcb911bed9d5 h1:VDXLzvY0Jxk4lzIntGXZuw0VH7S1JgQBmjWGkz7xphU=
github.com/minio/mc v0.0.0-20240430174448-dcb911bed9d5/go.mod h1:aOiBeSNmpfJn1yyz+EujrTM+XmUwkXiM69zSXg12VDM=
github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
github.com/minio/minio-go/v6 v6.0.46/go.mod h1:qD0lajrGW49lKZLtXKtCB4X/qkMf0a5tBvN2PaZg7Gg=
+8
View File
@@ -156,6 +156,14 @@ func (cred Credentials) IsServiceAccount() bool {
return cred.ParentUser != "" && ok
}
// IsImpliedPolicy - returns if the policy is implied via ParentUser or not.
func (cred Credentials) IsImpliedPolicy() bool {
if cred.IsServiceAccount() {
return cred.Claims[iamPolicyClaimNameSA] == "inherited-policy"
}
return false
}
// IsValid - returns whether credential is valid or not.
func (cred Credentials) IsValid() bool {
// Verify credentials if its enabled or not set.
+8
View File
@@ -149,4 +149,12 @@ var (
}
return fmt.Sprintf
}()
TurnOff = func() {
color.NoColor = true
}
TurnOn = func() {
color.NoColor = false
}
)
+28 -8
View File
@@ -29,14 +29,34 @@ type Opts struct {
} `yaml:"sftp"`
}
// ServerConfigVersion struct is used to extract the version
type ServerConfigVersion struct {
Version string `yaml:"version"`
}
// ServerConfigCommon struct for server config common options
type ServerConfigCommon struct {
RootUser string `yaml:"rootUser"`
RootPwd string `yaml:"rootPassword"`
Addr string `yaml:"address"`
ConsoleAddr string `yaml:"console-address"`
CertsDir string `yaml:"certs-dir"`
Options Opts `yaml:"options"`
}
// ServerConfigV1 represents a MinIO configuration file v1
type ServerConfigV1 struct {
ServerConfigVersion
ServerConfigCommon
Pools [][]string `yaml:"pools"`
}
// ServerConfig represents a MinIO configuration file
type ServerConfig struct {
Version string `yaml:"version"`
RootUser string `yaml:"rootUser"`
RootPwd string `yaml:"rootPassword"`
Addr string `yaml:"address"`
ConsoleAddr string `yaml:"console-address"`
CertsDir string `yaml:"certs-dir"`
Pools [][]string `yaml:"pools"`
Options Opts `yaml:"options"`
ServerConfigVersion
ServerConfigCommon
Pools []struct {
Args []string `yaml:"args"`
SetDriveCount uint64 `yaml:"set-drive-count"`
} `yaml:"pools"`
}
+10 -12
View File
@@ -25,7 +25,6 @@ import (
"time"
"github.com/minio/minio/internal/color"
c "github.com/minio/pkg/v2/console"
"github.com/minio/pkg/v2/logger/message/log"
)
@@ -99,8 +98,7 @@ func (f fatalMsg) json(msg string, args ...interface{}) {
if err != nil {
panic(err)
}
fmt.Println(string(logJSON))
fmt.Fprintln(Output, string(logJSON))
ExitFunc(1)
}
@@ -139,16 +137,16 @@ func (f fatalMsg) pretty(msg string, args ...interface{}) {
ansiSaveAttributes()
// Print banner with or without the log tag
if !tagPrinted {
c.Print(logBanner)
fmt.Fprint(Output, logBanner)
tagPrinted = true
} else {
c.Print(emptyBanner)
fmt.Fprint(Output, emptyBanner)
}
// Restore the text color of the error message
ansiRestoreAttributes()
ansiMoveRight(bannerWidth)
// Continue error message printing
c.Println(line)
fmt.Fprintln(Output, line)
break
}
}
@@ -176,7 +174,7 @@ func (i infoMsg) json(msg string, args ...interface{}) {
if err != nil {
panic(err)
}
fmt.Println(string(logJSON))
fmt.Fprintln(Output, string(logJSON))
}
func (i infoMsg) quiet(msg string, args ...interface{}) {
@@ -184,9 +182,9 @@ func (i infoMsg) quiet(msg string, args ...interface{}) {
func (i infoMsg) pretty(msg string, args ...interface{}) {
if msg == "" {
c.Println(args...)
fmt.Fprintln(Output, args...)
}
c.Printf(msg, args...)
fmt.Fprintf(Output, msg, args...)
}
type errorMsg struct{}
@@ -209,7 +207,7 @@ func (i errorMsg) json(msg string, args ...interface{}) {
if err != nil {
panic(err)
}
fmt.Println(string(logJSON))
fmt.Fprintln(Output, string(logJSON))
}
func (i errorMsg) quiet(msg string, args ...interface{}) {
@@ -218,9 +216,9 @@ func (i errorMsg) quiet(msg string, args ...interface{}) {
func (i errorMsg) pretty(msg string, args ...interface{}) {
if msg == "" {
c.Println(args...)
fmt.Fprintln(Output, args...)
}
c.Printf(msg, args...)
fmt.Fprintf(Output, msg, args...)
}
// Error :
+11 -2
View File
@@ -23,6 +23,8 @@ import (
"errors"
"fmt"
"go/build"
"io"
"os"
"path/filepath"
"reflect"
"runtime"
@@ -32,6 +34,7 @@ import (
"github.com/minio/highwayhash"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/color"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/pkg/v2/logger/message/log"
)
@@ -49,8 +52,12 @@ const (
InfoKind = madmin.LogKindInfo
)
// DisableErrorLog avoids printing error/event/info kind of logs
var DisableErrorLog = false
var (
// DisableErrorLog avoids printing error/event/info kind of logs
DisableErrorLog = false
// Output allows configuring custom writer, defaults to os.Stderr
Output io.Writer = os.Stderr
)
var trimStrings []string
@@ -73,11 +80,13 @@ var (
// EnableQuiet - turns quiet option on.
func EnableQuiet() {
color.TurnOff() // no colored outputs necessary in quiet mode.
quietFlag = true
}
// EnableJSON - outputs logs in json format.
func EnableJSON() {
color.TurnOff() // no colored outputs necessary in JSON mode.
jsonFlag = true
quietFlag = true
}
+240
View File
@@ -0,0 +1,240 @@
// Copyright (c) 2015-2024 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package logger
import (
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"time"
"github.com/klauspost/compress/gzip"
xioutil "github.com/minio/minio/internal/ioutil"
"github.com/minio/pkg/v2/logger/message/log"
)
func defaultFilenameFunc() string {
return fmt.Sprintf("minio-%s.log", fmt.Sprintf("%X", time.Now().UTC().UnixNano()))
}
// Options define configuration options for Writer
type Options struct {
// Directory defines the directory where log files will be written to.
// If the directory does not exist, it will be created.
Directory string
// MaximumFileSize defines the maximum size of each log file in bytes.
MaximumFileSize int64
// FileNameFunc specifies the name a new file will take.
// FileNameFunc must ensure collisions in filenames do not occur.
// Do not rely on timestamps to be unique, high throughput writes
// may fall on the same timestamp.
// Eg.
// 2020-03-28_15-00-945-<random-hash>.log
// When FileNameFunc is not specified, DefaultFilenameFunc will be used.
FileNameFunc func() string
// Compress specify if you want the logs to be compressed after rotation.
Compress bool
}
// Writer is a concurrency-safe writer with file rotation.
type Writer struct {
// opts are the configuration options for this Writer
opts Options
// f is the currently open file used for appends.
// Writes to f are only synchronized once Close() is called,
// or when files are being rotated.
f *os.File
pw *xioutil.PipeWriter
pr *xioutil.PipeReader
}
// Write writes p into the current file, rotating if necessary.
// Write is non-blocking, if the writer's queue is not full.
// Write is blocking otherwise.
func (w *Writer) Write(p []byte) (n int, err error) {
return w.pw.Write(p)
}
// Close closes the writer.
// Any accepted writes will be flushed. Any new writes will be rejected.
// Once Close() exits, files are synchronized to disk.
func (w *Writer) Close() error {
w.pw.CloseWithError(nil)
if w.f != nil {
if err := w.closeCurrentFile(); err != nil {
return err
}
}
return nil
}
var stdErrEnc = json.NewEncoder(os.Stderr)
func (w *Writer) listen() {
for {
var r io.Reader = w.pr
if w.opts.MaximumFileSize > 0 {
r = io.LimitReader(w.pr, w.opts.MaximumFileSize)
}
if _, err := io.Copy(w.f, r); err != nil {
msg := fmt.Sprintf("unable to write to log file %v: %v", w.f.Name(), err)
stdErrEnc.Encode(&log.Entry{
Level: ErrorKind,
Message: msg,
Time: time.Now().UTC(),
Trace: &log.Trace{Message: msg},
})
}
if err := w.rotate(); err != nil {
msg := fmt.Sprintf("unable to rotate log file %v: %v", w.f.Name(), err)
stdErrEnc.Encode(&log.Entry{
Level: ErrorKind,
Message: msg,
Time: time.Now().UTC(),
Trace: &log.Trace{Message: msg},
})
}
}
}
func (w *Writer) closeCurrentFile() error {
if err := w.f.Close(); err != nil {
return fmt.Errorf("unable to close current log file: %w", err)
}
return nil
}
func (w *Writer) compress() error {
if !w.opts.Compress {
return nil
}
oldLgFile := w.f.Name()
r, err := os.Open(oldLgFile)
if err != nil {
return err
}
defer r.Close()
gw, err := os.Create(oldLgFile + ".gz")
if err != nil {
return err
}
defer gw.Close()
var wc io.WriteCloser
wc = gzip.NewWriter(gw)
if _, err = io.Copy(wc, r); err != nil {
return err
}
if err = wc.Close(); err != nil {
return err
}
// Persist to disk any caches.
if err = gw.Sync(); err != nil {
return err
}
// close everything before we delete.
if err = gw.Close(); err != nil {
return err
}
if err = r.Close(); err != nil {
return err
}
// Attempt to remove after all fd's are closed.
return os.Remove(oldLgFile)
}
func (w *Writer) rotate() error {
if w.f != nil {
if err := w.closeCurrentFile(); err != nil {
return err
}
// This function is a no-op if opts.Compress is false
// writes an error in JSON form to stderr, if we cannot
// compress.
if err := w.compress(); err != nil {
msg := fmt.Sprintf("unable to compress log file %v: %v, ignoring and moving on", w.f.Name(), err)
stdErrEnc.Encode(&log.Entry{
Level: ErrorKind,
Message: msg,
Time: time.Now().UTC(),
Trace: &log.Trace{Message: msg},
})
}
}
path := filepath.Join(w.opts.Directory, w.opts.FileNameFunc())
f, err := newFile(path)
if err != nil {
return fmt.Errorf("unable to create new file at %v: %w", path, err)
}
w.f = f
return nil
}
// NewDir creates a new concurrency safe Writer which performs log rotation.
func NewDir(opts Options) (io.WriteCloser, error) {
if err := os.MkdirAll(opts.Directory, os.ModePerm); err != nil {
return nil, fmt.Errorf("directory %v does not exist and could not be created: %w", opts.Directory, err)
}
if opts.FileNameFunc == nil {
opts.FileNameFunc = defaultFilenameFunc
}
pr, pw := xioutil.WaitPipe()
w := &Writer{
opts: opts,
pw: pw,
pr: pr,
}
if w.f == nil {
if err := w.rotate(); err != nil {
return nil, fmt.Errorf("Failed to create log file: %w", err)
}
}
go w.listen()
return w, nil
}
func newFile(path string) (*os.File, error) {
return os.OpenFile(path, os.O_WRONLY|os.O_TRUNC|os.O_CREATE|os.O_SYNC, 0o666)
}
+9 -7
View File
@@ -20,18 +20,20 @@ package console
import (
"encoding/json"
"fmt"
"io"
"strconv"
"strings"
"github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/v2/console"
"github.com/minio/pkg/v2/logger/message/log"
)
// Target implements loggerTarget to send log
// in plain or json format to the standard output.
type Target struct{}
type Target struct {
output io.Writer
}
// Validate - validate if the tty can be written to
func (c *Target) Validate() error {
@@ -58,12 +60,12 @@ func (c *Target) Send(e interface{}) error {
if err != nil {
return err
}
fmt.Println(string(logJSON))
fmt.Fprintln(c.output, string(logJSON))
return nil
}
if entry.Level == logger.EventKind {
fmt.Println(entry.Message)
fmt.Fprintln(c.output, entry.Message)
return nil
}
@@ -146,13 +148,13 @@ func (c *Target) Send(e interface{}) error {
apiString, timeString, deploymentID, requestID, remoteHost, host, userAgent,
msg, tagString, strings.Join(trace, "\n"))
console.Println(output)
fmt.Fprintln(c.output, output)
return nil
}
// New initializes a new logger target
// which prints log directly in the standard
// output.
func New() *Target {
return &Target{}
func New(w io.Writer) *Target {
return &Target{output: w}
}