symious commented on code in PR #3874: URL: https://github.com/apache/ozone/pull/3874#discussion_r1105476733
########## hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java: ########## @@ -0,0 +1,374 @@ +/* + * 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.diskbalancer; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import org.apache.hadoop.hdds.conf.ConfigurationSource; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DiskBalancerReportProto; +import org.apache.hadoop.hdds.scm.storage.DiskBalancerConfiguration; +import org.apache.hadoop.hdds.server.ServerUtils; +import org.apache.hadoop.hdds.utils.BackgroundService; +import org.apache.hadoop.hdds.utils.BackgroundTask; +import org.apache.hadoop.hdds.utils.BackgroundTaskQueue; +import org.apache.hadoop.hdds.utils.BackgroundTaskResult; +import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.container.common.impl.ContainerData; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration; +import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil; +import org.apache.hadoop.ozone.container.common.volume.HddsVolume; +import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet; +import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; +import org.apache.hadoop.util.Time; +import org.apache.ratis.util.FileUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import static org.apache.hadoop.hdds.scm.storage.DiskBalancerConfiguration.HDDS_DATANODE_DISK_BALANCER_INFO_DIR; + + +/** + * A per-datanode disk balancing service takes in charge + * of moving contains among disks. + */ +public class DiskBalancerService extends BackgroundService { + + private static final Logger LOG = + LoggerFactory.getLogger(DiskBalancerService.class); + + public static final String DISK_BALANCER_DIR = "diskBalancer"; + + private OzoneContainer ozoneContainer; + private final ConfigurationSource conf; + + private boolean shouldRun = false; + private double threshold; + private long bandwidthInMB; + private int parallelThread; + + private AtomicLong totalBalancedBytes = new AtomicLong(0L); + private AtomicLong balancedBytesInLastWindow = new AtomicLong(0L); + private AtomicLong nextAvailableTime = new AtomicLong(Time.monotonicNow()); + + private Map<DiskBalancerTask, Integer> inProgressTasks; + private Set<Long> inProgressContainers; + private Map<HddsVolume, Long> deltaSizes; + private MutableVolumeSet volumeSet; + + private final File diskBalancerInfoFile; + + private DiskBalancerServiceMetrics metrics; + + public DiskBalancerService(OzoneContainer ozoneContainer, + long serviceCheckInterval, long serviceCheckTimeout, TimeUnit timeUnit, + int workerSize, ConfigurationSource conf) throws IOException { + super("DiskBalancerService", serviceCheckInterval, timeUnit, workerSize, + serviceCheckTimeout); + this.ozoneContainer = ozoneContainer; + this.conf = conf; + + String diskBalancerInfoPath = getDiskBalancerInfoPath(); + Preconditions.checkNotNull(diskBalancerInfoPath); + diskBalancerInfoFile = new File(diskBalancerInfoPath); + + inProgressTasks = new ConcurrentHashMap<>(); + inProgressContainers = ConcurrentHashMap.newKeySet(); + deltaSizes = new ConcurrentHashMap<>(); + volumeSet = ozoneContainer.getVolumeSet(); + + metrics = DiskBalancerServiceMetrics.create(); + + loadDiskBalancerInfo(); + + constructTmpDir(); + } + + /** + * Update DiskBalancerService based on new DiskBalancerInfo. + * @param diskBalancerInfo + * @throws IOException + */ + public void refresh(DiskBalancerInfo diskBalancerInfo) throws IOException { + applyDiskBalancerInfo(diskBalancerInfo); + } + + private void constructTmpDir() { + for (HddsVolume volume: + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())) { + Path tmpDir = getDiskBalancerTmpDir(volume); + try { + FileUtils.deleteFully(tmpDir); + FileUtils.createDirectories(tmpDir); + } catch (IOException ex) { + LOG.warn("Can not reconstruct tmp directory under volume {}", volume, + ex); + } + } + } + + /** + * If the diskBalancer.info file exists, load the file. If not exists, + * return the default config. + * @throws IOException + */ + private void loadDiskBalancerInfo() throws IOException { + DiskBalancerInfo diskBalancerInfo = null; + try { + if (diskBalancerInfoFile.exists()) { + diskBalancerInfo = readDiskBalancerInfoFile(diskBalancerInfoFile); + } + } catch (IOException e) { + LOG.warn("Can not load diskBalancerInfo from diskBalancer.info file. " + + "Falling back to default configs", e); + } finally { + if (diskBalancerInfo == null) { + boolean shouldRunDefault = conf.getObject(DatanodeConfiguration.class) Review Comment: > warn and throw exception Does that mean the datanode will quit when loading file is failed? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
