Skip to content

HADOOP-17995. Stale record should be remove when DataNodePeerMetrics#dumpSendPacketDownstreamAvgInfoAsJson #3630

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
ferhui merged 3 commits into from
Nov 17, 2021
Merged

HADOOP-17995. Stale record should be remove when DataNodePeerMetrics#dumpSendPacketDownstreamAvgInfoAsJson #3630

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
97e4d50
HADOOP-17995. Stale record should be remove when DataNodePeerMetrics#…
Nov 8, 2021
aff0c48
HADOOP-17995. Fix
Nov 9, 2021
a9fcd6a
HADOOP-17995. Fix
Nov 12, 2021
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
9 changes: 6 additions & 3 deletions ...ct/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRollingAverages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ synchronized void replaceScheduledTask(int windows, long interval,
}

@Override
public void snapshot(MetricsRecordBuilder builder, boolean all) {
public synchronized void snapshot(MetricsRecordBuilder builder, boolean all) {

@ferhui ferhui Nov 10, 2021

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.

Do we need synchronized here? Its super class is not.

@tomscut tomscut Nov 10, 2021

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.

I have the same question.

@haiyang1987 haiyang1987 Nov 10, 2021

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.

@ferhui @tomscut Thanks your comment.

SpotBugs module:hadoop-common-project/hadoop-common
  Inconsistent synchronization of org.apache.hadoop.metrics2.lib.MutableRollingAverages.recordValidityMs; locked 66% of time Unsynchronized access at MutableRollingAverages.java:66% of time Unsynchronized access at MutableRollingAverages.java:[line 182]

So consider add synchronized.
Reference code MutableRatesWithAggregation#snapshot, the method to add the synchronized

public class MutableRatesWithAggregation extends MutableMetric {
@OverRide
public synchronized void snapshot(MetricsRecordBuilder rb, boolean all) {
Iterator<WeakReference<ConcurrentMap<String, ThreadSafeSampleStat>>> iter =
weakReferenceQueue.iterator();
...
}
}

@hust-hhb hust-hhb Nov 11, 2021

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.

I'm not sure whether the synchronized is needed in here, the rest of part look good to me.

@haiyang1987 haiyang1987 Nov 12, 2021

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.

@ferhui @tomscut @AlphaGouGe Thanks your comment.
this logic is only called when DataNode JMX get SendPacketDownstreamAvgInfo Metrics,
so there is no concurrency security issue at the moment, no need to add synchronized method.
Update PR.

if (all || changed()) {
for (final Entry<String, LinkedBlockingDeque<SumAndCount>> entry
: averages.entrySet()) {
Expand All @@ -179,8 +179,11 @@ public void snapshot(MetricsRecordBuilder builder, boolean all) {
long totalCount = 0;

for (final SumAndCount sumAndCount : entry.getValue()) {
totalCount += sumAndCount.getCount();
totalSum += sumAndCount.getSum();
if (Time.monotonicNow() - sumAndCount.getSnapshotTimeStamp()
< recordValidityMs) {
totalCount += sumAndCount.getCount();
totalSum += sumAndCount.getSum();
}
}

if (totalCount != 0) {
Expand Down
10 changes: 10 additions & 0 deletions ...op-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodePeerMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,10 +122,16 @@ public void testRemoveStaleRecord() throws Exception {
GenericTestUtils.waitFor(
() -> rollingAverages.getStats(numSamples).size() > 0, 500, 5000);
assertEquals(3, rollingAverages.getStats(numSamples).size());
String json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
for (String peerAddr : peerAddrList) {
assertThat(json, containsString(peerAddr));
}
/* wait for stale report to be removed */
GenericTestUtils.waitFor(
() -> rollingAverages.getStats(numSamples).isEmpty(), 500, 10000);
assertEquals(0, rollingAverages.getStats(numSamples).size());
json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
assertEquals("{}", json);

/* dn can report peer metrics normally when it added back to cluster */
for (String peerAddr : peerAddrList) {
Expand All @@ -138,6 +144,10 @@ public void testRemoveStaleRecord() throws Exception {
GenericTestUtils.waitFor(
() -> rollingAverages.getStats(numSamples).size() > 0, 500, 10000);
assertEquals(3, rollingAverages.getStats(numSamples).size());
json = peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson();
for (String peerAddr : peerAddrList) {
assertThat(json, containsString(peerAddr));
}
}

/**
Expand Down