Skip to content

perf(plugin-base): batch namespace snapshot cache updates and invalidation (#7645) #7672

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

Open
dmsolr wants to merge 2 commits into
base: master
Choose a base branch
from
Open

perf(plugin-base): batch namespace snapshot cache updates and invalidation (#7645) #7672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c9fda96
perf(plugin-base): batch namespace snapshot cache updates and invalid…
Oct 16, 2026
b0074ef
fix(plugin-base): harden namespace snapshot against dangling and malf…
Oct 18, 2026
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
64 changes: 64 additions & 0 deletions ...n/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/cache/BaseDataCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -179,6 +181,44 @@ public void cleanSelectorData() {
public void cleanSelectorDataSelf(final List<SelectorData> selectorDataList) {
selectorDataList.forEach(this::removeSelectData);
}

/**
* Remove a batch of selectors with a single merge per affected plugin bucket.
* Selectors without a plugin name are matched against every bucket, mirroring
* {@link #removeSelectData(SelectorData)}.
*
* @param selectorDataList the selectors to remove
*/
void removeSelectDataBatch(final List<SelectorData> selectorDataList) {
if (selectorDataList.isEmpty()) {
return;
}
final Map<String, Set<String>> idsByPlugin = new HashMap<>();
final Set<String> danglingIds = new HashSet<>();
for (SelectorData data : selectorDataList) {
if (Objects.isNull(data) || Objects.isNull(data.getId())) {
continue;
}
if (StringUtils.isBlank(data.getPluginName())) {
danglingIds.add(data.getId());
} else {
idsByPlugin.computeIfAbsent(data.getPluginName(), key -> new HashSet<>()).add(data.getId());
}
}
if (!danglingIds.isEmpty()) {
selectorMap.keySet().forEach(pluginName -> removeSelectDataByIds(pluginName, danglingIds));
}
idsByPlugin.forEach(this::removeSelectDataByIds);
}

private void removeSelectDataByIds(final String pluginName, final Set<String> selectorIds) {
selectorMap.computeIfPresent(pluginName, (key, value) -> {
final List<SelectorData> result = value.stream()
.filter(selector -> !selectorIds.contains(selector.getId()))
.collect(Collectors.toList());
return result.isEmpty() ? null : List.copyOf(result);
});
}

/**
* Obtain selector data list list.
Expand Down Expand Up @@ -239,6 +279,30 @@ public void cleanRuleData() {
public void cleanRuleDataSelf(final List<RuleData> ruleDataList) {
ruleDataList.forEach(this::removeRuleData);
}

/**
* Remove a batch of rules with a single merge per affected selector bucket.
*
* @param ruleDataList the rules to remove
*/
void removeRuleDataBatch(final List<RuleData> ruleDataList) {
if (ruleDataList.isEmpty()) {
return;
}
final Map<String, Set<String>> idsBySelector = new HashMap<>();
for (RuleData data : ruleDataList) {
if (Objects.isNull(data) || Objects.isNull(data.getId()) || Objects.isNull(data.getSelectorId())) {
continue;
}
idsBySelector.computeIfAbsent(data.getSelectorId(), key -> new HashSet<>()).add(data.getId());
}
idsBySelector.forEach((selectorId, ruleIds) -> ruleMap.computeIfPresent(selectorId, (key, value) -> {
final List<RuleData> result = value.stream()
.filter(rule -> !ruleIds.contains(rule.getId()))
.collect(Collectors.toList());
return result.isEmpty() ? null : List.copyOf(result);
}));
}

/**
* Obtain rule data list list.
Expand Down
203 changes: 195 additions & 8 deletions ...in-base/src/main/java/org/apache/shenyu/plugin/base/cache/CommonPluginDataSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,16 @@
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.lang.NonNull;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
* The type Common plugin data subscriber.
Expand Down Expand Up @@ -160,26 +164,100 @@ public void applyPluginDataSnapshot(final String namespaceId, final List<PluginD

@Override
public void applySelectorDataSnapshot(final String namespaceId, final List<SelectorData> dataList) {
Set<String> selectorIds = dataList.stream().map(SelectorData::getId).collect(Collectors.toSet());
List<SelectorData> stale = BaseDataCache.getInstance().getSelectorMap().values().stream()
final BaseDataCache cache = BaseDataCache.getInstance();
// A row without a plugin name yields a null grouping key in refreshSelectorData, whose compute on a
// concurrent map throws and would abort the whole snapshot (and, over a websocket, loop resyncs).
// The previous per-row subscribe path isolated such a row; skip it up front so valid rows still converge.
final List<SelectorData> incoming = new ArrayList<>(dataList.size());
for (SelectorData data : dataList) {
if (Objects.nonNull(data) && Objects.nonNull(data.getPluginName())) {
incoming.add(data);
} else {
LOG.warn("skip malformed selector row without plugin name in snapshot for namespace [{}]", namespaceId);
}
}
final Set<String> selectorIds = incoming.stream().map(SelectorData::getId).collect(Collectors.toSet());
final List<SelectorData> stale = cache.getSelectorMap().values().stream()
.flatMap(List::stream)
.filter(data -> namespaceId.equals(data.getNamespaceId()))
.filter(data -> !selectorIds.contains(data.getId()))
.collect(Collectors.toList());
dataList.forEach(this::onSelectorSubscribe);
stale.forEach(this::unSelectorSubscribe);
// Merge and sort once per affected plugin bucket instead of per selector row.
if (!incoming.isEmpty()) {
cache.refreshSelectorData(incoming);
}
if (!stale.isEmpty()) {
cache.removeSelectDataBatch(stale);
}
try {
// Preserve per-item plugin handler callbacks and the initial-sync failure semantics
// that the previous per-row subscribe path provided.
incoming.forEach(data -> runGuarded(data, () -> {
if (StringUtils.isNotBlank(data.getPluginName())) {
Optional.ofNullable(handlerMap.get(data.getPluginName()))
.ifPresent(handler -> handler.handlerSelector(data));
}
}));
stale.forEach(data -> runGuarded(data, () -> {
// the concurrent handler map rejects null keys, and a dangling selector carries no plugin name
if (StringUtils.isNotBlank(data.getPluginName())) {
Optional.ofNullable(handlerMap.get(data.getPluginName()))
.ifPresent(handler -> handler.removeSelector(data));
}
}));
} finally {
// Invalidate even if a handler aborted during initial sync: the stale rows are already gone from
// BaseDataCache above, so a retry cannot rediscover them and their match cache would stay stale.
invalidateSelectorMatchCacheByPlugins(collectAffectedPlugins(incoming, stale, SelectorData::getPluginName));
invalidateDanglingSelectorMatchCache(incoming, stale);
}
}

@Override
public void applyRuleDataSnapshot(final String namespaceId, final List<RuleData> dataList) {
Set<String> ruleIds = dataList.stream().map(RuleData::getId).collect(Collectors.toSet());
List<RuleData> stale = BaseDataCache.getInstance().getRuleMap().values().stream()
final BaseDataCache cache = BaseDataCache.getInstance();
// A row without a selector id yields a null grouping key in refreshRuleData, whose compute on a
// concurrent map throws and would abort the whole snapshot; skip it up front so valid rows still converge.
final List<RuleData> incoming = new ArrayList<>(dataList.size());
for (RuleData data : dataList) {
if (Objects.nonNull(data) && Objects.nonNull(data.getSelectorId())) {
incoming.add(data);
} else {
LOG.warn("skip malformed rule row without selector id in snapshot for namespace [{}]", namespaceId);
}
}
final Set<String> ruleIds = incoming.stream().map(RuleData::getId).collect(Collectors.toSet());
final List<RuleData> stale = cache.getRuleMap().values().stream()
.flatMap(List::stream)
.filter(data -> namespaceId.equals(data.getNamespaceId()))
.filter(data -> !ruleIds.contains(data.getId()))
.collect(Collectors.toList());
dataList.forEach(this::onRuleSubscribe);
stale.forEach(this::unRuleSubscribe);
// Merge and sort once per affected selector bucket instead of per rule row.
if (!incoming.isEmpty()) {
cache.refreshRuleData(incoming);
}
if (!stale.isEmpty()) {
cache.removeRuleDataBatch(stale);
}
try {
incoming.forEach(data -> runGuarded(data, () -> {
if (StringUtils.isNotBlank(data.getPluginName())) {
Optional.ofNullable(handlerMap.get(data.getPluginName()))
.ifPresent(handler -> handler.handlerRule(data));
}
}));
stale.forEach(data -> runGuarded(data, () -> {
if (StringUtils.isNotBlank(data.getPluginName())) {
Optional.ofNullable(handlerMap.get(data.getPluginName()))
.ifPresent(handler -> handler.removeRule(data));
}
}));
} finally {
// Invalidate even if a handler aborted during initial sync, so a retry is not left with stale
// match-cache entries for rows already removed from BaseDataCache above.
invalidateRuleMatchCacheByPlugins(collectAffectedPlugins(incoming, stale, RuleData::getPluginName));
invalidateDanglingRuleMatchCache(incoming, stale);
}
}

@Override
Expand Down Expand Up @@ -402,6 +480,115 @@ private void invalidateRuleMatchCache(final RuleData ruleData) {
}
}

/**
* Run a per-row callback for a snapshot apply, preserving the exception semantics of
* {@link #subscribeDataHandler}: during initial sync failures abort the snapshot, otherwise
* they are logged so subsequent rows still converge.
*/
private <T> void runGuarded(final T data, final Runnable action) {
try {
action.run();
} catch (Exception e) {
if (InitialSyncApplication.isActive()) {
throw new IllegalStateException("Initial configuration application failed", e);
}
LOG.error("snapshot handler error, classData: {}", JsonUtils.toJson(data), e);
}
}

private <T> Set<String> collectAffectedPlugins(final List<T> incoming, final List<T> stale,
final Function<T, String> pluginNameOf) {
final Set<String> plugins = new HashSet<>();
incoming.forEach(data -> {
if (Objects.isNull(data)) {
return;
}
final String name = pluginNameOf.apply(data);
if (StringUtils.isNotBlank(name)) {
plugins.add(name);
}
});
stale.forEach(data -> {
if (Objects.isNull(data)) {
return;
}
final String name = pluginNameOf.apply(data);
if (StringUtils.isNotBlank(name)) {
plugins.add(name);
}
});
return plugins;
}

/**
* Mirror master's id-based cross-bucket sweep for dangling selectors that carry no plugin name:
* {@link #collectAffectedPlugins} drops blank names, so without this a deleted dangling selector would
* leave its cached matches (which may live under any plugin bucket) behind. Delegates to the blank-name
* branch of {@link MatchDataCache#removeSelectorData(String, String)} and
* {@link MatchDataCache#removeRuleDataBySelector(String, String)}.
*/
private void invalidateDanglingSelectorMatchCache(final List<SelectorData> incoming, final List<SelectorData> stale) {
if (!selectorMatchConfig.getCache().getEnabled() && !ruleMatchCacheConfig.getCache().getEnabled()) {
return;
}
final MatchDataCache matchDataCache = MatchDataCache.getInstance();
Stream.concat(incoming.stream(), stale.stream())
.filter(Objects::nonNull)
.filter(data -> StringUtils.isBlank(data.getPluginName()))
.map(SelectorData::getId)
.filter(Objects::nonNull)
.forEach(id -> {
if (selectorMatchConfig.getCache().getEnabled()) {
matchDataCache.removeSelectorData(null, id);
}
if (ruleMatchCacheConfig.getCache().getEnabled()) {
matchDataCache.removeRuleDataBySelector(null, id);
}
});
}

/**
* Rule counterpart of {@link #invalidateDanglingSelectorMatchCache}: a stale rule keyed by selectorId can
* carry a blank plugin name, so its cached matches are swept across every bucket by rule id via the
* blank-name branch of {@link MatchDataCache#removeRuleData(String, String)}.
*/
private void invalidateDanglingRuleMatchCache(final List<RuleData> incoming, final List<RuleData> stale) {
if (!ruleMatchCacheConfig.getCache().getEnabled()) {
return;
}
final MatchDataCache matchDataCache = MatchDataCache.getInstance();
Stream.concat(incoming.stream(), stale.stream())
.filter(Objects::nonNull)
.filter(data -> StringUtils.isBlank(data.getPluginName()))
.map(RuleData::getId)
.filter(Objects::nonNull)
.forEach(id -> matchDataCache.removeRuleData(null, id));
}

private void invalidateSelectorMatchCacheByPlugins(final Set<String> pluginNames) {
if (pluginNames.isEmpty()) {
return;
}
final MatchDataCache matchDataCache = MatchDataCache.getInstance();
for (String pluginName : pluginNames) {
if (selectorMatchConfig.getCache().getEnabled()) {
matchDataCache.removeSelectorData(pluginName);
}
if (ruleMatchCacheConfig.getCache().getEnabled()) {
// selector membership changed, so any cached rule match under this plugin may be stale
matchDataCache.removeRuleData(pluginName);
}
}
}

private void invalidateRuleMatchCacheByPlugins(final Set<String> pluginNames) {
if (pluginNames.isEmpty() || !ruleMatchCacheConfig.getCache().getEnabled()) {
return;
}
final MatchDataCache matchDataCache = MatchDataCache.getInstance();
pluginNames.forEach(matchDataCache::removeRuleData);
}

@Override
public void onPluginRefresh(final List<PluginData> dataList) {
if (CollectionUtils.isEmpty(dataList)) {
Expand Down
Loading
Loading