distbench.proto

// Copyright 2021 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

syntax = "proto2";

package distbench;

import "traffic_config.proto";

option cc_enable_arenas = true;

message TestsSetting {
  optional bool keep_instance_log = 1 [default = true];
  optional bool shutdown_after_tests = 2 [default = false];
}

message TestSequence {
  repeated DistributedSystemDescription tests = 1;
  optional TestsSetting tests_setting = 2;
}

message NodeServiceConfig {
  optional DistributedSystemDescription traffic_config = 1;
  repeated string services = 2;
}

message TraceContext {
  // For compactness the TraceContext fields are all packed repeated fields
  // with each node in the chain appending its local data to the end of each
  // repeated field.

  // This is set to the global ID of each engine along the RPC chain.
  repeated int64 engine_ids = 1 [packed = true];

  // The index of the actionlist that triggered this RPC within the
  // DistributedSystemDescription.
  repeated int64 actionlist_indices = 3 [packed = true];

  // The invocation count of the actionlist that triggered this RPC.
  // This number increments each time the actionlist is executed.
  repeated int64 actionlist_invocations = 4 [packed = true];

  // The index of the action within the actionlist that triggered this RPC.
  repeated int64 action_indices = 5 [packed = true];

  // The iteration number of the action that triggered this RPC.
  repeated int64 action_iterations = 6 [packed = true];

  // Each iteration of the action may send RPCs to multiple servers.
  // This tracks the index into the set of related individual RPCs.
  repeated int64 fanout_index = 7 [packed = true];

  // Obsolete field that was poorly/inconsistently defined.
  repeated int64 iterations = 2 [packed = true, deprecated = true];
}

message RpcReplayTraceLog {
  optional string rpc_replay_trace_name = 1;
  // All RPCs in the trace will share the same trace_context, and warmup
  // flag.
  optional TraceContext trace_context = 2;
  optional bool warmup = 3;

  optional int64 timestamp_offset_ns = 4;

  // The key of this map is the index into the rpc_trace.
  // The request_size and response_size fields are already in the
  // trace. Timestamps are relative to the above timestamp_offset field.
  map<int32, RpcSample> rpc_samples = 5;
}

message RpcSample {
  optional int64 request_size = 1;
  optional int64 response_size = 2;
  optional int64 start_timestamp_ns = 3;
  optional int64 latency_ns = 4;
  optional int64 latency_weight = 5 [default = 1];
  optional TraceContext trace_context = 6;
  // If true this was a warmup RPC, and should probably be ignored
  // by most analysis tools, except when tuning the warmup period
  // of a test.
  optional bool warmup = 7;
  optional int32 error_index = 8;

  // For multi-server channels during rpc replay, we record which server was
  // actually chosen:
  optional int32 server_instance = 9;
}

message RpcPerformanceLog {
  repeated RpcSample successful_rpc_samples = 1;
  repeated RpcSample failed_rpc_samples = 2;
}

message PeerPerformanceLog {
  // The key is the rpc index
  map<int32, RpcPerformanceLog> rpc_logs = 1;
}

message RUsage {
  optional double user_cpu_time_seconds = 1;
  optional double system_cpu_time_seconds = 2;
  optional int64 max_resident_set_size = 3;
  optional int64 integral_shared_memory_size = 4;
  optional int64 integral_unshared_data_size = 5;
  optional int64 integral_unshared_stack_size = 6;
  optional int64 page_reclaims_soft_page_faults = 7;
  optional int64 page_faults_hard_page_faults = 8;
  optional int64 swaps = 9;
  optional int64 block_input_operations = 10;
  optional int64 block_output_operations = 11;
  optional int64 ipc_messages_sent = 12;
  optional int64 ipc_messages_received = 13;
  optional int64 signals_received = 14;
  optional int64 voluntary_context_switches = 15;
  optional int64 involuntary_context_switches = 16;
}

message RUsageStats {
  optional RUsage rusage_start = 1;
  optional RUsage rusage_diff = 2;
}

message ActivityMetric {
  optional string name = 1;
  optional int64 value_int = 2;
}

message ActivityLog {
  repeated ActivityMetric activity_metrics = 1;
}

message ErrorDictionary {
  repeated string error_message = 1;
}

// Logs for an individual instance of a service:
message ServicePerformanceLog {
  // The key is service instance's name and the value contains performance
  // logs of RPCs initiated to the named service instance.
  map<string, PeerPerformanceLog> peer_logs = 1;

  repeated RpcReplayTraceLog replay_trace_logs = 5;

  // The key is activity name and value is activity logs.
  map<string, ActivityLog> activity_logs = 2;

  optional ErrorDictionary error_dictionary = 3;
  optional string engine_error_message = 4;
}

// Logs for multiple service instances:
message ServiceLogs {
  // The key is the client service instance name
  map<string, ServicePerformanceLog> instance_logs = 1;
}

message ResourceUsageLogs {
  optional RUsageStats test_sequencer_usage = 1;
  map<string, RUsageStats> node_usages = 2;
}

// Logs for all services in a test config:
message TestResult {
  optional DistributedSystemDescription traffic_config = 1;
  optional ServiceEndpointMap placement = 2;
  optional ServiceLogs service_logs = 3;
  optional ResourceUsageLogs resource_usage_logs = 5;
  repeated string log_summary = 100;
}

message GetTrafficResultResponse {
  optional ServiceLogs service_logs = 1;
  map<string, RUsageStats> node_usages = 2;
}

// Logs for all a test configs in a test sequence:
message TestSequenceResults {
  repeated TestResult test_results = 1;
}

message Attribute {
  optional string name = 1;
  optional string value = 2;
}

message NodeRegistration {
  optional string hostname = 1;
  optional string control_ip = 2;
  optional int32 control_port = 3;
  optional string service_address = 6;
  repeated Attribute attributes = 4;
  optional int32 preassigned_node_id = 5 [default = -1];
}

message NodeConfig {
  optional int32 node_id = 1;
  optional string node_alias = 2;
}

service DistBenchTestSequencer {
  // Registers a new node:
  rpc RegisterNode(NodeRegistration) returns (NodeConfig) {}

  // Runs a set of tests on the registered nodes:
  rpc RunTestSequence(TestSequence) returns (TestSequenceResults) {}
}

message ServiceEndpoint {
  optional string endpoint_address = 1;
  optional string hostname = 2;
  repeated Attribute attributes = 3;
}

message ServiceEndpointMap {
  map<string, ServiceEndpoint> service_endpoints = 1;
}

message IntroducePeersResult {}

message RunTrafficRequest {
  optional int64 start_timestamp_ns = 1;
}

message RunTrafficResponse {}

message GetTrafficResultRequest {
  optional bool clear_services = 1;
}

message CancelTrafficRequest {}

message CancelTrafficResult {}

message ShutdownNodeRequest {}

message ShutdownNodeResult {}

service DistBenchNodeManager {
  // Sets up services prior to running traffic:
  rpc ConfigureNode(NodeServiceConfig) returns (ServiceEndpointMap) {}

  // Introduces NodeEngines to each other.
  rpc IntroducePeers(ServiceEndpointMap) returns (IntroducePeersResult) {}

  // Runs traffic:
  rpc RunTraffic(RunTrafficRequest) returns (RunTrafficResponse) {}

  // Get the RunTraffic performance logs:
  rpc GetTrafficResult(GetTrafficResultRequest)
      returns (GetTrafficResultResponse) {}

  // Cancels a currently running traffic pattern immediately:
  rpc CancelTraffic(CancelTrafficRequest) returns (CancelTrafficResult) {}

  // Shuts down a node manager after any active test completes:
  rpc ShutdownNode(ShutdownNodeRequest) returns (ShutdownNodeResult) {}
}

message ConnectRequest {
  optional bytes initiator_info = 1;
}

message ConnectResponse {
  optional bytes responder_info = 1;
}

service ConnectionSetup {
  // Get a connection setup
  rpc SetupConnection(ConnectRequest) returns (ConnectResponse) {}
}

// GenericRequestResponse exists to make it possible to execute ping-pong
// traffic where the response payload is exactly the request payload.
// The request fields can be ignored by the code handling the response.
message GenericRequestResponse {
  //////////////////////////////
  // Request Metadata fields: //
  //////////////////////////////

  // rpc_index if present provides an index into the actionlist table for
  // synthesized RPCs. Replay RPCs do not speciy rpc_index.
  optional int32 rpc_index = 1;
  optional TraceContext trace_context = 2;
  optional bool warmup = 3;

  optional int64 response_payload_size = 4;

  // For replay RPCs we don't invoke a real service routine, but simulate
  // delay with this instead:
  optional int64 server_processing_time_ns = 5;

  ///////////////////////////////
  // Response Metadata fields: //
  ///////////////////////////////

  optional string error_message = 9;

  // For MultiServerChannels we need to know which server was chosen, even if
  // the RPC layer does not expose that.
  optional int32 server_instance = 10;

  //////////////////////////
  // Payload data fields: //
  //////////////////////////

  // This does not mean anything, and if present is always false.
  // It's only use is to produce protobufs that encode to an exact length.
  optional bool trim = 14;
  optional bytes payload = 15 [ctype = CORD];

  ////////////////////////////
  // Unit-Test Only fields: //
  ////////////////////////////

  optional fixed32 before_payload32 = 12;
  optional fixed64 before_payload64 = 13;
  optional fixed32 after_payload32 = 16;
  optional fixed64 after_payload64 = 17;

  // No new fields can be added with tags greater than 15. Do not do it.
  // It will break MetaDataLength().
  reserved 18 to 1000000;
}

message ServerAddress {
  optional string ip_address = 1;
  optional int32 port = 2;
  optional string socket_address = 3;
}

service Traffic {
  // One RPC to simulate them all:
  rpc GenericRpc(GenericRequestResponse) returns (GenericRequestResponse) {}
}