mirror of
https://github.com/minio/minio.git
synced 2024-12-24 06:05:55 -05:00
fix: Get/HeadObject return 404 on non quorum objects (#10753)
This commit is contained in:
parent
029758cb20
commit
eb95353cb1
@ -67,7 +67,7 @@ func TestHealing(t *testing.T) {
|
||||
}
|
||||
|
||||
disk := er.getDisks()[0]
|
||||
fileInfoPreHeal, err := disk.ReadVersion(context.Background(), bucket, object, "")
|
||||
fileInfoPreHeal, err := disk.ReadVersion(context.Background(), bucket, object, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -84,7 +84,7 @@ func TestHealing(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fileInfoPostHeal, err := disk.ReadVersion(context.Background(), bucket, object, "")
|
||||
fileInfoPostHeal, err := disk.ReadVersion(context.Background(), bucket, object, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -113,7 +113,7 @@ func TestHealing(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
fileInfoPostHeal, err = disk.ReadVersion(context.Background(), bucket, object, "")
|
||||
fileInfoPostHeal, err = disk.ReadVersion(context.Background(), bucket, object, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
@ -113,9 +113,21 @@ func hashOrder(key string, cardinality int) []int {
|
||||
return nums
|
||||
}
|
||||
|
||||
// Reads all `xl.meta` metadata as a FileInfo slice and checks if the data dir exists as well,
|
||||
// otherwise returns errFileNotFound (or errFileVersionNotFound)
|
||||
func getAllObjectFileInfo(ctx context.Context, disks []StorageAPI, bucket, object, versionID string) ([]FileInfo, []error) {
|
||||
return readVersionFromDisks(ctx, disks, bucket, object, versionID, true)
|
||||
}
|
||||
|
||||
// Reads all `xl.meta` metadata as a FileInfo slice.
|
||||
// Returns error slice indicating the failed metadata reads.
|
||||
func readAllFileInfo(ctx context.Context, disks []StorageAPI, bucket, object, versionID string) ([]FileInfo, []error) {
|
||||
return readVersionFromDisks(ctx, disks, bucket, object, versionID, false)
|
||||
}
|
||||
|
||||
// Reads all `xl.meta` metadata as a FileInfo slice and checks if the data dir
|
||||
// exists as well, if checkDataDir is set to true.
|
||||
func readVersionFromDisks(ctx context.Context, disks []StorageAPI, bucket, object, versionID string, checkDataDir bool) ([]FileInfo, []error) {
|
||||
metadataArray := make([]FileInfo, len(disks))
|
||||
|
||||
g := errgroup.WithNErrs(len(disks))
|
||||
@ -126,7 +138,7 @@ func readAllFileInfo(ctx context.Context, disks []StorageAPI, bucket, object, ve
|
||||
if disks[index] == nil {
|
||||
return errDiskNotFound
|
||||
}
|
||||
metadataArray[index], err = disks[index].ReadVersion(ctx, bucket, object, versionID)
|
||||
metadataArray[index], err = disks[index].ReadVersion(ctx, bucket, object, versionID, checkDataDir)
|
||||
if err != nil {
|
||||
if err != errFileNotFound && err != errVolumeNotFound && err != errFileVersionNotFound {
|
||||
logger.GetReqInfo(ctx).AppendTags("disk", disks[index].String())
|
||||
|
@ -43,7 +43,25 @@ func (er erasureObjects) getMultipartSHADir(bucket, object string) string {
|
||||
|
||||
// checkUploadIDExists - verify if a given uploadID exists and is valid.
|
||||
func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object, uploadID string) error {
|
||||
_, _, _, err := er.getObjectFileInfo(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID), ObjectOptions{})
|
||||
disks := er.getDisks()
|
||||
|
||||
// Read metadata associated with the object from all disks.
|
||||
metaArr, errs := readAllFileInfo(ctx, disks, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID), "")
|
||||
|
||||
readQuorum, _, err := objectQuorumFromMeta(ctx, er, metaArr, errs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if reducedErr := reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum); reducedErr != nil {
|
||||
return toObjectErr(reducedErr, bucket, object)
|
||||
}
|
||||
|
||||
// List all online disks.
|
||||
_, modTime := listOnlineDisks(disks, metaArr, errs)
|
||||
|
||||
// Pick latest valid metadata.
|
||||
_, err = pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
|
||||
return err
|
||||
}
|
||||
|
||||
@ -110,7 +128,7 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
|
||||
}
|
||||
for _, uploadIDDir := range uploadIDDirs {
|
||||
uploadIDPath := pathJoin(shaDir, uploadIDDir)
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaMultipartBucket, uploadIDPath, "")
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaMultipartBucket, uploadIDPath, "", false)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
@ -124,7 +142,7 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
|
||||
return
|
||||
}
|
||||
for _, tmpDir := range tmpDirs {
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaTmpBucket, tmpDir, "")
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaTmpBucket, tmpDir, "", false)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
@ -178,7 +196,7 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, objec
|
||||
if populatedUploadIds.Contains(uploadID) {
|
||||
continue
|
||||
}
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaMultipartBucket, pathJoin(er.getUploadIDDir(bucket, object, uploadID)), "")
|
||||
fi, err := disk.ReadVersion(ctx, minioMetaMultipartBucket, pathJoin(er.getUploadIDDir(bucket, object, uploadID)), "", false)
|
||||
if err != nil {
|
||||
return result, toObjectErr(err, bucket, object)
|
||||
}
|
||||
|
@ -355,7 +355,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
|
||||
disks := er.getDisks()
|
||||
|
||||
// Read metadata associated with the object from all disks.
|
||||
metaArr, errs := readAllFileInfo(ctx, disks, bucket, object, opts.VersionID)
|
||||
metaArr, errs := getAllObjectFileInfo(ctx, disks, bucket, object, opts.VersionID)
|
||||
|
||||
readQuorum, _, err := objectQuorumFromMeta(ctx, er, metaArr, errs)
|
||||
if err != nil {
|
||||
|
@ -259,11 +259,11 @@ func (d *naughtyDisk) DeleteVersion(ctx context.Context, volume, path string, fi
|
||||
return d.disk.DeleteVersion(ctx, volume, path, fi)
|
||||
}
|
||||
|
||||
func (d *naughtyDisk) ReadVersion(ctx context.Context, volume, path, versionID string) (fi FileInfo, err error) {
|
||||
func (d *naughtyDisk) ReadVersion(ctx context.Context, volume, path, versionID string, checkDataDir bool) (fi FileInfo, err error) {
|
||||
if err := d.calcError(); err != nil {
|
||||
return FileInfo{}, err
|
||||
}
|
||||
return d.disk.ReadVersion(ctx, volume, path, versionID)
|
||||
return d.disk.ReadVersion(ctx, volume, path, versionID, checkDataDir)
|
||||
}
|
||||
|
||||
func (d *naughtyDisk) WriteAll(ctx context.Context, volume string, path string, reader io.Reader) (err error) {
|
||||
|
@ -59,7 +59,7 @@ type StorageAPI interface {
|
||||
DeleteVersion(ctx context.Context, volume, path string, fi FileInfo) error
|
||||
DeleteVersions(ctx context.Context, volume string, versions []FileInfo) []error
|
||||
WriteMetadata(ctx context.Context, volume, path string, fi FileInfo) error
|
||||
ReadVersion(ctx context.Context, volume, path, versionID string) (FileInfo, error)
|
||||
ReadVersion(ctx context.Context, volume, path, versionID string, checkDataDir bool) (FileInfo, error)
|
||||
RenameData(ctx context.Context, srcVolume, srcPath, dataDir, dstVolume, dstPath string) error
|
||||
|
||||
// File operations.
|
||||
|
@ -366,11 +366,12 @@ func (client *storageRESTClient) RenameData(ctx context.Context, srcVolume, srcP
|
||||
return err
|
||||
}
|
||||
|
||||
func (client *storageRESTClient) ReadVersion(ctx context.Context, volume, path, versionID string) (fi FileInfo, err error) {
|
||||
func (client *storageRESTClient) ReadVersion(ctx context.Context, volume, path, versionID string, checkDataDir bool) (fi FileInfo, err error) {
|
||||
values := make(url.Values)
|
||||
values.Set(storageRESTVolume, volume)
|
||||
values.Set(storageRESTFilePath, path)
|
||||
values.Set(storageRESTVersionID, versionID)
|
||||
values.Set(storageRESTCheckDataDir, strconv.FormatBool(checkDataDir))
|
||||
|
||||
respBody, err := client.call(ctx, storageRESTMethodReadVersion, values, nil, -1)
|
||||
if err != nil {
|
||||
|
@ -17,7 +17,7 @@
|
||||
package cmd
|
||||
|
||||
const (
|
||||
storageRESTVersion = "v20" // Re-implementation of storage layer
|
||||
storageRESTVersion = "v21" // Add checkDataDir in ReadVersion API
|
||||
storageRESTVersionPrefix = SlashSeparator + storageRESTVersion
|
||||
storageRESTPrefix = minioReservedBucketPath + "/storage"
|
||||
)
|
||||
@ -60,6 +60,7 @@ const (
|
||||
storageRESTDirPath = "dir-path"
|
||||
storageRESTFilePath = "file-path"
|
||||
storageRESTVersionID = "version-id"
|
||||
storageRESTCheckDataDir = "check-data-dir"
|
||||
storageRESTTotalVersions = "total-versions"
|
||||
storageRESTSrcVolume = "source-volume"
|
||||
storageRESTSrcPath = "source-path"
|
||||
|
@ -326,8 +326,13 @@ func (s *storageRESTServer) ReadVersionHandler(w http.ResponseWriter, r *http.Re
|
||||
volume := vars[storageRESTVolume]
|
||||
filePath := vars[storageRESTFilePath]
|
||||
versionID := vars[storageRESTVersionID]
|
||||
checkDataDir, err := strconv.ParseBool(vars[storageRESTCheckDataDir])
|
||||
if err != nil {
|
||||
s.writeErrorResponse(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
fi, err := s.storage.ReadVersion(r.Context(), volume, filePath, versionID)
|
||||
fi, err := s.storage.ReadVersion(r.Context(), volume, filePath, versionID, checkDataDir)
|
||||
if err != nil {
|
||||
s.writeErrorResponse(w, err)
|
||||
return
|
||||
@ -925,7 +930,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpointServerSets Endpoint
|
||||
subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodDeleteVersion).HandlerFunc(httpTraceHdrs(server.DeleteVersionHandler)).
|
||||
Queries(restQueries(storageRESTVolume, storageRESTFilePath)...)
|
||||
subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodReadVersion).HandlerFunc(httpTraceHdrs(server.ReadVersionHandler)).
|
||||
Queries(restQueries(storageRESTVolume, storageRESTFilePath, storageRESTVersionID)...)
|
||||
Queries(restQueries(storageRESTVolume, storageRESTFilePath, storageRESTVersionID, storageRESTCheckDataDir)...)
|
||||
subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodRenameData).HandlerFunc(httpTraceHdrs(server.RenameDataHandler)).
|
||||
Queries(restQueries(storageRESTSrcVolume, storageRESTSrcPath, storageRESTDataDir,
|
||||
storageRESTDstVolume, storageRESTDstPath)...)
|
||||
|
@ -288,12 +288,12 @@ func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, volume, path s
|
||||
return p.storage.WriteMetadata(ctx, volume, path, fi)
|
||||
}
|
||||
|
||||
func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, volume, path, versionID string) (fi FileInfo, err error) {
|
||||
func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, volume, path, versionID string, checkDataDir bool) (fi FileInfo, err error) {
|
||||
if err = p.checkDiskStale(); err != nil {
|
||||
return fi, err
|
||||
}
|
||||
|
||||
return p.storage.ReadVersion(ctx, volume, path, versionID)
|
||||
return p.storage.ReadVersion(ctx, volume, path, versionID, checkDataDir)
|
||||
}
|
||||
|
||||
func (p *xlStorageDiskIDCheck) ReadAll(ctx context.Context, volume string, path string) (buf []byte, err error) {
|
||||
|
@ -1318,7 +1318,7 @@ func (s *xlStorage) renameLegacyMetadata(volume, path string) error {
|
||||
}
|
||||
|
||||
// ReadVersion - reads metadata and returns FileInfo at path `xl.meta`
|
||||
func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID string) (fi FileInfo, err error) {
|
||||
func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID string, checkDataDir bool) (fi FileInfo, err error) {
|
||||
buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
|
||||
if err != nil {
|
||||
if err == errFileNotFound {
|
||||
@ -1341,7 +1341,24 @@ func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID str
|
||||
return fi, errFileNotFound
|
||||
}
|
||||
|
||||
return getFileInfo(buf, volume, path, versionID)
|
||||
fi, err = getFileInfo(buf, volume, path, versionID)
|
||||
if err != nil {
|
||||
return fi, err
|
||||
}
|
||||
|
||||
if fi.DataDir != "" && checkDataDir {
|
||||
if _, err = s.StatVol(ctx, pathJoin(volume, path, fi.DataDir, slashSeparator)); err != nil {
|
||||
if err == errVolumeNotFound {
|
||||
if versionID != "" {
|
||||
return fi, errFileVersionNotFound
|
||||
}
|
||||
return fi, errFileNotFound
|
||||
}
|
||||
return fi, err
|
||||
}
|
||||
}
|
||||
|
||||
return fi, nil
|
||||
}
|
||||
|
||||
// ReadAll reads from r until an error or EOF and returns the data it read.
|
||||
|
Loading…
Reference in New Issue
Block a user