feat(storage): basic compactor scheduler policy

grafana/risingwave-dashboard.dashboard.py

@@ -562,6 +562,26 @@ def section_compaction(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_bytes(
+                    "Waiting compaction bytes",
+                    "Compaction bytes which need to be schedule in new task",
+                    [
+                        panels.target(
+                            f"sum({metric('storage_compactor_waiting_compaction_bytes')})",
+                            "waiting_compaction_bytes",
+                        ),
+                    ],
+                ),
+                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",
@@ -1553,7 +1573,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 {
@@ -562,7 +569,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);
@@ -580,6 +587,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);
 }
@@ -614,3 +622,12 @@ message HummockVersionStats {
   uint64 hummock_version_id = 1;
   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;
+}