diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index aae7259d813e..5e44f1ee19c0 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -73,6 +73,9 @@ public final class HddsConfigKeys { "hdds.heartbeat.report.max.limit"; public static final int HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT = 10000; + public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE = + "hdds.heartbeat.command.status.report.max.size"; + public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT = "64MB"; // Configuration to allow volume choosing policy. public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY = "hdds.datanode.volume.choosing.policy"; diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java index 9d3d7ef85cbe..7cdb81116c0d 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java @@ -112,7 +112,10 @@ public class ScmConfig extends ReconfigurableConfig { + "The block num is counted at the replica level." + "If SCM has 100000 blocks which need to be deleted and the " + "configuration is 5000 then it would only send 5000 blocks " - + "for deletion to the datanodes." + + "for deletion to the datanodes. Increasing this value can " + + "increase delete command acknowledgement size; each " + + "individual acknowledgement must fit within the IPC size " + + "limit (ipc.maximum.data.length)." ) private int blockDeletionLimit; diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index a4fd106c51a5..65da95572e1f 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -1805,8 +1805,22 @@ Maximum number of report messages (not containers) a datanode includes in a single heartbeat to an SCM or Recon endpoint. Reports beyond the limit - stay queued and are sent in an immediate follow-up heartbeat. Values below - 1 are treated as 1. + stay queued and are sent in later scheduled heartbeats. Values below 1 + are treated as 1. + + + + hdds.heartbeat.command.status.report.max.size + 64MB + DATANODE + + Maximum serialized size of command status reports included in one + heartbeat. The effective limit can be lower when other heartbeat fields + consume the safe RPC payload budget. Reports beyond the limit stay queued + for a later heartbeat. A single command status that exceeds this limit is + sent as the only command status report when it still fits within the + remaining safe RPC payload, ensuring that it does not block all later + command statuses. diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java index f376d640a3c4..6165c497db95 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java @@ -23,6 +23,7 @@ import com.google.common.base.CaseFormat; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.text.WordUtils; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.hdds.scm.net.HostAndPort; @@ -60,6 +61,14 @@ public final class DatanodeQueueMetrics implements MetricsSource { "ContainerActionQueue"; public static final String PIPELINE_ACTION_QUEUE_PREFIX = "PipelineActionQueue"; + public static final String COMMAND_STATUS_REPORT_QUEUE_PREFIX = + "CommandStatusReportQueue"; + public static final String COMMAND_STATUS_REPORT_SENT_PREFIX = + "CommandStatusReportSent"; + public static final String COMMAND_STATUS_REPORT_REQUEUED_PREFIX = + "CommandStatusReportRequeued"; + public static final String COMMAND_STATUS_REPORT_OVERSIZED_PREFIX = + "CommandStatusReportOversized"; private MetricsRegistry registry; @@ -71,6 +80,13 @@ public final class DatanodeQueueMetrics implements MetricsSource { private Map incrementalReportsQueueMap; private Map containerActionQueueMap; private Map pipelineActionQueueMap; + private Map commandStatusReportQueueBytesMap; + private Map commandStatusReportSentBytesMap; + private Map commandStatusReportRequeuedBytesMap; + private Map oversizedCommandStatusReportBytesMap; + private Map commandStatusReportSentBytes; + private Map commandStatusReportRequeuedBytes; + private Map oversizedCommandStatusReportBytes; public DatanodeQueueMetrics(DatanodeStateMachine datanodeStateMachine) { this.registry = new MetricsRegistry(METRICS_SOURCE_NAME); @@ -110,6 +126,13 @@ private void initializeQueues() { incrementalReportsQueueMap = new HashMap<>(); containerActionQueueMap = new HashMap<>(); pipelineActionQueueMap = new HashMap<>(); + commandStatusReportQueueBytesMap = new HashMap<>(); + commandStatusReportSentBytesMap = new HashMap<>(); + commandStatusReportRequeuedBytesMap = new HashMap<>(); + oversizedCommandStatusReportBytesMap = new HashMap<>(); + commandStatusReportSentBytes = new HashMap<>(); + commandStatusReportRequeuedBytes = new HashMap<>(); + oversizedCommandStatusReportBytes = new HashMap<>(); } @Override @@ -150,6 +173,28 @@ public void getMetrics(MetricsCollector collector, boolean b) { datanodeStateMachine.getContext().getPipelineActionQueueSize() .getOrDefault(entry.getKey(), 0)); } + Map queuedCommandStatusReportBytes = + datanodeStateMachine.getContext().getCommandStatusReportQueueBytes(); + for (Map.Entry entry: + commandStatusReportQueueBytesMap.entrySet()) { + builder.addGauge(entry.getValue(), + queuedCommandStatusReportBytes.getOrDefault(entry.getKey(), 0L)); + } + for (Map.Entry entry: + commandStatusReportSentBytesMap.entrySet()) { + builder.addCounter(entry.getValue(), + commandStatusReportSentBytes.get(entry.getKey()).get()); + } + for (Map.Entry entry: + commandStatusReportRequeuedBytesMap.entrySet()) { + builder.addCounter(entry.getValue(), + commandStatusReportRequeuedBytes.get(entry.getKey()).get()); + } + for (Map.Entry entry: + oversizedCommandStatusReportBytesMap.entrySet()) { + builder.addGauge(entry.getValue(), + oversizedCommandStatusReportBytes.get(entry.getKey()).get()); + } } public static synchronized void unRegister() { @@ -158,24 +203,84 @@ public static synchronized void unRegister() { } public void addEndpoint(HostAndPort endpoint) { + String endpointName = CaseFormat.UPPER_UNDERSCORE + .to(CaseFormat.UPPER_CAMEL, endpoint.getHostName()); incrementalReportsQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(INCREMENTAL_REPORT_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); containerActionQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(CONTAINER_ACTION_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); pipelineActionQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(PIPELINE_ACTION_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); + commandStatusReportQueueBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_QUEUE_PREFIX, + endpointName, "Queued command status report bytes for ")); + commandStatusReportSentBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_SENT_PREFIX, + endpointName, "Command status report bytes sent to ")); + commandStatusReportRequeuedBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_REQUEUED_PREFIX, + endpointName, "Command status report bytes requeued for ")); + oversizedCommandStatusReportBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_OVERSIZED_PREFIX, + endpointName, "Oversized command status report bytes for ")); + commandStatusReportSentBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); + commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); + oversizedCommandStatusReportBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); } public void removeEndpoint(HostAndPort endpoint) { incrementalReportsQueueMap.remove(endpoint); containerActionQueueMap.remove(endpoint); pipelineActionQueueMap.remove(endpoint); + commandStatusReportQueueBytesMap.remove(endpoint); + commandStatusReportSentBytesMap.remove(endpoint); + commandStatusReportRequeuedBytesMap.remove(endpoint); + oversizedCommandStatusReportBytesMap.remove(endpoint); + commandStatusReportSentBytes.remove(endpoint); + commandStatusReportRequeuedBytes.remove(endpoint); + oversizedCommandStatusReportBytes.remove(endpoint); + } + + public void incrCommandStatusReportBytesSent( + HostAndPort endpoint, long bytes) { + commandStatusReportSentBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).addAndGet(bytes); + } + + public void incrCommandStatusReportBytesRequeued( + HostAndPort endpoint, long bytes) { + commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).addAndGet(bytes); + } + + public void setOversizedCommandStatusReportBytes( + HostAndPort endpoint, long bytes) { + oversizedCommandStatusReportBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).set(bytes); + } + + @VisibleForTesting + public long getCommandStatusReportBytesSent(HostAndPort endpoint) { + AtomicLong value = commandStatusReportSentBytes.get(endpoint); + return value == null ? 0 : value.get(); + } + + @VisibleForTesting + public long getCommandStatusReportBytesRequeued(HostAndPort endpoint) { + AtomicLong value = commandStatusReportRequeuedBytes.get(endpoint); + return value == null ? 0 : value.get(); + } + + @VisibleForTesting + public long getOversizedCommandStatusReportBytes(HostAndPort endpoint) { + AtomicLong value = oversizedCommandStatusReportBytes.get(endpoint); + return value == null ? 0 : value.get(); } @VisibleForTesting @@ -198,4 +303,10 @@ private MetricsInfo getMetricsInfo(String prefix, String metricName) { String description = "Queue size of " + metricName + " from " + prefix; return info(metric, description); } + + private MetricsInfo getByteMetricsInfo(String prefix, String metricName, + String description) { + String metric = prefix + WordUtils.capitalize(metricName) + "Bytes"; + return info(metric, description + metricName); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index ae5352169eb9..d40e1e3a7c05 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -25,6 +25,7 @@ import static org.apache.hadoop.hdds.utils.HddsServerUtil.getScmInitialHeartbeatInterval; import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors.Descriptor; import com.google.protobuf.Message; import java.io.IOException; @@ -36,6 +37,7 @@ import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; +import java.util.ListIterator; import java.util.Map; import java.util.Objects; import java.util.OptionalLong; @@ -59,6 +61,7 @@ import org.apache.commons.collections4.CollectionUtils; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus.Status; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; @@ -69,6 +72,7 @@ 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.scm.net.HostAndPort; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ClosePipelineCommandHandler; @@ -375,38 +379,94 @@ public void putBackReports(List reportsToPutBack, */ public List getAllAvailableReports( HostAndPort endpoint) { - return getAllAvailableReportsUpToLimit(endpoint, Integer.MAX_VALUE); + List reports = getFullReports(endpoint, Integer.MAX_VALUE); + reports.addAll(getIncrementalReports(endpoint, Integer.MAX_VALUE - reports.size())); + return reports; } /** - * Returns up to {@code maxLimit} available reports for the endpoint, - * removing them from the queue. Any reports beyond the limit stay queued - * and are returned by a later call; callers can use {@link - * #hasPendingReports(HostAndPort)} to detect that case and schedule a - * follow-up heartbeat. - * - * @return List of reports + * Returns available reports other than command status reports. Full reports + * are selected before incremental reports, and the total number returned is + * capped by {@code maxLimit}. */ - public List getAllAvailableReports( + public List getAvailableNonCommandStatusReports( HostAndPort endpoint, int maxLimit) { - return getAllAvailableReportsUpToLimit(endpoint, maxLimit); + List reports = getFullReports(endpoint, maxLimit); + reports.addAll(getIncrementalReportsExcludingCommandStatus( + endpoint, maxLimit - reports.size())); + return reports; } /** - * Returns true if any reports are still queued for the endpoint. + * Selects command status reports within both the report count and serialized + * byte limits. An aggregate report may be split between complete command + * statuses, but an individual command status is never split. To guarantee + * progress, one status may exceed {@code maxBytes} when it still fits within + * {@code maxSingleStatusBytes}. */ - public boolean hasPendingReports(HostAndPort endpoint) { + public CommandStatusReportSelection getCommandStatusReports( + HostAndPort endpoint, int maxLimit, long maxBytes, + long maxSingleStatusBytes, long oversizedThreshold) { + List selected = new LinkedList<>(); + long selectedBytes = 0; + long oversizedBytes = 0; synchronized (incrementalReportsQueue) { - List reportsForEndpoint = incrementalReportsQueue.get(endpoint); - if (reportsForEndpoint != null && !reportsForEndpoint.isEmpty()) { - return true; + List reportsForEndpoint = + incrementalReportsQueue.get(endpoint); + if (reportsForEndpoint != null) { + ListIterator iterator = reportsForEndpoint.listIterator(); + while (iterator.hasNext() && selected.size() < maxLimit) { + Message message = iterator.next(); + if (!(message instanceof CommandStatusReportsProto)) { + continue; + } + + CommandStatusReportsProto report = + (CommandStatusReportsProto) message; + long remainingBytes = maxBytes - selectedBytes; + long reportBytes = getCommandStatusReportWireSize(report); + if (reportBytes <= remainingBytes) { + selected.add(report); + selectedBytes += reportBytes; + iterator.remove(); + continue; + } + + CommandStatusReportSplit split = + splitCommandStatusReport(report, remainingBytes); + if (split.getSelected() != null) { + selected.add(split.getSelected()); + selectedBytes += + getCommandStatusReportWireSize(split.getSelected()); + iterator.set(split.getRemainder()); + } else { + long firstStatusBytes = getFirstCommandStatusWireSize(report); + if (selected.isEmpty() && firstStatusBytes <= maxSingleStatusBytes) { + if (report.getCmdStatusCount() == 0) { + selected.add(report); + selectedBytes += firstStatusBytes; + iterator.remove(); + } else { + CommandStatusReportSplit singleStatus = + splitCommandStatusReport(report, firstStatusBytes); + selected.add(singleStatus.getSelected()); + selectedBytes += firstStatusBytes; + if (singleStatus.getRemainder().getCmdStatusCount() == 0) { + iterator.remove(); + } else { + iterator.set(singleStatus.getRemainder()); + } + } + } + if (firstStatusBytes > oversizedThreshold) { + oversizedBytes = firstStatusBytes; + } + } + break; + } } + return new CommandStatusReportSelection(selected, selectedBytes, oversizedBytes); } - Map fullReports = - isFullReportReadyToBeSent.get(endpoint); - return fullReports != null && fullReports.entrySet().stream() - .anyMatch(report -> report.getValue().get() - && type2Reports.get(report.getKey()).get() != null); } /** @@ -439,17 +499,6 @@ public ContainerReportsProto getFullContainerReportDiscardPendingICR() } } - @VisibleForTesting - List getAllAvailableReportsUpToLimit( - HostAndPort endpoint, - int limit) { - List reports = getFullReports(endpoint, limit); - List incrementalReports = getIncrementalReports(endpoint, - limit - reports.size()); // get all (MAX_VALUE) - reports.addAll(incrementalReports); - return reports; - } - List getIncrementalReports( HostAndPort endpoint, int maxLimit) { List reportsToReturn = new LinkedList<>(); @@ -466,6 +515,144 @@ List getIncrementalReports( return reportsToReturn; } + private List getIncrementalReportsExcludingCommandStatus( + HostAndPort endpoint, int maxLimit) { + List reportsToReturn = new LinkedList<>(); + synchronized (incrementalReportsQueue) { + List reportsForEndpoint = + incrementalReportsQueue.get(endpoint); + if (reportsForEndpoint != null) { + Iterator iterator = reportsForEndpoint.iterator(); + while (iterator.hasNext() && reportsToReturn.size() < maxLimit) { + Message report = iterator.next(); + if (!(report instanceof CommandStatusReportsProto)) { + reportsToReturn.add(report); + iterator.remove(); + } + } + } + } + return reportsToReturn; + } + + private CommandStatusReportSplit splitCommandStatusReport( + CommandStatusReportsProto report, long maxBytes) { + CommandStatusReportsProto.Builder selected = + CommandStatusReportsProto.newBuilder(); + CommandStatusReportsProto.Builder remainder = + CommandStatusReportsProto.newBuilder(); + long selectedReportBytes = 0; + boolean limitReached = false; + for (StorageContainerDatanodeProtocolProtos.CommandStatus status : + report.getCmdStatusList()) { + int statusBytes = CodedOutputStream.computeMessageSize( + CommandStatusReportsProto.CMDSTATUS_FIELD_NUMBER, status); + long candidateReportBytes = selectedReportBytes + statusBytes; + if (!limitReached && + getCommandStatusReportWireSize(candidateReportBytes) <= maxBytes) { + selected.addCmdStatus(status); + selectedReportBytes = candidateReportBytes; + } else { + limitReached = true; + remainder.addCmdStatus(status); + } + } + return new CommandStatusReportSplit( + selected.getCmdStatusCount() == 0 ? null : selected.build(), + remainder.build()); + } + + private long getFirstCommandStatusWireSize( + CommandStatusReportsProto report) { + if (report.getCmdStatusCount() == 0) { + return getCommandStatusReportWireSize(report); + } + int statusBytes = CodedOutputStream.computeMessageSize( + CommandStatusReportsProto.CMDSTATUS_FIELD_NUMBER, + report.getCmdStatus(0)); + return getCommandStatusReportWireSize(statusBytes); + } + + private static long getCommandStatusReportWireSize( + CommandStatusReportsProto report) { + return getCommandStatusReportWireSize(report.getSerializedSize()); + } + + private static long getCommandStatusReportWireSize( + long reportBytes) { + return CodedOutputStream.computeTagSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER) + + CodedOutputStream.computeUInt64SizeNoTag(reportBytes) + + reportBytes; + } + + public long getCommandStatusReportQueueBytes(HostAndPort endpoint) { + synchronized (incrementalReportsQueue) { + List reports = incrementalReportsQueue.get(endpoint); + return reports == null ? 0 : reports.stream() + .filter(CommandStatusReportsProto.class::isInstance) + .map(CommandStatusReportsProto.class::cast) + .mapToLong(StateContext::getCommandStatusReportWireSize) + .sum(); + } + } + + public Map getCommandStatusReportQueueBytes() { + synchronized (incrementalReportsQueue) { + return incrementalReportsQueue.keySet().stream() + .collect(Collectors.toMap( + endpoint -> endpoint, + this::getCommandStatusReportQueueBytes)); + } + } + + /** + * Command status reports selected for one heartbeat and their byte metrics. + */ + public static final class CommandStatusReportSelection { + private final List reports; + private final long selectedBytes; + private final long oversizedBytes; + + private CommandStatusReportSelection(List reports, long selectedBytes, + long oversizedBytes) { + this.reports = reports; + this.selectedBytes = selectedBytes; + this.oversizedBytes = oversizedBytes; + } + + public List getReports() { + return reports; + } + + public long getSelectedBytes() { + return selectedBytes; + } + + public long getOversizedBytes() { + return oversizedBytes; + } + } + + private static final class CommandStatusReportSplit { + private final CommandStatusReportsProto selected; + private final CommandStatusReportsProto remainder; + + private CommandStatusReportSplit(CommandStatusReportsProto selected, + CommandStatusReportsProto remainder) { + this.selected = selected; + this.remainder = remainder; + } + + private CommandStatusReportsProto getSelected() { + return selected; + } + + private CommandStatusReportsProto getRemainder() { + return remainder; + } + } + List getFullReports( HostAndPort endpoint, int maxLimit) { int count = 0; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 76358e4755c8..a093066e7697 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -21,15 +21,19 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD_DEFAULT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.hdds.conf.StorageUnit.BYTES; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toVersionProto; import com.google.common.base.Preconditions; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors; import com.google.protobuf.Message; import java.io.IOException; @@ -42,6 +46,7 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; 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.DatanodeVersionProto; @@ -57,6 +62,7 @@ import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine.EndPointStates; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; import org.apache.hadoop.ozone.protocol.commands.ClosePipelineCommand; @@ -80,6 +86,11 @@ public class HeartbeatEndpointTask implements Callable { private static final Logger LOG = LoggerFactory.getLogger(HeartbeatEndpointTask.class); + static final String HDDS_IPC_MAXIMUM_DATA_LENGTH = + "ipc.maximum.data.length"; + static final int HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT = + 128 * 1024 * 1024; + private static final double IPC_SIZE_SAFETY_FACTOR = 0.9; private final EndpointStateMachine rpcEndpoint; private final DatanodeDetails datanodeDetails; private StateContext context; @@ -91,6 +102,8 @@ public class HeartbeatEndpointTask private final HDDSVersion testCurrentVersion; private final boolean resolveOnFailureEnabled; private final int refreshThreshold; + private final long maxCommandStatusReportBytes; + private final long safeHeartbeatSize; /** * Constructs a SCM heart beat. @@ -109,6 +122,18 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT); this.maxReportsPerHB = Math.max(1, conf.getInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT)); + this.maxCommandStatusReportBytes = (long) conf.getStorageSize( + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT, BYTES); + Preconditions.checkArgument(maxCommandStatusReportBytes > 0, + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE + " must be positive"); + int ipcMaximumDataLength = conf.getInt( + HDDS_IPC_MAXIMUM_DATA_LENGTH, + HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT); + Preconditions.checkArgument(ipcMaximumDataLength > 0, + HDDS_IPC_MAXIMUM_DATA_LENGTH + " must be positive"); + this.safeHeartbeatSize = + (long) Math.floor(ipcMaximumDataLength * IPC_SIZE_SAFETY_FACTOR); this.datanodeDetails = context.getParent().getDatanodeDetails(); this.versionManager = context.getParent().getVersionManager(); @@ -136,6 +161,7 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, public EndpointStateMachine.EndPointStates call() throws Exception { rpcEndpoint.lock(); SCMHeartbeatRequestProto.Builder requestBuilder = null; + CommandStatusReportSelection commandStatusSelection = null; try { DatanodeVersionProto versionInfo = toVersionProto( versionManager.getApparentVersion(), @@ -148,10 +174,15 @@ public EndpointStateMachine.EndPointStates call() throws Exception { requestBuilder = SCMHeartbeatRequestProto.newBuilder() .setDatanodeDetails(datanodeDetailsProto) .setDatanodeVersion(versionInfo); - addReports(requestBuilder); - addContainerActions(requestBuilder); - addPipelineActions(requestBuilder); - addQueuedCommandCounts(requestBuilder); + SerializedSizeAccumulator heartbeatSize = new SerializedSizeAccumulator(); + heartbeatSize.add(SCMHeartbeatRequestProto.DATANODEDETAILS_FIELD_NUMBER, datanodeDetailsProto); + heartbeatSize.add(SCMHeartbeatRequestProto.DATANODEVERSION_FIELD_NUMBER, versionInfo); + int reportCount = addNonCommandStatusReports(requestBuilder, heartbeatSize); + addContainerActions(requestBuilder, heartbeatSize); + addPipelineActions(requestBuilder, heartbeatSize); + addQueuedCommandCounts(requestBuilder, heartbeatSize); + commandStatusSelection = addCommandStatusReports( + requestBuilder, maxReportsPerHB - reportCount, heartbeatSize.get()); SCMHeartbeatRequestProto request = requestBuilder.build(); LOG.debug("Sending heartbeat message : {}", request); SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint() @@ -159,14 +190,21 @@ public EndpointStateMachine.EndPointStates call() throws Exception { processResponse(response, datanodeDetailsProto); rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now()); rpcEndpoint.zeroMissedCount(); - if (context.hasPendingReports(rpcEndpoint.getAddress())) { - // addReports was capped at maxReportsPerHB and left reports queued. Trigger immediately. - context.getParent().setNextHB(Time.monotonicNow()); + if (context.getQueueMetrics() != null) { + context.getQueueMetrics().incrCommandStatusReportBytesSent( + rpcEndpoint.getAddress(), + commandStatusSelection.getSelectedBytes()); } } catch (IOException ex) { Preconditions.checkState(requestBuilder != null); // put back the reports which failed to be sent putBackIncrementalReports(requestBuilder); + if (commandStatusSelection != null && + context.getQueueMetrics() != null) { + context.getQueueMetrics().incrCommandStatusReportBytesRequeued( + rpcEndpoint.getAddress(), + commandStatusSelection.getSelectedBytes()); + } rpcEndpoint.logIfNeeded(ex); maybeRefreshScmAddress(ex); } finally { @@ -211,14 +249,13 @@ private void putBackIncrementalReports( context.putBackReports(reports, rpcEndpoint.getAddress()); } - /** - * Adds all the available reports to heartbeat. - * - * @param requestBuilder builder to which the report has to be added. - */ - private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { + private int addNonCommandStatusReports(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { + int reportCount = 0; for (Message report : - context.getAllAvailableReports(rpcEndpoint.getAddress(), maxReportsPerHB)) { + context.getAvailableNonCommandStatusReports( + rpcEndpoint.getAddress(), maxReportsPerHB)) { + reportCount++; String reportName = report.getDescriptorForType().getFullName(); for (Descriptors.FieldDescriptor descriptor : SCMHeartbeatRequestProto.getDescriptor().getFields()) { @@ -229,10 +266,45 @@ private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { } else { requestBuilder.setField(descriptor, report); } + heartbeatSize.add(descriptor.getNumber(), report); break; } } } + return reportCount; + } + + private CommandStatusReportSelection addCommandStatusReports( + SCMHeartbeatRequestProto.Builder requestBuilder, + int remainingReportCount, long currentHeartbeatSize) { + long remainingSafeBytes = Math.max( + 0, safeHeartbeatSize - currentHeartbeatSize); + long commandStatusReportBudget = Math.min(maxCommandStatusReportBytes, remainingSafeBytes); + long oversizedThreshold = Math.min(maxCommandStatusReportBytes, safeHeartbeatSize); + CommandStatusReportSelection selection = + context.getCommandStatusReports(rpcEndpoint.getAddress(), + remainingReportCount, commandStatusReportBudget, + remainingSafeBytes, oversizedThreshold); + for (CommandStatusReportsProto report : selection.getReports()) { + requestBuilder.addCommandStatusReports(report); + } + + if (currentHeartbeatSize > safeHeartbeatSize) { + LOG.warn("Heartbeat for {} is already {} bytes before command status " + + "reports, exceeding the safe size of {} bytes", + rpcEndpoint.getAddress(), currentHeartbeatSize, safeHeartbeatSize); + } + if (selection.getOversizedBytes() > 0) { + LOG.warn("A command status for {} is {} bytes, exceeding the effective " + + "limit of {} bytes (configured limit: {}, safe heartbeat size: {})", + rpcEndpoint.getAddress(), selection.getOversizedBytes(), + oversizedThreshold, maxCommandStatusReportBytes, safeHeartbeatSize); + } + if (context.getQueueMetrics() != null) { + context.getQueueMetrics().setOversizedCommandStatusReportBytes( + rpcEndpoint.getAddress(), selection.getOversizedBytes()); + } + return selection; } /** @@ -240,8 +312,8 @@ private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { * * @param requestBuilder builder to which the report has to be added. */ - private void addContainerActions( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addContainerActions(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { List actions = context.getPendingContainerAction( rpcEndpoint.getAddress(), maxContainerActionsPerHB); if (!actions.isEmpty()) { @@ -249,6 +321,7 @@ private void addContainerActions( .addAllContainerActions(actions) .build(); requestBuilder.setContainerActions(cap); + heartbeatSize.add(SCMHeartbeatRequestProto.CONTAINERACTIONS_FIELD_NUMBER, cap); } } @@ -257,8 +330,8 @@ private void addContainerActions( * * @param requestBuilder builder to which the report has to be added. */ - private void addPipelineActions( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addPipelineActions(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { List actions = context.getPendingPipelineAction( rpcEndpoint.getAddress(), maxPipelineActionsPerHB); if (!actions.isEmpty()) { @@ -266,6 +339,7 @@ private void addPipelineActions( .addAllPipelineActions(actions) .build(); requestBuilder.setPipelineActions(pap); + heartbeatSize.add(SCMHeartbeatRequestProto.PIPELINEACTIONS_FIELD_NUMBER, pap); } } @@ -273,8 +347,8 @@ private void addPipelineActions( * Adds the count of all queued commands to the heartbeat. * @param requestBuilder Builder to which the details will be added. */ - private void addQueuedCommandCounts( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addQueuedCommandCounts(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { EnumCounters commandCount = context.getParent().getQueuedCommandCount(); CommandQueueReportProto.Builder reportProto = @@ -286,7 +360,21 @@ private void addQueuedCommandCounts( .addCount((int) count); } } - requestBuilder.setCommandQueueReport(reportProto.build()); + CommandQueueReportProto commandQueueReport = reportProto.build(); + requestBuilder.setCommandQueueReport(commandQueueReport); + heartbeatSize.add(SCMHeartbeatRequestProto.COMMANDQUEUEREPORT_FIELD_NUMBER, commandQueueReport); + } + + private static final class SerializedSizeAccumulator { + private long size; + + private void add(int fieldNumber, Message message) { + size += CodedOutputStream.computeMessageSize(fieldNumber, message); + } + + private long get() { + return size; + } } /** diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 249b0ac788cf..76569663d518 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -30,6 +30,7 @@ import static org.mockito.Mockito.when; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors.Descriptor; import com.google.protobuf.Message; import java.io.IOException; @@ -52,18 +53,22 @@ import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineAction; 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.scm.net.HostAndPort; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.ContainerTestUtils; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.common.states.DatanodeState; import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; @@ -228,46 +233,148 @@ public void testReportQueueWithAddReports() throws IOException { } @Test - public void testGetAllAvailableReportsRespectsLimit() throws IOException { - StateContext ctx = createSubject(); - HostAndPort scm1 = new HostAndPort("scm1", 9001); - ctx.addEndpoint(scm1); - - // Queue 10 ICRs; no full report is refreshed, so only ICRs are returned. - batchAddIncrementalReport(ctx, - StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 10); - - // A limited collection returns at most the limit and leaves the rest. - assertEquals(4, ctx.getAllAvailableReports(scm1, 4).size()); - assertTrue(ctx.hasPendingReports(scm1)); - assertEquals(4, ctx.getAllAvailableReports(scm1, 4).size()); - assertTrue(ctx.hasPendingReports(scm1)); - // Last chunk drains the queue. - assertEquals(2, ctx.getAllAvailableReports(scm1, 4).size()); - assertFalse(ctx.hasPendingReports(scm1)); - assertEquals(0, ctx.getAllAvailableReports(scm1, 4).size()); + public void testCommandStatusReportsRespectByteLimit() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + long firstBytes = commandStatusReportWireSize(first); + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, firstBytes, firstBytes, firstBytes); + assertEquals(Collections.singletonList(first), selection.getReports()); + assertEquals(firstBytes, selection.getSelectedBytes()); + assertEquals(commandStatusReportWireSize(second), context.getCommandStatusReportQueueBytes(scm)); } @Test - public void testHasPendingReportsWithFullReports() throws IOException { - StateContext ctx = createSubject(); - HostAndPort scm1 = new HostAndPort("scm1", 9001); - ctx.addEndpoint(scm1); + public void testAggregateCommandStatusReportIsSplitAtStatusBoundary() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + + CommandStatus first = newCommandStatus(1, 128); + CommandStatus second = newCommandStatus(2, 128); + CommandStatus third = newCommandStatus(3, 128); + CommandStatusReportsProto aggregate = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(first) + .addCmdStatus(second) + .addCmdStatus(third) + .build(); + CommandStatusReportsProto firstBatch = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(first) + .addCmdStatus(second) + .build(); + CommandStatusReportsProto secondBatch = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(third) + .build(); + context.addIncrementalReport(aggregate); + + CommandStatusReportSelection selection = context.getCommandStatusReports( + scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch)); + assertEquals(Collections.singletonList(firstBatch), + selection.getReports()); + assertEquals(commandStatusReportWireSize(secondBatch), context.getCommandStatusReportQueueBytes(scm)); + + selection = context.getCommandStatusReports( + scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch)); + assertEquals(Collections.singletonList(secondBatch), + selection.getReports()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); + } - // addEndpoint marks the full-report ready flags true, but no full report - // value has been generated yet, so nothing is actually pending. - assertFalse(ctx.hasPendingReports(scm1)); + @Test + public void testNonCommandReportsAreSelectedIndependently() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatusReportsProto commandStatus = + newCommandStatusReport(1, 128); + IncrementalContainerReportProto icr = + IncrementalContainerReportProto.getDefaultInstance(); + context.addIncrementalReport(commandStatus); + context.addIncrementalReport(icr); + + assertEquals(Collections.singletonList(icr), + context.getAvailableNonCommandStatusReports(scm, 10)); + assertEquals(commandStatusReportWireSize(commandStatus), + context.getCommandStatusReportQueueBytes(scm)); + } - // Generate all three full reports; now values exist and flags are ready. - batchRefreshfullReports(ctx, StateContext.CONTAINER_REPORTS_PROTO_NAME, 1); - batchRefreshfullReports(ctx, StateContext.NODE_REPORT_PROTO_NAME, 1); - batchRefreshfullReports(ctx, StateContext.PIPELINE_REPORTS_PROTO_NAME, 1); - assertTrue(ctx.hasPendingReports(scm1)); + @Test + public void testOversizedCommandStatusMakesProgressWhenSafe() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatusReportsProto report = newCommandStatusReport(1, 128); + context.addIncrementalReport(report); + long reportBytes = commandStatusReportWireSize(report); + + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, reportBytes - 1, reportBytes, reportBytes - 1); + assertEquals(Collections.singletonList(report), selection.getReports()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); + assertEquals(reportBytes, selection.getOversizedBytes()); + + context.addIncrementalReport(report); + selection = context.getCommandStatusReports( + scm, 10, reportBytes - 1, reportBytes - 1, reportBytes - 1); + assertTrue(selection.getReports().isEmpty()); + assertEquals(reportBytes, context.getCommandStatusReportQueueBytes(scm)); + assertEquals(reportBytes, selection.getOversizedBytes()); + } - // With a cap of one, a heartbeat takes a single full report but the others - // stay ready, so a follow-up is still needed. - assertEquals(1, ctx.getAllAvailableReports(scm1, 1).size()); - assertTrue(ctx.hasPendingReports(scm1)); + @Test + public void testOversizedCommandStatusPreservesAggregateOrder() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatus firstSmall = newCommandStatus(1, 16); + CommandStatus oversized = newCommandStatus(2, 256); + CommandStatus lastSmall = newCommandStatus(3, 16); + CommandStatusReportsProto aggregate = CommandStatusReportsProto.newBuilder() + .addCmdStatus(firstSmall) + .addCmdStatus(oversized) + .addCmdStatus(lastSmall) + .build(); + CommandStatusReportsProto firstSmallReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(firstSmall).build(); + CommandStatusReportsProto oversizedReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(oversized).build(); + CommandStatusReportsProto lastSmallReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(lastSmall).build(); + CommandStatusReportsProto firstRemainder = CommandStatusReportsProto.newBuilder() + .addCmdStatus(oversized) + .addCmdStatus(lastSmall) + .build(); + long smallBytes = commandStatusReportWireSize(firstSmallReport); + long oversizedBytes = commandStatusReportWireSize(oversizedReport); + context.addIncrementalReport(aggregate); + + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(firstSmallReport), selection.getReports()); + assertEquals(commandStatusReportWireSize(firstRemainder), context.getCommandStatusReportQueueBytes(scm)); + + selection = context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(oversizedReport), selection.getReports()); + assertEquals(oversizedBytes, selection.getOversizedBytes()); + assertEquals(commandStatusReportWireSize(lastSmallReport), context.getCommandStatusReportQueueBytes(scm)); + + selection = context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(lastSmallReport), selection.getReports()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); } void batchRefreshfullReports(StateContext ctx, String reportName, int count) { @@ -294,6 +401,29 @@ void checkReportCount(List reports, assertEquals(expectedReportCount, reportCount); } + private static CommandStatusReportsProto newCommandStatusReport( + long commandId, int messageSize) { + return CommandStatusReportsProto.newBuilder() + .addCmdStatus(newCommandStatus(commandId, messageSize)) + .build(); + } + + private static CommandStatus newCommandStatus( + long commandId, int messageSize) { + return CommandStatus.newBuilder() + .setCmdId(commandId) + .setStatus(CommandStatus.Status.EXECUTED) + .setType(SCMCommandProto.Type.deleteBlocksCommand) + .setMsg(String.join("", Collections.nCopies(messageSize, "x"))) + .build(); + } + + private static long commandStatusReportWireSize( + CommandStatusReportsProto report) { + return CodedOutputStream.computeMessageSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER, report); + } + /** * Check if Container, Node and Pipeline report APIs work as expected. */ @@ -729,17 +859,17 @@ public void testGetReports() { // (100 - 3 non-incremental reports) expectedReportCount.put( StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 97); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm1, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm1, 100), expectedReportCount); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm2, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm2, 100), expectedReportCount); expectedReportCount.clear(); expectedReportCount.put( StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, totalIncrementalCount - 97); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm1, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm1, 100), expectedReportCount); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm2, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm2, 100), expectedReportCount); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 4f90b767cdcd..815a3bf97910 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -18,9 +18,11 @@ package org.apache.hadoop.ozone.container.common.states.endpoint; import static java.util.Collections.emptyList; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconcileContainerCommand; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconstructECContainersCommand; +import static org.apache.hadoop.ozone.container.common.states.endpoint.HeartbeatEndpointTask.HDDS_IPC_MAXIMUM_DATA_LENGTH; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -28,12 +30,16 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.anyLong; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.UnsafeByteOperations; +import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.OptionalLong; @@ -46,8 +52,10 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto; 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; @@ -58,10 +66,12 @@ import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.ContainerTestUtils; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeQueueMetrics; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.ReconcileContainerCommand; import org.apache.hadoop.ozone.protocol.commands.ReconstructECContainersCommand; @@ -405,7 +415,7 @@ public void testheartbeatWithAllReports() throws Exception { } @Test - public void testHeartbeatCapsReportsAndSchedulesFollowup() throws Exception { + public void testHeartbeatCapsReports() throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); @@ -441,16 +451,239 @@ public void testHeartbeatCapsReportsAndSchedulesFollowup() throws Exception { // Only the capped number of ICRs is sent in this heartbeat. SCMHeartbeatRequestProto heartbeat = argument.getValue(); assertEquals(2, heartbeat.getIncrementalContainerReportCount()); - // The remainder stays queued and a follow-up heartbeat is scheduled now. - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); - verify(datanodeStateMachine).setNextHB(anyLong()); + // The remainder stays queued for the next regular heartbeat. + assertEquals(3, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); - // The follow-ups drain the rest; the last one sends 1 ICR and schedules nothing. + // Subsequent regular heartbeats drain the rest. endpointTask.call(); endpointTask.call(); assertEquals(1, argument.getValue().getIncrementalContainerReportCount()); - assertFalse(context.hasPendingReports(TEST_SCM_ENDPOINT)); - verify(datanodeStateMachine, times(2)).setNextHB(anyLong()); + assertEquals(0, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); + } + + @Test + public void testHeartbeatCapsCommandStatusReportsByBytes() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 10); + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, reportBytes + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor + .forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + assertEquals(1, argument.getValue().getCommandStatusReportsCount()); + assertEquals(first, + argument.getValue().getCommandStatusReports(0)); + assertEquals(commandStatusReportWireSize(second), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesSent(TEST_SCM_ENDPOINT)); + assertEquals(0, queueMetrics.getOversizedCommandStatusReportBytes(TEST_SCM_ENDPOINT)); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); + } + + @Test + public void testNonCommandReportsConsumeReportCountLimit() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID(((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask(conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(IncrementalContainerReportProto.getDefaultInstance()); + context.addIncrementalReport(IncrementalContainerReportProto.getDefaultInstance()); + CommandStatusReportsProto commandStatus = newCommandStatusReport(1, 128); + context.addIncrementalReport(commandStatus); + + endpointTask.call(); + + assertEquals(2, argument.getValue().getIncrementalContainerReportCount()); + assertEquals(0, argument.getValue().getCommandStatusReportsCount()); + assertEquals(commandStatusReportWireSize(commandStatus), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + } + + @Test + public void testOversizedCommandStatusSentWhenSafe() throws Exception { + CommandStatusReportsProto report = newCommandStatusReport(1, 128); + long reportBytes = commandStatusReportWireSize(report); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, (reportBytes - 1) + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID(((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask(conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(report); + + endpointTask.call(); + + assertEquals(report, argument.getValue().getCommandStatusReports(0)); + assertEquals(0, context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesSent(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getOversizedCommandStatusReportBytes(TEST_SCM_ENDPOINT)); + } + + @Test + public void testExistingHeartbeatContentReducesCommandStatusBudget() throws Exception { + IncrementalContainerReportProto icr = + IncrementalContainerReportProto.newBuilder() + .addReport(ContainerReplicaProto.newBuilder() + .setContainerID(1) + .setState(ContainerReplicaProto.State.CLOSED) + .setOriginNodeId(String.join("", Collections.nCopies(1024, "x"))) + .build()) + .build(); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + OzoneConfiguration baselineConf = new OzoneConfiguration(); + StateContext baselineContext = new StateContext(baselineConf, + DatanodeStates.RUNNING, datanodeStateMachine, ""); + StorageContainerDatanodeProtocolClientSideTranslatorPB baselineScm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor baselineArgument = + ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(baselineScm.sendHeartbeat(baselineArgument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + HeartbeatEndpointTask baselineTask = getHeartbeatEndpointTask( + baselineConf, baselineContext, baselineScm); + baselineContext.addEndpoint(TEST_SCM_ENDPOINT); + baselineContext.addIncrementalReport(icr); + baselineTask.call(); + long nonCommandHeartbeatBytes = + baselineArgument.getValue().getSerializedSize(); + + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + long targetSafeSize = nonCommandHeartbeatBytes + reportBytes; + int ipcMaximumDataLength = + (int) Math.ceil(targetSafeSize / 0.9D); + long safeHeartbeatSize = + (long) Math.floor(ipcMaximumDataLength * 0.9D); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(HDDS_IPC_MAXIMUM_DATA_LENGTH, ipcMaximumDataLength); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, (reportBytes * 2) + "B"); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = + ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(icr); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + assertEquals(1, argument.getValue().getCommandStatusReportsCount()); + assertTrue(argument.getValue().getSerializedSize() <= safeHeartbeatSize); + assertEquals(commandStatusReportWireSize(second), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + } + + @Test + public void testFailedHeartbeatRestoresCommandStatusReports() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, reportBytes + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + when(scm.sendHeartbeat(any())).thenThrow(new IOException("failure")); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + CommandStatusReportSelection selection = context.getCommandStatusReports( + TEST_SCM_ENDPOINT, 10, reportBytes * 2, reportBytes * 2, reportBytes * 2); + assertEquals(Arrays.asList(first, second), selection.getReports()); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesRequeued(TEST_SCM_ENDPOINT)); } @ParameterizedTest @@ -490,7 +723,7 @@ public void testHeartbeatClampsNonPositiveReportLimit(int configuredLimit) // progress (one ICR) rather than sending nothing (0) or throwing (-1). endpointTask.call(); assertEquals(1, argument.getValue().getIncrementalContainerReportCount()); - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(2, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); } @Test @@ -626,4 +859,22 @@ private ContainerAction getContainerAction() { .setReason(ContainerAction.Reason.CONTAINER_FULL); return builder.build(); } + + private static CommandStatusReportsProto newCommandStatusReport( + long commandId, int messageSize) { + return CommandStatusReportsProto.newBuilder() + .addCmdStatus(CommandStatus.newBuilder() + .setCmdId(commandId) + .setStatus(CommandStatus.Status.EXECUTED) + .setType(SCMCommandProto.Type.deleteBlocksCommand) + .setMsg(String.join("", Collections.nCopies(messageSize, "x"))) + .build()) + .build(); + } + + private static long commandStatusReportWireSize( + CommandStatusReportsProto report) { + return CodedOutputStream.computeMessageSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER, report); + } }