| /** |
| * 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.hdds.scm.block; |
| |
| import javax.management.ObjectName; |
| import java.io.IOException; |
| import java.time.Duration; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.hadoop.hdds.client.BlockID; |
| import org.apache.hadoop.hdds.client.ContainerBlockID; |
| import org.apache.hadoop.hdds.conf.ConfigurationSource; |
| import org.apache.hadoop.hdds.conf.StorageUnit; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ScmOps; |
| import org.apache.hadoop.hdds.scm.PipelineRequestInformation; |
| import org.apache.hadoop.hdds.scm.ScmConfigKeys; |
| import org.apache.hadoop.hdds.scm.ScmUtils; |
| import org.apache.hadoop.hdds.scm.PipelineChoosePolicy; |
| import org.apache.hadoop.hdds.scm.ScmConfig; |
| import org.apache.hadoop.hdds.scm.container.ContainerInfo; |
| import org.apache.hadoop.hdds.scm.container.ContainerManager; |
| import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock; |
| import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; |
| import org.apache.hadoop.hdds.scm.exceptions.SCMException; |
| import org.apache.hadoop.hdds.scm.pipeline.Pipeline; |
| import org.apache.hadoop.hdds.scm.pipeline.PipelineManager; |
| import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException; |
| import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager.SafeModeStatus; |
| import org.apache.hadoop.hdds.scm.safemode.SafeModePrecheck; |
| import org.apache.hadoop.hdds.scm.server.StorageContainerManager; |
| import org.apache.hadoop.hdds.server.events.EventPublisher; |
| import org.apache.hadoop.hdds.utils.UniqueId; |
| import org.apache.hadoop.metrics2.util.MBeans; |
| import org.apache.hadoop.ozone.common.BlockGroup; |
| import org.apache.hadoop.util.StringUtils; |
| |
| import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.INVALID_BLOCK_SIZE; |
| import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_TIMEOUT; |
| import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_TIMEOUT_DEFAULT; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| |
| /** Block Manager manages the block access for SCM. */ |
| public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean { |
| private static final Logger LOG = |
| LoggerFactory.getLogger(BlockManagerImpl.class); |
| // TODO : FIX ME : Hard coding the owner. |
| // Currently only user of the block service is Ozone, CBlock manages blocks |
| // by itself and does not rely on the Block service offered by SCM. |
| |
| private final PipelineManager pipelineManager; |
| private final ContainerManager containerManager; |
| |
| private final long containerSize; |
| |
| private final DeletedBlockLog deletedBlockLog; |
| private final SCMBlockDeletingService blockDeletingService; |
| |
| private ObjectName mxBean; |
| private SafeModePrecheck safeModePrecheck; |
| private PipelineChoosePolicy pipelineChoosePolicy; |
| |
| /** |
| * Constructor. |
| * |
| * @param conf - configuration. |
| * @param scm |
| * @throws IOException |
| */ |
| public BlockManagerImpl(final ConfigurationSource conf, |
| final StorageContainerManager scm) { |
| Objects.requireNonNull(scm, "SCM cannot be null"); |
| this.pipelineManager = scm.getPipelineManager(); |
| this.containerManager = scm.getContainerManager(); |
| this.pipelineChoosePolicy = scm.getPipelineChoosePolicy(); |
| this.containerSize = (long)conf.getStorageSize( |
| ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, |
| ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, |
| StorageUnit.BYTES); |
| |
| mxBean = MBeans.register("BlockManager", "BlockManagerImpl", this); |
| |
| // SCM block deleting transaction log and deleting service. |
| deletedBlockLog = new DeletedBlockLogImpl(conf, scm.getContainerManager(), |
| scm.getScmMetadataStore()); |
| Duration svcInterval = conf.getObject( |
| ScmConfig.class).getBlockDeletionInterval(); |
| long serviceTimeout = |
| conf.getTimeDuration( |
| OZONE_BLOCK_DELETING_SERVICE_TIMEOUT, |
| OZONE_BLOCK_DELETING_SERVICE_TIMEOUT_DEFAULT, |
| TimeUnit.MILLISECONDS); |
| blockDeletingService = |
| new SCMBlockDeletingService(deletedBlockLog, containerManager, |
| scm.getScmNodeManager(), scm.getEventQueue(), svcInterval, |
| serviceTimeout, conf); |
| safeModePrecheck = new SafeModePrecheck(conf); |
| } |
| |
| /** |
| * Start block manager services. |
| * |
| * @throws IOException |
| */ |
| public void start() throws IOException { |
| this.blockDeletingService.start(); |
| } |
| |
| /** |
| * Shutdown block manager services. |
| * |
| * @throws IOException |
| */ |
| public void stop() throws IOException { |
| this.blockDeletingService.shutdown(); |
| this.close(); |
| } |
| |
| /** |
| * Allocates a block in a container and returns that info. |
| * |
| * @param size - Block Size |
| * @param type Replication Type |
| * @param factor - Replication Factor |
| * @param excludeList List of datanodes/containers to exclude during block |
| * allocation. |
| * @return Allocated block |
| * @throws IOException on failure. |
| */ |
| @Override |
| public AllocatedBlock allocateBlock(final long size, ReplicationType type, |
| ReplicationFactor factor, String owner, ExcludeList excludeList) |
| throws IOException { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Size : {} , type : {}, factor : {} ", size, type, factor); |
| } |
| ScmUtils.preCheck(ScmOps.allocateBlock, safeModePrecheck); |
| if (size < 0 || size > containerSize) { |
| LOG.warn("Invalid block size requested : {}", size); |
| throw new SCMException("Unsupported block size: " + size, |
| INVALID_BLOCK_SIZE); |
| } |
| |
| /* |
| Here is the high level logic. |
| |
| 1. We try to find pipelines in open state. |
| |
| 2. If there are no pipelines in OPEN state, then we try to create one. |
| |
| 3. We allocate a block from the available containers in the selected |
| pipeline. |
| |
| TODO : #CLUTIL Support random picking of two containers from the list. |
| So we can use different kind of policies. |
| */ |
| |
| ContainerInfo containerInfo; |
| |
| while (true) { |
| List<Pipeline> availablePipelines = |
| pipelineManager |
| .getPipelines(type, factor, Pipeline.PipelineState.OPEN, |
| excludeList.getDatanodes(), excludeList.getPipelineIds()); |
| Pipeline pipeline = null; |
| if (availablePipelines.size() == 0 && !excludeList.isEmpty()) { |
| // if no pipelines can be found, try finding pipeline without |
| // exclusion |
| availablePipelines = pipelineManager |
| .getPipelines(type, factor, Pipeline.PipelineState.OPEN); |
| } |
| if (availablePipelines.size() == 0) { |
| try { |
| // TODO: #CLUTIL Remove creation logic when all replication types and |
| // factors are handled by pipeline creator |
| pipeline = pipelineManager.createPipeline(type, factor); |
| |
| // wait until pipeline is ready |
| pipelineManager.waitPipelineReady(pipeline.getId(), 0); |
| } catch (SCMException se) { |
| LOG.warn("Pipeline creation failed for type:{} factor:{}. " + |
| "Datanodes may be used up.", type, factor, se); |
| break; |
| } catch (IOException e) { |
| LOG.warn("Pipeline creation failed for type:{} factor:{}. Retrying " + |
| "get pipelines call once.", type, factor, e); |
| availablePipelines = pipelineManager |
| .getPipelines(type, factor, Pipeline.PipelineState.OPEN, |
| excludeList.getDatanodes(), excludeList.getPipelineIds()); |
| if (availablePipelines.size() == 0 && !excludeList.isEmpty()) { |
| // if no pipelines can be found, try finding pipeline without |
| // exclusion |
| availablePipelines = pipelineManager |
| .getPipelines(type, factor, Pipeline.PipelineState.OPEN); |
| } |
| if (availablePipelines.size() == 0) { |
| LOG.info("Could not find available pipeline of type:{} and " + |
| "factor:{} even after retrying", type, factor); |
| break; |
| } |
| } |
| } |
| |
| if (null == pipeline) { |
| PipelineRequestInformation pri = |
| PipelineRequestInformation.Builder.getBuilder() |
| .setSize(size) |
| .build(); |
| pipeline = pipelineChoosePolicy.choosePipeline( |
| availablePipelines, pri); |
| } |
| |
| // look for OPEN containers that match the criteria. |
| containerInfo = containerManager.getMatchingContainer(size, owner, |
| pipeline, excludeList.getContainerIds()); |
| |
| if (containerInfo != null) { |
| return newBlock(containerInfo); |
| } |
| } |
| |
| // we have tried all strategies we know and but somehow we are not able |
| // to get a container for this block. Log that info and return a null. |
| LOG.error( |
| "Unable to allocate a block for the size: {}, type: {}, factor: {}", |
| size, type, factor); |
| return null; |
| } |
| |
| /** |
| * newBlock - returns a new block assigned to a container. |
| * |
| * @param containerInfo - Container Info. |
| * @return AllocatedBlock |
| */ |
| private AllocatedBlock newBlock(ContainerInfo containerInfo) { |
| try { |
| final Pipeline pipeline = pipelineManager |
| .getPipeline(containerInfo.getPipelineID()); |
| // TODO : Revisit this local ID allocation when HA is added. |
| long localID = UniqueId.next(); |
| long containerID = containerInfo.getContainerID(); |
| AllocatedBlock.Builder abb = new AllocatedBlock.Builder() |
| .setContainerBlockID(new ContainerBlockID(containerID, localID)) |
| .setPipeline(pipeline); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("New block allocated : {} Container ID: {}", localID, |
| containerID); |
| } |
| pipelineManager.incNumBlocksAllocatedMetric(pipeline.getId()); |
| return abb.build(); |
| } catch (PipelineNotFoundException ex) { |
| LOG.error("Pipeline Machine count is zero.", ex); |
| return null; |
| } |
| } |
| |
| /** |
| * Deletes a list of blocks in an atomic operation. Internally, SCM writes |
| * these blocks into a |
| * {@link DeletedBlockLog} and deletes them from SCM DB. If this is |
| * successful, given blocks are |
| * entering pending deletion state and becomes invisible from SCM namespace. |
| * |
| * @param keyBlocksInfoList . This is the list of BlockGroup which contains |
| * groupID of keys and list of BlockIDs associated with them. |
| * @throws IOException if exception happens, non of the blocks is deleted. |
| */ |
| @Override |
| public void deleteBlocks(List<BlockGroup> keyBlocksInfoList) |
| throws IOException { |
| ScmUtils.preCheck(ScmOps.deleteBlock, safeModePrecheck); |
| |
| Map<Long, List<Long>> containerBlocks = new HashMap<>(); |
| // TODO: track the block size info so that we can reclaim the container |
| // TODO: used space when the block is deleted. |
| for (BlockGroup bg : keyBlocksInfoList) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Deleting blocks {}", |
| StringUtils.join(",", bg.getBlockIDList())); |
| } |
| for (BlockID block : bg.getBlockIDList()) { |
| long containerID = block.getContainerID(); |
| if (containerBlocks.containsKey(containerID)) { |
| containerBlocks.get(containerID).add(block.getLocalID()); |
| } else { |
| List<Long> item = new ArrayList<>(); |
| item.add(block.getLocalID()); |
| containerBlocks.put(containerID, item); |
| } |
| } |
| } |
| |
| try { |
| deletedBlockLog.addTransactions(containerBlocks); |
| } catch (IOException e) { |
| throw new IOException("Skip writing the deleted blocks info to" |
| + " the delLog because addTransaction fails. " + keyBlocksInfoList |
| .size() + "Keys skipped", e); |
| } |
| // TODO: Container report handling of the deleted blocks: |
| // Remove tombstone and update open container usage. |
| // We will revisit this when the closed container replication is done. |
| } |
| |
| @Override |
| public DeletedBlockLog getDeletedBlockLog() { |
| return this.deletedBlockLog; |
| } |
| |
| /** |
| * Close the resources for BlockManager. |
| * |
| * @throws IOException |
| */ |
| @Override |
| public void close() throws IOException { |
| if (deletedBlockLog != null) { |
| deletedBlockLog.close(); |
| } |
| blockDeletingService.shutdown(); |
| if (mxBean != null) { |
| MBeans.unregister(mxBean); |
| mxBean = null; |
| } |
| } |
| |
| @Override |
| public int getOpenContainersNo() { |
| return 0; |
| // TODO : FIX ME : The open container being a single number does not make |
| // sense. |
| // We have to get open containers by Replication Type and Replication |
| // factor. Hence returning 0 for now. |
| // containers.get(HddsProtos.LifeCycleState.OPEN).size(); |
| } |
| |
| @Override |
| public SCMBlockDeletingService getSCMBlockDeletingService() { |
| return this.blockDeletingService; |
| } |
| |
| /** |
| * Returns status of scm safe mode determined by SAFE_MODE_STATUS event. |
| * */ |
| public boolean isScmInSafeMode() { |
| return this.safeModePrecheck.isInSafeMode(); |
| } |
| |
| /** |
| * Get class logger. |
| * */ |
| public static Logger getLogger() { |
| return LOG; |
| } |
| |
| @Override |
| public void onMessage(SafeModeStatus status, |
| EventPublisher publisher) { |
| this.safeModePrecheck.setInSafeMode(status.isInSafeMode()); |
| } |
| |
| /** |
| * This class uses system current time milliseconds to generate unique id. |
| */ |
| } |
