Skip to content

HDDS-11376. Improve ReplicationSupervisor to record replication metrics #7140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sodonnel merged 10 commits into from
Sep 10, 2024
Merged

HDDS-11376. Improve ReplicationSupervisor to record replication metrics #7140

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
7d04590
HDDS-11376. Improve ReplicationSupervisor to record replication metrics
jianghuazhu Aug 30, 2024
9627505
Fix checkstyle.
jianghuazhu Aug 30, 2024
e25bb91
Improve requestCounter to record metrics for different replication ty…
jianghuazhu Sep 5, 2024
9e6ff29
Update some code.
jianghuazhu Sep 6, 2024
d143ace
Fix some checkstyle
jianghuazhu Sep 6, 2024
ce67b30
Fix some checkstyle
jianghuazhu Sep 6, 2024
f03b420
Fix some checkstyle
jianghuazhu Sep 6, 2024
ae8fc2c
Improve some abstract methods.
jianghuazhu Sep 6, 2024
06f3db3
Simplify some code.
jianghuazhu Sep 7, 2024
8e06209
Improve some code.
jianghuazhu Sep 9, 2024
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
80 changes: 64 additions & 16 deletions ...ce/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,11 +71,11 @@ public final class ReplicationSupervisor {
private final StateContext context;
private final Clock clock;

private final AtomicLong requestCounter = new AtomicLong();
private final AtomicLong successCounter = new AtomicLong();
private final AtomicLong failureCounter = new AtomicLong();
private final AtomicLong timeoutCounter = new AtomicLong();
private final AtomicLong skippedCounter = new AtomicLong();
private final Map<Class<?>, AtomicLong> requestCounter = new ConcurrentHashMap<>();
private final Map<Class<?>, AtomicLong> successCounter = new ConcurrentHashMap<>();
private final Map<Class<?>, AtomicLong> failureCounter = new ConcurrentHashMap<>();
private final Map<Class<?>, AtomicLong> timeoutCounter = new ConcurrentHashMap<>();
private final Map<Class<?>, AtomicLong> skippedCounter = new ConcurrentHashMap<>();

/**
* A set of container IDs that are currently being downloaded
Expand Down Expand Up @@ -221,6 +221,18 @@ public void addTask(AbstractReplicationTask task) {
return;
}

if (requestCounter.get(task.getClass()) == null) {
synchronized (this) {
if (requestCounter.get(task.getClass()) == null) {
requestCounter.put(task.getClass(), new AtomicLong(0));
successCounter.put(task.getClass(), new AtomicLong(0));
failureCounter.put(task.getClass(), new AtomicLong(0));
timeoutCounter.put(task.getClass(), new AtomicLong(0));
skippedCounter.put(task.getClass(), new AtomicLong(0));
}
}
}

if (inFlight.add(task)) {
if (task.getPriority() != ReplicationCommandPriority.LOW) {
// Low priority tasks are not included in the replication queue sizes
Expand Down Expand Up @@ -330,14 +342,14 @@ public TaskRunner(AbstractReplicationTask task) {
@Override
public void run() {
try {
requestCounter.incrementAndGet();
requestCounter.get(task.getClass()).incrementAndGet();

final long now = clock.millis();
final long deadline = task.getDeadline();
if (deadline > 0 && now > deadline) {
LOG.info("Ignoring {} since the deadline has passed ({} < {})",
this, Instant.ofEpochMilli(deadline), Instant.ofEpochMilli(now));
timeoutCounter.incrementAndGet();
timeoutCounter.get(task.getClass()).incrementAndGet();
return;
}

Expand All @@ -364,18 +376,18 @@ public void run() {
task.runTask();
if (task.getStatus() == Status.FAILED) {
LOG.warn("Failed {}", this);
failureCounter.incrementAndGet();
failureCounter.get(task.getClass()).incrementAndGet();
} else if (task.getStatus() == Status.DONE) {
LOG.info("Successful {}", this);
successCounter.incrementAndGet();
successCounter.get(task.getClass()).incrementAndGet();
} else if (task.getStatus() == Status.SKIPPED) {
LOG.info("Skipped {}", this);
skippedCounter.incrementAndGet();
skippedCounter.get(task.getClass()).incrementAndGet();
}
} catch (Exception e) {
task.setStatus(Status.FAILED);
LOG.warn("Failed {}", this, e);
failureCounter.incrementAndGet();
failureCounter.get(task.getClass()).incrementAndGet();
} finally {
inFlight.remove(task);
decrementTaskCounter(task);
Expand Down Expand Up @@ -419,7 +431,12 @@ public boolean equals(Object o) {
}

public long getReplicationRequestCount() {
return requestCounter.get();
return getCount(requestCounter);
}

public long getReplicationRequestCount(Class<?> cls) {
AtomicLong counter = requestCounter.get(cls);
return counter != null ? counter.get() : 0;
}

public long getQueueSize() {
Expand All @@ -438,20 +455,51 @@ public long getMaxReplicationStreams() {
}
}

private long getCount(Map<Class<?>, AtomicLong> counter) {
if (counter.isEmpty()) {
return 0;
}
AtomicLong total = new AtomicLong(0);

@sodonnel sodonnel Sep 9, 2024

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.

This does not need to be an AtomicLong - a simple Long is fine. This is a local variable and we return only its value, so it does not need to be thread safe.

Not particularly important, but you could also remove the if (counter.isEmpty()) section:

long counter = 0;
counter.forEach((key, value) -> total += value.get();
return counter;

@jianghuazhu jianghuazhu Sep 9, 2024

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 @sodonnel for your comment and review.
I have updated it, can you review it again?
Thanks.

counter.forEach((key, value) -> {
total.set(total.get() + value.get());
});

@errose28 errose28 Sep 6, 2024

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.

Why are we using atomic long instead of primitive long for the local variable here?

Suggested change
counter.forEach((key, value) -> {
total.set(total.get() + value.get());
});
counter.forEach((key, value) -> total.addAndGet(value.get()));

@jianghuazhu jianghuazhu Sep 7, 2024

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 .
I have updated it.

return total.get();
}

public long getReplicationSuccessCount() {
return successCounter.get();
return getCount(successCounter);
}

public long getReplicationSuccessCount(Class<?> cls) {
AtomicLong counter = successCounter.get(cls);
return counter != null ? counter.get() : 0;
}

public long getReplicationFailureCount() {
return failureCounter.get();
return getCount(failureCounter);
}

public long getReplicationFailureCount(Class<?> cls) {
AtomicLong counter = failureCounter.get(cls);
return counter != null ? counter.get() : 0;
}

public long getReplicationTimeoutCount() {
return timeoutCounter.get();
return getCount(timeoutCounter);
}

public long getReplicationTimeoutCount(Class<?> cls) {
AtomicLong counter = timeoutCounter.get(cls);
return counter != null ? counter.get() : 0;
}

public long getReplicationSkippedCount() {
return skippedCounter.get();
return getCount(skippedCounter);
}

public long getReplicationSkippedCount(Class<?> cls) {
AtomicLong counter = skippedCounter.get(cls);
return counter != null ? counter.get() : 0;
}

}
42 changes: 41 additions & 1 deletion ...main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.metrics2.lib.Interns;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinatorTask;

import java.util.Map;

Expand Down Expand Up @@ -71,12 +72,51 @@ public void getMetrics(MetricsCollector collector, boolean all) {
.addGauge(Interns.info("numRequestedReplications",
"Number of requested replications"),
supervisor.getReplicationRequestCount())
.addGauge(Interns.info("numRequestedECReconstructions",

@sodonnel sodonnel Sep 5, 2024

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.

If a new task subclass was added in the future (I think reconciliation may be one), then the earlier code in this PR that populates the HashMaps will automatically start gathering metrics for the new class. But the code here will never present them.

Could we add all these gauges by iterating each map, so that if a new task appears in the future, it will automatically get added?

You may need to add some abstract method to AbstractTask to be able to get a meaningful name for each of the tasks that each sub class must implement.

Then instead of requestCounter.put(task.getClass(), new AtomicLong(0)); have:

requestCounter.put(task.getMetricName, new AtomicLong(0));

The metric names may be something like:

ECReconstructions
ContainerReplications

etc.

There may be a better way of doing this too - that is the first thought I came up with.

@jianghuazhu jianghuazhu Sep 6, 2024

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 @sodonnel for the comment and review.
I think this is a great idea, I will update it later.

"Number of requested EC reconstructions"),
supervisor.getReplicationRequestCount(ECReconstructionCoordinatorTask.class))
.addGauge(Interns.info("numRequestedContainerReplications",
"Number of requested container replications"),
supervisor.getReplicationRequestCount(ReplicationTask.class))
.addGauge(Interns.info("numSuccessReplications",
"Number of successful replications"),
supervisor.getReplicationSuccessCount())
.addGauge(Interns.info("numSuccessECReconstructions",
"Number of successful EC reconstructions"),
supervisor.getReplicationSuccessCount(ECReconstructionCoordinatorTask.class))
.addGauge(Interns.info("numSuccessContainerReplications",
"Number of successful container replications"),
supervisor.getReplicationSuccessCount(ReplicationTask.class))
.addGauge(Interns.info("numFailureReplications",
"Number of failure replications"),
supervisor.getReplicationFailureCount())
.addGauge(Interns.info("numFailureECReconstructions",
"Number of failure EC reconstructions"),
supervisor.getReplicationFailureCount(ECReconstructionCoordinatorTask.class))
.addGauge(Interns.info("numFailureContainerReplications",
"Number of failure container replications"),
supervisor.getReplicationFailureCount(ReplicationTask.class))
.addGauge(Interns.info("numTimeoutReplications",
"Number of replication requests timed out before being processed"),
supervisor.getReplicationTimeoutCount())
.addGauge(Interns.info("numTimeoutECReconstructions",
"Number of EC reconstructions timed out before being processed"),
supervisor.getReplicationTimeoutCount(ECReconstructionCoordinatorTask.class))
.addGauge(Interns.info("numTimeoutContainerReplications",
"Number of container replications timed out before being processed"),
supervisor.getReplicationTimeoutCount(ReplicationTask.class))
.addGauge(Interns.info("numSkippedReplications",
"Number of replication requests skipped as the container is "
+ "already present"), supervisor.getReplicationSkippedCount())
+ "already present"),
supervisor.getReplicationSkippedCount())
.addGauge(Interns.info("numSkippedECReconstructions",
"Number of EC reconstructions skipped as the container is "
+ "already present"),
supervisor.getReplicationSkippedCount(ECReconstructionCoordinatorTask.class))
.addGauge(Interns.info("numSkippedContainerReplications",
"Number of container replications skipped as the container is "
+ "already present"),
supervisor.getReplicationSkippedCount(ReplicationTask.class))
.addGauge(Interns.info("maxReplicationStreams", "Maximum number of "
+ "concurrent replication tasks which can run simultaneously"),
supervisor.getMaxReplicationStreams());
Expand Down
Loading