feat(storage): basic compactor scheduler policy (#6986)

Co-authored-by: Yuhao Su <31772373+yuhao-su@users.noreply.github.com>
Co-authored-by: Dylan <chenzl25@mail2.sysu.edu.cn>
Co-authored-by: Yuanxin Cao <60498509+xx01cyx@users.noreply.github.com>
Co-authored-by: ZENOTME <43447882+ZENOTME@users.noreply.github.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Co-authored-by: congyi wang <58715567+wcy-fdu@users.noreply.github.com>

grafana/risingwave-dashboard.dashboard.py

@@ -520,10 +520,16 @@ def section_cluster_node(panels):
             [
                 panels.target(
                     f"sum(rate({metric('process_cpu_seconds_total')}[$__rate_interval])) by (job,instance)",
-                    "{{job}} @ {{instance}}",
-                )
+                    "cpu - {{job}} @ {{instance}}",
+                ),
+
+                panels.target(
+                    f"sum(rate({metric('process_cpu_seconds_total')}[$__rate_interval])) by (job,instance) / avg({metric('process_cpu_core_num')}) by (job,instance)",
+                    "cpu usage -{{job}} @ {{instance}}",
+                ),
             ],
         ),
+
         panels.timeseries_count(
             "Meta Cluster",
             "",
@@ -562,6 +568,16 @@ def section_compaction(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "scale compactor core count",
+                    "compactor core resource need to scale out",
+                    [
+                        panels.target(
+                            f"sum({metric('storage_compactor_suggest_core_count')})",
+                            "suggest-core-count"
+                        ),
+                    ],
+                ),
                 panels.timeseries_count(
                     "Compaction Success & Failure Count",
                     "num of compactions from each level to next level",
@@ -1564,7 +1580,6 @@ def section_batch_exchange(outer_panels):
         ),
     ]
 
-
 def section_frontend(outer_panels):
     panels = outer_panels.sub_panel()
     return [

proto/hummock.proto

@@ -358,18 +358,25 @@ message CompactTaskProgress {
   uint32 num_ssts_uploaded = 3;
 }
 
-message ReportCompactionTaskProgressRequest {
+// The measurement of the workload on a compactor to determine whether it is idle.
+message CompactorWorkload {
+  uint32 cpu = 1;
+}
+
+message CompactorHeartbeatRequest {
   uint32 context_id = 1;
   repeated CompactTaskProgress progress = 2;
+  CompactorWorkload workload = 3;
 }
 
-message ReportCompactionTaskProgressResponse {
+message CompactorHeartbeatResponse {
   common.Status status = 1;
 }
 
 message SubscribeCompactTasksRequest {
   uint32 context_id = 1;
   uint64 max_concurrent_task_number = 2;
+  uint32 cpu_core_num = 3;
 }
 
 message ValidationTask {
@@ -563,7 +570,7 @@ service HummockManagerService {
   rpc TriggerCompactionDeterministic(TriggerCompactionDeterministicRequest) returns (TriggerCompactionDeterministicResponse);
   rpc DisableCommitEpoch(DisableCommitEpochRequest) returns (DisableCommitEpochResponse);
   rpc ReportCompactionTasks(ReportCompactionTasksRequest) returns (ReportCompactionTasksResponse);
-  rpc ReportCompactionTaskProgress(ReportCompactionTaskProgressRequest) returns (ReportCompactionTaskProgressResponse);
+  rpc CompactorHeartbeat(CompactorHeartbeatRequest) returns (CompactorHeartbeatResponse);
   rpc PinSnapshot(PinSnapshotRequest) returns (PinSnapshotResponse);
   rpc PinSpecificSnapshot(PinSpecificSnapshotRequest) returns (PinSnapshotResponse);
   rpc GetEpoch(GetEpochRequest) returns (GetEpochResponse);
@@ -581,6 +588,7 @@ service HummockManagerService {
   rpc RiseCtlUpdateCompactionConfig(RiseCtlUpdateCompactionConfigRequest) returns (RiseCtlUpdateCompactionConfigResponse);
   rpc InitMetadataForReplay(InitMetadataForReplayRequest) returns (InitMetadataForReplayResponse);
   rpc SetCompactorRuntimeConfig(SetCompactorRuntimeConfigRequest) returns (SetCompactorRuntimeConfigResponse);
+  rpc GetScaleCompactor(GetScaleCompactorRequest) returns (GetScaleCompactorResponse);
   rpc PinVersion(PinVersionRequest) returns (PinVersionResponse);
   rpc SplitCompactionGroup(SplitCompactionGroupRequest) returns (SplitCompactionGroupResponse);
 }
@@ -619,6 +627,15 @@ message HummockVersionStats {
   map<uint32, TableStats> table_stats = 2;
 }
 
+message GetScaleCompactorRequest {}
+
+message GetScaleCompactorResponse {
+  uint64 suggest_cores = 1;
+  uint64 running_cores = 2;
+  uint64 total_cores = 3;
+  uint64 waiting_compaction_bytes = 4;
+}
+
 message WriteLimits {
   message WriteLimit {
     repeated uint32 table_ids = 1;