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

audit: per object send pool number, set number and servers per operation (#11233)

Browse Source
This commit is contained in:
Anis Elleuch
2021-01-26 22:21:51 +01:00
committed by GitHub
parent 9722531817
commit 00cff1aac5
30 changed files with 275 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
cmd/erasure-sets.go
View File

@@ -83,6 +83,8 @@ type erasureSets struct {
setCount, setDriveCount int
defaultParityCount int
poolNumber int
disksConnectEvent chan diskConnectInfo
// Distribution algorithm of choice.
@@ -411,6 +413,7 @@ func newErasureSets(ctx context.Context, endpoints Endpoints, storageDisks []Sto
// Initialize erasure objects for a given set.
s.sets[i] = &erasureObjects{
setNumber: i,
setDriveCount: setDriveCount,
defaultParityCount: defaultParityCount,
getDisks: s.GetDisks(i),
@@ -449,6 +452,44 @@ func (s *erasureSets) cleanupStaleUploads(ctx context.Context, cleanupInterval,
}
}
const objectErasureMapKey = "objectErasureMap"
type auditObjectOp struct {
Pool int `json:"pool"`
Set int `json:"set"`
BackendServers []string `json:"backendServers"`
}
func auditObjectErasureSet(ctx context.Context, object string, set *erasureObjects, poolNum int) {
if len(logger.AuditTargets) == 0 {
return
}
object = decodeDirObject(object)
op := auditObjectOp{
Pool: poolNum + 1,
Set: set.setNumber + 1,
BackendServers: set.getEndpoints(),
}
var objectErasureSetTag map[string]auditObjectOp
reqInfo := logger.GetReqInfo(ctx)
for _, kv := range reqInfo.GetTags() {
if kv.Key == objectErasureMapKey {
objectErasureSetTag = kv.Val.(map[string]auditObjectOp)
break
}
}
if objectErasureSetTag == nil {
objectErasureSetTag = make(map[string]auditObjectOp)
}
objectErasureSetTag[object] = op
reqInfo.SetTags(objectErasureMapKey, objectErasureSetTag)
}
// NewNSLock - initialize a new namespace RWLocker instance.
func (s *erasureSets) NewNSLock(bucket string, objects ...string) RWLocker {
if len(objects) == 1 {
@@ -744,12 +785,16 @@ func (s *erasureSets) ListBuckets(ctx context.Context) (buckets []BucketInfo, er
// GetObjectNInfo - returns object info and locked object ReadCloser
func (s *erasureSets) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) {
return s.getHashedSet(object).GetObjectNInfo(ctx, bucket, object, rs, h, lockType, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.GetObjectNInfo(ctx, bucket, object, rs, h, lockType, opts)
}
// GetObject - reads an object from the hashedSet based on the object name.
func (s *erasureSets) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error {
return s.getHashedSet(object).GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts)
}
func (s *erasureSets) parentDirIsObject(ctx context.Context, bucket, parent string) bool {
@@ -761,18 +806,24 @@ func (s *erasureSets) parentDirIsObject(ctx context.Context, bucket, parent stri
// PutObject - writes an object to hashedSet based on the object name.
func (s *erasureSets) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) {
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
opts.ParentIsObject = s.parentDirIsObject
return s.getHashedSet(object).PutObject(ctx, bucket, object, data, opts)
return set.PutObject(ctx, bucket, object, data, opts)
}
// GetObjectInfo - reads object metadata from the hashedSet based on the object name.
func (s *erasureSets) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
return s.getHashedSet(object).GetObjectInfo(ctx, bucket, object, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.GetObjectInfo(ctx, bucket, object, opts)
}
// DeleteObject - deletes an object from the hashedSet based on the object name.
func (s *erasureSets) DeleteObject(ctx context.Context, bucket string, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
return s.getHashedSet(object).DeleteObject(ctx, bucket, object, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.DeleteObject(ctx, bucket, object, opts)
}
// DeleteObjects - bulk delete of objects
@@ -817,10 +868,14 @@ func (s *erasureSets) DeleteObjects(ctx context.Context, bucket string, objects
// Invoke bulk delete on objects per set and save
// the result of the delete operation
for _, objsGroup := range objSetMap {
dobjects, errs := s.getHashedSet(objsGroup[0].object.ObjectName).DeleteObjects(ctx, bucket, toNames(objsGroup), opts)
set := s.getHashedSet(objsGroup[0].object.ObjectName)
dobjects, errs := set.DeleteObjects(ctx, bucket, toNames(objsGroup), opts)
for i, obj := range objsGroup {
delErrs[obj.origIndex] = errs[i]
delObjects[obj.origIndex] = dobjects[i]
if errs[i] == nil {
auditObjectErasureSet(ctx, obj.object.ObjectName, set, s.poolNumber)
}
}
}
@@ -832,6 +887,8 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
srcSet := s.getHashedSet(srcObject)
dstSet := s.getHashedSet(dstObject)
auditObjectErasureSet(ctx, dstObject, dstSet, s.poolNumber)
cpSrcDstSame := srcSet == dstSet
// Check if this request is only metadata update.
if cpSrcDstSame && srcInfo.metadataOnly {
@@ -1013,46 +1070,60 @@ func (s *erasureSets) startMergeWalksVersionsN(ctx context.Context, bucket, pref
func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
// In list multipart uploads we are going to treat input prefix as the object,
// this means that we are not supporting directory navigation.
return s.getHashedSet(prefix).ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
set := s.getHashedSet(prefix)
auditObjectErasureSet(ctx, prefix, set, s.poolNumber)
return set.ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
}
// Initiate a new multipart upload on a hashedSet based on object name.
func (s *erasureSets) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) {
return s.getHashedSet(object).NewMultipartUpload(ctx, bucket, object, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.NewMultipartUpload(ctx, bucket, object, opts)
}
// Copies a part of an object from source hashedSet to destination hashedSet.
func (s *erasureSets) CopyObjectPart(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, uploadID string, partID int,
startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (partInfo PartInfo, err error) {
destSet := s.getHashedSet(destObject)
auditObjectErasureSet(ctx, destObject, destSet, s.poolNumber)
return destSet.PutObjectPart(ctx, destBucket, destObject, uploadID, partID, NewPutObjReader(srcInfo.Reader, nil, nil), dstOpts)
}
// PutObjectPart - writes part of an object to hashedSet based on the object name.
func (s *erasureSets) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) {
return s.getHashedSet(object).PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts)
}
// GetMultipartInfo - return multipart metadata info uploaded at hashedSet.
func (s *erasureSets) GetMultipartInfo(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (result MultipartInfo, err error) {
return s.getHashedSet(object).GetMultipartInfo(ctx, bucket, object, uploadID, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.GetMultipartInfo(ctx, bucket, object, uploadID, opts)
}
// ListObjectParts - lists all uploaded parts to an object in hashedSet.
func (s *erasureSets) ListObjectParts(ctx context.Context, bucket, object, uploadID string, partNumberMarker int, maxParts int, opts ObjectOptions) (result ListPartsInfo, err error) {
return s.getHashedSet(object).ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
}
// Aborts an in-progress multipart operation on hashedSet based on the object name.
func (s *erasureSets) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error {
return s.getHashedSet(object).AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
return set.AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
}
// CompleteMultipartUpload - completes a pending multipart transaction, on hashedSet based on object name.
func (s *erasureSets) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) {
set := s.getHashedSet(object)
auditObjectErasureSet(ctx, object, set, s.poolNumber)
opts.ParentIsObject = s.parentDirIsObject
return s.getHashedSet(object).CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
return set.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
}
/*
@@ -1328,17 +1399,20 @@ func (s *erasureSets) HealObject(ctx context.Context, bucket, object, versionID
// PutObjectTags - replace or add tags to an existing object
func (s *erasureSets) PutObjectTags(ctx context.Context, bucket, object string, tags string, opts ObjectOptions) error {
return s.getHashedSet(object).PutObjectTags(ctx, bucket, object, tags, opts)
er := s.getHashedSet(object)
return er.PutObjectTags(ctx, bucket, object, tags, opts)
}
// DeleteObjectTags - delete object tags from an existing object
func (s *erasureSets) DeleteObjectTags(ctx context.Context, bucket, object string, opts ObjectOptions) error {
return s.getHashedSet(object).DeleteObjectTags(ctx, bucket, object, opts)
er := s.getHashedSet(object)
return er.DeleteObjectTags(ctx, bucket, object, opts)
}
// GetObjectTags - get object tags from an existing object
func (s *erasureSets) GetObjectTags(ctx context.Context, bucket, object string, opts ObjectOptions) (*tags.Tags, error) {
return s.getHashedSet(object).GetObjectTags(ctx, bucket, object, opts)
er := s.getHashedSet(object)
return er.GetObjectTags(ctx, bucket, object, opts)
}
// maintainMRFList gathers the list of successful partial uploads