Skip to content
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

Enhance SegmentStatusChecker to honor no-queryable servers and instance assignment config #14536

Merged
merged 12 commits into from
Jan 10, 2025
Merged

Enhance SegmentStatusChecker to honor no-queryable servers and instance assignment config #14536

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
87308b4
Fix PERCENT_OF_REPLICAS metric
mqliang Nov 25, 2024
1f10cbf
Treat segments on not-queryable servers as down
mqliang Nov 26, 2024
28df5a8
fix lint
mqliang Nov 26, 2024
d77cc4e
fix typo in comments
mqliang Nov 26, 2024
bf098d3
pre build a queryable server set to reduce ZK acess
mqliang Nov 26, 2024
d5d5973
check if helix enabled
mqliang Nov 26, 2024
7ef229c
refactor code by adding ServerInfoCache
mqliang Nov 26, 2024
eec01cc
renaming
mqliang Nov 26, 2024
57e8bdc
fix integraiton test failure
mqliang Nov 27, 2024
bbba882
address comments
mqliang Dec 4, 2024
d49e8ab
address comments
mqliang Jan 7, 2025
9007e0a
don't explicitly enable an instance in testing
mqliang Jan 10, 2025
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
85 changes: 56 additions & 29 deletions pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
Expand Down Expand Up @@ -269,10 +270,12 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon

ExternalView externalView = _pinotHelixResourceManager.getTableExternalView(tableNameWithType);

// Maximum number of replicas in ideal state
int maxISReplicas = Integer.MIN_VALUE;
// Minimum number of replicas in external view
int minEVReplicas = Integer.MAX_VALUE;
// Maximum number of replicas that is up (ONLINE/CONSUMING) in ideal state
int maxISReplicasUp = Integer.MIN_VALUE;
// Minimum number of replicas that is up (ONLINE/CONSUMING) in external view
int minEVReplicasUp = Integer.MAX_VALUE;
// Minimum percentage of replicas that is up (ONLINE/CONSUMING) in external view
int minEVReplicasUpPercent = 100;
// Total compressed segment size in deep store
long tableCompressedSize = 0;
// Segments without ZK metadata
Expand All @@ -286,18 +289,19 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon
List<String> segmentsInvalidStartTime = new ArrayList<>();
List<String> segmentsInvalidEndTime = new ArrayList<>();
for (String segment : segments) {
int numISReplicas = 0;
// Number of replicas in ideal state that is in ONLINE/CONSUMING state
int numISReplicasUp = 0;
for (Map.Entry<String, String> entry : idealState.getInstanceStateMap(segment).entrySet()) {
String state = entry.getValue();
if (state.equals(SegmentStateModel.ONLINE) || state.equals(SegmentStateModel.CONSUMING)) {
numISReplicas++;
numISReplicasUp++;
}
}
// Skip segments not ONLINE/CONSUMING in ideal state
if (numISReplicas == 0) {
// Skip segments with no ONLINE/CONSUMING in ideal state
if (numISReplicasUp == 0) {
continue;
}
maxISReplicas = Math.max(maxISReplicas, numISReplicas);
maxISReplicasUp = Math.max(maxISReplicasUp, numISReplicasUp);

SegmentZKMetadata segmentZKMetadata = _pinotHelixResourceManager.getSegmentZKMetadata(tableNameWithType, segment);
// Skip the segment when it doesn't have ZK metadata. Most likely the segment is just deleted.
Expand Down Expand Up @@ -330,46 +334,46 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon
}
}

int numEVReplicas = 0;
int numEVReplicasUp = 0;
if (externalView != null) {
Map<String, String> stateMap = externalView.getStateMap(segment);
if (stateMap != null) {
for (Map.Entry<String, String> entry : stateMap.entrySet()) {
String serverInstanceId = entry.getKey();
String state = entry.getValue();
if (state.equals(SegmentStateModel.ONLINE) || state.equals(SegmentStateModel.CONSUMING)) {
numEVReplicas++;
if (isServerQueryable(serverInstanceId, context)
&& (state.equals(SegmentStateModel.ONLINE) || state.equals(SegmentStateModel.CONSUMING))) {
numEVReplicasUp++;
}
if (state.equals(SegmentStateModel.ERROR)) {
errorSegments.add(Pair.of(segment, entry.getKey()));
}
}
}
}
if (numEVReplicas == 0) {
if (numEVReplicasUp == 0) {
offlineSegments.add(segment);
} else if (numEVReplicas < numISReplicas) {
} else if (numEVReplicasUp < numISReplicasUp) {
partialOnlineSegments.add(segment);
} else {
// Do not allow nReplicasEV to be larger than nReplicasIS
numEVReplicas = numISReplicas;
}
minEVReplicas = Math.min(minEVReplicas, numEVReplicas);

minEVReplicasUp = Math.min(minEVReplicasUp, numEVReplicasUp);
// Total number of replicas in ideal state (including ERROR/OFFLINE states)
int numISReplicasTotal = idealState.getInstanceStateMap(segment).entrySet().size();
minEVReplicasUpPercent = Math.min(minEVReplicasUpPercent, numEVReplicasUp * 100 / numISReplicasTotal);
}

if (maxISReplicas == Integer.MIN_VALUE) {
if (maxISReplicasUp == Integer.MIN_VALUE) {
try {
maxISReplicas = Math.max(Integer.parseInt(idealState.getReplicas()), 1);
maxISReplicasUp = Math.max(Integer.parseInt(idealState.getReplicas()), 1);
} catch (NumberFormatException e) {
maxISReplicas = 1;
maxISReplicasUp = 1;
}
}
// Do not allow minEVReplicas to be larger than maxISReplicas
minEVReplicas = Math.min(minEVReplicas, maxISReplicas);

if (minEVReplicas < maxISReplicas) {
LOGGER.warn("Table {} has at least one segment running with only {} replicas, below replication threshold :{}",
tableNameWithType, minEVReplicas, maxISReplicas);
}
Copy link
Contributor Author

@mqliang mqliang Nov 26, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

@sajjad-moradi This log need to be deleted, as

  1. This log will always (and falsely) being printed out when online and consuming segments have different replica group.
  2. L388-L392 already log the partial online segments. This log is redundant

Copy link
Contributor

@jasperjiaguo jasperjiaguo Nov 26, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Can we add some new logging instead? Meanwhile, it would be great if we have a list of segments having min replicas and print them out as well? I feel sometimes the debuggability of alerts on this metric is not so great. nvm [L388-L392] should be good

// Do not allow minEVReplicasUp to be larger than maxISReplicasUp
minEVReplicasUp = Math.min(minEVReplicasUp, maxISReplicasUp);

int numSegmentsWithoutZKMetadata = segmentsWithoutZKMetadata.size();
if (numSegmentsWithoutZKMetadata > 0) {
LOGGER.warn("Table {} has {} segments without ZK metadata: {}", tableNameWithType, numSegmentsWithoutZKMetadata,
Expand Down Expand Up @@ -402,9 +406,9 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon
}

// Synchronization provided by Controller Gauge to make sure that only one thread updates the gauge
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.NUMBER_OF_REPLICAS, minEVReplicas);
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.NUMBER_OF_REPLICAS, minEVReplicasUp);
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.PERCENT_OF_REPLICAS,
minEVReplicas * 100L / maxISReplicas);
minEVReplicasUpPercent);
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.SEGMENTS_IN_ERROR_STATE,
numErrorSegments);
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.PERCENT_SEGMENTS_AVAILABLE,
Expand All @@ -426,6 +430,28 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon
}
}

private boolean isServerQueryable(String instanceId, Context context) {
if (context._serverQueryableMap.containsKey(instanceId)) {
return context._serverQueryableMap.get(instanceId);
}
InstanceConfig instanceConfig = _pinotHelixResourceManager.getHelixInstanceConfig(instanceId);
// Instance not found in Helix cluster or not Server we assume it is down
if (instanceConfig == null || !instanceConfig.getInstanceName().startsWith("Server_")) {
LOGGER.warn("Instance {} not found in Helix cluster, or not a server instance", instanceId);
LOGGER.warn("will assume instance is down!");
context._serverQueryableMap.put(instanceId, false);
return false;
}

ZNRecord record = instanceConfig.getRecord();
boolean queriesDisabled = Boolean.valueOf(record.getSimpleField("queriesDisabled"));
boolean shutdownInProgress = Boolean.valueOf(record.getSimpleField("shutdownInProgress"));
boolean queryable = !queriesDisabled && !shutdownInProgress;

context._serverQueryableMap.put(instanceId, queryable);
return queryable;
}

private static String logSegments(List<?> segments) {
if (segments.size() <= MAX_SEGMENTS_TO_LOG) {
return segments.toString();
Expand Down Expand Up @@ -473,5 +499,6 @@ public static final class Context {
private final Set<String> _processedTables = new HashSet<>();
private final Set<String> _disabledTables = new HashSet<>();
private final Set<String> _pausedTables = new HashSet<>();
private final Map<String, Boolean> _serverQueryableMap = new HashMap<>();
}
}
Loading
Loading