| /** |
| * 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.container.common.statemachine.commandhandler; |
| |
| import com.google.common.primitives.Longs; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; |
| import org.apache.hadoop.hdds.protocol.proto |
| .StorageContainerDatanodeProtocolProtos.SCMCommandProto; |
| import org.apache.hadoop.hdds.scm.container.common.helpers |
| .StorageContainerException; |
| import org.apache.hadoop.hdfs.DFSUtil; |
| import org.apache.hadoop.hdds.protocol.proto |
| .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto; |
| import org.apache.hadoop.hdds.protocol.proto |
| .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto |
| .DeleteBlockTransactionResult; |
| import org.apache.hadoop.hdds.protocol.proto |
| .StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction; |
| import org.apache.hadoop.ozone.OzoneConsts; |
| import org.apache.hadoop.ozone.container.common.helpers |
| .DeletedContainerBlocksSummary; |
| import org.apache.hadoop.ozone.container.common.interfaces.Container; |
| import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; |
| import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils; |
| import org.apache.hadoop.ozone.container.common.impl.ContainerSet; |
| import org.apache.hadoop.ozone.container.common.statemachine |
| .SCMConnectionManager; |
| import org.apache.hadoop.ozone.container.common.statemachine.StateContext; |
| import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; |
| import org.apache.hadoop.ozone.protocol.commands.CommandStatus; |
| import org.apache.hadoop.ozone.protocol.commands.DeleteBlockCommandStatus; |
| import org.apache.hadoop.ozone.protocol.commands.DeleteBlocksCommand; |
| import org.apache.hadoop.ozone.protocol.commands.SCMCommand; |
| import org.apache.hadoop.util.Time; |
| import org.apache.hadoop.hdds.utils.BatchOperation; |
| import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.List; |
| import java.util.function.Consumer; |
| |
| import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos |
| .Result.CONTAINER_NOT_FOUND; |
| |
| /** |
| * Handle block deletion commands. |
| */ |
| public class DeleteBlocksCommandHandler implements CommandHandler { |
| |
| public static final Logger LOG = |
| LoggerFactory.getLogger(DeleteBlocksCommandHandler.class); |
| |
| private final ContainerSet containerSet; |
| private final Configuration conf; |
| private int invocationCount; |
| private long totalTime; |
| private boolean cmdExecuted; |
| |
| public DeleteBlocksCommandHandler(ContainerSet cset, |
| Configuration conf) { |
| this.containerSet = cset; |
| this.conf = conf; |
| } |
| |
| @Override |
| public void handle(SCMCommand command, OzoneContainer container, |
| StateContext context, SCMConnectionManager connectionManager) { |
| cmdExecuted = false; |
| long startTime = Time.monotonicNow(); |
| ContainerBlocksDeletionACKProto blockDeletionACK = null; |
| try { |
| if (command.getType() != SCMCommandProto.Type.deleteBlocksCommand) { |
| LOG.warn("Skipping handling command, expected command " |
| + "type {} but found {}", |
| SCMCommandProto.Type.deleteBlocksCommand, command.getType()); |
| return; |
| } |
| LOG.debug("Processing block deletion command."); |
| invocationCount++; |
| |
| // move blocks to deleting state. |
| // this is a metadata update, the actual deletion happens in another |
| // recycling thread. |
| DeleteBlocksCommand cmd = (DeleteBlocksCommand) command; |
| List<DeletedBlocksTransaction> containerBlocks = cmd.blocksTobeDeleted(); |
| |
| DeletedContainerBlocksSummary summary = |
| DeletedContainerBlocksSummary.getFrom(containerBlocks); |
| LOG.info("Start to delete container blocks, TXIDs={}, " |
| + "numOfContainers={}, numOfBlocks={}", |
| summary.getTxIDSummary(), |
| summary.getNumOfContainers(), |
| summary.getNumOfBlocks()); |
| |
| ContainerBlocksDeletionACKProto.Builder resultBuilder = |
| ContainerBlocksDeletionACKProto.newBuilder(); |
| containerBlocks.forEach(entry -> { |
| DeleteBlockTransactionResult.Builder txResultBuilder = |
| DeleteBlockTransactionResult.newBuilder(); |
| txResultBuilder.setTxID(entry.getTxID()); |
| long containerId = entry.getContainerID(); |
| try { |
| Container cont = containerSet.getContainer(containerId); |
| if (cont == null) { |
| throw new StorageContainerException("Unable to find the container " |
| + containerId, CONTAINER_NOT_FOUND); |
| } |
| ContainerProtos.ContainerType containerType = cont.getContainerType(); |
| switch (containerType) { |
| case KeyValueContainer: |
| KeyValueContainerData containerData = (KeyValueContainerData) |
| cont.getContainerData(); |
| cont.writeLock(); |
| try { |
| deleteKeyValueContainerBlocks(containerData, entry); |
| } finally { |
| cont.writeUnlock(); |
| } |
| txResultBuilder.setContainerID(containerId) |
| .setSuccess(true); |
| break; |
| default: |
| LOG.error( |
| "Delete Blocks Command Handler is not implemented for " + |
| "containerType {}", containerType); |
| } |
| } catch (IOException e) { |
| LOG.warn("Failed to delete blocks for container={}, TXID={}", |
| entry.getContainerID(), entry.getTxID(), e); |
| txResultBuilder.setContainerID(containerId) |
| .setSuccess(false); |
| } |
| resultBuilder.addResults(txResultBuilder.build()) |
| .setDnId(context.getParent().getDatanodeDetails() |
| .getUuid().toString()); |
| }); |
| blockDeletionACK = resultBuilder.build(); |
| |
| // Send ACK back to SCM as long as meta updated |
| // TODO Or we should wait until the blocks are actually deleted? |
| if (!containerBlocks.isEmpty()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Sending following block deletion ACK to SCM"); |
| for (DeleteBlockTransactionResult result : blockDeletionACK |
| .getResultsList()) { |
| LOG.debug(result.getTxID() + " : " + result.getSuccess()); |
| } |
| } |
| } |
| cmdExecuted = true; |
| } finally { |
| final ContainerBlocksDeletionACKProto deleteAck = |
| blockDeletionACK; |
| Consumer<CommandStatus> statusUpdater = (cmdStatus) -> { |
| cmdStatus.setStatus(cmdExecuted); |
| ((DeleteBlockCommandStatus) cmdStatus).setBlocksDeletionAck(deleteAck); |
| }; |
| updateCommandStatus(context, command, statusUpdater, LOG); |
| long endTime = Time.monotonicNow(); |
| totalTime += endTime - startTime; |
| } |
| } |
| |
| /** |
| * Move a bunch of blocks from a container to deleting state. This is a meta |
| * update, the actual deletes happen in async mode. |
| * |
| * @param containerData - KeyValueContainerData |
| * @param delTX a block deletion transaction. |
| * @throws IOException if I/O error occurs. |
| */ |
| private void deleteKeyValueContainerBlocks( |
| KeyValueContainerData containerData, DeletedBlocksTransaction delTX) |
| throws IOException { |
| long containerId = delTX.getContainerID(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Processing Container : {}, DB path : {}", containerId, |
| containerData.getMetadataPath()); |
| } |
| |
| if (delTX.getTxID() < containerData.getDeleteTransactionId()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(String.format("Ignoring delete blocks for containerId: %d." |
| + " Outdated delete transactionId %d < %d", containerId, |
| delTX.getTxID(), containerData.getDeleteTransactionId())); |
| } |
| return; |
| } |
| |
| int newDeletionBlocks = 0; |
| try(ReferenceCountedDB containerDB = |
| BlockUtils.getDB(containerData, conf)) { |
| for (Long blk : delTX.getLocalIDList()) { |
| BatchOperation batch = new BatchOperation(); |
| byte[] blkBytes = Longs.toByteArray(blk); |
| byte[] blkInfo = containerDB.getStore().get(blkBytes); |
| if (blkInfo != null) { |
| byte[] deletingKeyBytes = |
| DFSUtil.string2Bytes(OzoneConsts.DELETING_KEY_PREFIX + blk); |
| byte[] deletedKeyBytes = |
| DFSUtil.string2Bytes(OzoneConsts.DELETED_KEY_PREFIX + blk); |
| if (containerDB.getStore().get(deletingKeyBytes) != null |
| || containerDB.getStore().get(deletedKeyBytes) != null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(String.format( |
| "Ignoring delete for block %d in container %d." |
| + " Entry already added.", blk, containerId)); |
| } |
| continue; |
| } |
| // Found the block in container db, |
| // use an atomic update to change its state to deleting. |
| batch.put(deletingKeyBytes, blkInfo); |
| batch.delete(blkBytes); |
| try { |
| containerDB.getStore().writeBatch(batch); |
| newDeletionBlocks++; |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Transited Block {} to DELETING state in container {}", |
| blk, containerId); |
| } |
| } catch (IOException e) { |
| // if some blocks failed to delete, we fail this TX, |
| // without sending this ACK to SCM, SCM will resend the TX |
| // with a certain number of retries. |
| throw new IOException( |
| "Failed to delete blocks for TXID = " + delTX.getTxID(), e); |
| } |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Block {} not found or already under deletion in" |
| + " container {}, skip deleting it.", blk, containerId); |
| } |
| } |
| } |
| |
| containerDB.getStore() |
| .put(DFSUtil.string2Bytes(OzoneConsts.DELETE_TRANSACTION_KEY_PREFIX), |
| Longs.toByteArray(delTX.getTxID())); |
| containerData |
| .updateDeleteTransactionId(delTX.getTxID()); |
| // update pending deletion blocks count in in-memory container status |
| containerData.incrPendingDeletionBlocks(newDeletionBlocks); |
| } |
| } |
| |
| @Override |
| public SCMCommandProto.Type getCommandType() { |
| return SCMCommandProto.Type.deleteBlocksCommand; |
| } |
| |
| @Override |
| public int getInvocationCount() { |
| return this.invocationCount; |
| } |
| |
| @Override |
| public long getAverageRunTime() { |
| if (invocationCount > 0) { |
| return totalTime / invocationCount; |
| } |
| return 0; |
| } |
| } |
