add ru guard for LAC (pingcap#119)

guo-shaoge · Nov 9, 2023 · e9b1fae · e9b1fae
1 parent 7570f61
commit e9b1fae
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/dbms/src/Flash/ResourceControl/LocalAdmissionController.cpp b/dbms/src/Flash/ResourceControl/LocalAdmissionController.cpp
@@ -248,7 +248,7 @@ void LocalAdmissionController::fetchTokensFromGAC(
             req_ru->set_type(resource_manager::RequestUnitType::RU);
             req_ru->set_value(info.acquire_tokens);
         }
-        if (info.ru_consumption_delta > 0.0)
+        if (info.ru_consumption_delta > 0.0 || final_report)
         {
             single_group_req->set_is_tiflash(true);
             auto * tiflash_consumption = single_group_req->mutable_consumption_since_last_request();

diff --git a/dbms/src/Flash/ResourceControl/LocalAdmissionController.h b/dbms/src/Flash/ResourceControl/LocalAdmissionController.h
@@ -319,6 +319,15 @@ class ResourceGroup final : private boost::noncopyable
             if (elapsed < dura)
                 return {.speed = ru_consumption_speed, .delta = ru_consumption_delta, .updated = false};
 
+            // TODO: Add this to avoid bug that cause RU usage is very large. Remove it after tiflash resource control is stable.
+            // For serverless tier, the fillrate is 2M, so set it as 2M*5sec, and multiple 5 to avoid limit speed.
+            if unlikely(ru_consumption_delta >= 50'000'000L)
+            {
+                LOG_ERROR(log, "too much ru consumption({}), reset it as zero!", ru_consumption_delta);
+                ru_consumption_delta = 0;
+                return {.speed = ru_consumption_speed, .delta = ru_consumption_delta, .updated = false};
+            }
+
             ru_consumption_speed = ru_consumption_delta / elapsed.count();
             total_ru_consumption += ru_consumption_delta;
 
@@ -543,6 +552,7 @@ class LocalAdmissionController final : private boost::noncopyable
                  .acquire_tokens = 0,
                  .ru_consumption_delta = consumption_update_info.delta});
         }
+        LOG_DEBUG(log, "final report done(report size: {})", acquire_infos.size());
         fetchTokensFromGAC(acquire_infos, "before stop", true);
 
         if (need_reset_unique_client_id.load())