Skip to content

LAC: fallback to legacy group when pd not support keyspace #10929

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
yongman wants to merge 7 commits into
base: master
Choose a base branch
from
+366 −47
Open

LAC: fallback to legacy group when pd not support keyspace #10929

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
788a0c3
LAC: fallback to legacy group when pd not support keyspace
yongman Jun 25, 2026
04608f6
fix default keyspace
yongman Jun 29, 2026
ad9815b
simplify the legacy logic
yongman Jul 1, 2026
dd09b4e
format
yongman Jul 1, 2026
82147c7
polish
yongman Jul 1, 2026
add89e1
refactor for simplify
yongman Jul 3, 2026
2524798
format
yongman Jul 3, 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
180 changes: 154 additions & 26 deletions dbms/src/Flash/ResourceControl/LocalAdmissionController.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,9 @@ uint64_t ResourceGroup::getPriority(uint64_t max_ru_per_sec) const

std::optional<GACRequestInfo> ResourceGroup::buildRequestInfoIfNecessary(const SteadyClock::time_point & now)
{
if (!enable_gac)
return {};
Comment thread
yongman marked this conversation as resolved.
Outdated

std::lock_guard lock(mu);
if (!beginRequestWithoutLock(now))
{
Expand Down Expand Up @@ -292,6 +295,24 @@ LACRUConsumptionDeltaInfo ResourceGroup::updateRUConsumptionDeltaInfoWithoutLock
return info;
}

resource_manager::ResourceGroup LocalAdmissionController::buildReservedDefaultResourceGroup(const KeyspaceID & keyspace_id)
{
resource_manager::ResourceGroup group_pb;
group_pb.set_name("default");
group_pb.set_mode(resource_manager::GroupMode::RUMode);
group_pb.set_priority(ResourceGroup::UserMediumPriority);
group_pb.mutable_keyspace_id()->set_value(keyspace_id);
auto * settings = group_pb.mutable_r_u_settings()->mutable_r_u()->mutable_settings();
settings->set_fill_rate(RESERVED_DEFAULT_RESOURCE_GROUP_RU_PER_SEC);
settings->set_burst_limit(-1);
return group_pb;
}

void LocalAdmissionController::addReservedDefaultResourceGroup(const KeyspaceID & keyspace_id)
{
addResourceGroup(keyspace_id, buildReservedDefaultResourceGroup(keyspace_id), false);
}

void LocalAdmissionController::warmupResourceGroupInfoCache(const KeyspaceID & keyspace_id, const std::string & name)
{
if (unlikely(stopped))
Expand All @@ -300,9 +321,11 @@ void LocalAdmissionController::warmupResourceGroupInfoCache(const KeyspaceID & k
if (name.empty())
return;

ResourceGroupPtr group = findResourceGroup(keyspace_id, name);
if (group != nullptr)
return;
{
std::lock_guard lock(mu);
if (findResourceGroupWithCompatWithoutLock(keyspace_id, name).group != nullptr)
return;
}

resource_manager::GetResourceGroupRequest req;
req.mutable_keyspace_id()->set_value(keyspace_id);
Expand All @@ -323,12 +346,54 @@ void LocalAdmissionController::warmupResourceGroupInfoCache(const KeyspaceID & k
getCurrentExceptionMessage(false));
}

RUNTIME_CHECK_MSG(
!resp.has_error(),
"warmupResourceGroupInfoCache: {}(keyspace={}) failed: {}",
name,
keyspace_id,
resp.error().message());
if (resp.has_error())

@windtalker windtalker Jul 3, 2026

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 was wondering, isn't keyspace support a cluster-level configuration for PD? If a cluster's PD doesn't support keyspace, it likely won't support it for a long time. If that's the case, do we really need to handle the fallback at the RPC level?

@yongman yongman Jul 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Here it means the pd dose not support keyspace scope resource group (not the keyspace itself).

@JaySon-Huang JaySon-Huang Jul 4, 2026

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.

In the "starter" tier, currently the pd-cse is running with keyspace support but without keyspace scope resource group. And we don't actually know when pd-cse will be upgraded to pd. But we plan to upgrade the tiflash columnar binary on the "starter" tier to unify the codebase in late July, so tiflash need to adapt to the pd-cse.

{
resource_manager::GetResourceGroupRequest legacy_req;
legacy_req.set_resource_group_name(name);
resource_manager::GetResourceGroupResponse legacy_resp;
try
{
legacy_resp = cluster->pd_client->getResourceGroup(legacy_req);
}
catch (...)
{
LOG_WARNING(
log,
"warmupResourceGroupInfoCache failed to fetch legacy rg for keyspace={}, rg={}, err={}",
keyspace_id,
name,
getCurrentExceptionMessage(false));
}
if (!legacy_resp.has_error())
Comment thread
yongman marked this conversation as resolved.
Outdated
{
checkGACRespValid(legacy_resp.group());
const auto rg_keyspace_id
= legacy_resp.group().has_keyspace_id() ? legacy_resp.group().keyspace_id().value() : NullspaceID;
addResourceGroup(rg_keyspace_id, legacy_resp.group());
return;
}
if (isReservedDefaultResourceGroup(name))
{
{
std::lock_guard lock(mu);
if (findResourceGroupWithoutLock(NullspaceID, name) != nullptr)
return;
}
LOG_WARNING(
log,
"warmupResourceGroupInfoCache fallback to reserved default rg for keyspace={}, err={}",
keyspace_id,
resp.error().message());
addReservedDefaultResourceGroup(keyspace_id);
return;
}
RUNTIME_CHECK_MSG(
false,
"warmupResourceGroupInfoCache: {}(keyspace={}) failed: {}",
name,
keyspace_id,
resp.error().message());
}

checkGACRespValid(resp.group());

Expand Down Expand Up @@ -470,7 +535,8 @@ std::optional<resource_manager::TokenBucketsRequest> LocalAdmissionController::b
for (const auto & info : request_infos)
{
auto * group_request = gac_req.add_requests();
group_request->mutable_keyspace_id()->set_value(info.keyspace_id);
if (info.keyspace_id != NullspaceID)
group_request->mutable_keyspace_id()->set_value(info.keyspace_id);
group_request->set_resource_group_name(info.resource_group_name);
assert(info.acquire_tokens > 0.0 || info.ru_consumption_delta > 0.0 || is_final_report);
if (info.acquire_tokens > 0.0 || is_final_report)
Expand Down Expand Up @@ -535,7 +601,7 @@ static std::vector<std::pair<KeyspaceID, std::string>> extractGACReqNames(
std::vector<std::pair<KeyspaceID, std::string>> res;
res.reserve(gac_req.requests_size());
for (const auto & req : gac_req.requests())
res.push_back({req.keyspace_id().value(), req.resource_group_name()});
res.push_back({req.has_keyspace_id() ? req.keyspace_id().value() : NullspaceID, req.resource_group_name()});
return res;
}

Expand Down Expand Up @@ -567,7 +633,7 @@ void LocalAdmissionController::doRequestGAC()
const auto resp = cluster->pd_client->acquireTokenBuckets(req);
LOG_DEBUG(log, "request to GAC done, req: {}. resp: {}", req.ShortDebugString(), resp.ShortDebugString());

auto handled = handleTokenBucketsResp(resp);
auto handled = handleTokenBucketsResp(resp, req_rg_names);

std::vector<std::pair<KeyspaceID, std::string>> not_found;
// not_found includes resource group names that appears in gac_req but not found in resp.
Expand Down Expand Up @@ -607,7 +673,8 @@ void LocalAdmissionController::checkDegradeMode()
}

std::vector<std::pair<KeyspaceID, std::string>> LocalAdmissionController::handleTokenBucketsResp(
const resource_manager::TokenBucketsResponse & resp)
const resource_manager::TokenBucketsResponse & resp,
const std::vector<std::pair<KeyspaceID, std::string>> & req_rg_names)
{
if unlikely (resp.has_error())
{
Expand All @@ -617,6 +684,13 @@ std::vector<std::pair<KeyspaceID, std::string>> LocalAdmissionController::handle

std::vector<std::pair<KeyspaceID, std::string>> handled_resource_group_names;
handled_resource_group_names.reserve(resp.responses_size());

// Older PD may omit keyspace_id in TokenBucketResponse. Keep request order per resource-group name so
// duplicate names across keyspaces can still be matched to the corresponding requests in order.
std::unordered_map<std::string, std::vector<KeyspaceID>> pending_req_keyspaces_by_name;
for (const auto & [req_keyspace_id, req_name] : req_rg_names)
pending_req_keyspaces_by_name[req_name].push_back(req_keyspace_id);

if (resp.responses().empty())
{
LOG_ERROR(log, "got empty TokenBuckets resp from GAC, {}", resp.ShortDebugString());
Expand All @@ -633,19 +707,61 @@ std::vector<std::pair<KeyspaceID, std::string>> LocalAdmissionController::handle
}

const auto & name = one_resp.resource_group_name();
KeyspaceID keyspace_id = NullspaceID;
KeyspaceID resp_keyspace_id = NullspaceID;
if (one_resp.has_keyspace_id())
resp_keyspace_id = one_resp.keyspace_id().value();

std::optional<KeyspaceID> matched_req_keyspace_id;
ResourceGroupPtr resource_group = nullptr;
if (one_resp.has_keyspace_id())
keyspace_id = one_resp.keyspace_id().value();
auto resource_group = findResourceGroup(keyspace_id, name);
{
resource_group = findResourceGroup(resp_keyspace_id, name);
}
else
{
auto iter = pending_req_keyspaces_by_name.find(name);
if (iter != pending_req_keyspaces_by_name.end())
{
for (const auto req_keyspace_id : iter->second)
{
resource_group = findResourceGroup(req_keyspace_id, name);
if (resource_group != nullptr)
{
matched_req_keyspace_id = req_keyspace_id;
break;
}
}
}
if (resource_group == nullptr)
resource_group = findResourceGroup(resp_keyspace_id, name);
}
if (resource_group == nullptr)
{
LOG_ERROR(log, "cannot find resource group: {}(keyspace={})", name, keyspace_id);
LOG_ERROR(log, "cannot find resource group: {}(keyspace={})", name, resp_keyspace_id);
continue;
}

handled_resource_group_names.emplace_back(keyspace_id, name);
const KeyspaceID handled_keyspace_id = matched_req_keyspace_id.value_or(resp_keyspace_id);
handled_resource_group_names.emplace_back(handled_keyspace_id, name);
if (auto iter = pending_req_keyspaces_by_name.find(name); iter != pending_req_keyspaces_by_name.end())
{
auto & pending_keyspaces = iter->second;
if (const auto matched_iter
= std::find(pending_keyspaces.begin(), pending_keyspaces.end(), handled_keyspace_id);
matched_iter != pending_keyspaces.end())
{
pending_keyspaces.erase(matched_iter);
if (pending_keyspaces.empty())
pending_req_keyspaces_by_name.erase(iter);
}
}

const String err_msg = fmt::format("handle acquire token resp failed: rg: {}(keyspace={})", name, keyspace_id);
const auto metric_keyspace_id = resource_group->getKeyspaceID();
const String err_msg = fmt::format(
"handle acquire token resp failed: rg: {}(keyspace={}, cached_keyspace={})",
name,
handled_keyspace_id,
metric_keyspace_id);
// It's possible for one_resp.granted_r_u_tokens() to be empty
// when the acquire_token_req is only for report RU consumption or GAC got error(like nan token).
if (one_resp.granted_r_u_tokens().empty())
Expand Down Expand Up @@ -719,7 +835,7 @@ std::vector<std::pair<KeyspaceID, std::string>> LocalAdmissionController::handle
// https://github.com/tikv/pd/blob/e9757fbe03260775262763c67f62296fcb26b3c2/pkg/mcs/resourcemanager/server/token_buckets.go#L47
int64_t capacity = granted_token_bucket.granted_tokens().settings().burst_limit();

const auto name_with_keyspace_id = getResourceGroupMetricName(keyspace_id, name);
const auto name_with_keyspace_id = getResourceGroupMetricName(metric_keyspace_id, name);
if (added_tokens > 0)
GET_RESOURCE_GROUP_METRIC(tiflash_resource_group, type_gac_resp_tokens, name_with_keyspace_id)
.Set(added_tokens);
Expand Down Expand Up @@ -757,11 +873,21 @@ std::vector<std::pair<KeyspaceID, std::string>> LocalAdmissionController::handle

void LocalAdmissionController::watchGACLoop(const std::string & etcd_path)
{
auto grpc_context = std::make_unique<grpc::ClientContext>();
{
std::lock_guard lock(mu);
active_watch_gac_grpc_contexts.insert(grpc_context.get());
}
SCOPE_EXIT({
std::lock_guard lock(mu);
active_watch_gac_grpc_contexts.erase(grpc_context.get());
});

while (!stopped.load())
{
try
{
doWatch(etcd_path);
doWatch(etcd_path, grpc_context.get());
}
catch (...)
{
Expand All @@ -782,15 +908,16 @@ void LocalAdmissionController::watchGACLoop(const std::string & etcd_path)
}))
return;

// Create new grpc_context for each reader/writer.
watch_gac_grpc_context = std::make_unique<grpc::ClientContext>();
active_watch_gac_grpc_contexts.erase(grpc_context.get());
grpc_context = std::make_unique<grpc::ClientContext>();
active_watch_gac_grpc_contexts.insert(grpc_context.get());
}
}
}

void LocalAdmissionController::doWatch(const std::string & etcd_path)
void LocalAdmissionController::doWatch(const std::string & etcd_path, grpc::ClientContext * grpc_context)
{
auto stream = etcd_client->watch(watch_gac_grpc_context.get());
auto stream = etcd_client->watch(grpc_context);
auto watch_req = setupWatchReq(etcd_path);
LOG_DEBUG(log, "watchGAC req: {}", watch_req.ShortDebugString());
const bool write_ok = stream->Write(watch_req);
Expand Down Expand Up @@ -973,7 +1100,8 @@ void LocalAdmissionController::stop()
// So still need to lock.
{
std::lock_guard lock(mu);
watch_gac_grpc_context->TryCancel();
for (auto * grpc_context : active_watch_gac_grpc_contexts)
grpc_context->TryCancel();
cv.notify_all();
}
{
Expand Down
Loading