hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneChaosCluster.java - hadoop - Git at Google

 /**
  * 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
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * 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;

 import org.apache.commons.lang3.RandomUtils;
 import org.apache.hadoop.conf.StorageUnit;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.ozone.om.OzoneManager;
 import org.apache.hadoop.security.authentication.client.AuthenticationException;
 import org.apache.log4j.Level;
 import org.apache.ratis.grpc.client.GrpcClientProtocolClient;
 import org.apache.ratis.util.LogUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.IOException;
 import java.util.List;

 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.Executors;

 /**
  * This class causes random failures in the chaos cluster.
  */
 public class MiniOzoneChaosCluster extends MiniOzoneClusterImpl {

   static final Logger LOG =
       LoggerFactory.getLogger(MiniOzoneChaosCluster.class);

   private final int numDatanodes;
   private final ScheduledExecutorService executorService;

   private ScheduledFuture scheduledFuture;

   private enum FailureMode {
     NODES_RESTART,
     NODES_SHUTDOWN
   }

   public MiniOzoneChaosCluster(OzoneConfiguration conf,
                                OzoneManager ozoneManager,
                        StorageContainerManager scm,
                        List<HddsDatanodeService> hddsDatanodes) {
     super(conf, ozoneManager, scm, hddsDatanodes);

     this.executorService =  Executors.newSingleThreadScheduledExecutor();
     this.numDatanodes = getHddsDatanodes().size();
     LOG.info("Starting MiniOzoneChaosCluster with {} datanodes", numDatanodes);
     LogUtils.setLogLevel(GrpcClientProtocolClient.LOG, Level.WARN);
   }

   // Get the number of datanodes to fail in the cluster.
   private int getNumberOfNodesToFail() {
     return RandomUtils.nextBoolean() ? 1 : 2;
   }

   // Should the failed node wait for SCM to register the even before
   // restart, i.e fast restart or not.
   private boolean isFastRestart() {
     return RandomUtils.nextBoolean();
   }

   // Should the selected node be stopped or started.
   private boolean shouldStop() {
     return RandomUtils.nextBoolean();
   }

   // Get the datanode index of the datanode to fail.
   private int getNodeToFail() {
     return RandomUtils.nextInt() % numDatanodes;
   }

   private void restartNodes() {
     final int numNodesToFail = getNumberOfNodesToFail();
     LOG.info("Will restart {} nodes to simulate failure", numNodesToFail);
     for (int i = 0; i < numNodesToFail; i++) {
       boolean failureMode = isFastRestart();
       int failedNodeIndex = getNodeToFail();
       String failString = failureMode ? "Fast" : "Slow";
       DatanodeDetails dn =
           getHddsDatanodes().get(failedNodeIndex).getDatanodeDetails();
       try {
         LOG.info("{} Restarting DataNode: {}", failString, dn.getUuid());
         restartHddsDatanode(failedNodeIndex, failureMode);
         LOG.info("{} Completed restarting Datanode: {}", failString,
             dn.getUuid());
       } catch (Exception e) {
         LOG.error("Failed to restartNodes Datanode {}", dn.getUuid(), e);
       }
     }
   }

   private void shutdownNodes() {
     final int numNodesToFail = getNumberOfNodesToFail();
     LOG.info("Will shutdown {} nodes to simulate failure", numNodesToFail);
     for (int i = 0; i < numNodesToFail; i++) {
       boolean shouldStop = shouldStop();
       int failedNodeIndex = getNodeToFail();
       String stopString = shouldStop ? "Stopping" : "Restarting";
       DatanodeDetails dn =
           getHddsDatanodes().get(failedNodeIndex).getDatanodeDetails();
       try {
         LOG.info("{} DataNode {}", stopString, dn.getUuid());

         if (shouldStop) {
           shutdownHddsDatanode(failedNodeIndex);
         } else {
           restartHddsDatanode(failedNodeIndex, true);
         }
         LOG.info("Completed {} DataNode {}", stopString, dn.getUuid());

       } catch (Exception e) {
         LOG.error("Failed {} Datanode {}", stopString, dn.getUuid(), e);
       }
     }
   }

   private FailureMode getFailureMode() {
     return FailureMode.
         values()[RandomUtils.nextInt() % FailureMode.values().length];
   }

   // Fail nodes randomly at configured timeout period.
   private void fail() {
     FailureMode mode = getFailureMode();
     switch (mode) {
     case NODES_RESTART:
       restartNodes();
       break;
     case NODES_SHUTDOWN:
       shutdownNodes();
       break;

     default:
       LOG.error("invalid failure mode:{}", mode);
       break;
     }
   }

   void startChaos(long initialDelay, long period, TimeUnit timeUnit) {
     LOG.info("Starting Chaos with failure period:{} unit:{} numDataNodes:{}",
         period, timeUnit, numDatanodes);
     scheduledFuture = executorService.scheduleAtFixedRate(this::fail,
         initialDelay, period, timeUnit);
   }

   void stopChaos() throws Exception {
     if (scheduledFuture != null) {
       scheduledFuture.cancel(false);
       scheduledFuture.get();
     }
   }

   public void shutdown() {
     try {
       stopChaos();
       executorService.shutdown();
       executorService.awaitTermination(1, TimeUnit.DAYS);
       //this should be called after stopChaos to be sure that the
       //datanode collection is not modified during the shutdown
       super.shutdown();
     } catch (Exception e) {
       LOG.error("failed to shutdown MiniOzoneChaosCluster", e);
     }
   }

   /**
    * Builder for configuring the MiniOzoneChaosCluster to run.
    */
   public static class Builder extends MiniOzoneClusterImpl.Builder {

     /**
      * Creates a new Builder.
      *
      * @param conf configuration
      */
     public Builder(OzoneConfiguration conf) {
       super(conf);
     }

     /**
      * Sets the number of HddsDatanodes to be started as part of
      * MiniOzoneChaosCluster.
      *
      * @param val number of datanodes
      *
      * @return MiniOzoneChaosCluster.Builder
      */
     public Builder setNumDatanodes(int val) {
       super.setNumDatanodes(val);
       return this;
     }

     @Override
     void initializeConfiguration() throws IOException {
       super.initializeConfiguration();
       conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_KEY,
           2, StorageUnit.KB);
       conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE,
           16, StorageUnit.KB);
       conf.setStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_FLUSH_SIZE,
           4, StorageUnit.KB);
       conf.setStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_MAX_SIZE,
           8, StorageUnit.KB);
       conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
           1, StorageUnit.MB);
       conf.setTimeDuration(ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT, 1000,
           TimeUnit.MILLISECONDS);
       conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL, 10,
           TimeUnit.SECONDS);
       conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL, 20,
           TimeUnit.SECONDS);
       conf.setTimeDuration(HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL, 1,
           TimeUnit.SECONDS);
       conf.setTimeDuration(HddsConfigKeys.HDDS_PIPELINE_REPORT_INTERVAL, 1,
           TimeUnit.SECONDS);
       conf.setTimeDuration(
           ScmConfigKeys.OZONE_SCM_CONTAINER_CREATION_LEASE_TIMEOUT, 5,
           TimeUnit.SECONDS);
       conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
           1, TimeUnit.SECONDS);
       conf.setTimeDuration(HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL, 1,
           TimeUnit.SECONDS);
       conf.setInt(OzoneConfigKeys.OZONE_CONTAINER_CACHE_SIZE, 2);
       conf.setInt("hdds.scm.replication.thread.interval", 10 * 1000);
       conf.setInt("hdds.scm.replication.event.timeout", 20 * 1000);
       conf.setInt(OzoneConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_KEY, 100);
       conf.setInt(OzoneConfigKeys.DFS_CONTAINER_RATIS_LOG_PURGE_GAP, 100);
     }

     @Override
     public MiniOzoneChaosCluster build() throws IOException {
       DefaultMetricsSystem.setMiniClusterMode(true);
       initializeConfiguration();
       StorageContainerManager scm;
       OzoneManager om;
       try {
         scm = createSCM();
         scm.start();
         om = createOM();
         if(certClient != null) {
           om.setCertClient(certClient);
         }
       } catch (AuthenticationException ex) {
         throw new IOException("Unable to build MiniOzoneCluster. ", ex);
       }

       om.start();
       final List<HddsDatanodeService> hddsDatanodes = createHddsDatanodes(scm);
       MiniOzoneChaosCluster cluster =
           new MiniOzoneChaosCluster(conf, om, scm, hddsDatanodes);
       if (startDataNodes) {
         cluster.startHddsDatanodes();
       }
       return cluster;
     }
   }
 }
	/**
	* 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
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* 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;

	import org.apache.commons.lang3.RandomUtils;
	import org.apache.hadoop.conf.StorageUnit;
	import org.apache.hadoop.hdds.HddsConfigKeys;
	import org.apache.hadoop.hdds.conf.OzoneConfiguration;
	import org.apache.hadoop.hdds.protocol.DatanodeDetails;
	import org.apache.hadoop.hdds.scm.ScmConfigKeys;
	import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
	import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
	import org.apache.hadoop.ozone.om.OzoneManager;
	import org.apache.hadoop.security.authentication.client.AuthenticationException;
	import org.apache.log4j.Level;
	import org.apache.ratis.grpc.client.GrpcClientProtocolClient;
	import org.apache.ratis.util.LogUtils;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.IOException;
	import java.util.List;

	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ScheduledFuture;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.Executors;

	/**
	* This class causes random failures in the chaos cluster.
	*/
	public class MiniOzoneChaosCluster extends MiniOzoneClusterImpl {

	static final Logger LOG =
	LoggerFactory.getLogger(MiniOzoneChaosCluster.class);

	private final int numDatanodes;
	private final ScheduledExecutorService executorService;

	private ScheduledFuture scheduledFuture;

	private enum FailureMode {
	NODES_RESTART,
	NODES_SHUTDOWN
	}

	public MiniOzoneChaosCluster(OzoneConfiguration conf,
	OzoneManager ozoneManager,
	StorageContainerManager scm,
	List<HddsDatanodeService> hddsDatanodes) {
	super(conf, ozoneManager, scm, hddsDatanodes);

	this.executorService = Executors.newSingleThreadScheduledExecutor();
	this.numDatanodes = getHddsDatanodes().size();
	LOG.info("Starting MiniOzoneChaosCluster with {} datanodes", numDatanodes);
	LogUtils.setLogLevel(GrpcClientProtocolClient.LOG, Level.WARN);
	}

	// Get the number of datanodes to fail in the cluster.
	private int getNumberOfNodesToFail() {
	return RandomUtils.nextBoolean() ? 1 : 2;
	}

	// Should the failed node wait for SCM to register the even before
	// restart, i.e fast restart or not.
	private boolean isFastRestart() {
	return RandomUtils.nextBoolean();
	}

	// Should the selected node be stopped or started.
	private boolean shouldStop() {
	return RandomUtils.nextBoolean();
	}

	// Get the datanode index of the datanode to fail.
	private int getNodeToFail() {
	return RandomUtils.nextInt() % numDatanodes;
	}

	private void restartNodes() {
	final int numNodesToFail = getNumberOfNodesToFail();
	LOG.info("Will restart {} nodes to simulate failure", numNodesToFail);
	for (int i = 0; i < numNodesToFail; i++) {
	boolean failureMode = isFastRestart();
	int failedNodeIndex = getNodeToFail();
	String failString = failureMode ? "Fast" : "Slow";
	DatanodeDetails dn =
	getHddsDatanodes().get(failedNodeIndex).getDatanodeDetails();
	try {
	LOG.info("{} Restarting DataNode: {}", failString, dn.getUuid());
	restartHddsDatanode(failedNodeIndex, failureMode);
	LOG.info("{} Completed restarting Datanode: {}", failString,
	dn.getUuid());
	} catch (Exception e) {
	LOG.error("Failed to restartNodes Datanode {}", dn.getUuid(), e);
	}
	}
	}

	private void shutdownNodes() {
	final int numNodesToFail = getNumberOfNodesToFail();
	LOG.info("Will shutdown {} nodes to simulate failure", numNodesToFail);
	for (int i = 0; i < numNodesToFail; i++) {
	boolean shouldStop = shouldStop();
	int failedNodeIndex = getNodeToFail();
	String stopString = shouldStop ? "Stopping" : "Restarting";
	DatanodeDetails dn =
	getHddsDatanodes().get(failedNodeIndex).getDatanodeDetails();
	try {
	LOG.info("{} DataNode {}", stopString, dn.getUuid());

	if (shouldStop) {
	shutdownHddsDatanode(failedNodeIndex);
	} else {
	restartHddsDatanode(failedNodeIndex, true);
	}
	LOG.info("Completed {} DataNode {}", stopString, dn.getUuid());

	} catch (Exception e) {
	LOG.error("Failed {} Datanode {}", stopString, dn.getUuid(), e);
	}
	}
	}

	private FailureMode getFailureMode() {
	return FailureMode.
	values()[RandomUtils.nextInt() % FailureMode.values().length];
	}

	// Fail nodes randomly at configured timeout period.
	private void fail() {
	FailureMode mode = getFailureMode();
	switch (mode) {
	case NODES_RESTART:
	restartNodes();
	break;
	case NODES_SHUTDOWN:
	shutdownNodes();
	break;

	default:
	LOG.error("invalid failure mode:{}", mode);
	break;
	}
	}

	void startChaos(long initialDelay, long period, TimeUnit timeUnit) {
	LOG.info("Starting Chaos with failure period:{} unit:{} numDataNodes:{}",
	period, timeUnit, numDatanodes);
	scheduledFuture = executorService.scheduleAtFixedRate(this::fail,
	initialDelay, period, timeUnit);
	}

	void stopChaos() throws Exception {
	if (scheduledFuture != null) {
	scheduledFuture.cancel(false);
	scheduledFuture.get();
	}
	}

	public void shutdown() {
	try {
	stopChaos();
	executorService.shutdown();
	executorService.awaitTermination(1, TimeUnit.DAYS);
	//this should be called after stopChaos to be sure that the
	//datanode collection is not modified during the shutdown
	super.shutdown();
	} catch (Exception e) {
	LOG.error("failed to shutdown MiniOzoneChaosCluster", e);
	}
	}

	/**
	* Builder for configuring the MiniOzoneChaosCluster to run.
	*/
	public static class Builder extends MiniOzoneClusterImpl.Builder {

	/**
	* Creates a new Builder.
	*
	* @param conf configuration
	*/
	public Builder(OzoneConfiguration conf) {
	super(conf);
	}

	/**
	* Sets the number of HddsDatanodes to be started as part of
	* MiniOzoneChaosCluster.
	*
	* @param val number of datanodes
	*
	* @return MiniOzoneChaosCluster.Builder
	*/
	public Builder setNumDatanodes(int val) {
	super.setNumDatanodes(val);
	return this;
	}

	@Override
	void initializeConfiguration() throws IOException {
	super.initializeConfiguration();
	conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_KEY,
	2, StorageUnit.KB);
	conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE,
	16, StorageUnit.KB);
	conf.setStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_FLUSH_SIZE,
	4, StorageUnit.KB);
	conf.setStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_MAX_SIZE,
	8, StorageUnit.KB);
	conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
	1, StorageUnit.MB);
	conf.setTimeDuration(ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT, 1000,
	TimeUnit.MILLISECONDS);
	conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL, 10,
	TimeUnit.SECONDS);
	conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL, 20,
	TimeUnit.SECONDS);
	conf.setTimeDuration(HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL, 1,
	TimeUnit.SECONDS);
	conf.setTimeDuration(HddsConfigKeys.HDDS_PIPELINE_REPORT_INTERVAL, 1,
	TimeUnit.SECONDS);
	conf.setTimeDuration(
	ScmConfigKeys.OZONE_SCM_CONTAINER_CREATION_LEASE_TIMEOUT, 5,
	TimeUnit.SECONDS);
	conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
	1, TimeUnit.SECONDS);
	conf.setTimeDuration(HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL, 1,
	TimeUnit.SECONDS);
	conf.setInt(OzoneConfigKeys.OZONE_CONTAINER_CACHE_SIZE, 2);
	conf.setInt("hdds.scm.replication.thread.interval", 10 * 1000);
	conf.setInt("hdds.scm.replication.event.timeout", 20 * 1000);
	conf.setInt(OzoneConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_KEY, 100);
	conf.setInt(OzoneConfigKeys.DFS_CONTAINER_RATIS_LOG_PURGE_GAP, 100);
	}

	@Override
	public MiniOzoneChaosCluster build() throws IOException {
	DefaultMetricsSystem.setMiniClusterMode(true);
	initializeConfiguration();
	StorageContainerManager scm;
	OzoneManager om;
	try {
	scm = createSCM();
	scm.start();
	om = createOM();
	if(certClient != null) {
	om.setCertClient(certClient);
	}
	} catch (AuthenticationException ex) {
	throw new IOException("Unable to build MiniOzoneCluster. ", ex);
	}

	om.start();
	final List<HddsDatanodeService> hddsDatanodes = createHddsDatanodes(scm);
	MiniOzoneChaosCluster cluster =
	new MiniOzoneChaosCluster(conf, om, scm, hddsDatanodes);
	if (startDataNodes) {
	cluster.startHddsDatanodes();
	}
	return cluster;
	}
	}
	}