syntax = "proto3";

package monitor_service;

option java_package = "com.risingwave.proto";
option optimize_for = SPEED;

message StackTraceRequest {
  enum ActorTracesFormat {
    UNSPECIFIED = 0;
    JSON = 1;
    TEXT = 2;
  }
  ActorTracesFormat actor_traces_format = 1;
}

message StackTraceResponse {
  map<uint32, string> actor_traces = 1;
  map<string, string> rpc_traces = 2;
  map<string, string> compaction_task_traces = 3;
  map<uint64, string> inflight_barrier_traces = 4;
  map<uint32, string> barrier_worker_state = 5; // key: worker id
  map<uint32, string> jvm_stack_traces = 6; // key: worker id. Might be empty if the worker doesn't run JVM.
  map<string, string> meta_traces = 7;
  // Per-worker error messages when collecting stack traces from the cluster.
  // Key: worker id.
  map<uint32, string> node_errors = 8;
}

// CPU profiling
message ProfilingRequest {
  // How long the profiling should last.
  uint64 sleep_s = 1;
}

message ProfilingResponse {
  bytes result = 1;
}

// Heap profiling
message HeapProfilingRequest {
  // The directory that the dumped file in
  string dir = 1;
}

message HeapProfilingResponse {}

message ListHeapProfilingRequest {}
message ListHeapProfilingResponse {
  string dir = 1;
  repeated string name_manually = 2;
  repeated string name_auto = 3;
}

// Analyze dumped files
message AnalyzeHeapRequest {
  // The file path
  string path = 1;
}

message AnalyzeHeapResponse {
  bytes result = 1;
}

// Streaming Runtime Stats
message GetStreamingStatsRequest {}

message ChannelStats {
  // Total number of actors. This field is very important because the rest fields are `SUM` of all the actors.
  uint32 actor_count = 1 [json_name = "n"]; /* Use shorter name in JSON to reduce message size. */
  // Sum of output blocking duration of all actors in nanoseconds
  double output_blocking_duration = 2 [json_name = "b"];
  // Sum of input row count of all actors
  uint64 recv_row_count = 3 [json_name = "i"];
  // Sum of output row count of all actors
  uint64 send_row_count = 4 [json_name = "o"];
}

message ChannelDeltaStats {
  uint32 actor_count = 1 [json_name = "n"];
  // Sum of output blocking duration of all actors in nanoseconds
  double backpressure_rate = 2 [json_name = "b"];
  // Sum of input row count of all actors
  double recv_throughput = 3 [json_name = "i"];
  // Sum of output row count of all actors
  double send_throughput = 4 [json_name = "o"];
}

message FragmentStats {
  uint32 actor_count = 2 [json_name = "n"];
  uint64 current_epoch = 3 [json_name = "e"];
}

message RelationStats {
  uint32 actor_count = 2 [json_name = "n"];
  uint64 current_epoch = 3 [json_name = "e"];
}

message GetStreamingStatsResponse {
  // Key: "<upstream_fragment_id>_<downstream_fragment_id>"
  map<string, ChannelStats> channel_stats = 1;
  map<uint32, FragmentStats> fragment_stats = 2;
  map<uint32, RelationStats> relation_stats = 3;
}

message GetStreamingPrometheusStatsResponse {
  // Key: "<upstream_fragment_id>_<downstream_fragment_id>"
  map<string, ChannelDeltaStats> channel_stats = 1;
  map<uint32, FragmentStats> fragment_stats = 2;
  map<uint32, RelationStats> relation_stats = 3;
}

message TieredCacheTracingRequest {
  bool enable = 1;
  optional uint32 record_hybrid_insert_threshold_ms = 2;
  optional uint32 record_hybrid_get_threshold_ms = 3;
  optional uint32 record_hybrid_obtain_threshold_ms = 4;
  optional uint32 record_hybrid_remove_threshold_ms = 5;
  optional uint32 record_hybrid_fetch_threshold_ms = 6;
}

message TieredCacheTracingResponse {}

message GetProfileStatsRequest {
  // Executors to fetch statistics for.
  repeated uint64 executor_ids = 1;
  // Dispatchers do not have executors.
  // We have to fetch their statistics separately.
  repeated uint32 dispatcher_fragment_ids = 2;
}

message GetProfileStatsResponse {
  map<uint64, uint64> stream_node_output_row_count = 1;
  map<uint64, uint64> stream_node_output_blocking_duration_ns = 2;
  map<uint32, uint64> dispatch_fragment_output_row_count = 3;
  map<uint32, uint64> dispatch_fragment_output_blocking_duration_ns = 4;
}

service MonitorService {
  rpc StackTrace(StackTraceRequest) returns (StackTraceResponse);
  rpc Profiling(ProfilingRequest) returns (ProfilingResponse);
  rpc HeapProfiling(HeapProfilingRequest) returns (HeapProfilingResponse);
  rpc ListHeapProfiling(ListHeapProfilingRequest) returns (ListHeapProfilingResponse);
  rpc AnalyzeHeap(AnalyzeHeapRequest) returns (AnalyzeHeapResponse);
  rpc GetStreamingStats(GetStreamingStatsRequest) returns (GetStreamingStatsResponse);
  rpc TieredCacheTracing(TieredCacheTracingRequest) returns (TieredCacheTracingResponse);
  rpc GetProfileStats(GetProfileStatsRequest) returns (GetProfileStatsResponse);
}
