Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.TotalRunTimeMs;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.QueueWaitingTaskCount;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.InvocationCount;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.AvgRunTimeMs;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.ThreadPoolActivePoolSize;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.ThreadPoolMaxPoolSize;
import static org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics.CommandMetricsMetricsInfo.CommandReceivedCount;
Expand All @@ -46,6 +47,7 @@ public final class CommandHandlerMetrics implements MetricsSource {
enum CommandMetricsMetricsInfo implements MetricsInfo {
Command("The type of the SCM command"),
TotalRunTimeMs("The total runtime of the command handler in milliseconds"),
AvgRunTimeMs("Average run time of the command handler in milliseconds"),
QueueWaitingTaskCount("The number of queued tasks waiting for execution"),
InvocationCount("The number of times the command handler has been invoked"),
ThreadPoolActivePoolSize("The number of active threads in the thread pool"),
Expand Down Expand Up @@ -108,6 +110,7 @@ public void getMetrics(MetricsCollector collector, boolean all) {
commandHandler.getCommandType().name());

builder.addGauge(TotalRunTimeMs, commandHandler.getTotalRunTime());
builder.addGauge(AvgRunTimeMs, commandHandler.getAverageRunTime());
builder.addGauge(QueueWaitingTaskCount, commandHandler.getQueuedCount());
builder.addGauge(InvocationCount, commandHandler.getInvocationCount());
int activePoolSize = commandHandler.getThreadPoolActivePoolSize();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class ReconstructECContainersCommandHandler implements CommandHandler {
private final ReplicationSupervisor supervisor;
private final ECReconstructionCoordinator coordinator;
private final ConfigurationSource conf;
private String metricsName;

public ReconstructECContainersCommandHandler(ConfigurationSource conf,
ReplicationSupervisor supervisor,
Expand All @@ -52,8 +53,16 @@ public void handle(SCMCommand command, OzoneContainer container,
(ReconstructECContainersCommand) command;
ECReconstructionCommandInfo reconstructionCommandInfo =
new ECReconstructionCommandInfo(ecContainersCommand);
this.supervisor.addTask(new ECReconstructionCoordinatorTask(
coordinator, reconstructionCommandInfo));
ECReconstructionCoordinatorTask task = new ECReconstructionCoordinatorTask(
coordinator, reconstructionCommandInfo);
if (this.metricsName == null) {
this.metricsName = task.getMetricName();
}
this.supervisor.addTask(task);
}

public String getMetricsName() {
return this.metricsName;
}

@Override
Expand All @@ -63,17 +72,23 @@ public Type getCommandType() {

@Override
public int getInvocationCount() {
return 0;
return this.metricsName == null ? 0 : (int) this.supervisor
.getReplicationRequestCount(metricsName);
}

@Override
public long getAverageRunTime() {
long invocationCount = getInvocationCount();
if (invocationCount > 0) {
return getTotalRunTime() / invocationCount;
}
return 0;
}

@Override
public long getTotalRunTime() {
return 0;
return this.metricsName == null ? 0 : this.supervisor
.getReplicationRequestTotalTime(metricsName);
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,29 +43,28 @@ public class ReplicateContainerCommandHandler implements CommandHandler {
static final Logger LOG =
LoggerFactory.getLogger(ReplicateContainerCommandHandler.class);

private int invocationCount;

private long totalTime;

private ConfigurationSource conf;

private ReplicationSupervisor supervisor;

private ContainerReplicator downloadReplicator;

private ContainerReplicator pushReplicator;

private String metricsName;

public ReplicateContainerCommandHandler(
ConfigurationSource conf,
ReplicationSupervisor supervisor,
ContainerReplicator downloadReplicator,
ContainerReplicator pushReplicator) {
this.conf = conf;
this.supervisor = supervisor;
this.downloadReplicator = downloadReplicator;
this.pushReplicator = pushReplicator;
}

public String getMetricsName() {
return this.metricsName;
}

@Override
public void handle(SCMCommand command, OzoneContainer container,
StateContext context, SCMConnectionManager connectionManager) {
Expand All @@ -86,6 +85,9 @@ public void handle(SCMCommand command, OzoneContainer container,
downloadReplicator : pushReplicator;

ReplicationTask task = new ReplicationTask(replicateCommand, replicator);
if (metricsName == null) {
metricsName = task.getMetricName();
}
supervisor.addTask(task);
}

Expand All @@ -101,19 +103,22 @@ public SCMCommandProto.Type getCommandType() {

@Override
public int getInvocationCount() {
return this.invocationCount;
return this.metricsName == null ? 0 : (int) this.supervisor
.getReplicationRequestCount(metricsName);
}

@Override
public long getAverageRunTime() {
long invocationCount = getInvocationCount();
if (invocationCount > 0) {
return totalTime / invocationCount;
return getTotalRunTime() / invocationCount;
}
return 0;
}

@Override
public long getTotalRunTime() {
return totalTime;
return this.metricsName == null ? 0 : this.supervisor
.getReplicationRequestTotalTime(metricsName);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@

import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.util.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -77,6 +78,7 @@ public final class ReplicationSupervisor {
private final Map<String, AtomicLong> failureCounter = new ConcurrentHashMap<>();
private final Map<String, AtomicLong> timeoutCounter = new ConcurrentHashMap<>();
private final Map<String, AtomicLong> skippedCounter = new ConcurrentHashMap<>();
private final Map<String, AtomicLong> requestTotalTime = new ConcurrentHashMap<>();

private static final Map<String, String> METRICS_MAP;

Expand Down Expand Up @@ -240,6 +242,7 @@ public void addTask(AbstractReplicationTask task) {
failureCounter.put(task.getMetricName(), new AtomicLong(0));
timeoutCounter.put(task.getMetricName(), new AtomicLong(0));
skippedCounter.put(task.getMetricName(), new AtomicLong(0));
requestTotalTime.put(task.getMetricName(), new AtomicLong(0));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we add queued count here as well? The replication supervisor already has this information, but it is exposed via the individual commands instead. It might be more intuitive to have it here as well.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @errose28 for the comment and review.
I will update it later.
Right now, taskCounter records the number of queues.

private final Map<Class<?>, AtomicInteger> taskCounter =
      new ConcurrentHashMap<>();

I will add the QueueCount metric to ReplicationSupervisorMetrics later.
In addition, I suggest improving taskCounter so that it is consistent with requestCounter and successCounter.

private final Map<String, AtomicLong> taskCounter = new ConcurrentHashMap<>();

What do you think?

METRICS_MAP.put(task.getMetricName(), task.getMetricDescriptionSegment());
}
}
Expand Down Expand Up @@ -353,6 +356,7 @@ public TaskRunner(AbstractReplicationTask task) {

@Override
public void run() {
final long startTime = Time.monotonicNow();
try {
requestCounter.get(task.getMetricName()).incrementAndGet();

Expand Down Expand Up @@ -401,6 +405,8 @@ public void run() {
LOG.warn("Failed {}", this, e);
failureCounter.get(task.getMetricName()).incrementAndGet();
} finally {
requestTotalTime.get(task.getMetricName()).addAndGet(
Time.monotonicNow() - startTime);
inFlight.remove(task);
decrementTaskCounter(task);
}
Expand Down Expand Up @@ -511,4 +517,9 @@ public long getReplicationSkippedCount(String metricsName) {
return counter != null ? counter.get() : 0;
}

public long getReplicationRequestTotalTime(String metricsName) {
AtomicLong counter = requestTotalTime.get(metricsName);
return counter != null ? counter.get() : 0;
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/
package org.apache.hadoop.ozone.container.common.statemachine.commandhandler;

import com.google.protobuf.ByteString;
import com.google.protobuf.Proto2Utils;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
import org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics;
import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
import org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinator;
import org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinatorTask;
import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
import org.apache.hadoop.ozone.container.replication.ReplicationSupervisor;
import org.apache.hadoop.ozone.protocol.commands.ReconstructECContainersCommand;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.when;

/**
* Test cases to verify {@link ReconstructECContainersCommandHandler}.
*/
public class TestReconstructECContainersCommandHandler {
private OzoneConfiguration conf;
private ReplicationSupervisor supervisor;
private ECReconstructionCoordinator coordinator;
private OzoneContainer ozoneContainer;
private StateContext stateContext;
private SCMConnectionManager connectionManager;

@BeforeEach
public void setUp() {
supervisor = mock(ReplicationSupervisor.class);
coordinator = mock(ECReconstructionCoordinator.class);
conf = new OzoneConfiguration();
ozoneContainer = mock(OzoneContainer.class);
connectionManager = mock(SCMConnectionManager.class);
stateContext = mock(StateContext.class);
}

@Test
public void testMetrics() {
ReconstructECContainersCommandHandler commandHandler =
new ReconstructECContainersCommandHandler(conf, supervisor, coordinator);
doNothing().when(supervisor).addTask(any());
Map<SCMCommandProto.Type, CommandHandler> handlerMap = new HashMap<>();
handlerMap.put(commandHandler.getCommandType(), commandHandler);
CommandHandlerMetrics metrics = CommandHandlerMetrics.create(handlerMap);
try {
byte[] missingIndexes = {1, 2};
ByteString missingContainerIndexes = Proto2Utils.unsafeByteString(missingIndexes);
ECReplicationConfig ecReplicationConfig = new ECReplicationConfig(3, 2);
List<DatanodeDetails> dnDetails = getDNDetails(5);
List<ReconstructECContainersCommand.DatanodeDetailsAndReplicaIndex> sources =
dnDetails.stream().map(a -> new ReconstructECContainersCommand
.DatanodeDetailsAndReplicaIndex(a, dnDetails.indexOf(a)))
.collect(Collectors.toList());
List<DatanodeDetails> targets = getDNDetails(2);
ReconstructECContainersCommand reconstructECContainersCommand =
new ReconstructECContainersCommand(1L, sources, targets,
missingContainerIndexes, ecReplicationConfig);

commandHandler.handle(reconstructECContainersCommand, ozoneContainer,
stateContext, connectionManager);
String metricsName = "ECReconstructions";
assertEquals(commandHandler.getMetricsName(), metricsName);
when(supervisor.getReplicationRequestCount(metricsName)).thenReturn(1L);
assertEquals(commandHandler.getInvocationCount(), 1);

commandHandler.handle(new ReconstructECContainersCommand(2L, sources,
targets, missingContainerIndexes, ecReplicationConfig), ozoneContainer,
stateContext, connectionManager);
commandHandler.handle(new ReconstructECContainersCommand(3L, sources,
targets, missingContainerIndexes, ecReplicationConfig), ozoneContainer,
stateContext, connectionManager);
commandHandler.handle(new ReconstructECContainersCommand(4L, sources,
targets, missingContainerIndexes, ecReplicationConfig), ozoneContainer,
stateContext, connectionManager);
commandHandler.handle(new ReconstructECContainersCommand(5L, sources,
targets, missingContainerIndexes, ecReplicationConfig), ozoneContainer,
stateContext, connectionManager);
commandHandler.handle(new ReconstructECContainersCommand(6L, sources,
targets, missingContainerIndexes, ecReplicationConfig), ozoneContainer,
stateContext, connectionManager);

when(supervisor.getReplicationRequestCount(metricsName)).thenReturn(5L);
when(supervisor.getReplicationRequestTotalTime(metricsName)).thenReturn(10L);
when(supervisor.getInFlightReplications(ECReconstructionCoordinatorTask.class))
.thenReturn(1);
assertEquals(commandHandler.getInvocationCount(), 5);
assertEquals(commandHandler.getQueuedCount(), 1);
assertEquals(commandHandler.getTotalRunTime(), 10);

MetricsCollectorImpl metricsCollector = new MetricsCollectorImpl();
metrics.getMetrics(metricsCollector, true);
assertEquals(1, metricsCollector.getRecords().size());
} finally {
metrics.unRegister();
}
}

private List<DatanodeDetails> getDNDetails(int numDns) {
List<DatanodeDetails> dns = new ArrayList<>();
for (int i = 0; i < numDns; i++) {
dns.add(MockDatanodeDetails.randomDatanodeDetails());
}
return dns;
}
}
Loading