rich7420 commented on code in PR #11349: URL: https://github.com/apache/ozone/pull/11349#discussion_r4129128624
########## hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/simulation/SimDatanode.java: ########## @@ -0,0 +1,628 @@ +/* + * 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.hdds.scm.simulation; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port; +import org.apache.hadoop.hdds.protocol.DatanodeID; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerActionsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto; +import org.apache.hadoop.hdds.scm.HddsTestUtils; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; +import org.apache.hadoop.ozone.container.upgrade.UpgradeUtils; +import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.ClosePipelineCommand; +import org.apache.hadoop.ozone.protocol.commands.CreatePipelineCommand; +import org.apache.hadoop.ozone.protocol.commands.DeleteContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.SCMCommand; +import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand; + +/** + * Simulated datanode. It registers, heartbeats and reports like the datanode state machine, and applies SCM commands to + * its replicas the way the datanode command handlers do. Its replicas are the ground truth the simulation checks SCM + * against. For example, this is how a lost replica gets repaired; every step on the SCM side is real SCM code: + * <pre>{@code + * SimDatanode dn3 real SCM SimDatanode dn7 + * (holds #42) (target) + * | | | + * |--- heartbeat ------------->| SCMDatanodeProtocolServer.sendHeartbeat | + * | | | + * | | RM timer: #42 is under-replicated | + * | | under-replication timer: placement picks | + * | | dn7 (seeded), queues "replicate #42 | + * | | to dn7" for dn3 | + * |<-- heartbeat response -----| | + * | [replicate #42 to dn7] | | + * | | | + * | decode the proto like | | + * | HeartbeatEndpointTask, | | + * | apply after a delay | | + * |-------------------- push copy (transfer time) ----------------------->| + * | | | + * | |<-- heartbeat with incremental report ----| + * | | ContainerReportHandler (report lane): | + * | | replica on dn7 recorded, pending op | + * | | cleared | + * }</pre> + */ +final class SimDatanode { + + private static final int MAX_ACTIONS_PER_HEARTBEAT = 20; + + private final ScmSimulation sim; + private final String name; + private final DatanodeDetails details; + private final TreeMap<Long, SimReplica> replicas = new TreeMap<>(); + private final Set<PipelineID> pipelines = new LinkedHashSet<>(); + private final List<ContainerReplicaProto> pendingIcrs = new ArrayList<>(); + private final Map<Long, ContainerAction> pendingContainerActions = new LinkedHashMap<>(); + private final Map<SCMCommandProto.Type, Integer> queuedCommands = new EnumMap<>(SCMCommandProto.Type.class); + private final List<SimScheduler.Timer> timers = new ArrayList<>(); + private final Map<String, SimScheduler.Timer> reportTimers = new HashMap<>(); + private final SimScheduler.WakeUp heartbeatNow; + private boolean running; + private boolean partitioned; + private boolean registered; + private boolean containerReportDue; + private boolean pipelineReportDue; + private boolean nodeReportDue; + private long generation; + private long latestTerm; + private NodeOperationalState persistedOpState = NodeOperationalState.IN_SERVICE; + private long persistedOpStateExpiry; + + SimDatanode(ScmSimulation sim, String name, DatanodeDetails details) { + this.sim = sim; + this.name = name; + this.details = details; + this.heartbeatNow = sim.scheduler().newWakeUp(name + ".triggeredHeartbeat", this::heartbeat); + } + + DatanodeID getId() { + return details.getID(); + } + + String getName() { + return name; + } + + DatanodeDetails getDetails() { + return details; + } + + String getRack() { + return details.getNetworkLocation(); + } + + boolean isRunning() { + return running; + } + + boolean isPartitioned() { + return partitioned; + } + + boolean hasPipeline(PipelineID id) { + return pipelines.contains(id); + } + + SimReplica getReplica(long containerId) { + return replicas.get(containerId); + } + + Map<Long, SimReplica> getReplicas() { + return Collections.unmodifiableMap(replicas); + } + + NodeOperationalState getPersistedOpState() { + return persistedOpState; + } + + // ---- process lifecycle and faults ---- + + /** Starts the datanode process; it registers with SCM on its first heartbeat. */ + void start() { + if (running) { + return; + } + running = true; + generation++; + registered = false; + SimConfig config = sim.config(); + Random random = sim.random(SimRandom.DATANODE); + timers.add(sim.scheduler().scheduleEvery(name + ".heartbeat", + random.nextInt((int) config.getHeartbeatIntervalMs()), config.getHeartbeatIntervalMs(), this::heartbeat)); + timers.add(sim.scheduler().scheduleEvery(name + ".nodeReport", + config.getNodeReportIntervalMs(), config.getNodeReportIntervalMs(), () -> nodeReportDue = true)); + scheduleReport("containerReport", config.getContainerReportIntervalMs(), () -> containerReportDue = true); + scheduleReport("pipelineReport", config.getPipelineReportIntervalMs(), () -> pipelineReportDue = true); + for (PipelineID id : pipelines) { + sim.ratisGroup(id).memberUp(this); + } + } + + /** Stops the process: queued commands and unsent reports are lost, replicas and Ratis groups stay on disk. */ + void stop() { + if (!running) { + return; + } + running = false; + generation++; + timers.forEach(sim.scheduler()::cancel); + timers.clear(); + reportTimers.values().forEach(sim.scheduler()::cancel); + reportTimers.clear(); + heartbeatNow.cancel(); + queuedCommands.clear(); + pendingIcrs.clear(); + pendingContainerActions.clear(); + for (PipelineID id : pipelines) { + sim.ratisGroup(id).memberDown(this); + } + } + + /** Restarts the process with different ports, like after a configuration change or an upgrade. */ + void restartWithNewPorts() { + stop(); + for (Port.Name port : new Port.Name[] {Port.Name.STANDALONE, Port.Name.RATIS}) { + details.setPort(port, details.getPort(port).getValue() + 1); + } + start(); + } + + /** Cuts (or restores) the network between this datanode and SCM. */ + void setPartitioned(boolean value) { + partitioned = value; + } + + /** A failed volume loses the given replicas; the datanode sends a full report soon. */ + void loseReplicas(List<Long> containerIds) { + for (Long id : containerIds) { + replicas.remove(id); + } + containerReportDue = true; + triggerHeartbeat(); + } + + /** The replica fails a scan and is marked unhealthy. */ + void markUnhealthy(long containerId) { + SimReplica replica = replicas.get(containerId); + if (replica != null && replica.getState() != State.UNHEALTHY) { + replica.setState(State.UNHEALTHY); + queueIcr(replica.toProto()); + } + } + + // ---- data path ---- + + /** Applies state committed through the Ratis group of a pipeline, creating the replica if needed. */ + void applyCommitted(PipelineID pipelineId, long containerId, long bcsid, long used, long keys, boolean closed) { + SimReplica replica = replicas.get(containerId); + if (replica == null) { + replica = new SimReplica(containerId, getId().toString(), 0, pipelineId, false, State.OPEN); + replicas.put(containerId, replica); + queueIcr(replica.toProto()); + } + State state = replica.getState(); + if (state != State.OPEN && state != State.CLOSING) { + return; + } + replica.setData(bcsid, used, keys); + if (closed) { + replica.setState(State.CLOSED); + queueIcr(replica.toProto()); + } else if (state == State.OPEN && used >= sim.config().getContainerCloseThresholdBytes()) { + addContainerAction(containerId, ContainerAction.Reason.CONTAINER_FULL); + } + } + + void onBecomeLeader(PipelineID id) { + pipelineReportDue = true; + triggerHeartbeat(); + } + + private void addContainerAction(long containerId, ContainerAction.Reason reason) { + if (!pendingContainerActions.containsKey(containerId)) { + pendingContainerActions.put(containerId, ContainerAction.newBuilder() + .setContainerID(containerId) + .setAction(ContainerAction.Action.CLOSE) + .setReason(reason) + .build()); + triggerHeartbeat(); + } + } + + // ---- heartbeat and reports ---- + + /** Marks a full report due after the interval plus up to one more interval of jitter, over and over. */ + private void scheduleReport(String report, long interval, Runnable markDue) { + long delay = interval + (long) (sim.random(SimRandom.DATANODE).nextDouble() * interval); + reportTimers.put(report, sim.scheduler().schedule(name + "." + report, delay, () -> { + markDue.run(); + scheduleReport(report, interval, markDue); + })); + } + + private void queueIcr(ContainerReplicaProto report) { + if (!running) { + return; + } + pendingIcrs.add(report); + triggerHeartbeat(); + } + + /** Datanodes send a heartbeat right away when they have urgent reports. */ + private void triggerHeartbeat() { + if (running) { + heartbeatNow.request(sim.random(SimRandom.DATANODE).nextInt(200)); + } + } + + private void heartbeat() { + if (!running) { + return; + } + if (!registered) { + register(); + return; + } + SCMHeartbeatRequestProto.Builder hb = SCMHeartbeatRequestProto.newBuilder() + .setDatanodeDetails(currentDetails().getProtoBufMessage()) + .setDataNodeLayoutVersion(UpgradeUtils.defaultLayoutVersionProto()) + .setCommandQueueReport(commandQueueReport()); + if (nodeReportDue) { + hb.setNodeReport(nodeReport()); + nodeReportDue = false; + } + if (containerReportDue) { + // Building a full report drops the pending incremental reports. + hb.setContainerReport(containerReport()); + containerReportDue = false; + pendingIcrs.clear(); + } + List<ContainerReplicaProto> icrs = new ArrayList<>(pendingIcrs); + pendingIcrs.clear(); + for (ContainerReplicaProto icr : icrs) { + hb.addIncrementalContainerReport(IncrementalContainerReportProto.newBuilder().addReport(icr)); + } + if (!pendingContainerActions.isEmpty()) { + ContainerActionsProto.Builder actions = ContainerActionsProto.newBuilder(); + List<Long> sent = new ArrayList<>(); + for (Map.Entry<Long, ContainerAction> e : pendingContainerActions.entrySet()) { + if (sent.size() == MAX_ACTIONS_PER_HEARTBEAT) { + break; + } + actions.addContainerActions(e.getValue()); + sent.add(e.getKey()); + } + sent.forEach(pendingContainerActions::remove); + hb.setContainerActions(actions); + } + if (pipelineReportDue) { + hb.setPipelineReports(pipelineReports()); + pipelineReportDue = false; + } + if (partitioned) { + // The RPC fails: incremental reports are queued again, everything else is lost. + pendingIcrs.addAll(0, icrs); + return; + } + SCMHeartbeatResponseProto response = sim.heartbeat(this, hb.build()); + if (response.hasTerm()) { + latestTerm = Math.max(latestTerm, response.getTerm()); + } + for (SCMCommandProto command : response.getCommandsList()) { + receive(command); + } + } + + private void register() { + if (partitioned) { + return; + } + registered = sim.register(this, currentDetails().getExtendedProtoBufMessage(), nodeReport(), containerReport(), + pipelineReports(), UpgradeUtils.defaultLayoutVersionProto()); + if (registered) { + nodeReportDue = false; + containerReportDue = false; + pipelineReportDue = false; + } + } + + /** The details a datanode sends over RPC, including its persisted operational state. */ + private DatanodeDetails currentDetails() { + DatanodeDetails copy = DatanodeDetails.getFromProtoBuf(details.getExtendedProtoBufMessage()); + copy.setPersistedOpState(persistedOpState); + copy.setPersistedOpStateExpiryEpochSec(persistedOpStateExpiry); + return copy; + } + + private NodeReportProto nodeReport() { + long capacity = sim.config().getDatanodeCapacityBytes(); + long used = 0; + for (SimReplica r : replicas.values()) { + used += r.getUsed(); + } + long metaCapacity = sim.config().getDatanodeCapacityBytes() / 10; + return HddsTestUtils.createNodeReport( + Collections.singletonList(HddsTestUtils.createStorageReport(getId(), "/data/" + name, + capacity, used, capacity - used, StorageTypeProto.DISK)), + Collections.singletonList(HddsTestUtils.createMetadataStorageReport("/ratis/" + name, + metaCapacity, 0, metaCapacity, StorageTypeProto.DISK))); + } + + private ContainerReportsProto containerReport() { + ContainerReportsProto.Builder report = ContainerReportsProto.newBuilder(); + for (SimReplica replica : replicas.values()) { + report.addReports(replica.toProto()); + } + return report.build(); + } + + private PipelineReportsProto pipelineReports() { + PipelineReportsProto.Builder report = PipelineReportsProto.newBuilder(); + for (PipelineID id : pipelines) { + report.addPipelineReport(PipelineReport.newBuilder() + .setPipelineID(id.getProtobuf()) + .setIsLeader(getId().equals(sim.ratisGroup(id).getLeader()))); + } + return report.build(); + } + + private CommandQueueReportProto commandQueueReport() { + CommandQueueReportProto.Builder report = CommandQueueReportProto.newBuilder(); + for (Map.Entry<SCMCommandProto.Type, Integer> e : queuedCommands.entrySet()) { + report.addCommand(e.getKey()); + report.addCount(e.getValue()); + } + return report.build(); + } + + // ---- commands ---- + + private void receive(SCMCommandProto proto) { + SCMCommandProto.Type type = proto.getCommandType(); + if (type == SCMCommandProto.Type.reregisterCommand) { + registered = false; + triggerHeartbeat(); + return; + } + SCMCommand<?> command = decode(proto); + if (command == null) { + sim.count("datanode.ignored." + type); + return; + } + queuedCommands.merge(type, 1, Integer::sum); + long expectedGeneration = generation; + long delay = sim.config().getCommandDelayMs(type, sim.random(SimRandom.DATANODE)); + sim.scheduler().schedule(name + ".execute " + type, delay, () -> { + if (generation != expectedGeneration) { + return; + } + queuedCommands.merge(type, -1, Integer::sum); + queuedCommands.remove(type, 0); + if (command.getTerm() < latestTerm) { + sim.count("datanode.staleTermCommandDropped"); + return; + } + execute(command); + }); + } + + /** Reads a command from a heartbeat response like HeartbeatEndpointTask; null for commands not simulated. */ + private static SCMCommand<?> decode(SCMCommandProto proto) { + final SCMCommand<?> command; + switch (proto.getCommandType()) { + case createPipelineCommand: + command = CreatePipelineCommand.getFromProtobuf(proto.getCreatePipelineCommandProto()); + break; + case closePipelineCommand: + command = ClosePipelineCommand.getFromProtobuf(proto.getClosePipelineCommandProto()); + break; + case closeContainerCommand: + command = CloseContainerCommand.getFromProtobuf(proto.getCloseContainerCommandProto()); + break; + case replicateContainerCommand: + command = ReplicateContainerCommand.getFromProtobuf(proto.getReplicateContainerCommandProto()); + break; + case deleteContainerCommand: + command = DeleteContainerCommand.getFromProtobuf(proto.getDeleteContainerCommandProto()); + break; + case setNodeOperationalStateCommand: + command = SetNodeOperationalStateCommand.getFromProtobuf(proto.getSetNodeOperationalStateCommandProto()); + break; + default: + return null; + } + if (proto.hasTerm()) { + command.setTerm(proto.getTerm()); + } + if (proto.hasDeadlineMsSinceEpoch()) { + command.setDeadline(proto.getDeadlineMsSinceEpoch()); + } + return command; + } + + private void execute(SCMCommand<?> command) { + switch (command.getType()) { + case createPipelineCommand: + createPipeline((CreatePipelineCommand) command); + break; + case closePipelineCommand: + closePipeline(((ClosePipelineCommand) command).getPipelineID()); + break; + case closeContainerCommand: + closeContainer((CloseContainerCommand) command); + break; + case replicateContainerCommand: + replicate((ReplicateContainerCommand) command); + break; + case deleteContainerCommand: + delete((DeleteContainerCommand) command); + break; + case setNodeOperationalStateCommand: + SetNodeOperationalStateCommand op = (SetNodeOperationalStateCommand) command; + persistedOpState = op.getOpState(); + persistedOpStateExpiry = op.getStateExpiryEpochSeconds(); + break; + default: + throw new IllegalStateException("Not simulated: " + command.getType()); + } + } + + private void createPipeline(CreatePipelineCommand command) { + PipelineID id = command.getPipelineID(); + List<DatanodeID> members = new ArrayList<>(); + for (DatanodeDetails dn : command.getNodeList()) { + members.add(dn.getID()); + } + SimRatisGroup group = sim.ratisGroup(id, members, command.getPriorityList()); + if (pipelines.add(id)) { + // The group starts before a leader is known; report it right away. + pipelineReportDue = true; + triggerHeartbeat(); + group.join(this); + } + } + + private void closePipeline(PipelineID id) { + if (!pipelines.remove(id)) { + return; + } + SimRatisGroup group = sim.ratisGroup(id); + group.leave(this); + // Removing the group quasi-closes the open containers it wrote to. + for (Long containerId : group.getContainers()) { + SimReplica replica = replicas.get(containerId); + if (replica != null && (replica.getState() == State.OPEN || replica.getState() == State.CLOSING)) { + if (replica.getState() == State.OPEN) { + replica.setState(State.CLOSING); + queueIcr(replica.toProto()); + } + replica.setState(State.QUASI_CLOSED); + queueIcr(replica.toProto()); + } + } + } + + private void closeContainer(CloseContainerCommand command) { + SimReplica replica = replicas.get(command.getContainerID()); + if (replica == null) { + return; + } + if (replica.getState() == State.OPEN) { + replica.setState(State.CLOSING); + queueIcr(replica.toProto()); + } + if (replica.getState() == State.CLOSING) { + PipelineID pipelineId = command.getPipelineID(); + if (pipelineId != null && pipelines.contains(pipelineId)) { + // Submitted through Ratis; followers get NotLeader, the leader's close applies to all replicas. + sim.ratisGroup(pipelineId).closeContainer(this, command.getContainerID()); + } else { + replica.setState(command.isForce() ? State.CLOSED : State.QUASI_CLOSED); + queueIcr(replica.toProto()); + } + } else if (replica.getState() == State.QUASI_CLOSED && command.isForce()) { + replica.setState(State.CLOSED); + queueIcr(replica.toProto()); + } + } + + private void replicate(ReplicateContainerCommand command) { + if (command.hasExpired(sim.now())) { + sim.count("datanode.replicate.expired"); + return; + } + SimReplica source = replicas.get(command.getContainerID()); + if (source == null || !(source.getState() == State.CLOSED || source.getState() == State.QUASI_CLOSED + || source.getState() == State.UNHEALTHY)) { + sim.count("datanode.replicate.noSource"); + return; + } + SimDatanode target = sim.datanode(command.getTargetDatanode().getID()); + SimReplica copy = source.importedCopy(); + long transferMs = sim.config().getReplicationTransferMs(copy.getUsed(), sim.random(SimRandom.DATANODE)); + sim.scheduler().schedule(name + ".push #" + command.getContainerID() + " to " + target.getName(), transferMs, Review Comment: The transfer still succeeds after the source stops or the target restarts; both cases reproduced locally. Please check both datanode generations before importing the replica. ########## hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/simulation/SimDatanode.java: ########## @@ -0,0 +1,628 @@ +/* + * 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.hdds.scm.simulation; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port; +import org.apache.hadoop.hdds.protocol.DatanodeID; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerActionsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto; +import org.apache.hadoop.hdds.scm.HddsTestUtils; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; +import org.apache.hadoop.ozone.container.upgrade.UpgradeUtils; +import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.ClosePipelineCommand; +import org.apache.hadoop.ozone.protocol.commands.CreatePipelineCommand; +import org.apache.hadoop.ozone.protocol.commands.DeleteContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.SCMCommand; +import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand; + +/** + * Simulated datanode. It registers, heartbeats and reports like the datanode state machine, and applies SCM commands to + * its replicas the way the datanode command handlers do. Its replicas are the ground truth the simulation checks SCM + * against. For example, this is how a lost replica gets repaired; every step on the SCM side is real SCM code: + * <pre>{@code + * SimDatanode dn3 real SCM SimDatanode dn7 + * (holds #42) (target) + * | | | + * |--- heartbeat ------------->| SCMDatanodeProtocolServer.sendHeartbeat | + * | | | + * | | RM timer: #42 is under-replicated | + * | | under-replication timer: placement picks | + * | | dn7 (seeded), queues "replicate #42 | + * | | to dn7" for dn3 | + * |<-- heartbeat response -----| | + * | [replicate #42 to dn7] | | + * | | | + * | decode the proto like | | + * | HeartbeatEndpointTask, | | + * | apply after a delay | | + * |-------------------- push copy (transfer time) ----------------------->| + * | | | + * | |<-- heartbeat with incremental report ----| + * | | ContainerReportHandler (report lane): | + * | | replica on dn7 recorded, pending op | + * | | cleared | + * }</pre> + */ +final class SimDatanode { + + private static final int MAX_ACTIONS_PER_HEARTBEAT = 20; + + private final ScmSimulation sim; + private final String name; + private final DatanodeDetails details; + private final TreeMap<Long, SimReplica> replicas = new TreeMap<>(); + private final Set<PipelineID> pipelines = new LinkedHashSet<>(); + private final List<ContainerReplicaProto> pendingIcrs = new ArrayList<>(); + private final Map<Long, ContainerAction> pendingContainerActions = new LinkedHashMap<>(); + private final Map<SCMCommandProto.Type, Integer> queuedCommands = new EnumMap<>(SCMCommandProto.Type.class); + private final List<SimScheduler.Timer> timers = new ArrayList<>(); + private final Map<String, SimScheduler.Timer> reportTimers = new HashMap<>(); + private final SimScheduler.WakeUp heartbeatNow; + private boolean running; + private boolean partitioned; + private boolean registered; + private boolean containerReportDue; + private boolean pipelineReportDue; + private boolean nodeReportDue; + private long generation; + private long latestTerm; + private NodeOperationalState persistedOpState = NodeOperationalState.IN_SERVICE; + private long persistedOpStateExpiry; + + SimDatanode(ScmSimulation sim, String name, DatanodeDetails details) { + this.sim = sim; + this.name = name; + this.details = details; + this.heartbeatNow = sim.scheduler().newWakeUp(name + ".triggeredHeartbeat", this::heartbeat); + } + + DatanodeID getId() { + return details.getID(); + } + + String getName() { + return name; + } + + DatanodeDetails getDetails() { + return details; + } + + String getRack() { + return details.getNetworkLocation(); + } + + boolean isRunning() { + return running; + } + + boolean isPartitioned() { + return partitioned; + } + + boolean hasPipeline(PipelineID id) { + return pipelines.contains(id); + } + + SimReplica getReplica(long containerId) { + return replicas.get(containerId); + } + + Map<Long, SimReplica> getReplicas() { + return Collections.unmodifiableMap(replicas); + } + + NodeOperationalState getPersistedOpState() { + return persistedOpState; + } + + // ---- process lifecycle and faults ---- + + /** Starts the datanode process; it registers with SCM on its first heartbeat. */ + void start() { + if (running) { + return; + } + running = true; + generation++; + registered = false; + SimConfig config = sim.config(); + Random random = sim.random(SimRandom.DATANODE); + timers.add(sim.scheduler().scheduleEvery(name + ".heartbeat", + random.nextInt((int) config.getHeartbeatIntervalMs()), config.getHeartbeatIntervalMs(), this::heartbeat)); + timers.add(sim.scheduler().scheduleEvery(name + ".nodeReport", + config.getNodeReportIntervalMs(), config.getNodeReportIntervalMs(), () -> nodeReportDue = true)); + scheduleReport("containerReport", config.getContainerReportIntervalMs(), () -> containerReportDue = true); + scheduleReport("pipelineReport", config.getPipelineReportIntervalMs(), () -> pipelineReportDue = true); + for (PipelineID id : pipelines) { + sim.ratisGroup(id).memberUp(this); + } + } + + /** Stops the process: queued commands and unsent reports are lost, replicas and Ratis groups stay on disk. */ + void stop() { + if (!running) { + return; + } + running = false; + generation++; + timers.forEach(sim.scheduler()::cancel); + timers.clear(); + reportTimers.values().forEach(sim.scheduler()::cancel); + reportTimers.clear(); + heartbeatNow.cancel(); + queuedCommands.clear(); + pendingIcrs.clear(); + pendingContainerActions.clear(); + for (PipelineID id : pipelines) { + sim.ratisGroup(id).memberDown(this); + } + } + + /** Restarts the process with different ports, like after a configuration change or an upgrade. */ + void restartWithNewPorts() { + stop(); + for (Port.Name port : new Port.Name[] {Port.Name.STANDALONE, Port.Name.RATIS}) { + details.setPort(port, details.getPort(port).getValue() + 1); + } + start(); + } + + /** Cuts (or restores) the network between this datanode and SCM. */ + void setPartitioned(boolean value) { + partitioned = value; + } + + /** A failed volume loses the given replicas; the datanode sends a full report soon. */ + void loseReplicas(List<Long> containerIds) { + for (Long id : containerIds) { + replicas.remove(id); + } + containerReportDue = true; + triggerHeartbeat(); + } + + /** The replica fails a scan and is marked unhealthy. */ + void markUnhealthy(long containerId) { + SimReplica replica = replicas.get(containerId); + if (replica != null && replica.getState() != State.UNHEALTHY) { + replica.setState(State.UNHEALTHY); + queueIcr(replica.toProto()); + } + } + + // ---- data path ---- + + /** Applies state committed through the Ratis group of a pipeline, creating the replica if needed. */ + void applyCommitted(PipelineID pipelineId, long containerId, long bcsid, long used, long keys, boolean closed) { + SimReplica replica = replicas.get(containerId); + if (replica == null) { + replica = new SimReplica(containerId, getId().toString(), 0, pipelineId, false, State.OPEN); + replicas.put(containerId, replica); + queueIcr(replica.toProto()); + } + State state = replica.getState(); + if (state != State.OPEN && state != State.CLOSING) { + return; + } + replica.setData(bcsid, used, keys); + if (closed) { + replica.setState(State.CLOSED); + queueIcr(replica.toProto()); + } else if (state == State.OPEN && used >= sim.config().getContainerCloseThresholdBytes()) { + addContainerAction(containerId, ContainerAction.Reason.CONTAINER_FULL); + } + } + + void onBecomeLeader(PipelineID id) { + pipelineReportDue = true; + triggerHeartbeat(); + } + + private void addContainerAction(long containerId, ContainerAction.Reason reason) { + if (!pendingContainerActions.containsKey(containerId)) { + pendingContainerActions.put(containerId, ContainerAction.newBuilder() + .setContainerID(containerId) + .setAction(ContainerAction.Action.CLOSE) + .setReason(reason) + .build()); + triggerHeartbeat(); + } + } + + // ---- heartbeat and reports ---- + + /** Marks a full report due after the interval plus up to one more interval of jitter, over and over. */ + private void scheduleReport(String report, long interval, Runnable markDue) { + long delay = interval + (long) (sim.random(SimRandom.DATANODE).nextDouble() * interval); + reportTimers.put(report, sim.scheduler().schedule(name + "." + report, delay, () -> { + markDue.run(); + scheduleReport(report, interval, markDue); + })); + } + + private void queueIcr(ContainerReplicaProto report) { + if (!running) { + return; + } + pendingIcrs.add(report); + triggerHeartbeat(); + } + + /** Datanodes send a heartbeat right away when they have urgent reports. */ + private void triggerHeartbeat() { + if (running) { + heartbeatNow.request(sim.random(SimRandom.DATANODE).nextInt(200)); + } + } + + private void heartbeat() { + if (!running) { + return; + } + if (!registered) { + register(); + return; + } + SCMHeartbeatRequestProto.Builder hb = SCMHeartbeatRequestProto.newBuilder() + .setDatanodeDetails(currentDetails().getProtoBufMessage()) + .setDataNodeLayoutVersion(UpgradeUtils.defaultLayoutVersionProto()) + .setCommandQueueReport(commandQueueReport()); + if (nodeReportDue) { + hb.setNodeReport(nodeReport()); + nodeReportDue = false; + } + if (containerReportDue) { + // Building a full report drops the pending incremental reports. + hb.setContainerReport(containerReport()); + containerReportDue = false; + pendingIcrs.clear(); + } + List<ContainerReplicaProto> icrs = new ArrayList<>(pendingIcrs); + pendingIcrs.clear(); + for (ContainerReplicaProto icr : icrs) { + hb.addIncrementalContainerReport(IncrementalContainerReportProto.newBuilder().addReport(icr)); + } + if (!pendingContainerActions.isEmpty()) { + ContainerActionsProto.Builder actions = ContainerActionsProto.newBuilder(); + List<Long> sent = new ArrayList<>(); + for (Map.Entry<Long, ContainerAction> e : pendingContainerActions.entrySet()) { + if (sent.size() == MAX_ACTIONS_PER_HEARTBEAT) { + break; + } + actions.addContainerActions(e.getValue()); + sent.add(e.getKey()); + } + sent.forEach(pendingContainerActions::remove); + hb.setContainerActions(actions); + } + if (pipelineReportDue) { + hb.setPipelineReports(pipelineReports()); + pipelineReportDue = false; + } + if (partitioned) { + // The RPC fails: incremental reports are queued again, everything else is lost. + pendingIcrs.addAll(0, icrs); + return; + } + SCMHeartbeatResponseProto response = sim.heartbeat(this, hb.build()); + if (response.hasTerm()) { + latestTerm = Math.max(latestTerm, response.getTerm()); + } + for (SCMCommandProto command : response.getCommandsList()) { + receive(command); + } + } + + private void register() { + if (partitioned) { + return; + } + registered = sim.register(this, currentDetails().getExtendedProtoBufMessage(), nodeReport(), containerReport(), + pipelineReports(), UpgradeUtils.defaultLayoutVersionProto()); + if (registered) { + nodeReportDue = false; + containerReportDue = false; + pipelineReportDue = false; + } + } + + /** The details a datanode sends over RPC, including its persisted operational state. */ + private DatanodeDetails currentDetails() { + DatanodeDetails copy = DatanodeDetails.getFromProtoBuf(details.getExtendedProtoBufMessage()); + copy.setPersistedOpState(persistedOpState); + copy.setPersistedOpStateExpiryEpochSec(persistedOpStateExpiry); + return copy; + } + + private NodeReportProto nodeReport() { + long capacity = sim.config().getDatanodeCapacityBytes(); + long used = 0; + for (SimReplica r : replicas.values()) { + used += r.getUsed(); + } + long metaCapacity = sim.config().getDatanodeCapacityBytes() / 10; + return HddsTestUtils.createNodeReport( + Collections.singletonList(HddsTestUtils.createStorageReport(getId(), "/data/" + name, + capacity, used, capacity - used, StorageTypeProto.DISK)), + Collections.singletonList(HddsTestUtils.createMetadataStorageReport("/ratis/" + name, + metaCapacity, 0, metaCapacity, StorageTypeProto.DISK))); + } + + private ContainerReportsProto containerReport() { + ContainerReportsProto.Builder report = ContainerReportsProto.newBuilder(); + for (SimReplica replica : replicas.values()) { + report.addReports(replica.toProto()); + } + return report.build(); + } + + private PipelineReportsProto pipelineReports() { + PipelineReportsProto.Builder report = PipelineReportsProto.newBuilder(); + for (PipelineID id : pipelines) { + report.addPipelineReport(PipelineReport.newBuilder() + .setPipelineID(id.getProtobuf()) + .setIsLeader(getId().equals(sim.ratisGroup(id).getLeader()))); + } + return report.build(); + } + + private CommandQueueReportProto commandQueueReport() { + CommandQueueReportProto.Builder report = CommandQueueReportProto.newBuilder(); + for (Map.Entry<SCMCommandProto.Type, Integer> e : queuedCommands.entrySet()) { + report.addCommand(e.getKey()); + report.addCount(e.getValue()); + } + return report.build(); + } + + // ---- commands ---- + + private void receive(SCMCommandProto proto) { + SCMCommandProto.Type type = proto.getCommandType(); + if (type == SCMCommandProto.Type.reregisterCommand) { + registered = false; + triggerHeartbeat(); + return; + } + SCMCommand<?> command = decode(proto); + if (command == null) { + sim.count("datanode.ignored." + type); + return; + } + queuedCommands.merge(type, 1, Integer::sum); + long expectedGeneration = generation; + long delay = sim.config().getCommandDelayMs(type, sim.random(SimRandom.DATANODE)); + sim.scheduler().schedule(name + ".execute " + type, delay, () -> { + if (generation != expectedGeneration) { + return; + } + queuedCommands.merge(type, -1, Integer::sum); Review Comment: Please keep running transfers in the replication count, as `ReplicationSupervisor` does. The local test reports zero before the transfer finishes, bypassing SCM's replication limit. ########## hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/simulation/ScmSimulation.java: ########## @@ -0,0 +1,961 @@ +/* + * 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.hdds.scm.simulation; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; +import java.nio.file.Path; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import java.util.UUID; +import java.util.concurrent.TimeoutException; +import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.client.ReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.DatanodeID; +import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ExtendedDatanodeDetailsProto; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto; +import org.apache.hadoop.hdds.scm.container.ContainerHealthState; +import org.apache.hadoop.hdds.scm.container.ContainerID; +import org.apache.hadoop.hdds.scm.container.ContainerInfo; +import org.apache.hadoop.hdds.scm.container.ContainerManager; +import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException; +import org.apache.hadoop.hdds.scm.container.ContainerReplica; +import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport; +import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager; +import org.apache.hadoop.hdds.scm.container.replication.ReplicationSimSupport; +import org.apache.hadoop.hdds.scm.events.SCMEvents; +import org.apache.hadoop.hdds.scm.ha.BackgroundSCMService; +import org.apache.hadoop.hdds.scm.ha.SCMContext; +import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub; +import org.apache.hadoop.hdds.scm.ha.SCMService; +import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore; +import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStoreImpl; +import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl; +import org.apache.hadoop.hdds.scm.node.NodeDecommissionManager; +import org.apache.hadoop.hdds.scm.node.NodeSimSupport; +import org.apache.hadoop.hdds.scm.node.NodeStatus; +import org.apache.hadoop.hdds.scm.node.SCMNodeManager; +import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; +import org.apache.hadoop.hdds.scm.pipeline.BackgroundPipelineCreator; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; +import org.apache.hadoop.hdds.scm.pipeline.PipelineManagerImpl; +import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException; +import org.apache.hadoop.hdds.scm.pipeline.PipelineSimSupport; +import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager; +import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager; +import org.apache.hadoop.hdds.scm.server.SCMConfigurator; +import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.ReportFromDatanode; +import org.apache.hadoop.hdds.scm.server.SCMStorageConfig; +import org.apache.hadoop.hdds.scm.server.StorageContainerManager; +import org.apache.hadoop.hdds.server.events.EventHandler; +import org.apache.hadoop.hdds.server.events.EventPublisher; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.ozone.common.Storage; +import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode; +import org.apache.hadoop.security.authentication.client.AuthenticationException; +import org.apache.ozone.test.MockClock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Deterministic simulation of SCM. + * <p> + * The real {@link StorageContainerManager} runs in-process with simulated datanodes ({@link SimDatanode}), which + * register and heartbeat through its datanode protocol server. SCM is built with a simulated clock, seeded random + * sources and an event queue whose handlers run on the simulation's scheduler, and the simulation runs the services SCM + * would run on their own threads. A single thread runs everything through {@link SimScheduler}, so a run is fully + * determined by its {@link SimConfig}: rerunning a seed reproduces the exact same trace. + * <pre>{@code + * +-------------------------------------------------------------------------+ + * | real StorageContainerManager | + * | | + * | SCMDatanodeProtocolServer SCMNodeManager ReplicationManager | + * | ContainerManager PipelineManager DatanodeAdminMonitor | + * | event handlers: DeadNodeHandler, ContainerReportHandler, ... | + * | | + * | injected through SCMConfigurator: | + * | SimEventQueue ... each handler call becomes a scheduler task | + * | MockClock ....... simulated time | + * | seeded Random ... topology, placement, pipeline IDs, RM shuffle | + * +-------------------^-----------------------------------+-----------------+ + * | | + * register and heartbeat, commands in the heartbeat + * carrying node, container response (replicate, delete, + * and pipeline reports close, create pipeline, ...) + * | v + * +-------------------+-----------------------------------+-----------------+ + * | SimDatanode x 8..12 on 3 racks | + * | SimReplica ...... ground truth: which replicas really exist | + * | SimRatisGroup ... one per pipeline: leader election, commits | + * +-------------------^-----------------------------------^-----------------+ + * | | + * +---------+---------+ +-----------+------------+ + * | SimWorkload | | SimChaos | + * | client writes | | crash, partition, lose | + * | every ~2 s | | or corrupt replicas, | + * +-------------------+ | decommission (node or | + * | rack), maintenance, | + * | add node, restart SCM | + * +------------------------+ + * }</pre> + * A run has two phases. During the chaos phase a client writes to containers while {@link SimChaos} crashes, + * partitions, decommissions and adds datanodes and damages replicas. Then faults stop, every datanode comes back, and + * the settle phase gives SCM time to repair the cluster. Safety invariants are checked throughout. At the end SCM must + * have repaired everything, and then stay quiet: no more replicate or delete commands. With the default + * {@link SimConfig}: + * <pre>{@code + * simulated time + * 0 2h 3h 3h15m + * +---------------------+----------------------------+-------------+ + * | chaos | settle | quiet | + * +---------------------+----------------------------+-------------+ + * | client writes; | writes and faults stop; | SCM must | + * | a fault every 1-3 | every datanode comes back; | send no | + * | min, at most 2 | SCM repairs the cluster | replicate / | + * | datanodes down | | delete cmds | + * +---------------------+----------------------------+-------------+ + * |<-------------- safety checks every 30 s, whole run ----------->| + * ^ + * | convergence check: + * | - every container correctly replicated + * | - no CLOSING container, no ALLOCATED pipeline + * | - every datanode registered, IN_SERVICE, HEALTHY + * | - SCM's replicas == datanodes' replicas + * + * pass: next seed fail: violations + seed + full trace in target/scm-simulation + * }</pre> + * Not simulated: SCM HA (Ratis replication of SCM state), block deletion, EC containers, the container balancer and + * security. + */ +final class ScmSimulation implements AutoCloseable { + + /** Simulated time starts far in the future, so stray wall clock timestamps always look old. */ + static final Instant START = Instant.parse("2100-01-01T00:00:00Z"); + + private static final Logger LOG = LoggerFactory.getLogger(ScmSimulation.class); + /** The simulation running on this thread, for the plugins SCM creates from its configuration. */ + private static final ThreadLocal<ScmSimulation> RUNNING = new ThreadLocal<>(); + private static final ReplicationConfig RATIS_THREE = RatisReplicationConfig.getInstance(ReplicationFactor.THREE); + private static final String OWNER = "sim"; + private static final int MAX_VIOLATIONS = 20; + private static final long SAFETY_CHECK_INTERVAL_MS = 30_000; + /** How long SCM is watched after settling, when it must not move replicas any more. */ + private static final long QUIET_WINDOW_MS = 15 * 60_000; + private static final int REPORT_LANES = 10; + /** Handlers that repair node, topology and pipeline state after a node state change. */ + private static final List<String> NODE_LIFECYCLE_HANDLERS = Arrays.asList( + "StaleNodeHandler", "DeadNodeHandler", "HealthyReadOnlyNodeHandler", "NewNodeHandler", + "NodeAddressUpdateHandler", "StartDatanodeAdminHandler", "PipelineReportHandler"); + + private final SimConfig config; + private final Path dir; + private final MockClock clock = new MockClock(START, ZoneOffset.UTC); + private final Map<SimRandom, Random> randoms = new EnumMap<>(SimRandom.class); + private final SimTrace trace; + private final SimScheduler scheduler; + private final Map<DatanodeID, SimDatanode> datanodes = new LinkedHashMap<>(); + private final Map<PipelineID, SimRatisGroup> ratisGroups = new LinkedHashMap<>(); + private final Map<PipelineID, String> pipelineNames = new HashMap<>(); + private final Map<String, Long> counters = new TreeMap<>(); + private final List<String> violations = new ArrayList<>(); + private final List<SimScheduler.Timer> scmTimers = new ArrayList<>(); + private final List<SimScheduler.WakeUp> scmWakeUps = new ArrayList<>(); + private final String clusterId; + private final String scmId; + + private StorageContainerManager scm; + private SimEventQueue eventQueue; + private SCMNodeManager nodeManager; + private SimScheduler.WakeUp replicationWakeUp; + private SimScheduler.WakeUp pipelineCreatorWakeUp; + private long term = 1; + private int handlerFailuresSeen; + private int timerFailuresSeen; + /** Replicate and delete commands SCM sends while it should be quiet; null outside that window. */ + private List<String> replicaMoves; + + ScmSimulation(SimConfig config, Path dir) { + this.config = config; + this.dir = dir; + for (SimRandom stream : SimRandom.values()) { + randoms.put(stream, new Random(mix(config.getSeed() + 0x9E3779B97F4A7C15L * (stream.ordinal() + 1)))); + } + this.trace = new SimTrace(2000, config.isKeepFullTrace()); + this.scheduler = new SimScheduler(clock, random(SimRandom.SCHEDULER), trace, + config.getSlowHandlerProbability(), config.getMaxSlowHandlerMs()); + Random ids = random(SimRandom.IDS); + this.clusterId = new UUID(ids.nextLong(), ids.nextLong()).toString(); + this.scmId = new UUID(ids.nextLong(), ids.nextLong()).toString(); + // Several SCM instances register the same metrics sources in one JVM. + DefaultMetricsSystem.setMiniClusterMode(true); + } + + /** SplitMix64 finalizer, so that nearby seeds give unrelated streams. */ + private static long mix(long z) { + z = (z ^ (z >>> 30)) * 0xBF58476D1CE4E5B9L; + z = (z ^ (z >>> 27)) * 0x94D049BB133111EBL; + return z ^ (z >>> 31); + } + + // ---- running ---- + + SimResult run() throws IOException { + LOG.info("Starting SCM simulation {}", config); + RUNNING.set(this); + startScm(); + for (int i = 0; i < config.getInitialDatanodes(); i++) { + addDatanode().start(); + } + SimWorkload workload = new SimWorkload(this); + workload.start(); + SimChaos chaos = config.isFaults() ? new SimChaos(this) : null; + if (chaos != null) { + chaos.start(); + } + scheduler.scheduleEvery("sim.checkSafety", SAFETY_CHECK_INTERVAL_MS, SAFETY_CHECK_INTERVAL_MS, this::checkSafety); + + long chaosEnd = clock.millis() + config.getChaosDurationMs(); + runUntil(chaosEnd); + trace.record(scheduler.elapsed(), "phase", "settle"); + workload.stop(); + if (chaos != null) { + chaos.stopAndHeal(); + } + runUntil(chaosEnd + config.getSettleDurationMs()); + if (violations.size() < MAX_VIOLATIONS) { + checkConvergence(); + checkQuiet(); + } + return result(); + } + + private void runUntil(long time) { + while (clock.millis() < time && violations.size() < MAX_VIOLATIONS) { + if (!scheduler.step()) { + break; + } + collectHandlerFailures(); + } + } + + private SimResult result() throws IOException { + Path traceFile = null; + if (!violations.isEmpty() || config.isKeepFullTrace()) { + traceFile = config.getTraceDir().resolve("trace-" + config.getSeed() + ".txt").toAbsolutePath(); + trace.writeTo(traceFile); + } + return new SimResult(config, new ArrayList<>(violations), new TreeMap<>(counters), trace.hash(), + scheduler.steps(), scheduler.elapsed(), trace.recentLines(), traceFile); + } + + @Override + public void close() throws IOException { + try { + stopScm(); + } finally { + RUNNING.remove(); + } + } + + static ScmSimulation running() { + return RUNNING.get(); + } + + // ---- SCM ---- + + /** + * Restarts SCM. Committed state survives in its DB (in production the Ratis log guarantees that); events being + * handled, pending commands and all other in-memory state are lost, and datanodes register again. + */ + void restartScm() throws IOException { + record("scm.restart", "term " + (term + 1)); + scm.getScmHAManager().asSCMHADBTransactionBuffer().flush(); + stopScm(); + term++; + handlerFailuresSeen = 0; + startScm(); + } + + private void stopScm() throws IOException { + scmTimers.forEach(scheduler::cancel); + scmTimers.clear(); + scmWakeUps.forEach(SimScheduler.WakeUp::cancel); + scmWakeUps.clear(); + scheduler.clearLanes(); + if (scm != null) { + scm.stop(); + nodeManager.close(); + scm = null; + } + } + + private void startScm() throws IOException { + OzoneConfiguration conf = config.toOzoneConfiguration(dir.resolve("scm"), random(SimRandom.SCM).nextLong()); + SCMStorageConfig storage = new SCMStorageConfig(conf); + if (storage.getState() != Storage.StorageState.INITIALIZED) { + storage.setClusterId(clusterId); + storage.setScmId(scmId); + storage.setSCMHAFlag(true); + storage.initialize(); + } + SCMMetadataStore metadataStore = new SCMMetadataStoreImpl(conf); + SCMContext scmContext = new SCMContext.Builder() + .setLeader(true) + .setTerm(term) + .setSafeModeStatus(SCMSafeModeManager.SafeModeStatus.INITIAL) + .setSCM(scmProxy()) + .build(); + scmContext.setLeaderReady(); + eventQueue = new SimEventQueue(scheduler, this::describe, REPORT_LANES); + NetworkTopology topology = new NetworkTopologyImpl(conf, random(SimRandom.SCM)); + // Built here to track heartbeats on the simulated clock, and to take rack locations from the datanodes. + nodeManager = new SCMNodeManager(conf, storage, eventQueue, topology, scmContext, + new HDDSLayoutVersionManager(storage.getLayoutVersion()), host -> null, clock); + NodeSimSupport.stopHealthCheckThread(nodeManager); + + SCMConfigurator configurator = new SCMConfigurator(); + configurator.setMetadataStore(metadataStore); + configurator.setSCMHAManager(SCMHAManagerStub.getInstance(true, metadataStore.getStore())); + configurator.setScmContext(scmContext); + configurator.setNetworkTopology(topology); + configurator.setScmNodeManager(nodeManager); + configurator.setLeaseManager(new SimLeaseManager(scheduler, config.getCloseContainerWaitMs())); + configurator.setEventQueue(eventQueue); + configurator.setSystemClock(clock); + Random ids = random(SimRandom.IDS); + configurator.setPipelineIdGenerator(() -> PipelineID.valueOf(new UUID(ids.nextLong(), ids.nextLong()))); + configurator.setRandom(random(SimRandom.SCM)); + try { + scm = StorageContainerManager.createSCM(conf, configurator); + } catch (AuthenticationException e) { + throw new IOException(e); + } + runScmServices(); + } + + /** + * SCM runs its periodic services on their own threads. Stop those threads, and run each service's task on the + * scheduler instead, at the same interval and behind the same checks, woken up by the same requests. + */ + private void runScmServices() { + scm.getSCMServiceManager().stop(); + NodeDecommissionManager decommissionManager = scm.getScmDecommissionManager(); + decommissionManager.stop(); + ReplicationManager rm = scm.getReplicationManager(); + PipelineManagerImpl pipelineManager = (PipelineManagerImpl) scm.getPipelineManager(); + BackgroundPipelineCreator creator = pipelineManager.getBackgroundPipelineCreator(); + // Like the pipeline scrubber, the expired replica op scrubber is a BackgroundSCMService, which waits as long. + BackgroundSCMService scrubber = pipelineManager.getBackgroundPipelineScrubber(); + + replicationWakeUp = newWakeUp("scm.replicationMonitor.wakeUp", rm::processAll); + pipelineCreatorWakeUp = newWakeUp("scm.pipelineCreator.wakeUp", + () -> PipelineSimSupport.runPipelineCreator(creator)); + scm.getSCMServiceManager().register(new PipelineCreatorTrigger()); + eventQueue.addHandler(SCMEvents.REPLICATION_MANAGER_NOTIFY, new ReplicationMonitorTrigger()); + + every("scm.nodeHealthCheck", config.getHeartbeatProcessIntervalMs(), + () -> NodeSimSupport.checkNodesHealth(nodeManager)); + every("scm.replicationMonitor", config.getReplicationIntervalMs(), rm::processAll); + every("scm.underReplicatedProcessor", config.getUnderReplicatedIntervalMs(), + () -> ReplicationSimSupport.processUnderReplicated(rm)); + every("scm.overReplicatedProcessor", config.getOverReplicatedIntervalMs(), + () -> ReplicationSimSupport.processOverReplicated(rm)); + every("scm.pipelineCreator", config.getPipelineCreationIntervalMs(), + () -> PipelineSimSupport.runPipelineCreator(creator)); + every("scm.pipelineScrubber", config.getPipelineScrubIntervalMs(), () -> { + if (scrubber.shouldRun()) { + pipelineManager.scrubAndClosePipelinesMissingDataStreamPort(); + } + }); + every("scm.replicaOpScrubber", config.getReplicaOpScrubIntervalMs(), () -> { + if (scrubber.shouldRun()) { + rm.getContainerReplicaPendingOps().removeExpiredEntries(); + } + }); + every("scm.datanodeAdminMonitor", config.getAdminMonitorIntervalMs(), decommissionManager.getMonitor()::run); + } + + private void every(String name, long intervalMs, Runnable task) { + scmTimers.add(scheduler.scheduleEvery(name, intervalMs, intervalMs, task)); + } + + private SimScheduler.WakeUp newWakeUp(String name, Runnable task) { + SimScheduler.WakeUp wakeUp = scheduler.newWakeUp(name, task); + scmWakeUps.add(wakeUp); + return wakeUp; + } + + /** + * The SCM that managers reach through {@link SCMContext#getScm()}. The context has to exist before SCM, which is + * looked up only when called. + */ + private OzoneStorageContainerManager scmProxy() { + return (OzoneStorageContainerManager) Proxy.newProxyInstance(getClass().getClassLoader(), + new Class<?>[] {OzoneStorageContainerManager.class}, (proxy, method, args) -> { + try { + return method.invoke(scm, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + }); + } + + // ---- datanode boundary, through SCM's datanode protocol server ---- + + boolean register(SimDatanode dn, ExtendedDatanodeDetailsProto details, NodeReportProto nodeReport, + ContainerReportsProto containerReport, PipelineReportsProto pipelineReports, LayoutVersionProto layout) { + SCMRegisteredResponseProto response; + try { + response = scm.getDatanodeProtocolServer().register(details, nodeReport, containerReport, pipelineReports, + layout); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + if (response.getErrorCode() != SCMRegisteredResponseProto.ErrorCode.success) { + count("datanode.register.failed"); + return false; + } + record("register", dn.getName()); + return true; + } + + SCMHeartbeatResponseProto heartbeat(SimDatanode dn, SCMHeartbeatRequestProto request) { + SCMHeartbeatResponseProto response; + try { + response = scm.getDatanodeProtocolServer().sendHeartbeat(request); + } catch (IOException e) { + throw new UncheckedIOException(e); + } catch (TimeoutException e) { + throw new IllegalStateException(e); + } + for (SCMCommandProto command : response.getCommandsList()) { + record("command", dn.getName() + " " + command.getCommandType()); + if (replicaMoves != null) { + recordReplicaMove(dn, command); + } + } + return response; + } + + private void recordReplicaMove(SimDatanode dn, SCMCommandProto command) { + if (command.getCommandType() == SCMCommandProto.Type.replicateContainerCommand) { + replicaMoves.add("replicate #" + command.getReplicateContainerCommandProto().getContainerID() + " from " + + dn.getName()); + } else if (command.getCommandType() == SCMCommandProto.Type.deleteContainerCommand) { + replicaMoves.add("delete #" + command.getDeleteContainerCommandProto().getContainerID() + " on " + dn.getName()); + } + } + + // ---- datanodes and Ratis groups ---- + + SimDatanode addDatanode() { + int index = datanodes.size() + 1; + Random ids = random(SimRandom.IDS); + DatanodeID id = DatanodeID.of(new UUID(ids.nextLong(), ids.nextLong())); + int rack = (index - 1) % config.getRacks(); + String name = "dn" + index; + DatanodeDetails details = MockDatanodeDetails.createDatanodeDetails(id, name, "10.0." + rack + "." + index, + "/rack-" + rack); + SimDatanode dn = new SimDatanode(this, name, details); + datanodes.put(id, dn); + record("datanode.add", name + " /rack-" + rack); + return dn; + } + + SimDatanode datanode(DatanodeID id) { + return datanodes.get(id); + } + + Collection<SimDatanode> datanodes() { + return datanodes.values(); + } + + SimRatisGroup ratisGroup(PipelineID id, List<DatanodeID> members, List<Integer> priorities) { + return ratisGroups.computeIfAbsent(id, k -> new SimRatisGroup(this, id, members, priorities)); + } + + SimRatisGroup ratisGroup(PipelineID id) { + return ratisGroups.get(id); + } + + // ---- client and admin operations ---- + + boolean isOutOfSafeMode() { + return !scm.getScmContext().isInSafeMode(); + } + + /** + * Picks the container for a client write of the given size the way block allocation does: on a random open + * RATIS/THREE pipeline, a container of this client with enough room, allocated if there is none. + */ + ContainerInfo containerForWrite(Random random, long size) { + List<Pipeline> open = scm.getPipelineManager().getPipelines(RATIS_THREE, Pipeline.PipelineState.OPEN); + if (open.isEmpty()) { + count("client.allocate.noPipeline"); + return null; + } + Pipeline pipeline = open.get(random.nextInt(open.size())); + ContainerInfo container = scm.getContainerManager().getMatchingContainer(size, OWNER, pipeline); + if (container == null) { + count("client.allocate.failed"); + } + return container; + } + + NodeStatus nodeStatus(SimDatanode dn) { + try { + return nodeManager.getNodeStatus(dn.getDetails()); + } catch (NodeNotFoundException e) { + return null; + } + } + + boolean startDecommission(SimDatanode dn) { + return admin("decommission", dn, node -> scm.getScmDecommissionManager().startDecommission(node)); + } + + boolean startMaintenance(SimDatanode dn) { + return admin("maintenance", dn, node -> scm.getScmDecommissionManager().startMaintenance(node, 0)); + } + + /** @return false if SCM rejected it, for example because the node has not registered since SCM restarted. */ + boolean recommission(SimDatanode dn) { + return admin("recommission", dn, node -> scm.getScmDecommissionManager().recommission(node)); + } + + private boolean admin(String operation, SimDatanode dn, AdminCommand command) { + try { + command.run(nodeManager.getNode(dn.getId())); + record("admin." + operation, dn.getName()); + return true; + } catch (Exception e) { Review Comment: An injected `IllegalStateException` from `startDecommission` is silently treated as a rejection. Please catch only expected rejection exceptions and handle unregistered nodes explicitly. -- 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]
