| /** |
| * 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.mapred; |
| |
| import java.io.IOException; |
| import java.io.PrintStream; |
| import java.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.regex.Pattern; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.conf.Configured; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.mapred.SimulatorEvent; |
| import org.apache.hadoop.mapred.SimulatorEventQueue; |
| import org.apache.hadoop.mapred.JobCompleteEvent; |
| import org.apache.hadoop.mapred.SimulatorJobClient; |
| import org.apache.hadoop.mapred.SimulatorJobTracker; |
| import org.apache.hadoop.mapred.SimulatorTaskTracker; |
| import org.apache.hadoop.net.DNSToSwitchMapping; |
| import org.apache.hadoop.net.StaticMapping; |
| import org.apache.hadoop.tools.rumen.ClusterStory; |
| import org.apache.hadoop.tools.rumen.ClusterTopologyReader; |
| import org.apache.hadoop.tools.rumen.JobStoryProducer; |
| import org.apache.hadoop.tools.rumen.LoggedNetworkTopology; |
| import org.apache.hadoop.tools.rumen.MachineNode; |
| import org.apache.hadoop.tools.rumen.RackNode; |
| import org.apache.hadoop.tools.rumen.ZombieCluster; |
| import org.apache.hadoop.util.Tool; |
| import org.apache.hadoop.util.ToolRunner; |
| |
| /** |
| * {@link SimulatorEngine} is the main class of the simulator. To launch the |
| * simulator, user can either run the main class directly with two parameters, |
| * input trace file and corresponding topology file, or use the script |
| * "bin/mumak.sh trace.json topology.json". Trace file and topology file are |
| * produced by rumen. |
| */ |
| public class SimulatorEngine extends Configured implements Tool { |
| public static final List<SimulatorEvent> EMPTY_EVENTS = new ArrayList<SimulatorEvent>(); |
| private static final int DEFAULT_MAP_SLOTS_PER_NODE = 2; |
| private static final int DEFAULT_REDUCE_SLOTS_PER_NODE = 2; |
| |
| protected final SimulatorEventQueue queue = new SimulatorEventQueue(); |
| String traceFile; |
| String topologyFile; |
| SimulatorJobTracker jt; |
| SimulatorJobClient jc; |
| boolean shutdown = false; |
| long terminateTime = Long.MAX_VALUE; |
| long currentTime; |
| |
| /** |
| * Start simulated task trackers based on topology. |
| * @param clusterStory The cluster topology. |
| * @param now |
| * time stamp when the simulator is started, {@link SimulatorTaskTracker}s |
| * are started shortly after this time stamp |
| */ |
| void startTaskTrackers(ClusterStory clusterStory, long now) { |
| /** port assigned to TTs, incremented by 1 for each TT */ |
| int port = 10000; |
| long ms = now + 100; |
| |
| for (MachineNode node : clusterStory.getMachines()) { |
| String hostname = node.getName(); |
| String taskTrackerName = "tracker_" + hostname + ":localhost/127.0.0.1:" |
| + port; |
| port++; |
| SimulatorTaskTracker tt = new SimulatorTaskTracker(jt, taskTrackerName, |
| hostname, node.getMapSlots(), node.getReduceSlots()); |
| queue.addAll(tt.init(ms++)); |
| } |
| } |
| |
| /** |
| * Initiate components in the simulation. |
| * @throws InterruptedException |
| * @throws IOException if trace or topology files cannot be open |
| */ |
| @SuppressWarnings("deprecation") |
| void init() throws InterruptedException, IOException { |
| long now = System.currentTimeMillis(); |
| |
| JobConf jobConf = new JobConf(getConf()); |
| jobConf.setClass("topology.node.switch.mapping.impl", |
| StaticMapping.class, DNSToSwitchMapping.class); |
| jobConf.set("fs.default.name", "file:///"); |
| jobConf.set("mapred.job.tracker", "localhost:8012"); |
| jobConf.setInt("mapred.jobtracker.job.history.block.size", 512); |
| jobConf.setInt("mapred.jobtracker.job.history.buffer.size", 512); |
| jobConf.setLong("mapred.tasktracker.expiry.interval", 5000); |
| jobConf.setInt("mapred.reduce.copy.backoff", 4); |
| jobConf.setLong("mapred.job.reuse.jvm.num.tasks", -1); |
| jobConf.setUser("mumak"); |
| jobConf.set("mapred.system.dir", |
| jobConf.get("hadoop.log.dir", "/tmp/hadoop-"+jobConf.getUser()) + "/mapred/system"); |
| jobConf.set("mapred.jobtracker.taskScheduler", JobQueueTaskScheduler.class.getName()); |
| |
| FileSystem lfs = FileSystem.getLocal(getConf()); |
| Path logPath = |
| new Path(System.getProperty("hadoop.log.dir")).makeQualified(lfs); |
| jobConf.set("mapred.system.dir", logPath.toString()); |
| jobConf.set("hadoop.job.history.location", (new Path(logPath, "history") |
| .toString())); |
| |
| jt = SimulatorJobTracker.startTracker(jobConf, now, this); |
| jt.offerService(); |
| |
| // max Map/Reduce tasks per node |
| int maxMaps = getConf().getInt("mapred.tasktracker.map.tasks.maximum", |
| DEFAULT_MAP_SLOTS_PER_NODE); |
| int maxReduces = getConf().getInt( |
| "mapred.tasktracker.reduce.tasks.maximum", |
| DEFAULT_REDUCE_SLOTS_PER_NODE); |
| |
| MachineNode defaultNode = new MachineNode.Builder("default", 2) |
| .setMapSlots(maxMaps).setReduceSlots(maxReduces).build(); |
| |
| LoggedNetworkTopology topology = new ClusterTopologyReader(new Path( |
| topologyFile), jobConf).get(); |
| // Setting the static mapping before removing numeric IP hosts. |
| setStaticMapping(topology); |
| if (getConf().getBoolean("mumak.topology.filter-numeric-ips", true)) { |
| removeIpHosts(topology); |
| } |
| ZombieCluster cluster = new ZombieCluster(topology, defaultNode); |
| long firstJobStartTime = now + 60000; |
| JobStoryProducer jobStoryProducer = new SimulatorJobStoryProducer( |
| new Path(traceFile), cluster, firstJobStartTime, jobConf); |
| |
| final SimulatorJobSubmissionPolicy submissionPolicy = SimulatorJobSubmissionPolicy |
| .getPolicy(jobConf); |
| |
| jc = new SimulatorJobClient(jt, jobStoryProducer, submissionPolicy); |
| queue.addAll(jc.init(firstJobStartTime)); |
| |
| // create TTs based on topology.json |
| startTaskTrackers(cluster, now); |
| |
| terminateTime = getConf().getLong("mumak.terminate.time", Long.MAX_VALUE); |
| if (terminateTime <= 0) { |
| throw new IllegalArgumentException("Terminate time must be positive: " |
| + terminateTime); |
| } |
| } |
| |
| /** |
| * The main loop of the simulation. First call init() to get objects ready, |
| * then go into the main loop, where {@link SimulatorEvent}s are handled removed from |
| * the {@link SimulatorEventQueue}, and new {@link SimulatorEvent}s are created and inserted |
| * into the {@link SimulatorEventQueue}. |
| * @throws IOException |
| * @throws InterruptedException |
| */ |
| void run() throws IOException, InterruptedException { |
| init(); |
| |
| for (SimulatorEvent next = queue.get(); next != null |
| && next.getTimeStamp() < terminateTime && !shutdown; next = queue.get()) { |
| currentTime = next.getTimeStamp(); |
| assert(currentTime == queue.getCurrentTime()); |
| SimulatorEventListener listener = next.getListener(); |
| List<SimulatorEvent> response = listener.accept(next); |
| queue.addAll(response); |
| } |
| |
| summary(System.out); |
| } |
| |
| /** |
| * Run after the main loop. |
| * @param out stream to output information about the simulation |
| */ |
| void summary(PrintStream out) { |
| out.println("Done, total events processed: " + queue.getEventCount()); |
| } |
| |
| public static void main(String[] args) throws Exception { |
| int res = ToolRunner.run(new Configuration(), new SimulatorEngine(), args); |
| System.exit(res); |
| } |
| |
| @Override |
| public int run(String[] args) throws Exception { |
| parseParameters(args); |
| try { |
| run(); |
| return 0; |
| } finally { |
| if (jt != null) { |
| jt.getTaskScheduler().terminate(); |
| } |
| } |
| } |
| |
| void parseParameters(String[] args) { |
| if (args.length != 2) { |
| throw new IllegalArgumentException("Usage: java ... SimulatorEngine trace.json topology.json"); |
| } |
| traceFile = args[0]; |
| topologyFile = args[1]; |
| } |
| |
| /** |
| * Called when a job is completed. Insert a {@link JobCompleteEvent} into the |
| * {@link SimulatorEventQueue}. This event will be picked up by |
| * {@link SimulatorJobClient}, which will in turn decide whether the |
| * simulation is done. |
| * @param jobStatus final status of a job, SUCCEEDED or FAILED |
| * @param timestamp time stamp when the job is completed |
| */ |
| void markCompletedJob(JobStatus jobStatus, long timestamp) { |
| queue.add(new JobCompleteEvent(jc, timestamp, jobStatus, this)); |
| } |
| |
| /** |
| * Called by {@link SimulatorJobClient} when the simulation is completed and |
| * should be stopped. |
| */ |
| void shutdown() { |
| shutdown = true; |
| } |
| |
| /** |
| * Get the current virtual time of the on-going simulation. It is defined by |
| * the time stamp of the last event handled. |
| * @return the current virtual time |
| */ |
| long getCurrentTime() { |
| return currentTime; |
| } |
| |
| // Due to HDFS-778, a node may appear in job history logs as both numeric |
| // ips and as host names. We remove them from the parsed network topology |
| // before feeding it to ZombieCluster. |
| static void removeIpHosts(LoggedNetworkTopology topology) { |
| for (Iterator<LoggedNetworkTopology> rackIt = topology.getChildren() |
| .iterator(); rackIt.hasNext();) { |
| LoggedNetworkTopology rack = rackIt.next(); |
| List<LoggedNetworkTopology> nodes = rack.getChildren(); |
| for (Iterator<LoggedNetworkTopology> it = nodes.iterator(); it.hasNext();) { |
| LoggedNetworkTopology node = it.next(); |
| if (isIPAddress(node.getName())) { |
| it.remove(); |
| } |
| } |
| if (nodes.isEmpty()) { |
| rackIt.remove(); |
| } |
| } |
| } |
| |
| static Pattern IP_PATTERN; |
| |
| static { |
| // 0-255 |
| String IPV4BK1 = "(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)"; |
| // .b.c.d - where b/c/d are 0-255, and optionally adding two more |
| // backslashes before each period |
| String IPV4BKN = "(?:\\\\?\\." + IPV4BK1 + "){3}"; |
| String IPV4_PATTERN = IPV4BK1 + IPV4BKN; |
| |
| // first hexadecimal number |
| String IPV6BK1 = "(?:[0-9a-fA-F]{1,4})"; |
| // remaining 7 hexadecimal numbers, each preceded with ":". |
| String IPV6BKN = "(?::" + IPV6BK1 + "){7}"; |
| String IPV6_PATTERN = IPV6BK1 + IPV6BKN; |
| |
| IP_PATTERN = Pattern.compile( |
| "^(?:" + IPV4_PATTERN + "|" + IPV6_PATTERN + ")$"); |
| } |
| |
| |
| static boolean isIPAddress(String hostname) { |
| return IP_PATTERN.matcher(hostname).matches(); |
| } |
| |
| static void setStaticMapping(LoggedNetworkTopology topology) { |
| for (LoggedNetworkTopology rack : topology.getChildren()) { |
| for (LoggedNetworkTopology node : rack.getChildren()) { |
| StaticMapping.addNodeToRack(node.getName(), |
| new RackNode(rack.getName(), 1).getName()); |
| } |
| } |
| } |
| } |