-
Notifications
You must be signed in to change notification settings - Fork 135
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.
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
[Log Improvement] Output the registering/lost/exclude nodes in log #148
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -30,6 +30,7 @@ | |
import java.util.concurrent.ScheduledExecutorService; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.atomic.AtomicLong; | ||
import java.util.stream.Collectors; | ||
|
||
import com.google.common.annotations.VisibleForTesting; | ||
import com.google.common.collect.Lists; | ||
|
@@ -60,12 +61,17 @@ public class SimpleClusterManager implements ClusterManager { | |
private ScheduledExecutorService checkNodesExecutorService; | ||
private FileSystem hadoopFileSystem; | ||
|
||
private long outputAliveServerCount = 0; | ||
private final long periodicOutputIntervalTimes; | ||
|
||
public SimpleClusterManager(CoordinatorConf conf, Configuration hadoopConf) throws IOException { | ||
this.shuffleNodesMax = conf.getInteger(CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX); | ||
this.heartbeatTimeout = conf.getLong(CoordinatorConf.COORDINATOR_HEARTBEAT_TIMEOUT); | ||
// the thread for checking if shuffle server report heartbeat in time | ||
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor( | ||
ThreadUtils.getThreadFactory("SimpleClusterManager-%d")); | ||
|
||
periodicOutputIntervalTimes = conf.get(CoordinatorConf.COORDINATOR_NODES_PERIODIC_OUTPUT_INTERVAL_TIMES); | ||
scheduledExecutorService.scheduleAtFixedRate( | ||
() -> nodesCheck(), heartbeatTimeout / 3, | ||
heartbeatTimeout / 3, TimeUnit.MILLISECONDS); | ||
|
@@ -99,6 +105,13 @@ void nodesCheck() { | |
} | ||
} | ||
} | ||
if (!deleteIds.isEmpty() || outputAliveServerCount % periodicOutputIntervalTimes == 0) { | ||
LOG.info("Alive servers number: {}, ids: {}", | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do we need this? We have metrics tell us how many there are alive servers. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Number is not enough. I want to know which servers are lost from the coordinator side. And the number shown in log will be easy to find. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ok, I got it. Could There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Updated.
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do you judge There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. when deleteIds is not empty, it means the server list changed, we should output the latest alive servers. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok. |
||
servers.size(), | ||
servers.keySet().stream().collect(Collectors.toList()) | ||
); | ||
} | ||
outputAliveServerCount++; | ||
|
||
CoordinatorMetrics.gaugeTotalServerNum.set(servers.size()); | ||
} catch (Exception e) { | ||
|
@@ -107,6 +120,7 @@ void nodesCheck() { | |
} | ||
|
||
private void updateExcludeNodes(String path) { | ||
int originalExcludeNodesNumber = excludeNodes.size(); | ||
try { | ||
Path hadoopPath = new Path(path); | ||
FileStatus fileStatus = hadoopFileSystem.getFileStatus(hadoopPath); | ||
|
@@ -118,12 +132,16 @@ private void updateExcludeNodes(String path) { | |
} else { | ||
excludeNodes = Sets.newConcurrentHashSet(); | ||
} | ||
CoordinatorMetrics.gaugeExcludeServerNum.set(excludeNodes.size()); | ||
} catch (FileNotFoundException fileNotFoundException) { | ||
excludeNodes = Sets.newConcurrentHashSet(); | ||
} catch (Exception e) { | ||
LOG.warn("Error when updating exclude nodes, the exclude nodes file path: " + path, e); | ||
} | ||
int newlyExcludeNodesNumber = excludeNodes.size(); | ||
if (newlyExcludeNodesNumber != originalExcludeNodesNumber) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ditto. |
||
LOG.info("Exclude nodes number: {}, nodes list: {}", newlyExcludeNodesNumber, excludeNodes); | ||
} | ||
CoordinatorMetrics.gaugeExcludeServerNum.set(excludeNodes.size()); | ||
} | ||
|
||
private void parseExcludeNodesFile(DataInputStream fsDataInputStream) throws IOException { | ||
|
@@ -143,6 +161,9 @@ private void parseExcludeNodesFile(DataInputStream fsDataInputStream) throws IOE | |
|
||
@Override | ||
public void add(ServerNode node) { | ||
if (!servers.containsKey(node.getId())) { | ||
LOG.info("Newly registering node: {}", node.getId()); | ||
} | ||
servers.put(node.getId(), node); | ||
Set<String> tags = node.getTags(); | ||
// remove node with all tags to deal with the situation of tag change | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we add
checkValue
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes. I will add check.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we add some document?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's OK. But currently the file of doc/coordinator_guide.md is empty and the README about coordinator conf is missing lots of params.
So do i add this conf into README?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you migrate the cooridinator's document from README to doc/coordinator_guide.md and add the document of this config option in the doc/coordinator_guide.md in a new pr?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Got it. I'm glad to do this.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
OK, let's merge this pr first.