| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.ozone.om.request.s3.multipart; |
| |
| import org.apache.commons.codec.digest.DigestUtils; |
| import org.apache.commons.lang3.StringUtils; |
| |
| import java.io.IOException; |
| import java.nio.file.InvalidPathException; |
| import java.nio.file.Paths; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.BiFunction; |
| |
| import org.apache.hadoop.hdds.client.ReplicationConfig; |
| import org.apache.hadoop.ozone.om.OzoneConfigUtil; |
| import org.apache.hadoop.ozone.om.request.file.OMDirectoryCreateRequestWithFSO; |
| import org.apache.hadoop.ozone.om.request.file.OMFileRequest; |
| import org.apache.hadoop.ozone.protocolPB.OMPBHelper; |
| import org.apache.ratis.server.protocol.TermIndex; |
| import org.apache.hadoop.hdds.utils.db.cache.CacheKey; |
| import org.apache.hadoop.hdds.utils.db.cache.CacheValue; |
| import org.apache.hadoop.ozone.OzoneConsts; |
| import org.apache.hadoop.ozone.audit.OMAction; |
| import org.apache.hadoop.ozone.om.OMMetadataManager; |
| import org.apache.hadoop.ozone.om.OzoneManager; |
| import org.apache.hadoop.ozone.om.exceptions.OMException; |
| import org.apache.hadoop.ozone.om.helpers.KeyValueUtil; |
| import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; |
| import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo; |
| import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; |
| import org.apache.hadoop.ozone.om.helpers.BucketLayout; |
| import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; |
| import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; |
| import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; |
| import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; |
| import org.apache.hadoop.ozone.om.request.key.OMKeyRequest; |
| import org.apache.hadoop.ozone.om.request.util.OmResponseUtil; |
| import org.apache.hadoop.ozone.om.request.validation.RequestFeatureValidator; |
| import org.apache.hadoop.ozone.om.request.validation.RequestProcessingPhase; |
| import org.apache.hadoop.ozone.om.request.validation.ValidationCondition; |
| import org.apache.hadoop.ozone.om.request.validation.ValidationContext; |
| import org.apache.hadoop.ozone.om.response.OMClientResponse; |
| import org.apache.hadoop.ozone.om.response.s3.multipart.S3MultipartUploadCompleteResponse; |
| import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartUploadCompleteRequest; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartUploadCompleteResponse; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo; |
| import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type; |
| import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType; |
| import org.apache.hadoop.util.Time; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import jakarta.annotation.Nullable; |
| |
| import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.NOT_A_FILE; |
| import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK; |
| |
| /** |
| * Handle Multipart upload complete request. |
| */ |
| public class S3MultipartUploadCompleteRequest extends OMKeyRequest { |
| |
| private static final Logger LOG = |
| LoggerFactory.getLogger(S3MultipartUploadCompleteRequest.class); |
| |
| private BiFunction<OzoneManagerProtocolProtos.Part, PartKeyInfo, MultipartCommitRequestPart> eTagBasedValidator = |
| (part, partKeyInfo) -> { |
| String eTag = part.getETag(); |
| AtomicReference<String> dbPartETag = new AtomicReference<>(); |
| String dbPartName = null; |
| if (partKeyInfo != null) { |
| partKeyInfo.getPartKeyInfo().getMetadataList() |
| .stream() |
| .filter(keyValue -> keyValue.getKey().equals(OzoneConsts.ETAG)) |
| .findFirst().ifPresent(kv -> dbPartETag.set(kv.getValue())); |
| dbPartName = partKeyInfo.getPartName(); |
| } |
| return new MultipartCommitRequestPart(eTag, partKeyInfo == null ? null : |
| dbPartETag.get(), StringUtils.equals(eTag, dbPartETag.get()) || StringUtils.equals(eTag, dbPartName)); |
| }; |
| private BiFunction<OzoneManagerProtocolProtos.Part, PartKeyInfo, MultipartCommitRequestPart> partNameBasedValidator = |
| (part, partKeyInfo) -> { |
| String partName = part.getPartName(); |
| String dbPartName = null; |
| if (partKeyInfo != null) { |
| dbPartName = partKeyInfo.getPartName(); |
| } |
| return new MultipartCommitRequestPart(partName, partKeyInfo == null ? null : |
| dbPartName, StringUtils.equals(partName, dbPartName)); |
| }; |
| |
| public S3MultipartUploadCompleteRequest(OMRequest omRequest, |
| BucketLayout bucketLayout) { |
| super(omRequest, bucketLayout); |
| } |
| |
| @Override |
| public OMRequest preExecute(OzoneManager ozoneManager) throws IOException { |
| MultipartUploadCompleteRequest multipartUploadCompleteRequest = |
| super.preExecute(ozoneManager).getCompleteMultiPartUploadRequest(); |
| |
| KeyArgs keyArgs = multipartUploadCompleteRequest.getKeyArgs(); |
| String keyPath = keyArgs.getKeyName(); |
| keyPath = validateAndNormalizeKey(ozoneManager.getEnableFileSystemPaths(), |
| keyPath, getBucketLayout()); |
| |
| KeyArgs newKeyArgs = keyArgs.toBuilder().setModificationTime(Time.now()) |
| .setKeyName(keyPath).build(); |
| KeyArgs resolvedArgs = resolveBucketAndCheckKeyAcls(newKeyArgs, |
| ozoneManager, ACLType.WRITE); |
| |
| return getOmRequest().toBuilder().setCompleteMultiPartUploadRequest( |
| multipartUploadCompleteRequest.toBuilder().setKeyArgs( |
| resolvedArgs)).setUserInfo(getUserInfo()).build(); |
| } |
| |
| @Override |
| @SuppressWarnings("methodlength") |
| public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIndex termIndex) { |
| final long trxnLogIndex = termIndex.getIndex(); |
| MultipartUploadCompleteRequest multipartUploadCompleteRequest = |
| getOmRequest().getCompleteMultiPartUploadRequest(); |
| |
| KeyArgs keyArgs = multipartUploadCompleteRequest.getKeyArgs(); |
| |
| List<OzoneManagerProtocolProtos.Part> partsList = |
| multipartUploadCompleteRequest.getPartsListList(); |
| Map<String, String> auditMap = buildKeyArgsAuditMap(keyArgs); |
| |
| String volumeName = keyArgs.getVolumeName(); |
| String bucketName = keyArgs.getBucketName(); |
| final String requestedVolume = volumeName; |
| final String requestedBucket = bucketName; |
| String keyName = keyArgs.getKeyName(); |
| String uploadID = keyArgs.getMultipartUploadID(); |
| String multipartKey = null; |
| |
| ozoneManager.getMetrics().incNumCompleteMultipartUploads(); |
| |
| OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager(); |
| |
| boolean acquiredLock = false; |
| OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder( |
| getOmRequest()); |
| OMClientResponse omClientResponse = null; |
| Exception exception = null; |
| Result result = null; |
| try { |
| multipartKey = omMetadataManager.getMultipartKey(volumeName, |
| bucketName, keyName, uploadID); |
| |
| mergeOmLockDetails(omMetadataManager.getLock() |
| .acquireWriteLock(BUCKET_LOCK, volumeName, bucketName)); |
| acquiredLock = getOmLockDetails().isLockAcquired(); |
| |
| validateBucketAndVolume(omMetadataManager, volumeName, bucketName); |
| OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager, |
| volumeName, bucketName); |
| |
| List<OmDirectoryInfo> missingParentInfos; |
| OMFileRequest.OMPathInfoWithFSO pathInfoFSO = OMFileRequest |
| .verifyDirectoryKeysInPath(omMetadataManager, volumeName, bucketName, |
| keyName, Paths.get(keyName)); |
| missingParentInfos = OMDirectoryCreateRequestWithFSO |
| .getAllMissingParentDirInfo(ozoneManager, keyArgs, omBucketInfo, |
| pathInfoFSO, trxnLogIndex); |
| |
| if (missingParentInfos != null) { |
| final long volumeId = omMetadataManager.getVolumeId(volumeName); |
| final long bucketId = omMetadataManager.getBucketId(volumeName, |
| bucketName); |
| |
| // add all missing parents to directory table |
| addMissingParentsToCache(omBucketInfo, missingParentInfos, |
| omMetadataManager, volumeId, bucketId, trxnLogIndex); |
| |
| String multipartOpenKey = omMetadataManager |
| .getMultipartKey(volumeId, bucketId, |
| pathInfoFSO.getLastKnownParentId(), |
| pathInfoFSO.getLeafNodeName(), |
| keyArgs.getMultipartUploadID()); |
| |
| if (getOmKeyInfoFromOpenKeyTable(multipartOpenKey, |
| keyName, omMetadataManager) == null) { |
| |
| final ReplicationConfig replicationConfig = OzoneConfigUtil |
| .resolveReplicationConfigPreference(keyArgs.getType(), |
| keyArgs.getFactor(), keyArgs.getEcReplicationConfig(), |
| omBucketInfo != null ? |
| omBucketInfo.getDefaultReplicationConfig() : |
| null, ozoneManager); |
| |
| OmMultipartKeyInfo multipartKeyInfoFromArgs = |
| new OmMultipartKeyInfo.Builder() |
| .setUploadID(keyArgs.getMultipartUploadID()) |
| .setCreationTime(keyArgs.getModificationTime()) |
| .setReplicationConfig(replicationConfig) |
| .setObjectID(pathInfoFSO.getLeafNodeObjectId()) |
| .setUpdateID(trxnLogIndex) |
| .setParentID(pathInfoFSO.getLastKnownParentId()) |
| .build(); |
| |
| OmKeyInfo keyInfoFromArgs = new OmKeyInfo.Builder() |
| .setVolumeName(volumeName) |
| .setBucketName(bucketName) |
| .setKeyName(keyName) |
| .setCreationTime(keyArgs.getModificationTime()) |
| .setModificationTime(keyArgs.getModificationTime()) |
| .setReplicationConfig(replicationConfig) |
| .setOmKeyLocationInfos(Collections.singletonList( |
| new OmKeyLocationInfoGroup(0, new ArrayList<>(), true))) |
| .setAcls(getAclsForKey(keyArgs, omBucketInfo, pathInfoFSO, |
| ozoneManager.getPrefixManager())) |
| .setObjectID(pathInfoFSO.getLeafNodeObjectId()) |
| .setUpdateID(trxnLogIndex) |
| .setFileEncryptionInfo(keyArgs.hasFileEncryptionInfo() ? |
| OMPBHelper.convert(keyArgs.getFileEncryptionInfo()) : null) |
| .setParentObjectID(pathInfoFSO.getLastKnownParentId()) |
| .build(); |
| |
| // Add missing multi part info to open key table |
| addMultiPartToCache(omMetadataManager, multipartOpenKey, |
| pathInfoFSO, keyInfoFromArgs, trxnLogIndex); |
| } |
| } |
| |
| String dbMultipartOpenKey = |
| getDBMultipartOpenKey(volumeName, bucketName, keyName, uploadID, |
| omMetadataManager); |
| |
| OmMultipartKeyInfo multipartKeyInfo = omMetadataManager |
| .getMultipartInfoTable().get(multipartKey); |
| |
| String ozoneKey = omMetadataManager.getOzoneKey( |
| volumeName, bucketName, keyName); |
| |
| String dbOzoneKey = |
| getDBOzoneKey(omMetadataManager, volumeName, bucketName, keyName); |
| |
| // Check for directory exists with same name for the LEGACY_FS, |
| // if it exists throw error. |
| checkDirectoryAlreadyExists(ozoneManager, omBucketInfo, keyName, |
| omMetadataManager); |
| |
| if (multipartKeyInfo == null) { |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName), |
| OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); |
| } |
| |
| if (partsList.size() > 0) { |
| final OmMultipartKeyInfo.PartKeyInfoMap partKeyInfoMap |
| = multipartKeyInfo.getPartKeyInfoMap(); |
| if (partKeyInfoMap.size() == 0) { |
| LOG.error("Complete MultipartUpload failed for key {} , MPU Key has" + |
| " no parts in OM, parts given to upload are {}", ozoneKey, |
| partsList); |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName), |
| OMException.ResultCodes.INVALID_PART); |
| } |
| |
| // First Check for Invalid Part Order. |
| List< Integer > partNumbers = new ArrayList<>(); |
| int partsListSize = getPartsListSize(requestedVolume, |
| requestedBucket, keyName, ozoneKey, partNumbers, partsList); |
| |
| List<OmKeyLocationInfo> partLocationInfos = new ArrayList<>(); |
| long dataSize = getMultipartDataSize(requestedVolume, requestedBucket, |
| keyName, ozoneKey, partKeyInfoMap, partsListSize, |
| partLocationInfos, partsList, ozoneManager); |
| |
| // All parts have same replication information. Here getting from last |
| // part. |
| OmKeyInfo omKeyInfo = |
| getOmKeyInfo(ozoneManager, trxnLogIndex, keyArgs, volumeName, |
| bucketName, keyName, dbMultipartOpenKey, omMetadataManager, |
| dbOzoneKey, partKeyInfoMap, partLocationInfos, dataSize); |
| |
| long usedBytesDiff = 0; |
| //Find all unused parts. |
| List<OmKeyInfo> allKeyInfoToRemove = new ArrayList<>(); |
| for (PartKeyInfo partKeyInfo : partKeyInfoMap) { |
| if (!partNumbers.contains(partKeyInfo.getPartNumber())) { |
| OmKeyInfo delPartKeyInfo = |
| OmKeyInfo.getFromProtobuf(partKeyInfo.getPartKeyInfo()); |
| allKeyInfoToRemove.add(delPartKeyInfo); |
| usedBytesDiff -= delPartKeyInfo.getReplicatedSize(); |
| } |
| } |
| |
| // If bucket versioning is turned on during the update, between key |
| // creation and key commit, old versions will be just overwritten and |
| // not kept. Bucket versioning will be effective from the first key |
| // creation after the knob turned on. |
| OmKeyInfo keyToDelete = |
| omMetadataManager.getKeyTable(getBucketLayout()).get(dbOzoneKey); |
| boolean isNamespaceUpdate = false; |
| if (keyToDelete != null && !omBucketInfo.getIsVersionEnabled()) { |
| RepeatedOmKeyInfo oldKeyVersionsToDelete = getOldVersionsToCleanUp( |
| keyToDelete, trxnLogIndex, ozoneManager.isRatisEnabled()); |
| allKeyInfoToRemove.addAll(oldKeyVersionsToDelete.getOmKeyInfoList()); |
| usedBytesDiff -= keyToDelete.getReplicatedSize(); |
| } else { |
| checkBucketQuotaInNamespace(omBucketInfo, 1L); |
| omBucketInfo.incrUsedNamespace(1L); |
| isNamespaceUpdate = true; |
| } |
| |
| String dbBucketKey = omMetadataManager.getBucketKey( |
| omBucketInfo.getVolumeName(), omBucketInfo.getBucketName()); |
| if (usedBytesDiff != 0) { |
| omBucketInfo.incrUsedBytes(usedBytesDiff); |
| } else if (!isNamespaceUpdate) { |
| // If no bucket size and Namespace changed, prevent from updating |
| // bucket object. |
| omBucketInfo = null; |
| } |
| |
| updateCache(omMetadataManager, dbBucketKey, omBucketInfo, dbOzoneKey, |
| dbMultipartOpenKey, multipartKey, omKeyInfo, trxnLogIndex); |
| |
| omResponse.setCompleteMultiPartUploadResponse( |
| MultipartUploadCompleteResponse.newBuilder() |
| .setVolume(requestedVolume) |
| .setBucket(requestedBucket) |
| .setKey(keyName) |
| .setHash(omKeyInfo.getMetadata().get(OzoneConsts.ETAG))); |
| |
| long volumeId = omMetadataManager.getVolumeId(volumeName); |
| long bucketId = omMetadataManager.getBucketId(volumeName, bucketName); |
| omClientResponse = |
| getOmClientResponse(multipartKey, omResponse, dbMultipartOpenKey, |
| omKeyInfo, allKeyInfoToRemove, omBucketInfo, |
| volumeId, bucketId, missingParentInfos, multipartKeyInfo); |
| |
| result = Result.SUCCESS; |
| } else { |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName) + |
| " because of empty part list", |
| OMException.ResultCodes.INVALID_REQUEST); |
| } |
| |
| } catch (IOException | InvalidPathException ex) { |
| result = Result.FAILURE; |
| exception = ex; |
| omClientResponse = getOmClientResponse(omResponse, exception); |
| } finally { |
| if (acquiredLock) { |
| mergeOmLockDetails(omMetadataManager.getLock() |
| .releaseWriteLock(BUCKET_LOCK, volumeName, bucketName)); |
| } |
| if (omClientResponse != null) { |
| omClientResponse.setOmLockDetails(getOmLockDetails()); |
| } |
| } |
| |
| logResult(ozoneManager, multipartUploadCompleteRequest, partsList, |
| auditMap, volumeName, bucketName, keyName, exception, result); |
| |
| return omClientResponse; |
| } |
| |
| protected S3MultipartUploadCompleteResponse getOmClientResponse( |
| OMResponse.Builder omResponse, Exception exception) { |
| return new S3MultipartUploadCompleteResponse( |
| createErrorOMResponse(omResponse, exception), getBucketLayout()); |
| } |
| |
| @SuppressWarnings("parameternumber") |
| protected OMClientResponse getOmClientResponse(String multipartKey, |
| OMResponse.Builder omResponse, String dbMultipartOpenKey, |
| OmKeyInfo omKeyInfo, List<OmKeyInfo> allKeyInfoToRemove, |
| OmBucketInfo omBucketInfo, |
| long volumeId, long bucketId, List<OmDirectoryInfo> missingParentInfos, |
| OmMultipartKeyInfo multipartKeyInfo) { |
| |
| return new S3MultipartUploadCompleteResponse(omResponse.build(), |
| multipartKey, dbMultipartOpenKey, omKeyInfo, allKeyInfoToRemove, |
| getBucketLayout(), omBucketInfo); |
| } |
| |
| protected void checkDirectoryAlreadyExists(OzoneManager ozoneManager, |
| OmBucketInfo omBucketInfo, String keyName, |
| OMMetadataManager omMetadataManager) throws IOException { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("BucketName: {}, BucketLayout: {}", |
| omBucketInfo.getBucketName(), omBucketInfo.getBucketLayout()); |
| } |
| if (omBucketInfo.getBucketLayout() |
| .shouldNormalizePaths(ozoneManager.getEnableFileSystemPaths())) { |
| if (checkDirectoryAlreadyExists(omBucketInfo.getVolumeName(), |
| omBucketInfo.getBucketName(), keyName, omMetadataManager)) { |
| throw new OMException("Can not Complete MPU for file: " + keyName + |
| " as there is already directory in the given path", |
| NOT_A_FILE); |
| } |
| } |
| } |
| |
| @SuppressWarnings("checkstyle:ParameterNumber") |
| protected void logResult(OzoneManager ozoneManager, |
| MultipartUploadCompleteRequest multipartUploadCompleteRequest, |
| List<OzoneManagerProtocolProtos.Part> partsList, |
| Map<String, String> auditMap, String volumeName, |
| String bucketName, String keyName, Exception exception, |
| Result result) { |
| auditMap.put(OzoneConsts.MULTIPART_LIST, partsList.toString() |
| .replaceAll("\\n", " ")); |
| |
| // audit log |
| auditLog(ozoneManager.getAuditLogger(), buildAuditMessage( |
| OMAction.COMPLETE_MULTIPART_UPLOAD, auditMap, exception, |
| getOmRequest().getUserInfo())); |
| |
| switch (result) { |
| case SUCCESS: |
| LOG.debug("MultipartUpload Complete request is successful for Key: {} " + |
| "in Volume/Bucket {}/{}", keyName, volumeName, bucketName); |
| break; |
| case FAILURE: |
| ozoneManager.getMetrics().incNumCompleteMultipartUploadFails(); |
| LOG.error("MultipartUpload Complete request failed for Key: {} " + |
| "in Volume/Bucket {}/{}", keyName, volumeName, bucketName, |
| exception); |
| break; |
| default: |
| LOG.error("Unrecognized Result for S3MultipartUploadCommitRequest: {}", |
| multipartUploadCompleteRequest); |
| } |
| } |
| |
| @SuppressWarnings("checkstyle:ParameterNumber") |
| protected OmKeyInfo getOmKeyInfo(OzoneManager ozoneManager, long trxnLogIndex, |
| KeyArgs keyArgs, String volumeName, String bucketName, String keyName, |
| String multipartOpenKey, OMMetadataManager omMetadataManager, |
| String ozoneKey, OmMultipartKeyInfo.PartKeyInfoMap partKeyInfoMap, |
| List<OmKeyLocationInfo> partLocationInfos, long dataSize) |
| throws IOException { |
| OzoneManagerProtocolProtos.KeyInfo partKeyInfo = |
| partKeyInfoMap.lastEntry().getPartKeyInfo(); |
| |
| OmKeyInfo omKeyInfo = getOmKeyInfoFromKeyTable(ozoneKey, keyName, |
| omMetadataManager); |
| if (omKeyInfo == null) { |
| // This is a newly added key, it does not have any versions. |
| OmKeyLocationInfoGroup keyLocationInfoGroup = new |
| OmKeyLocationInfoGroup(0, partLocationInfos, true); |
| |
| // Get the objectID of the key from OpenKeyTable |
| OmKeyInfo dbOpenKeyInfo = getOmKeyInfoFromOpenKeyTable(multipartOpenKey, |
| keyName, omMetadataManager); |
| |
| // A newly created key, this is the first version. |
| OmKeyInfo.Builder builder = |
| new OmKeyInfo.Builder().setVolumeName(volumeName) |
| .setBucketName(bucketName).setKeyName(dbOpenKeyInfo.getKeyName()) |
| .setReplicationConfig(ReplicationConfig.fromProto( |
| partKeyInfo.getType(), partKeyInfo.getFactor(), |
| partKeyInfo.getEcReplicationConfig())) |
| .setCreationTime(keyArgs.getModificationTime()) |
| .setModificationTime(keyArgs.getModificationTime()) |
| .setDataSize(dataSize) |
| .setFileEncryptionInfo(dbOpenKeyInfo.getFileEncryptionInfo()) |
| .setOmKeyLocationInfos( |
| Collections.singletonList(keyLocationInfoGroup)) |
| .setAcls(dbOpenKeyInfo.getAcls()) |
| .addMetadata(OzoneConsts.ETAG, |
| multipartUploadedKeyHash(partKeyInfoMap)); |
| // Check if db entry has ObjectID. This check is required because |
| // it is possible that between multipart key uploads and complete, |
| // we had an upgrade. |
| if (dbOpenKeyInfo.getObjectID() != 0) { |
| builder.setObjectID(dbOpenKeyInfo.getObjectID()); |
| } |
| updatePrefixFSOInfo(dbOpenKeyInfo, builder); |
| omKeyInfo = builder.build(); |
| } else { |
| OmKeyInfo dbOpenKeyInfo = getOmKeyInfoFromOpenKeyTable(multipartOpenKey, |
| keyName, omMetadataManager); |
| |
| // Already a version exists, so we should add it as a new version. |
| // But now as versioning is not supported, just following the commit |
| // key approach. When versioning support comes, then we can uncomment |
| // below code keyInfo.addNewVersion(locations); |
| |
| // As right now versioning is not supported, we can set encryption info |
| // at KeyInfo level, but once we start supporting versioning, |
| // encryption info needs to be set at KeyLocation level, as each version |
| // will have it's own file encryption info. |
| if (dbOpenKeyInfo.getFileEncryptionInfo() != null) { |
| omKeyInfo.setFileEncryptionInfo(dbOpenKeyInfo.getFileEncryptionInfo()); |
| } |
| omKeyInfo.updateLocationInfoList(partLocationInfos, true, true); |
| omKeyInfo.setModificationTime(keyArgs.getModificationTime()); |
| omKeyInfo.setDataSize(dataSize); |
| omKeyInfo.setReplicationConfig(dbOpenKeyInfo.getReplicationConfig()); |
| omKeyInfo.getMetadata().put(OzoneConsts.ETAG, |
| multipartUploadedKeyHash(partKeyInfoMap)); |
| } |
| omKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled()); |
| return omKeyInfo; |
| } |
| |
| protected void updatePrefixFSOInfo(OmKeyInfo dbOpenKeyInfo, |
| OmKeyInfo.Builder builder) { |
| // FSO is disabled. Do nothing. |
| } |
| |
| protected String getDBOzoneKey(OMMetadataManager omMetadataManager, |
| String volumeName, String bucketName, String keyName) throws IOException { |
| return omMetadataManager.getOzoneKey(volumeName, bucketName, keyName); |
| } |
| |
| protected void addMissingParentsToCache(OmBucketInfo omBucketInfo, |
| List<OmDirectoryInfo> missingParentInfos, |
| OMMetadataManager omMetadataManager, |
| long volumeId, long bucketId, long transactionLogIndex |
| ) throws IOException { |
| // FSO is disabled. Do nothing. |
| } |
| |
| protected void addMultiPartToCache( |
| OMMetadataManager omMetadataManager, String multipartOpenKey, |
| OMFileRequest.OMPathInfoWithFSO pathInfoFSO, OmKeyInfo omKeyInfo, |
| long transactionLogIndex |
| ) throws IOException { |
| // FSO is disabled. Do nothing. |
| } |
| |
| protected OmKeyInfo getOmKeyInfoFromKeyTable(String dbOzoneKey, |
| String keyName, OMMetadataManager omMetadataManager) throws IOException { |
| return omMetadataManager.getKeyTable(getBucketLayout()).get(dbOzoneKey); |
| } |
| |
| protected OmKeyInfo getOmKeyInfoFromOpenKeyTable(String dbMultipartKey, |
| String keyName, OMMetadataManager omMetadataManager) throws IOException { |
| return omMetadataManager.getOpenKeyTable(getBucketLayout()) |
| .get(dbMultipartKey); |
| } |
| |
| protected void addKeyTableCacheEntry(OMMetadataManager omMetadataManager, |
| String dbOzoneKey, OmKeyInfo omKeyInfo, long transactionLogIndex) |
| throws IOException { |
| |
| // Add key entry to file table. |
| omMetadataManager.getKeyTable(getBucketLayout()) |
| .addCacheEntry(new CacheKey<>(dbOzoneKey), |
| CacheValue.get(transactionLogIndex, omKeyInfo)); |
| } |
| |
| private int getPartsListSize(String requestedVolume, |
| String requestedBucket, String keyName, String ozoneKey, |
| List<Integer> partNumbers, |
| List<OzoneManagerProtocolProtos.Part> partsList) throws OMException { |
| int prevPartNumber = partsList.get(0).getPartNumber(); |
| int partsListSize = partsList.size(); |
| partNumbers.add(prevPartNumber); |
| for (int i = 1; i < partsListSize; i++) { |
| int currentPartNumber = partsList.get(i).getPartNumber(); |
| if (prevPartNumber >= currentPartNumber) { |
| LOG.error("PartNumber at index {} is {}, and its previous " + |
| "partNumber at index {} is {} for ozonekey is " + |
| "{}", i, currentPartNumber, i - 1, prevPartNumber, |
| ozoneKey); |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName) + |
| " because parts are in Invalid order.", |
| OMException.ResultCodes.INVALID_PART_ORDER); |
| } |
| prevPartNumber = currentPartNumber; |
| partNumbers.add(prevPartNumber); |
| } |
| return partsListSize; |
| } |
| |
| @SuppressWarnings("checkstyle:ParameterNumber") |
| private long getMultipartDataSize(String requestedVolume, |
| String requestedBucket, String keyName, String ozoneKey, |
| OmMultipartKeyInfo.PartKeyInfoMap partKeyInfoMap, |
| int partsListSize, List<OmKeyLocationInfo> partLocationInfos, |
| List<OzoneManagerProtocolProtos.Part> partsList, |
| OzoneManager ozoneManager) throws OMException { |
| long dataSize = 0; |
| int currentPartCount = 0; |
| boolean eTagBasedValidationAvailable = partsList.stream().allMatch(OzoneManagerProtocolProtos.Part::hasETag); |
| // Now do actual logic, and check for any Invalid part during this. |
| for (OzoneManagerProtocolProtos.Part part : partsList) { |
| currentPartCount++; |
| int partNumber = part.getPartNumber(); |
| PartKeyInfo partKeyInfo = partKeyInfoMap.get(partNumber); |
| MultipartCommitRequestPart requestPart = eTagBasedValidationAvailable ? |
| eTagBasedValidator.apply(part, partKeyInfo) : partNameBasedValidator.apply(part, partKeyInfo); |
| if (!requestPart.isValid()) { |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName) + |
| ". Provided Part info is { " + requestPart.getRequestPartId() + ", " + partNumber + |
| "}, whereas OM has eTag " + requestPart.getOmPartId(), |
| OMException.ResultCodes.INVALID_PART); |
| } |
| |
| OmKeyInfo currentPartKeyInfo = null; |
| try { |
| currentPartKeyInfo = |
| OmKeyInfo.getFromProtobuf(partKeyInfo.getPartKeyInfo()); |
| } catch (IOException ioe) { |
| throw new OMException(ioe, OMException.ResultCodes.INTERNAL_ERROR); |
| } |
| |
| // Except for last part all parts should have minimum size. |
| if (currentPartCount != partsListSize) { |
| if (currentPartKeyInfo.getDataSize() < |
| ozoneManager.getMinMultipartUploadPartSize()) { |
| LOG.error("MultipartUpload: {} Part number: {} size {} is less" + |
| " than minimum part size {}", ozoneKey, |
| partKeyInfo.getPartNumber(), currentPartKeyInfo.getDataSize(), |
| ozoneManager.getMinMultipartUploadPartSize()); |
| throw new OMException( |
| failureMessage(requestedVolume, requestedBucket, keyName) + |
| ". Entity too small.", |
| OMException.ResultCodes.ENTITY_TOO_SMALL); |
| } |
| } |
| |
| // As all part keys will have only one version. |
| OmKeyLocationInfoGroup currentKeyInfoGroup = currentPartKeyInfo |
| .getKeyLocationVersions().get(0); |
| |
| // Set partNumber in each block. |
| currentKeyInfoGroup.getLocationList().forEach( |
| omKeyLocationInfo -> omKeyLocationInfo.setPartNumber(partNumber)); |
| |
| partLocationInfos.addAll(currentKeyInfoGroup.getLocationList()); |
| dataSize += currentPartKeyInfo.getDataSize(); |
| } |
| return dataSize; |
| } |
| |
| private static String failureMessage(String volume, String bucket, |
| String keyName) { |
| return "Complete Multipart Upload Failed: volume: " + |
| volume + " bucket: " + bucket + " key: " + keyName; |
| } |
| |
| @SuppressWarnings("parameternumber") |
| private void updateCache(OMMetadataManager omMetadataManager, |
| String dbBucketKey, @Nullable OmBucketInfo omBucketInfo, |
| String dbOzoneKey, String dbMultipartOpenKey, String dbMultipartKey, |
| OmKeyInfo omKeyInfo, long transactionLogIndex) throws IOException { |
| // Update cache. |
| // 1. Add key entry to key table. |
| // 2. Delete multipartKey entry from openKeyTable and multipartInfo table. |
| // 3. If the bucket size has changed (omBucketInfo is not null), |
| // update bucket cache |
| addKeyTableCacheEntry(omMetadataManager, dbOzoneKey, omKeyInfo, |
| transactionLogIndex); |
| |
| omMetadataManager.getOpenKeyTable(getBucketLayout()) |
| .addCacheEntry( |
| new CacheKey<>(dbMultipartOpenKey), |
| CacheValue.get(transactionLogIndex)); |
| omMetadataManager.getMultipartInfoTable().addCacheEntry( |
| new CacheKey<>(dbMultipartKey), |
| CacheValue.get(transactionLogIndex)); |
| |
| // Here, omBucketInfo can be null if its size has not changed. No need to |
| // update the bucket info unless its size has changed. We never want to |
| // delete the bucket info here, but just avoiding unnecessary update. |
| if (omBucketInfo != null) { |
| omMetadataManager.getBucketTable().addCacheEntry( |
| new CacheKey<>(dbBucketKey), |
| CacheValue.get(transactionLogIndex, omBucketInfo)); |
| } |
| } |
| |
| @RequestFeatureValidator( |
| conditions = ValidationCondition.CLUSTER_NEEDS_FINALIZATION, |
| processingPhase = RequestProcessingPhase.PRE_PROCESS, |
| requestType = Type.CompleteMultiPartUpload |
| ) |
| public static OMRequest |
| disallowCompleteMultiPartUploadWithECReplicationConfig( |
| OMRequest req, ValidationContext ctx) throws OMException { |
| if (!ctx.versionManager().isAllowed( |
| OMLayoutFeature.ERASURE_CODED_STORAGE_SUPPORT)) { |
| if (req.getCompleteMultiPartUploadRequest().getKeyArgs() |
| .hasEcReplicationConfig()) { |
| throw new OMException("Cluster does not have the Erasure Coded" |
| + " Storage support feature finalized yet, but the request contains" |
| + " an Erasure Coded replication type. Rejecting the request," |
| + " please finalize the cluster upgrade and then try again.", |
| OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); |
| } |
| } |
| return req; |
| } |
| |
| /** |
| * Validates S3 MPU complete requests. |
| * We do not want to allow older clients to upload MPU keys to buckets which |
| * use non LEGACY layouts. |
| * |
| * @param req - the request to validate |
| * @param ctx - the validation context |
| * @return the validated request |
| * @throws OMException if the request is invalid |
| */ |
| @RequestFeatureValidator( |
| conditions = ValidationCondition.OLDER_CLIENT_REQUESTS, |
| processingPhase = RequestProcessingPhase.PRE_PROCESS, |
| requestType = Type.CompleteMultiPartUpload |
| ) |
| public static OMRequest blockMPUCompleteWithBucketLayoutFromOldClient( |
| OMRequest req, ValidationContext ctx) throws IOException { |
| if (req.getCompleteMultiPartUploadRequest().hasKeyArgs()) { |
| KeyArgs keyArgs = req.getCompleteMultiPartUploadRequest().getKeyArgs(); |
| |
| if (keyArgs.hasVolumeName() && keyArgs.hasBucketName()) { |
| BucketLayout bucketLayout = ctx.getBucketLayout( |
| keyArgs.getVolumeName(), keyArgs.getBucketName()); |
| bucketLayout.validateSupportedOperation(); |
| } |
| } |
| return req; |
| } |
| |
| private String multipartUploadedKeyHash( |
| OmMultipartKeyInfo.PartKeyInfoMap partsList) { |
| StringBuffer keysConcatenated = new StringBuffer(); |
| for (PartKeyInfo partKeyInfo: partsList) { |
| String partPropertyToComputeHash = KeyValueUtil.getFromProtobuf(partKeyInfo.getPartKeyInfo().getMetadataList()) |
| .get(OzoneConsts.ETAG); |
| if (partPropertyToComputeHash == null) { |
| partPropertyToComputeHash = partKeyInfo.getPartName(); |
| } |
| keysConcatenated.append(partPropertyToComputeHash); |
| } |
| return DigestUtils.md5Hex(keysConcatenated.toString()) + "-" |
| + partsList.size(); |
| } |
| |
| private static class MultipartCommitRequestPart { |
| private String requestPartId; |
| |
| private String omPartId; |
| |
| private boolean isValid; |
| |
| MultipartCommitRequestPart(String requestPartId, String omPartId, boolean isValid) { |
| this.requestPartId = requestPartId; |
| this.omPartId = omPartId; |
| this.isValid = isValid; |
| } |
| |
| public String getRequestPartId() { |
| return requestPartId; |
| } |
| |
| public String getOmPartId() { |
| return omPartId; |
| } |
| |
| public boolean isValid() { |
| return isValid; |
| } |
| } |
| |
| } |
