diff --git a/benchmarking/README.md b/benchmarking/README.md index 3f53801d8..469eb7002 100644 --- a/benchmarking/README.md +++ b/benchmarking/README.md @@ -62,6 +62,53 @@ You can also configure things like the number of users, how quickly those users are spawned, the frequency with which requests are made and whether or not tracing is enabled. +User classes implemented in boomer rather than Python are selected at deploy +time — the stack runs one per deployment: + +```bash +./benchmarking/locust/deploy.sh --deploy --user-class durdir +``` + +### Headless (automation only) + +`runner.py` runs a test without the web UI, writing CSVs, logs and traces to +`--dest`. The nightly automation submits it as a Job on the test cluster; it is +not a local entry point. See [automation/README.md](automation/README.md). + +```bash +python3 runner.py -f tests/.py -t 1m -u 1 --name --dest /tmp/bench +``` + +Test-specific flags are appended to the same command; see the sections below. + +### DurDir Benchmark + +The DurDir benchmark evaluates actor suspend/resume performance, disk persistence overhead, +and state restoration latency when a durable directory is attached to the actor. + +#### DurDir Configuration Knobs + +* `--durdir-file-size-bytes`: Size in bytes of the data file (default `8388608` = 8 MiB). +* `--resume-mode`: Resume trigger mode: + * `explicit` (default): Client invokes the `ResumeActor` RPC before sending traffic. + * `implicit`: Client sends traffic through the router without an explicit wake RPC, testing traffic-triggered resume. +* `--durdir-read-mode`: Verification read mode: + * `data` (default): Server returns full payload bytes for client-side SHA-256 verification. + * `digest`: Server hashes the file and returns size and digest, reducing network transfer. +* `--durdir-template`: ActorTemplate name: + * `glutton-durdir-data` (default): Attaches a durable data directory without memory snapshot restore. + * `glutton-durdir-full`: Attaches a durable data directory and performs a full memory snapshot restore. + +#### DurDir Reported Metrics + +* `DurDirWrite`: Initial truncate-write creating the data file. +* `DurDirServeInitial`: First read immediately following file creation. +* `SuspendActor`: Actor suspend latency (snapshot creation + persistence upload). +* `ResumeActor`: Actor resume latency. +* `DurDirServeAfterResume`: First read after resume (measures page faults / lazy load overhead on restored volume). +* `DurDirServeWarm`: Subsequent read within the same active cycle (cached state baseline). +* `DurDirOverwrite`: In-place file overwrite with checksum verification. + ### Viewing Traces You must have enabled otel tracing for your cluster to view traces. diff --git a/benchmarking/automation/tests.yaml b/benchmarking/automation/tests.yaml index 7cd1627b3..a8970e323 100644 --- a/benchmarking/automation/tests.yaml +++ b/benchmarking/automation/tests.yaml @@ -83,3 +83,117 @@ tests: - "0.5" - "--max-wait-time" - "1.0" + - name: durdir_data_baseline + description: "DurDir data baseline: 1 concurrent user, 8 MiB file, explicit resume" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-data" + - "--resume-mode" + - "explicit" + - "--durdir-file-size-bytes" + - "8388608" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" + - name: durdir_full_baseline + description: "DurDir full baseline: 1 concurrent user, 8 MiB file, full memory restore comparison" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-full" + - "--resume-mode" + - "explicit" + - "--durdir-file-size-bytes" + - "8388608" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" + - name: durdir_implicit_resume + description: "DurDir implicit resume: 1 concurrent user, 8 MiB file, traffic-triggered resume" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-data" + - "--resume-mode" + - "implicit" + - "--durdir-file-size-bytes" + - "8388608" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" + - name: durdir_size_5mb + description: "DurDir file size sweep (5 MiB): 1 concurrent user, digest read mode" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-data" + - "--resume-mode" + - "explicit" + - "--durdir-read-mode" + - "digest" + - "--durdir-file-size-bytes" + - "5242880" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" + - name: durdir_size_10mb + description: "DurDir file size sweep (10 MiB): 1 concurrent user, digest read mode" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-data" + - "--resume-mode" + - "explicit" + - "--durdir-read-mode" + - "digest" + - "--durdir-file-size-bytes" + - "10485760" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" + - name: durdir_size_64mb + description: "DurDir file size sweep (64 MiB): 1 concurrent user, digest read mode" + targetCluster: dev + file: /app/tests/durdir.py + duration: 1m + users: 1 + workerCount: 1 + flags: + - "--durdir-template" + - "glutton-durdir-data" + - "--resume-mode" + - "explicit" + - "--durdir-read-mode" + - "digest" + - "--durdir-file-size-bytes" + - "67108864" + - "--min-wait-time" + - "1.0" + - "--max-wait-time" + - "1.0" diff --git a/benchmarking/locust/common/boomer_config.py b/benchmarking/locust/common/boomer_config.py index 2f0194a74..3c58fc89b 100644 --- a/benchmarking/locust/common/boomer_config.py +++ b/benchmarking/locust/common/boomer_config.py @@ -17,6 +17,8 @@ Flag registration lives in the modules that own each flag: * --trace-probability → common.trace.init_tracing * --min-wait-time / --max-wait-time → common.wait_time.init_wait_time + * --resume-mode → common.resume_mode.add_resume_mode_arguments + * --durdir-* → common.durdir_config.add_durdir_arguments This module ties them together so boomer-Go workers can pick up the values the operator set in the web UI form: @@ -34,17 +36,19 @@ import logging from collections.abc import Iterable -from locust import events -from locust.env import Environment - -from common.trace import init_tracing -from common.wait_time import init_wait_time - logger = logging.getLogger(__name__) -# Boomer-tunable flags. CLI form ("--foo-bar") is converted to the -# attribute / JSON-key form ("foo_bar") by _attr(). -_FLAGS = ("--trace-probability", "--min-wait-time", "--max-wait-time") +# Boomer-tunable flags and their types. CLI form ("--foo-bar") is converted +# to the attribute / JSON-key form ("foo_bar") by _attr(). +_FLAGS = { + "--trace-probability": float, + "--min-wait-time": float, + "--max-wait-time": float, + "--durdir-file-size-bytes": int, + "--resume-mode": str, + "--durdir-read-mode": str, + "--durdir-template": str, +} def _attr(flag: str) -> str: @@ -56,8 +60,8 @@ def build_config_json(argv: Iterable[str]) -> str: --config-json flag. Unknown args are ignored; unset flags are omitted so boomer falls back to its own defaults.""" p = argparse.ArgumentParser(add_help=False) - for flag in _FLAGS: - p.add_argument(flag, type=float) + for flag, type_func in _FLAGS.items(): + p.add_argument(flag, type=type_func) parsed, _ = p.parse_known_args(argv) cfg = { _attr(f): getattr(parsed, _attr(f)) @@ -71,6 +75,14 @@ def init_boomer_config() -> None: """Ensure the owning modules have registered the boomer-tunable flags, then expose their current values at /boomer-config so boomer-Go workers can fetch them at runtime.""" + from locust import events + from locust.env import Environment + + from common.durdir_config import add_durdir_arguments + from common.resume_mode import add_resume_mode_arguments + from common.trace import init_tracing + from common.wait_time import init_wait_time + init_tracing() init_wait_time() @@ -82,7 +94,7 @@ def on_init(environment: Environment, **kwargs) -> None: return @environment.web_ui.app.route("/boomer-config") - def boomer_config() -> dict[str, float | None]: + def boomer_config() -> dict[str, float | int | str | None]: opts = environment.parsed_options return {_attr(f): getattr(opts, _attr(f), None) for f in _FLAGS} diff --git a/benchmarking/locust/common/durdir_config.py b/benchmarking/locust/common/durdir_config.py new file mode 100644 index 000000000..09059c3d0 --- /dev/null +++ b/benchmarking/locust/common/durdir_config.py @@ -0,0 +1,43 @@ +# Copyright 2026 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 +# +# http://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. + +"""DurDir benchmark runtime flags.""" + +from locust import events +from locust.argument_parser import LocustArgumentParser + + +@events.init_command_line_parser.add_listener +def add_durdir_arguments(parser: LocustArgumentParser) -> None: + group = parser.add_argument_group("DurDir Benchmark") + group.add_argument( + "--durdir-file-size-bytes", + type=int, + default=8388608, + help="Size of the test file written and read during the DurDir benchmark (default: 8388608 = 8 MiB)", + ) + group.add_argument( + "--durdir-read-mode", + type=str, + default="data", + choices=["data", "digest"], + help="Read mode for DurDir serves: 'data' (default) returns and client-verifies the full payload; " + "'digest' returns only size+sha256 for reduced network overhead", + ) + group.add_argument( + "--durdir-template", + type=str, + default="glutton-durdir-data", + help="ActorTemplate name to benchmark (default: glutton-durdir-data)", + ) diff --git a/benchmarking/locust/common/glutton_pb2.py b/benchmarking/locust/common/glutton_pb2.py index 9cf0f37ba..34d0f26b7 100644 --- a/benchmarking/locust/common/glutton_pb2.py +++ b/benchmarking/locust/common/glutton_pb2.py @@ -38,7 +38,7 @@ -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\rglutton.proto\x12\x07glutton\"T\n\x0fWriteRAMRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x0c\n\x04size\x18\x02 \x01(\x05\x12&\n\nwrite_mode\x18\x03 \x01(\x0e\x32\x12.glutton.WriteMode\"\x12\n\x10WriteRAMResponse\"U\n\x10WriteDiskRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x0c\n\x04size\x18\x02 \x01(\x05\x12&\n\nwrite_mode\x18\x03 \x01(\x0e\x32\x12.glutton.WriteMode\"\x13\n\x11WriteDiskResponse\"\x1e\n\rOpenFDRequest\x12\r\n\x05\x63ount\x18\x01 \x01(\x05\"\x10\n\x0eOpenFDResponse\"\x1e\n\x0bPingRequest\x12\x0f\n\x07message\x18\x01 \x01(\t\"\x1f\n\x0cPingResponse\x12\x0f\n\x07message\x18\x01 \x01(\t\"-\n\rGossipRequest\x12\x1c\n\x05peers\x18\x01 \x03(\x0b\x32\r.glutton.Peer\"\x10\n\x0eGossipResponse\"&\n\x04Peer\x12\x0c\n\x04host\x18\x01 \x01(\t\x12\x10\n\x08\x64\x65lay_ms\x18\x02 \x01(\x05*>\n\tWriteMode\x12\x17\n\x13WRITE_MODE_TRUNCATE\x10\x00\x12\x18\n\x14WRITE_MODE_OVERWRITE\x10\x01\x32\xc3\x02\n\x07Glutton\x12\x41\n\x08WriteRAM\x12\x18.glutton.WriteRAMRequest\x1a\x19.glutton.WriteRAMResponse\"\x00\x12\x44\n\tWriteDisk\x12\x19.glutton.WriteDiskRequest\x1a\x1a.glutton.WriteDiskResponse\"\x00\x12;\n\x06OpenFD\x12\x16.glutton.OpenFDRequest\x1a\x17.glutton.OpenFDResponse\"\x00\x12\x35\n\x04Ping\x12\x14.glutton.PingRequest\x1a\x15.glutton.PingResponse\"\x00\x12;\n\x06Gossip\x12\x16.glutton.GossipRequest\x1a\x17.glutton.GossipResponse\"\x00\x42=Z;github.com/agent-substrate/substrate/internal/proto/gluttonb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\rglutton.proto\x12\x07glutton\"T\n\x0fWriteRAMRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x0c\n\x04size\x18\x02 \x01(\x05\x12&\n\nwrite_mode\x18\x03 \x01(\x0e\x32\x12.glutton.WriteMode\"\x12\n\x10WriteRAMResponse\"U\n\x10WriteDiskRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x0c\n\x04size\x18\x02 \x01(\x05\x12&\n\nwrite_mode\x18\x03 \x01(\x0e\x32\x12.glutton.WriteMode\"1\n\x11WriteDiskResponse\x12\x0c\n\x04size\x18\x01 \x01(\x03\x12\x0e\n\x06sha256\x18\x02 \x01(\x0c\"D\n\x0fReadDiskRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12$\n\tread_mode\x18\x02 \x01(\x0e\x32\x11.glutton.ReadMode\">\n\x10ReadDiskResponse\x12\x0c\n\x04size\x18\x01 \x01(\x03\x12\x0e\n\x06sha256\x18\x02 \x01(\x0c\x12\x0c\n\x04\x64\x61ta\x18\x03 \x01(\x0c\"\x1e\n\rOpenFDRequest\x12\r\n\x05\x63ount\x18\x01 \x01(\x05\"\x10\n\x0eOpenFDResponse\"\x1e\n\x0bPingRequest\x12\x0f\n\x07message\x18\x01 \x01(\t\"\x1f\n\x0cPingResponse\x12\x0f\n\x07message\x18\x01 \x01(\t\"-\n\rGossipRequest\x12\x1c\n\x05peers\x18\x01 \x03(\x0b\x32\r.glutton.Peer\"\x10\n\x0eGossipResponse\"&\n\x04Peer\x12\x0c\n\x04host\x18\x01 \x01(\t\x12\x10\n\x08\x64\x65lay_ms\x18\x02 \x01(\x05*>\n\tWriteMode\x12\x17\n\x13WRITE_MODE_TRUNCATE\x10\x00\x12\x18\n\x14WRITE_MODE_OVERWRITE\x10\x01*9\n\x08ReadMode\x12\x12\n\x0eREAD_MODE_DATA\x10\x00\x12\x19\n\x15READ_MODE_DIGEST_ONLY\x10\x01\x32\x86\x03\n\x07Glutton\x12\x41\n\x08WriteRAM\x12\x18.glutton.WriteRAMRequest\x1a\x19.glutton.WriteRAMResponse\"\x00\x12\x44\n\tWriteDisk\x12\x19.glutton.WriteDiskRequest\x1a\x1a.glutton.WriteDiskResponse\"\x00\x12\x41\n\x08ReadDisk\x12\x18.glutton.ReadDiskRequest\x1a\x19.glutton.ReadDiskResponse\"\x00\x12;\n\x06OpenFD\x12\x16.glutton.OpenFDRequest\x1a\x17.glutton.OpenFDResponse\"\x00\x12\x35\n\x04Ping\x12\x14.glutton.PingRequest\x1a\x15.glutton.PingResponse\"\x00\x12;\n\x06Gossip\x12\x16.glutton.GossipRequest\x1a\x17.glutton.GossipResponse\"\x00\x42=Z;github.com/agent-substrate/substrate/internal/proto/gluttonb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -46,8 +46,10 @@ if not _descriptor._USE_C_DESCRIPTORS: _globals['DESCRIPTOR']._loaded_options = None _globals['DESCRIPTOR']._serialized_options = b'Z;github.com/agent-substrate/substrate/internal/proto/glutton' - _globals['_WRITEMODE']._serialized_start=460 - _globals['_WRITEMODE']._serialized_end=522 + _globals['_WRITEMODE']._serialized_start=624 + _globals['_WRITEMODE']._serialized_end=686 + _globals['_READMODE']._serialized_start=688 + _globals['_READMODE']._serialized_end=745 _globals['_WRITERAMREQUEST']._serialized_start=26 _globals['_WRITERAMREQUEST']._serialized_end=110 _globals['_WRITERAMRESPONSE']._serialized_start=112 @@ -55,21 +57,25 @@ _globals['_WRITEDISKREQUEST']._serialized_start=132 _globals['_WRITEDISKREQUEST']._serialized_end=217 _globals['_WRITEDISKRESPONSE']._serialized_start=219 - _globals['_WRITEDISKRESPONSE']._serialized_end=238 - _globals['_OPENFDREQUEST']._serialized_start=240 - _globals['_OPENFDREQUEST']._serialized_end=270 - _globals['_OPENFDRESPONSE']._serialized_start=272 - _globals['_OPENFDRESPONSE']._serialized_end=288 - _globals['_PINGREQUEST']._serialized_start=290 - _globals['_PINGREQUEST']._serialized_end=320 - _globals['_PINGRESPONSE']._serialized_start=322 - _globals['_PINGRESPONSE']._serialized_end=353 - _globals['_GOSSIPREQUEST']._serialized_start=355 - _globals['_GOSSIPREQUEST']._serialized_end=400 - _globals['_GOSSIPRESPONSE']._serialized_start=402 - _globals['_GOSSIPRESPONSE']._serialized_end=418 - _globals['_PEER']._serialized_start=420 - _globals['_PEER']._serialized_end=458 - _globals['_GLUTTON']._serialized_start=525 - _globals['_GLUTTON']._serialized_end=848 + _globals['_WRITEDISKRESPONSE']._serialized_end=268 + _globals['_READDISKREQUEST']._serialized_start=270 + _globals['_READDISKREQUEST']._serialized_end=338 + _globals['_READDISKRESPONSE']._serialized_start=340 + _globals['_READDISKRESPONSE']._serialized_end=402 + _globals['_OPENFDREQUEST']._serialized_start=404 + _globals['_OPENFDREQUEST']._serialized_end=434 + _globals['_OPENFDRESPONSE']._serialized_start=436 + _globals['_OPENFDRESPONSE']._serialized_end=452 + _globals['_PINGREQUEST']._serialized_start=454 + _globals['_PINGREQUEST']._serialized_end=484 + _globals['_PINGRESPONSE']._serialized_start=486 + _globals['_PINGRESPONSE']._serialized_end=517 + _globals['_GOSSIPREQUEST']._serialized_start=519 + _globals['_GOSSIPREQUEST']._serialized_end=564 + _globals['_GOSSIPRESPONSE']._serialized_start=566 + _globals['_GOSSIPRESPONSE']._serialized_end=582 + _globals['_PEER']._serialized_start=584 + _globals['_PEER']._serialized_end=622 + _globals['_GLUTTON']._serialized_start=748 + _globals['_GLUTTON']._serialized_end=1138 # @@protoc_insertion_point(module_scope) diff --git a/benchmarking/locust/common/glutton_pb2_grpc.py b/benchmarking/locust/common/glutton_pb2_grpc.py index 351017c38..57a7fae6d 100644 --- a/benchmarking/locust/common/glutton_pb2_grpc.py +++ b/benchmarking/locust/common/glutton_pb2_grpc.py @@ -60,6 +60,11 @@ def __init__(self, channel): request_serializer=glutton__pb2.WriteDiskRequest.SerializeToString, response_deserializer=glutton__pb2.WriteDiskResponse.FromString, _registered_method=True) + self.ReadDisk = channel.unary_unary( + '/glutton.Glutton/ReadDisk', + request_serializer=glutton__pb2.ReadDiskRequest.SerializeToString, + response_deserializer=glutton__pb2.ReadDiskResponse.FromString, + _registered_method=True) self.OpenFD = channel.unary_unary( '/glutton.Glutton/OpenFD', request_serializer=glutton__pb2.OpenFDRequest.SerializeToString, @@ -99,6 +104,13 @@ def WriteDisk(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def ReadDisk(self, request, context): + """Tells glutton to read from disk using the specified mode. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def OpenFD(self, request, context): """Tells glutton to make sure it has the specified number of file descriptors open. It will open or close file descriptors to @@ -137,6 +149,11 @@ def add_GluttonServicer_to_server(servicer, server): request_deserializer=glutton__pb2.WriteDiskRequest.FromString, response_serializer=glutton__pb2.WriteDiskResponse.SerializeToString, ), + 'ReadDisk': grpc.unary_unary_rpc_method_handler( + servicer.ReadDisk, + request_deserializer=glutton__pb2.ReadDiskRequest.FromString, + response_serializer=glutton__pb2.ReadDiskResponse.SerializeToString, + ), 'OpenFD': grpc.unary_unary_rpc_method_handler( servicer.OpenFD, request_deserializer=glutton__pb2.OpenFDRequest.FromString, @@ -219,6 +236,33 @@ def WriteDisk(request, metadata, _registered_method=True) + @staticmethod + def ReadDisk(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/glutton.Glutton/ReadDisk', + glutton__pb2.ReadDiskRequest.SerializeToString, + glutton__pb2.ReadDiskResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + @staticmethod def OpenFD(request, target, diff --git a/benchmarking/locust/common/resume_mode.py b/benchmarking/locust/common/resume_mode.py new file mode 100644 index 000000000..7e6d3ab45 --- /dev/null +++ b/benchmarking/locust/common/resume_mode.py @@ -0,0 +1,31 @@ +# Copyright 2026 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 +# +# http://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. + +"""Locust custom arguments for the actor resume mode.""" + +from locust import events +from locust.argument_parser import LocustArgumentParser + + +@events.init_command_line_parser.add_listener +def add_resume_mode_arguments(parser: LocustArgumentParser) -> None: + group = parser.add_argument_group("Resume Mode") + group.add_argument( + "--resume-mode", + type=str, + default="explicit", + choices=["explicit", "implicit"], + help="Resume mode: 'explicit' issues ResumeActor RPC before sending traffic; " + "'implicit' sends traffic directly and lets the router wake the actor (default: explicit)", + ) diff --git a/benchmarking/locust/deploy.sh b/benchmarking/locust/deploy.sh index fe92b8cb6..37df81415 100755 --- a/benchmarking/locust/deploy.sh +++ b/benchmarking/locust/deploy.sh @@ -29,15 +29,20 @@ if [ -z "${PROJECT_ID:-}" ]; then exit 1 fi -MANIFEST="benchmarking/locust/manifests/locust.yaml" +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +MANIFEST="${SCRIPT_DIR}/manifests/locust.yaml" + +# Substituted into the boomer container's --user-class argument and the master's -f. +BENCHMARK_USER_CLASS=glutton usage() { echo "Usage: $0 [options]" echo "" echo "Options:" - echo " --deploy Deploy the locust workers" - echo " --delete Delete the locust workers" - echo " -h|--help Show this help message" + echo " --deploy Deploy the locust workers" + echo " --delete Delete the locust workers" + echo " --user-class NAME Locust user class, lowercase; runs tests/NAME.py (default: glutton)" + echo " -h|--help Show this help message" } deploy() { @@ -46,7 +51,7 @@ deploy() { # benchmarking/monitoring.yaml is otherwise optional. echo "Ensuring benchmarking namespace exists..." kubectl create namespace benchmarking --dry-run=client -o yaml | kubectl apply -f - - echo "Deploying Locust load (PROJECT_ID=${PROJECT_ID})..." + echo "Deploying Locust load (PROJECT_ID=${PROJECT_ID}, user_class=${BENCHMARK_USER_CLASS})..." envsubst < "${MANIFEST}" | kubectl apply -f - } @@ -65,6 +70,8 @@ while [[ "$#" -gt 0 ]]; do case "$1" in --deploy) action="deploy" ;; --delete) action="delete" ;; + --user-class) shift; BENCHMARK_USER_CLASS="$(printf '%s' "$1" | tr '[:upper:]' '[:lower:]')" ;; + --user-class=*) BENCHMARK_USER_CLASS="$(printf '%s' "${1#*=}" | tr '[:upper:]' '[:lower:]')" ;; -h|--help) usage; exit 0 ;; *) echo "Error: Unknown option: $1" >&2 @@ -75,6 +82,12 @@ while [[ "$#" -gt 0 ]]; do shift done +if [[ ! -f "${SCRIPT_DIR}/tests/${BENCHMARK_USER_CLASS}.py" ]]; then + echo "Error: no tests/${BENCHMARK_USER_CLASS}.py; --user-class must name a test file" >&2 + exit 1 +fi +export BENCHMARK_USER_CLASS + if [[ "${action}" == "deploy" ]]; then deploy elif [[ "${action}" == "delete" ]]; then diff --git a/benchmarking/locust/manifests/locust.yaml b/benchmarking/locust/manifests/locust.yaml index 101f4aa0e..de2e1da26 100644 --- a/benchmarking/locust/manifests/locust.yaml +++ b/benchmarking/locust/manifests/locust.yaml @@ -20,7 +20,7 @@ # LOCUST_NO_GLUTTON_USER=1 prevents tests/glutton.py from declaring the # stub User on this worker so GluttonUser spawns are owned exclusively # by the boomer container. -# * boomer-glutton: Go re-implementation of GluttonUser. Connects to the +# * boomer-glutton: Go re-implementation of GluttonUser and DurdirUser. Connects to the # master at localhost:5557 via the locust worker ZMQ protocol; exposes # its own per-worker diagnostics at /metrics on :8001 (aggregate stats # flow through the master via boomer.RecordSuccess). @@ -52,7 +52,7 @@ spec: - "-m" - "locust" - "-f" - - "/app/tests/glutton.py" + - "/app/tests/${BENCHMARK_USER_CLASS}.py" - "--master" - "--master-bind-host=0.0.0.0" env: @@ -134,6 +134,7 @@ spec: # are honored. The endpoint is served by # benchmarking/locust/common/boomer_config.py on the master. - "--master-web-port=8089" + - "--user-class=${BENCHMARK_USER_CLASS}" env: - name: OTEL_EXPORTER_OTLP_ENDPOINT value: opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317 diff --git a/benchmarking/locust/runner.py b/benchmarking/locust/runner.py index 52d24a282..5d84d58c0 100644 --- a/benchmarking/locust/runner.py +++ b/benchmarking/locust/runner.py @@ -80,10 +80,20 @@ def parse_args() -> argparse.Namespace: return args +# Tests whose User class is implemented in Python. Closed set: new load +# generators are written in boomer, so anything else runs on boomer. Remove +# entries as these are retired; when empty, drop needs_boomer entirely. +PYTHON_TESTS = frozenset({ + "ate_api.py", + "counter_demo.py", + "sleep.py", + "usermem.py", + "kernelmem.py", +}) + + def needs_boomer(test_file: str) -> bool: - """Return True if the test file is the glutton stub; the real GluttonUser - implementation lives in the boomer-glutton binary.""" - return os.path.basename(test_file) == "glutton.py" + return os.path.basename(test_file) not in PYTHON_TESTS def tee(logs: TextIO, msg: str) -> None: @@ -208,7 +218,7 @@ def run_test(args: argparse.Namespace, csv_prefix: Path, logs: TextIO, traces: T boomer_proc = None if with_boomer: - boomer_cmd = [BOOMER_BINARY] + boomer_cmd = [BOOMER_BINARY, "--user-class", Path(args.file).stem] cfg_json = build_config_json(args.locust_extra) if cfg_json: boomer_cmd += ["--config-json", cfg_json] diff --git a/benchmarking/locust/tests/durdir.py b/benchmarking/locust/tests/durdir.py new file mode 100644 index 000000000..c97285d38 --- /dev/null +++ b/benchmarking/locust/tests/durdir.py @@ -0,0 +1,46 @@ +# Copyright 2026 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 +# +# http://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. + +"""Stub DurdirUser declaration. + +The real load implementation lives in the boomer-Go worker at +cmd/benchmarking/boomer-glutton/; this Python class is declared only so the +master recognizes the name and attributes boomer's stats rows to it. The +master loads this stub file (or glutton.py), selected by +${BENCHMARK_USER_CLASS} in locust/manifests/locust.yaml. The Python worker +container sets LOCUST_NO_DURDIR_USER=1 to skip loading this file, leaving +boomer as the sole owner of DurdirUser load. +""" + +import os + +if os.environ.get("LOCUST_NO_DURDIR_USER") != "1": + from locust import User, task + from common.boomer_config import init_boomer_config + + # Master serves /boomer-config so the boomer-glutton workers can fetch + # runtime flag values (trace probability, wait times, durdir config) the + # operator set in the web UI form. No-op on workers without a web UI. + init_boomer_config() + + class DurdirUser(User): + host = "api.ate-system.svc.cluster.local:443" + + @task + def noop(self) -> None: + # Unreached under normal operation: the Python worker container + # does not load this file (LOCUST_NO_DURDIR_USER=1). Body is + # required because locust validates that every User has at least + # one @task method. + pass diff --git a/benchmarking/workloads/manifests/workloads.yaml.tmpl b/benchmarking/workloads/manifests/workloads.yaml.tmpl index 989b9597b..2200649a4 100644 --- a/benchmarking/workloads/manifests/workloads.yaml.tmpl +++ b/benchmarking/workloads/manifests/workloads.yaml.tmpl @@ -97,3 +97,81 @@ spec: workload: benchmark-ateom snapshotsConfig: location: gs://${BUCKET_NAME}/benchmark-workloads/glutton/ + +--- + +apiVersion: ate.dev/v1alpha1 +kind: ActorTemplate +metadata: + name: glutton-durdir-data + namespace: benchmark-workloads +spec: + # Must match the WorkerPool's sandboxClass so snapshots stay within a class. + sandboxClass: ${SANDBOX_CLASS} + containers: + - name: glutton + image: ko://github.com/agent-substrate/substrate/cmd/benchmarking/glutton + command: + - "/ko-app/glutton" + - "--grpc-listen-addr=:80" + - "--metrics-listen-addr=:9090" + - "--data-dir=/var/lib/glutton" + - "--mode=http" + # Data-scope resume is a cold boot; without this the router can reach the + # sandbox before glutton is listening. + readyz: + httpGet: + path: /readyz + port: 80 + volumeMounts: + - name: data + mountPath: /var/lib/glutton + volumes: + - name: data + durableDir: {} + workerSelector: + matchLabels: + workload: benchmark-ateom + snapshotsConfig: + onPause: Full + onCommit: Data # suspend uploads ONLY the durable dir + onResume: + fromData: ColdBoot + location: gs://${BUCKET_NAME}/benchmark-workloads/glutton-durdir-data/ + +--- + +apiVersion: ate.dev/v1alpha1 +kind: ActorTemplate +metadata: + name: glutton-durdir-full + namespace: benchmark-workloads +spec: + # Must match the WorkerPool's sandboxClass so snapshots stay within a class. + sandboxClass: ${SANDBOX_CLASS} + containers: + - name: glutton + image: ko://github.com/agent-substrate/substrate/cmd/benchmarking/glutton + command: + - "/ko-app/glutton" + - "--grpc-listen-addr=:80" + - "--metrics-listen-addr=:9090" + - "--data-dir=/var/lib/glutton" + - "--mode=http" + readyz: + httpGet: + path: /readyz + port: 80 + volumeMounts: + - name: data + mountPath: /var/lib/glutton + volumes: + - name: data + durableDir: {} + workerSelector: + matchLabels: + workload: benchmark-ateom + snapshotsConfig: + onPause: Full + onCommit: Full + location: gs://${BUCKET_NAME}/benchmark-workloads/glutton-durdir-full/ diff --git a/cmd/benchmarking/boomer-glutton/main.go b/cmd/benchmarking/boomer-glutton/main.go index f341b5cd5..65683dd9d 100644 --- a/cmd/benchmarking/boomer-glutton/main.go +++ b/cmd/benchmarking/boomer-glutton/main.go @@ -25,12 +25,14 @@ import ( "log/slog" "net/http" "os" + "strings" "time" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/glutton" bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" btrace "github.com/agent-substrate/substrate/internal/benchmarking/boomer/trace" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" "github.com/myzhan/boomer" ) @@ -40,14 +42,17 @@ func main() { routerURL = flag.String("router-url", "http://atenet-router.ate-system.svc.cluster.local", "atenet HTTP router base URL (no trailing slash).") atespace = flag.String("atespace", "benchmark", "Atespace every actor this worker creates lives in. Ensured (CreateAtespace, AlreadyExists is ok) at startup.") promAddr = flag.String("prometheus-addr", ":8001", "Address for the Prometheus /metrics endpoint.") - configJSON = flag.String("config-json", "", "Initial dynconfig as a JSON object (keys: trace_probability, min_wait_time, max_wait_time in seconds). Unset fields keep their built-in defaults.") + configJSON = flag.String("config-json", "", "Initial dynconfig as a JSON object (keys: trace_probability, min_wait_time, max_wait_time in seconds, durdir_file_size_bytes, resume_mode, durdir_read_mode, durdir_template). Unset fields keep their built-in defaults.") masterWebPort = flag.Int("master-web-port", 0, "If non-zero, fetch dynconfig from http://{master-host}:{master-web-port}/boomer-config on each spawn message and fail fatally on error. {master-host} comes from boomer's existing --master-host flag.") useTokenAuth = flag.Bool("use-token-auth", false, "Use Kubernetes ServiceAccount token for ateapi auth instead of client certificate.") + userClass = flag.String("user-class", "glutton", fmt.Sprintf("Locust user class to run, lowercase; one of %s.", strings.Join(userclass.Names(), "|"))) ) // boomer.Run will call flag.Parse() if we haven't yet; calling here so // our flag-derived values are usable before that. flag.Parse() + class := strings.ToLower(*userClass) + slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil))) initialCfg, err := dynconfig.Parse([]byte(*configJSON), dynconfig.Config{MaxWait: 500 * time.Millisecond}) @@ -95,14 +100,27 @@ func main() { slog.Info("dynconfig fetch enabled", slog.String("url", configURL)) } - cfg := &glutton.Config{ + cfg := &userclass.Config{ APIStub: apiStub, HTTPClient: httpClient, RouterURL: *routerURL, Atespace: *atespace, Dyn: dyn, } - taskFn, shutdownFn := glutton.Register(cfg) + + entry, ok := userclass.Lookup(class) + if !ok { + slog.Error("fatal: unknown --user-class value", + slog.String("user_class", *userClass), + slog.String("known", strings.Join(userclass.Names(), ","))) + os.Exit(1) + } + taskFn, shutdownFn := entry.Init(cfg) + + slog.Info("registered boomer task", + slog.String("user_class", entry.UserClass), + slog.String("user_class_flag", class), + ) metricsCtx, metricsCancel := context.WithCancel(context.Background()) defer metricsCancel() @@ -115,7 +133,7 @@ func main() { // Blocks until SIGINT/SIGTERM or master quit. Boomer registers its own // signal handlers; we do cleanup after it returns. boomer.Run(&boomer.Task{ - Name: "GluttonUser", + Name: entry.UserClass, Weight: 1, Fn: taskFn, }) diff --git a/cmd/benchmarking/glutton/main.go b/cmd/benchmarking/glutton/main.go index 5dc0f697a..afc70547e 100644 --- a/cmd/benchmarking/glutton/main.go +++ b/cmd/benchmarking/glutton/main.go @@ -20,6 +20,7 @@ package main import ( "context" "crypto/rand" + "crypto/sha256" "errors" "fmt" "io" @@ -29,6 +30,7 @@ import ( "os" "path/filepath" "regexp" + "strconv" "strings" "sync" "time" @@ -47,6 +49,7 @@ import ( "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/proto/glutton" "github.com/agent-substrate/substrate/internal/serverboot" "github.com/agent-substrate/substrate/internal/version" @@ -119,14 +122,6 @@ func main() { slog.String("mode", *mode), ) - // ateom blocks RestoreWorkload until /readyz returns 200, so ResumeActor - // cannot report success before this listener is reachable. The probe is - // an HTTP GET, so both modes must serve it. - mux := http.NewServeMux() - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - var handler http.Handler switch *mode { case "grpc": @@ -135,17 +130,14 @@ func main() { ) glutton.RegisterGluttonServer(srv, svc) reflection.Register(srv) - handler = splitGRPC(srv, mux) + // The readiness probe is an HTTP GET, so gRPC mode serves it next to + // the gRPC handler on the same listener. + handler = splitGRPC(srv, readyzMux()) case "http": - // HTTP/1.1 mode: a single /ping route that consumes - // proto.Marshal(PingRequest) and returns proto.Marshal(PingResponse). - // Only Ping is exposed in HTTP mode; the other RPCs remain gRPC-only - // (re-exposable as additional routes if/when needed). - mux.HandleFunc("/ping", httpPingHandler(svc)) // otelhttp at the mux level + per-handler span follows // docs/dev/best-practices/tracing.md: extract incoming context, // then name the span after the operation in each handler. - handler = otelhttp.NewHandler(mux, "/") + handler = otelhttp.NewHandler(newMux(svc), "/") default: serverboot.Fatal(ctx, "Invalid --mode", fmt.Errorf("must be grpc or http: %q", *mode)) } @@ -176,11 +168,34 @@ func splitGRPC(grpcSrv, rest http.Handler) http.Handler { }) } -// httpPingHandler accepts a POST whose body is proto.Marshal(PingRequest) and -// returns proto.Marshal(PingResponse) (same Ping handler the gRPC server -// uses, so the per-call stats stay comparable across protocols). -func httpPingHandler(svc *gluttonService) http.HandlerFunc { +// readyzMux serves the readiness probe both modes need: ateom blocks +// RestoreWorkload until /readyz returns 200, so ResumeActor cannot report +// success before this listener is reachable. +func readyzMux() *http.ServeMux { + mux := http.NewServeMux() + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + return mux +} + +// newMux builds the HTTP-mode route table on top of the readiness probe. +func newMux(svc *gluttonService) *http.ServeMux { + mux := readyzMux() + mux.HandleFunc("/ping", protoRoute("Ping", svc.Ping)) + mux.HandleFunc("/writedisk", protoRoute("WriteDisk", svc.WriteDisk)) + mux.HandleFunc("/readdisk", protoRoute("ReadDisk", svc.ReadDisk)) + return mux +} + +// protoRoute wraps a protobuf handler with POST-only routing, protobuf +// unmarshaling, status code mapping, and server-timing headers. +func protoRoute[Req any, Resp proto.Message, PtrReq interface { + *Req + proto.Message +}](spanName string, handler func(context.Context, PtrReq) (Resp, error)) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { + start := time.Now() if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return @@ -190,16 +205,28 @@ func httpPingHandler(svc *gluttonService) http.HandlerFunc { http.Error(w, err.Error(), http.StatusBadRequest) return } - var req glutton.PingRequest - if err := proto.Unmarshal(body, &req); err != nil { + var req Req + ptrReq := PtrReq(&req) + if err := proto.Unmarshal(body, ptrReq); err != nil { http.Error(w, "unmarshal: "+err.Error(), http.StatusBadRequest) return } - ctx, span := otel.Tracer("glutton").Start(r.Context(), "Ping") + ctx, span := otel.Tracer("glutton").Start(r.Context(), spanName) defer span.End() - resp, err := svc.Ping(ctx, &req) + resp, err := handler(ctx, ptrReq) if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if st, ok := status.FromError(err); ok { + switch st.Code() { + case codes.InvalidArgument: + http.Error(w, st.Message(), http.StatusBadRequest) + case codes.NotFound: + http.Error(w, st.Message(), http.StatusNotFound) + default: + http.Error(w, st.Message(), http.StatusInternalServerError) + } + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } return } out, err := proto.Marshal(resp) @@ -207,6 +234,11 @@ func httpPingHandler(svc *gluttonService) http.HandlerFunc { http.Error(w, err.Error(), http.StatusInternalServerError) return } + // Glutton does not run ateinterceptors, so without this the serve path has no + // server-side timing at all. Mirrors the control-plane gRPC trailer so boomer's + // elapsedFromMD logic (source=server) works identically over HTTP. + w.Header().Set(ateinterceptors.ServerElapsedTrailer, + strconv.FormatInt(time.Since(start).Microseconds(), 10)) w.Header().Set("Content-Type", "application/x-protobuf") _, _ = w.Write(out) } @@ -230,6 +262,7 @@ type gluttonService struct { ramWriteBytes metric.Int64Counter diskWriteBytes metric.Int64Counter + diskReadBytes metric.Int64Counter pingsReceived metric.Int64Counter gossipSent metric.Int64Counter gossipLatency metric.Float64Histogram @@ -268,6 +301,14 @@ func newGluttonService(dir string) (*gluttonService, error) { if err != nil { return nil, fmt.Errorf("create glutton.disk.write.bytes counter: %w", err) } + s.diskReadBytes, err = m.Int64Counter( + "glutton.disk.read.bytes", + metric.WithUnit("By"), + metric.WithDescription("Total bytes read from disk via ReadDisk over the process lifetime."), + ) + if err != nil { + return nil, fmt.Errorf("create glutton.disk.read.bytes counter: %w", err) + } s.pingsReceived, err = m.Int64Counter( "glutton.ping.requests", metric.WithDescription("Number of Ping requests received."), @@ -391,10 +432,10 @@ func (s *gluttonService) WriteDisk(ctx context.Context, req *glutton.WriteDiskRe var flag int switch req.GetWriteMode() { case glutton.WriteMode_WRITE_MODE_TRUNCATE: - flag = os.O_WRONLY | os.O_CREATE | os.O_TRUNC + flag = os.O_RDWR | os.O_CREATE | os.O_TRUNC case glutton.WriteMode_WRITE_MODE_OVERWRITE: // No O_TRUNC: writes go from offset 0 but any bytes beyond size remain. - flag = os.O_WRONLY | os.O_CREATE + flag = os.O_RDWR | os.O_CREATE default: return nil, status.Errorf(codes.InvalidArgument, "unknown write_mode %v", req.GetWriteMode()) } @@ -405,12 +446,68 @@ func (s *gluttonService) WriteDisk(ctx context.Context, req *glutton.WriteDiskRe } defer f.Close() - if err := streamRandomBytes(f, int64(req.GetSize())); err != nil { + h := sha256.New() + size := int64(req.GetSize()) + if err := streamRandomBytes(io.MultiWriter(f, h), size); err != nil { return nil, status.Errorf(codes.Internal, "write %s: %v", path, err) } + // OVERWRITE has no O_TRUNC, bytes from a larger, earlier write will persist. + // The cursor is already at size, so folding the remainder into the + // same digest completes it without re-reading the prefix. + if req.GetWriteMode() == glutton.WriteMode_WRITE_MODE_OVERWRITE { + tail, err := io.Copy(h, f) + if err != nil { + return nil, status.Errorf(codes.Internal, "hash tail %s: %v", path, err) + } + size += tail + } + s.diskWriteBytes.Add(ctx, int64(req.GetSize())) - return &glutton.WriteDiskResponse{}, nil + return &glutton.WriteDiskResponse{Size: size, Sha256: h.Sum(nil)}, nil +} + +func (s *gluttonService) ReadDisk(ctx context.Context, req *glutton.ReadDiskRequest) (*glutton.ReadDiskResponse, error) { + if !diskKeyRE.MatchString(req.GetKey()) { + return nil, status.Errorf(codes.InvalidArgument, "key %q must match %s", req.GetKey(), diskKeyRE) + } + + path := filepath.Join(s.dataDir, req.GetKey()) + + f, err := os.Open(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, status.Errorf(codes.NotFound, "file %q not found", req.GetKey()) + } + return nil, status.Errorf(codes.Internal, "open %s: %v", path, err) + } + defer f.Close() + + h := sha256.New() + + if req.GetReadMode() == glutton.ReadMode_READ_MODE_DIGEST_ONLY { + n, err := io.Copy(h, f) + if err != nil { + return nil, status.Errorf(codes.Internal, "read %s: %v", path, err) + } + s.diskReadBytes.Add(ctx, n) + return &glutton.ReadDiskResponse{ + Size: n, + Sha256: h.Sum(nil), + }, nil + } + + data, err := io.ReadAll(io.TeeReader(f, h)) + if err != nil { + return nil, status.Errorf(codes.Internal, "read %s: %v", path, err) + } + + s.diskReadBytes.Add(ctx, int64(len(data))) + return &glutton.ReadDiskResponse{ + Size: int64(len(data)), + Sha256: h.Sum(nil), + Data: data, + }, nil } // Make sure it has the specified number of file descriptors open. It will open or diff --git a/cmd/benchmarking/glutton/main_test.go b/cmd/benchmarking/glutton/main_test.go index 354770d64..34d8db271 100644 --- a/cmd/benchmarking/glutton/main_test.go +++ b/cmd/benchmarking/glutton/main_test.go @@ -15,16 +15,25 @@ package main import ( + "bytes" "context" + "crypto/sha256" + "io" "net" "net/http" + "net/http/httptest" + "os" + "path/filepath" "testing" "time" + "github.com/agent-substrate/substrate/internal/ateinterceptors" + "github.com/agent-substrate/substrate/internal/proto/glutton" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" - - "github.com/agent-substrate/substrate/internal/proto/glutton" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) // TestSplitGRPCServesReadyzAndGRPCOnOneListener starts the grpc-mode handler @@ -111,3 +120,334 @@ func TestSplitGRPCRoutesOnContentType(t *testing.T) { t.Error("HTTP/1.1 GET reached the gRPC handler") } } + +func TestWriteDiskReadDiskRoundTrip(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ctx := context.Background() + tests := []struct { + name string + key string + size int32 + }{ + {name: "zero size", key: "zero", size: 0}, + {name: "small size", key: "small", size: 1024}, + {name: "chunk unaligned size", key: "unaligned", size: (1 << 20) + 1}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + writeResp, err := svc.WriteDisk(ctx, &glutton.WriteDiskRequest{ + Key: tt.key, + Size: tt.size, + WriteMode: glutton.WriteMode_WRITE_MODE_TRUNCATE, + }) + if err != nil { + t.Fatalf("WriteDisk failed: %v", err) + } + if writeResp.GetSize() != int64(tt.size) { + t.Errorf("WriteDisk size mismatch: got %d, want %d", writeResp.GetSize(), tt.size) + } + + // 1. Full data read + readResp, err := svc.ReadDisk(ctx, &glutton.ReadDiskRequest{ + Key: tt.key, + ReadMode: glutton.ReadMode_READ_MODE_DATA, + }) + if err != nil { + t.Fatalf("ReadDisk (DATA) failed: %v", err) + } + + if readResp.GetSize() != int64(tt.size) { + t.Errorf("ReadDisk size mismatch: got %d, want %d", readResp.GetSize(), tt.size) + } + if !bytes.Equal(readResp.GetSha256(), writeResp.GetSha256()) { + t.Errorf("sha256 mismatch between WriteDisk and ReadDisk") + } + if len(readResp.GetData()) != int(tt.size) { + t.Errorf("ReadDisk data length mismatch: got %d, want %d", len(readResp.GetData()), tt.size) + } + + computedDigest := sha256.Sum256(readResp.GetData()) + if !bytes.Equal(readResp.GetSha256(), computedDigest[:]) { + t.Errorf("ReadDisk returned sha256 does not match computed digest of returned data") + } + + // 2. Digest-only read + digestResp, err := svc.ReadDisk(ctx, &glutton.ReadDiskRequest{ + Key: tt.key, + ReadMode: glutton.ReadMode_READ_MODE_DIGEST_ONLY, + }) + if err != nil { + t.Fatalf("ReadDisk (DIGEST_ONLY) failed: %v", err) + } + if digestResp.GetSize() != int64(tt.size) { + t.Errorf("ReadDisk (DIGEST_ONLY) size mismatch: got %d, want %d", digestResp.GetSize(), tt.size) + } + if !bytes.Equal(digestResp.GetSha256(), writeResp.GetSha256()) { + t.Errorf("sha256 mismatch between WriteDisk and ReadDisk (DIGEST_ONLY)") + } + if len(digestResp.GetData()) != 0 { + t.Errorf("ReadDisk (DIGEST_ONLY) should not return data payload, got %d bytes", len(digestResp.GetData())) + } + }) + } +} + +func TestWriteDiskTruncateProducesExactSize(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ctx := context.Background() + key := "testfile" + size := int32(2048) + + _, err = svc.WriteDisk(ctx, &glutton.WriteDiskRequest{ + Key: key, + Size: size, + WriteMode: glutton.WriteMode_WRITE_MODE_TRUNCATE, + }) + if err != nil { + t.Fatalf("WriteDisk failed: %v", err) + } + + filePath := filepath.Join(tempDir, key) + fi, err := os.Stat(filePath) + if err != nil { + t.Fatalf("os.Stat failed: %v", err) + } + if fi.Size() != int64(size) { + t.Errorf("file size on disk mismatch: got %d, want %d", fi.Size(), size) + } +} + +func TestWriteDiskOverwriteDigestMatchesReadDisk(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ctx := context.Background() + key := "overwrittenfile" + + // 1. Initial write of large file (4096 bytes) + _, err = svc.WriteDisk(ctx, &glutton.WriteDiskRequest{ + Key: key, + Size: 4096, + WriteMode: glutton.WriteMode_WRITE_MODE_TRUNCATE, + }) + if err != nil { + t.Fatalf("WriteDisk (large) failed: %v", err) + } + + // 2. Overwrite prefix with smaller size (1024 bytes) without truncation + overwriteResp, err := svc.WriteDisk(ctx, &glutton.WriteDiskRequest{ + Key: key, + Size: 1024, + WriteMode: glutton.WriteMode_WRITE_MODE_OVERWRITE, + }) + if err != nil { + t.Fatalf("WriteDisk (overwrite) failed: %v", err) + } + + if overwriteResp.GetSize() != 4096 { + t.Errorf("expected WriteDisk under OVERWRITE to report total file size 4096, got %d", overwriteResp.GetSize()) + } + + // 3. ReadDisk reads the entire file (4096 bytes) + readResp, err := svc.ReadDisk(ctx, &glutton.ReadDiskRequest{ + Key: key, + ReadMode: glutton.ReadMode_READ_MODE_DATA, + }) + if err != nil { + t.Fatalf("ReadDisk failed: %v", err) + } + + if readResp.GetSize() != 4096 { + t.Errorf("expected ReadDisk size 4096, got %d", readResp.GetSize()) + } + if !bytes.Equal(readResp.GetSha256(), overwriteResp.GetSha256()) { + t.Errorf("expected WriteDisk(OVERWRITE) whole-file digest to match ReadDisk digest") + } +} + +func TestReadDiskRejectsInvalidKey(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ctx := context.Background() + _, err = svc.ReadDisk(ctx, &glutton.ReadDiskRequest{Key: "../escape"}) + if err == nil { + t.Error("expected error for invalid key with path traversal, got nil") + } + if s, ok := status.FromError(err); !ok || s.Code() != codes.InvalidArgument { + t.Errorf("expected InvalidArgument code, got %v", err) + } +} + +func TestReadDiskNotFound(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ctx := context.Background() + _, err = svc.ReadDisk(ctx, &glutton.ReadDiskRequest{Key: "nonexistent"}) + if err == nil { + t.Error("expected error for nonexistent file, got nil") + } + if s, ok := status.FromError(err); !ok || s.Code() != codes.NotFound { + t.Errorf("expected NotFound code, got %v", err) + } +} + +func TestHTTPRoutes(t *testing.T) { + tempDir := t.TempDir() + svc, err := newGluttonService(tempDir) + if err != nil { + t.Fatalf("failed to create glutton service: %v", err) + } + defer svc.Close() + + ts := httptest.NewServer(newMux(svc)) + defer ts.Close() + + // 1. /readyz GET -> 200 OK + res, err := http.Get(ts.URL + "/readyz") + if err != nil { + t.Fatalf("GET /readyz failed: %v", err) + } + if res.StatusCode != http.StatusOK { + t.Errorf("GET /readyz status: got %d, want 200", res.StatusCode) + } + res.Body.Close() + + // 2. GET on /ping -> 405 Method Not Allowed + res, err = http.Get(ts.URL + "/ping") + if err != nil { + t.Fatalf("GET /ping failed: %v", err) + } + if res.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("GET /ping status: got %d, want 405", res.StatusCode) + } + res.Body.Close() + + // 3. POST bad body -> 400 Bad Request + res, err = http.Post(ts.URL+"/ping", "application/x-protobuf", bytes.NewReader([]byte("garbage"))) + if err != nil { + t.Fatalf("POST /ping garbage failed: %v", err) + } + if res.StatusCode != http.StatusBadRequest { + t.Errorf("POST /ping garbage status: got %d, want 400", res.StatusCode) + } + res.Body.Close() + + // 4. POST /ping -> 200 OK & protobuf Content-Type & ServerElapsedTrailer & echo message + pingReqBytes, _ := proto.Marshal(&glutton.PingRequest{Message: "hello"}) + res, err = http.Post(ts.URL+"/ping", "application/x-protobuf", bytes.NewReader(pingReqBytes)) + if err != nil { + t.Fatalf("POST /ping failed: %v", err) + } + if res.StatusCode != http.StatusOK { + t.Errorf("POST /ping status: got %d, want 200", res.StatusCode) + } + if ct := res.Header.Get("Content-Type"); ct != "application/x-protobuf" { + t.Errorf("POST /ping Content-Type: got %q, want application/x-protobuf", ct) + } + if elapsed := res.Header.Get(ateinterceptors.ServerElapsedTrailer); elapsed == "" { + t.Errorf("POST /ping missing header %q", ateinterceptors.ServerElapsedTrailer) + } + body, _ := io.ReadAll(res.Body) + res.Body.Close() + var pingResp glutton.PingResponse + if err := proto.Unmarshal(body, &pingResp); err != nil { + t.Fatalf("unmarshal PingResponse failed: %v", err) + } + if pingResp.GetMessage() != "hello" { + t.Errorf("PingResponse message: got %q, want 'hello'", pingResp.GetMessage()) + } + + // 5. POST /writedisk -> 200 OK & protobuf Content-Type + writeReqBytes, _ := proto.Marshal(&glutton.WriteDiskRequest{ + Key: "httpfile", + Size: 512, + WriteMode: glutton.WriteMode_WRITE_MODE_TRUNCATE, + }) + res, err = http.Post(ts.URL+"/writedisk", "application/x-protobuf", bytes.NewReader(writeReqBytes)) + if err != nil { + t.Fatalf("POST /writedisk failed: %v", err) + } + if res.StatusCode != http.StatusOK { + t.Errorf("POST /writedisk status: got %d, want 200", res.StatusCode) + } + body, _ = io.ReadAll(res.Body) + res.Body.Close() + var writeResp glutton.WriteDiskResponse + if err := proto.Unmarshal(body, &writeResp); err != nil { + t.Fatalf("unmarshal WriteDiskResponse failed: %v", err) + } + if writeResp.GetSize() != 512 { + t.Errorf("WriteDiskResponse size: got %d, want 512", writeResp.GetSize()) + } + + // 6. POST /readdisk -> 200 OK & matching size & digest + readReqBytes, _ := proto.Marshal(&glutton.ReadDiskRequest{Key: "httpfile"}) + res, err = http.Post(ts.URL+"/readdisk", "application/x-protobuf", bytes.NewReader(readReqBytes)) + if err != nil { + t.Fatalf("POST /readdisk failed: %v", err) + } + if res.StatusCode != http.StatusOK { + t.Errorf("POST /readdisk status: got %d, want 200", res.StatusCode) + } + body, _ = io.ReadAll(res.Body) + res.Body.Close() + var readResp glutton.ReadDiskResponse + if err := proto.Unmarshal(body, &readResp); err != nil { + t.Fatalf("unmarshal ReadDiskResponse failed: %v", err) + } + if readResp.GetSize() != 512 { + t.Errorf("ReadDiskResponse size: got %d, want 512", readResp.GetSize()) + } + if !bytes.Equal(readResp.GetSha256(), writeResp.GetSha256()) { + t.Errorf("sha256 mismatch over HTTP between writedisk and readdisk") + } + + // 7. unknown key -> 404 (NotFound mapping) + missBytes, _ := proto.Marshal(&glutton.ReadDiskRequest{Key: "nosuchfile"}) + res, err = http.Post(ts.URL+"/readdisk", "application/x-protobuf", bytes.NewReader(missBytes)) + if err != nil { + t.Fatalf("POST /readdisk miss failed: %v", err) + } + if res.StatusCode != http.StatusNotFound { + t.Errorf("POST /readdisk miss status: got %d, want 404", res.StatusCode) + } + res.Body.Close() + + // 8. traversal key -> 400 (InvalidArgument mapping) + badBytes, _ := proto.Marshal(&glutton.ReadDiskRequest{Key: "../etc/passwd"}) + res, err = http.Post(ts.URL+"/readdisk", "application/x-protobuf", bytes.NewReader(badBytes)) + if err != nil { + t.Fatalf("POST /readdisk bad key failed: %v", err) + } + if res.StatusCode != http.StatusBadRequest { + t.Errorf("POST /readdisk bad key status: got %d, want 400", res.StatusCode) + } + res.Body.Close() +} diff --git a/internal/benchmarking/boomer/dynconfig/dynconfig.go b/internal/benchmarking/boomer/dynconfig/dynconfig.go index 2e65f863e..5aa86218e 100644 --- a/internal/benchmarking/boomer/dynconfig/dynconfig.go +++ b/internal/benchmarking/boomer/dynconfig/dynconfig.go @@ -24,6 +24,7 @@ import ( "encoding/json" "fmt" "log/slog" + "math" "net/http" "sync/atomic" "time" @@ -31,12 +32,28 @@ import ( "github.com/myzhan/boomer" ) +// Resume modes. Explicit issues a ResumeActor RPC before sending traffic. +// Implicit issues no wake request at all: the actor stays suspended until a +// request reaches the atenet router, which wakes it while the request is +// parked. +const ( + ResumeModeExplicit = "explicit" + ResumeModeImplicit = "implicit" + + ReadModeData = "data" + ReadModeDigest = "digest" +) + // Config is the dynamic-mutable subset of boomer's behavior. Holder swaps // it atomically so task goroutines read a consistent snapshot. type Config struct { MinWait time.Duration MaxWait time.Duration TraceProbability float64 + DurDirFileSize int64 // bytes + ResumeMode string // ResumeModeExplicit | ResumeModeImplicit + DurDirReadMode string // ReadModeData | ReadModeDigest + DurDirTemplate string // ActorTemplate name } // Holder lets readers Load() the current Config and writers Store() a new @@ -70,6 +87,10 @@ type payload struct { TraceProbability *float64 `json:"trace_probability"` MinWaitTime *float64 `json:"min_wait_time"` MaxWaitTime *float64 `json:"max_wait_time"` + DurDirFileSize *float64 `json:"durdir_file_size_bytes"` + ResumeMode *string `json:"resume_mode"` + DurDirReadMode *string `json:"durdir_read_mode"` + DurDirTemplate *string `json:"durdir_template"` } // Parse decodes a JSON blob (typically from a CLI flag) and merges its @@ -83,7 +104,11 @@ func Parse(jsonBytes []byte, current Config) (Config, error) { if err := json.Unmarshal(jsonBytes, &p); err != nil { return current, fmt.Errorf("decode config json: %w", err) } - return p.merge(current), nil + merged := p.merge(current) + if err := merged.Validate(); err != nil { + return current, fmt.Errorf("validate config: %w", err) + } + return merged, nil } // Fetch GETs `url` and merges any returned fields into `current`. Returns @@ -105,7 +130,40 @@ func Fetch(ctx context.Context, url string, current Config) (Config, error) { if err := json.NewDecoder(resp.Body).Decode(&p); err != nil { return current, fmt.Errorf("decode %s: %w", url, err) } - return p.merge(current), nil + merged := p.merge(current) + if err := merged.Validate(); err != nil { + return current, fmt.Errorf("validate %s: %w", url, err) + } + return merged, nil +} + +// Validate checks that the config values are within legal ranges. +func (c Config) Validate() error { + if c.MinWait < 0 { + return fmt.Errorf("min_wait_time cannot be negative: %v", c.MinWait) + } + if c.MaxWait < 0 { + return fmt.Errorf("max_wait_time cannot be negative: %v", c.MaxWait) + } + if c.MaxWait < c.MinWait { + return fmt.Errorf("max_wait_time (%v) cannot be less than min_wait_time (%v)", c.MaxWait, c.MinWait) + } + if c.TraceProbability < 0 || c.TraceProbability > 1 { + return fmt.Errorf("trace_probability must be between 0.0 and 1.0, got: %f", c.TraceProbability) + } + if c.DurDirFileSize < 0 { + return fmt.Errorf("durdir_file_size_bytes cannot be negative: %d", c.DurDirFileSize) + } + if c.DurDirFileSize > math.MaxInt32 { + return fmt.Errorf("durdir_file_size_bytes cannot exceed %d (2 GiB), got: %d", math.MaxInt32, c.DurDirFileSize) + } + if c.ResumeMode != "" && c.ResumeMode != ResumeModeExplicit && c.ResumeMode != ResumeModeImplicit { + return fmt.Errorf("invalid resume_mode %q: must be %q or %q", c.ResumeMode, ResumeModeExplicit, ResumeModeImplicit) + } + if c.DurDirReadMode != "" && c.DurDirReadMode != ReadModeData && c.DurDirReadMode != ReadModeDigest { + return fmt.Errorf("invalid durdir_read_mode %q: must be %q or %q", c.DurDirReadMode, ReadModeData, ReadModeDigest) + } + return nil } // merge folds the payload's set fields into `current`, leaving unset fields @@ -122,6 +180,18 @@ func (p payload) merge(current Config) Config { if p.MaxWaitTime != nil { out.MaxWait = time.Duration(*p.MaxWaitTime * float64(time.Second)) } + if p.DurDirFileSize != nil { + out.DurDirFileSize = int64(*p.DurDirFileSize) + } + if p.ResumeMode != nil { + out.ResumeMode = *p.ResumeMode + } + if p.DurDirReadMode != nil { + out.DurDirReadMode = *p.DurDirReadMode + } + if p.DurDirTemplate != nil { + out.DurDirTemplate = *p.DurDirTemplate + } return out } @@ -148,6 +218,10 @@ func SubscribeSpawn(url string, holder *Holder, sampler ProbabilityUpdater, fetc slog.Float64("trace_probability", next.TraceProbability), slog.Duration("min_wait", next.MinWait), slog.Duration("max_wait", next.MaxWait), + slog.Int64("durdir_file_size_bytes", next.DurDirFileSize), + slog.String("resume_mode", next.ResumeMode), + slog.String("durdir_read_mode", next.DurDirReadMode), + slog.String("durdir_template", next.DurDirTemplate), ) }) } diff --git a/internal/benchmarking/boomer/dynconfig/dynconfig_test.go b/internal/benchmarking/boomer/dynconfig/dynconfig_test.go new file mode 100644 index 000000000..afcf29b86 --- /dev/null +++ b/internal/benchmarking/boomer/dynconfig/dynconfig_test.go @@ -0,0 +1,144 @@ +// Copyright 2026 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 +// +// http://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. + +package dynconfig + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestParseValid(t *testing.T) { + jsonBlob := []byte(`{ + "trace_probability": 0.5, + "min_wait_time": 0.1, + "max_wait_time": 0.5, + "durdir_file_size_bytes": 1048576, + "resume_mode": "explicit", + "durdir_read_mode": "data", + "durdir_template": "glutton-durdir-data" + }`) + + cfg, err := Parse(jsonBlob, Config{}) + if err != nil { + t.Fatalf("Parse failed: %v", err) + } + + if cfg.TraceProbability != 0.5 { + t.Errorf("TraceProbability: got %f, want 0.5", cfg.TraceProbability) + } + if cfg.MinWait != 100*time.Millisecond { + t.Errorf("MinWait: got %v, want 100ms", cfg.MinWait) + } + if cfg.MaxWait != 500*time.Millisecond { + t.Errorf("MaxWait: got %v, want 500ms", cfg.MaxWait) + } + if cfg.DurDirFileSize != 1048576 { + t.Errorf("DurDirFileSize: got %d, want 1048576", cfg.DurDirFileSize) + } + if cfg.ResumeMode != ResumeModeExplicit { + t.Errorf("ResumeMode: got %q, want %q", cfg.ResumeMode, ResumeModeExplicit) + } + if cfg.DurDirReadMode != ReadModeData { + t.Errorf("DurDirReadMode: got %q, want %q", cfg.DurDirReadMode, ReadModeData) + } + if cfg.DurDirTemplate != "glutton-durdir-data" { + t.Errorf("DurDirTemplate: got %q, want glutton-durdir-data", cfg.DurDirTemplate) + } +} + +func TestParseInvalidValues(t *testing.T) { + tests := []struct { + name string + json string + }{ + { + name: "negative trace probability", + json: `{"trace_probability": -0.1}`, + }, + { + name: "trace probability > 1.0", + json: `{"trace_probability": 1.5}`, + }, + { + name: "negative min wait", + json: `{"min_wait_time": -1.0}`, + }, + { + name: "negative max wait", + json: `{"max_wait_time": -1.0}`, + }, + { + name: "max wait less than min wait", + json: `{"min_wait_time": 2.0, "max_wait_time": 1.0}`, + }, + { + name: "negative file size", + json: `{"durdir_file_size_bytes": -100}`, + }, + { + name: "file size exceeds 2 GiB", + json: `{"durdir_file_size_bytes": 2147483648}`, + }, + { + name: "invalid resume mode", + json: `{"resume_mode": "invalid_mode"}`, + }, + { + name: "invalid read mode", + json: `{"durdir_read_mode": "invalid_read"}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := Parse([]byte(tt.json), Config{}) + if err == nil { + t.Errorf("expected Parse to fail for %s, got nil error", tt.name) + } + }) + } +} + +func TestFetchValidAndInvalid(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/valid", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"resume_mode": "implicit", "durdir_read_mode": "digest"}`)) + }) + mux.HandleFunc("/invalid", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"resume_mode": "bogus"}`)) + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + ctx := context.Background() + + cfg, err := Fetch(ctx, ts.URL+"/valid", Config{}) + if err != nil { + t.Fatalf("Fetch valid failed: %v", err) + } + if cfg.ResumeMode != ResumeModeImplicit || cfg.DurDirReadMode != ReadModeDigest { + t.Errorf("Fetch valid values mismatch: got %+v", cfg) + } + + _, err = Fetch(ctx, ts.URL+"/invalid", Config{}) + if err == nil { + t.Errorf("expected Fetch invalid to fail, got nil") + } +} diff --git a/internal/benchmarking/boomer/glutton/durdir.go b/internal/benchmarking/boomer/glutton/durdir.go new file mode 100644 index 000000000..9ddcd0ae8 --- /dev/null +++ b/internal/benchmarking/boomer/glutton/durdir.go @@ -0,0 +1,471 @@ +// Copyright 2026 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 +// +// http://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. + +package glutton + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "log/slog" + "math/rand/v2" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/agent-substrate/substrate/internal/ateinterceptors" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" + bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" + gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/uuid" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +const ( + // Locust class name from tests/durdir.py; must match boomer.Task.Name. + durDirUserClass = "DurdirUser" + defaultDurTemplate = "glutton-durdir-data" + + writeDiskRoute = "/writedisk" + readDiskRoute = "/readdisk" + + durDirTestFile = "bench-data" + + defaultFileSize int64 = 8388608 // 8 MiB +) + +func init() { + userclass.Add(userclass.Entry{ + Name: "durdir", + LocustFile: "durdir.py", + UserClass: durDirUserClass, + Init: initDurDir, + }) +} + +// initDurDir creates a runtime tied to cfg and returns a boomer-compatible task +// function plus a Shutdown hook the caller should run before exit. +func initDurDir(cfg *userclass.Config) (taskFn func(), shutdown func(context.Context)) { + if cfg.Tracer == nil { + cfg.Tracer = otel.Tracer("substrate-boomer/glutton-durdir") + } + rt := &durDirRuntime{cfg: cfg} + return rt.iterate, rt.shutdown +} + +type durDirRuntime struct { + cfg *userclass.Config + users sync.Map // goroutineID -> *durDirUser +} + +func (r *durDirRuntime) dynamicWait() time.Duration { + cfg := r.cfg.Dyn.Load() + if cfg.MaxWait <= cfg.MinWait { + return cfg.MinWait + } + jitter := cfg.MaxWait - cfg.MinWait + return cfg.MinWait + time.Duration(rand.Float64()*float64(jitter)) +} + +func (r *durDirRuntime) iterate() { + gid := goroutineID() + val, loaded := r.users.Load(gid) + if !loaded { + dynCfg := r.cfg.Dyn.Load() + u, err := r.startUser(context.Background(), dynCfg) + if err != nil { + slog.Warn("durdir on_start failed; goroutine will retry next iter", + slog.String("err", err.Error())) + time.Sleep(r.dynamicWait()) + return + } + val, _ = r.users.LoadOrStore(gid, u) + } + user := val.(*durDirUser) + + dynCfg := r.cfg.Dyn.Load() + ctx := context.Background() + user.step(ctx, dynCfg) + + time.Sleep(r.dynamicWait()) +} + +func (r *durDirRuntime) startUser(ctx context.Context, dynCfg dynconfig.Config) (*durDirUser, error) { + tmpl := dynCfg.DurDirTemplate + if tmpl == "" { + tmpl = defaultDurTemplate + } + + u := &durDirUser{ + cfg: r.cfg, + actorName: "sb-" + uuid.NewString(), + templateName: tmpl, + userClass: durDirUserClass, + } + u.hostHeader = u.actorName + "." + u.cfg.Atespace + "." + actorDomain + bmetrics.UpdateUsers(durDirUserClass, 1) + if err := u.ensureAtespace(ctx); err != nil { + bmetrics.UpdateUsers(durDirUserClass, -1) + return nil, err + } + if err := u.create(ctx); err != nil { + bmetrics.UpdateUsers(durDirUserClass, -1) + return nil, err + } + if err := u.bootstrap(ctx, dynCfg); err != nil { + u.suspendAndDelete(ctx) + bmetrics.UpdateUsers(durDirUserClass, -1) + return nil, err + } + return u, nil +} + +func (r *durDirRuntime) shutdown(ctx context.Context) { + r.users.Range(func(_, val any) bool { + u := val.(*durDirUser) + u.suspendAndDelete(ctx) + bmetrics.UpdateUsers(durDirUserClass, -1) + return true + }) +} + +type durDirUser struct { + cfg *userclass.Config + actorName string + hostHeader string + templateName string + userClass string + expectedDigest string + expectedSize int64 +} + +func (u *durDirUser) ref() *ateapipb.ObjectRef { + return &ateapipb.ObjectRef{Atespace: u.cfg.Atespace, Name: u.actorName} +} + +func (u *durDirUser) ensureAtespace(ctx context.Context) error { + return u.tracedCall(ctx, "CreateAtespace", func(callCtx context.Context, tr *metadata.MD) error { + _, err := u.cfg.APIStub.CreateAtespace(callCtx, &ateapipb.CreateAtespaceRequest{ + Atespace: &ateapipb.Atespace{ + Metadata: &ateapipb.ResourceMetadata{ + Name: u.cfg.Atespace, + }, + }, + }, grpc.Trailer(tr)) + if err == nil { + return nil + } + if s, ok := status.FromError(err); ok && s.Code() == codes.AlreadyExists { + return nil + } + return err + }) +} + +func (u *durDirUser) create(ctx context.Context) error { + return u.tracedCall(ctx, "CreateActor", func(callCtx context.Context, tr *metadata.MD) error { + _, err := u.cfg.APIStub.CreateActor(callCtx, &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: u.cfg.Atespace, Name: u.actorName}, + ActorTemplateNamespace: templateNS, + ActorTemplateName: u.templateName, + }, + }, grpc.Trailer(tr)) + return err + }) +} + +func (u *durDirUser) resume(ctx context.Context, mode string) bool { + // In implicit mode, the actor stays suspended until router traffic wakes it. + if mode == dynconfig.ResumeModeImplicit { + return true + } + err := u.tracedCall(ctx, "ResumeActor", func(callCtx context.Context, tr *metadata.MD) error { + _, err := u.cfg.APIStub.ResumeActor(callCtx, &ateapipb.ResumeActorRequest{ + Actor: u.ref(), + }, grpc.Trailer(tr)) + return err + }) + return err == nil +} + +func (u *durDirUser) suspend(ctx context.Context) { + _ = u.tracedCall(ctx, "SuspendActor", func(callCtx context.Context, tr *metadata.MD) error { + _, err := u.cfg.APIStub.SuspendActor(callCtx, &ateapipb.SuspendActorRequest{ + Actor: u.ref(), + }, grpc.Trailer(tr)) + return err + }) +} + +// suspendAndDelete suspends the actor before deleting it. DeleteActor requires +// SUSPENDED or CRASHED; deleting a running actor leaks it. The suspend is +// unmetered (teardown precondition, not benchmark latency), while the delete +// is metered so true leaks still surface in failures.csv. +func (u *durDirUser) suspendAndDelete(ctx context.Context) { + _, _ = u.cfg.APIStub.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ + Actor: u.ref(), + }) + u.delete(ctx) +} + +func (u *durDirUser) delete(ctx context.Context) { + _ = u.tracedCall(ctx, "DeleteActor", func(callCtx context.Context, tr *metadata.MD) error { + _, err := u.cfg.APIStub.DeleteActor(callCtx, &ateapipb.DeleteActorRequest{ + Actor: u.ref(), + }, grpc.Trailer(tr)) + return err + }) +} + +func (u *durDirUser) tracedCall(ctx context.Context, name string, do func(context.Context, *metadata.MD) error) error { + ctx, span := u.cfg.Tracer.Start(ctx, name) + defer span.End() + + start := time.Now() + var tr metadata.MD + err := do(ctx, &tr) + clientLatency := time.Since(start) + + latency, source := elapsedFromMD(tr, ateinterceptors.ServerElapsedTrailer, clientLatency) + if source == sourceServer { + span.SetAttributes(attribute.Float64("server.elapsed_ms", msFloat(latency))) + } + logSampledTrace(span, name, latency, source, err) + if err != nil { + bmetrics.RecordFailure("grpc", name, u.userClass, latency, err.Error()) + return err + } + bmetrics.RecordSuccess("grpc", name, u.userClass, latency, 0) + return nil +} + +func (u *durDirUser) params(dynCfg dynconfig.Config) (int64, gluttonpb.ReadMode) { + fileSize := dynCfg.DurDirFileSize + if fileSize <= 0 { + fileSize = defaultFileSize + } + readMode := gluttonpb.ReadMode_READ_MODE_DATA + if dynCfg.DurDirReadMode == dynconfig.ReadModeDigest { + readMode = gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY + } + return fileSize, readMode +} + +func (u *durDirUser) step(ctx context.Context, dynCfg dynconfig.Config) { + fileSize, readMode := u.params(dynCfg) + + // 1. Suspend actor + u.suspend(ctx) + + // 2. Resume — a no-op in implicit mode, where router traffic wakes the actor. + if !u.resume(ctx, dynCfg.ResumeMode) { + return + } + + // 3. Serve after resume (durability assertion: verify restored bytes) + if err := u.readDisk(ctx, "DurDirServeAfterResume", readMode); err != nil { + return + } + + // 4. Serve warm (immediate second read: measures page cache warming delta) + if err := u.readDisk(ctx, "DurDirServeWarm", readMode); err != nil { + return + } + + // 5. Overwrite file with fresh random bytes + if err := u.writeDisk(ctx, "DurDirOverwrite", fileSize, gluttonpb.WriteMode_WRITE_MODE_TRUNCATE); err != nil { + return + } +} + +func (u *durDirUser) bootstrap(ctx context.Context, dynCfg dynconfig.Config) error { + fileSize, readMode := u.params(dynCfg) + + if !u.resume(ctx, dynCfg.ResumeMode) { + return fmt.Errorf("initial resume failed") + } + + // Initial write to create DurDir file + if err := u.writeDisk(ctx, "DurDirWrite", fileSize, gluttonpb.WriteMode_WRITE_MODE_TRUNCATE); err != nil { + return fmt.Errorf("initial WriteDisk failed: %w", err) + } + + // Initial read to verify file + if err := u.readDisk(ctx, "DurDirServeInitial", readMode); err != nil { + return fmt.Errorf("initial ReadDisk failed: %w", err) + } + + return nil +} + +func (u *durDirUser) writeDisk(ctx context.Context, metricName string, size int64, mode gluttonpb.WriteMode) error { + req := &gluttonpb.WriteDiskRequest{ + Key: durDirTestFile, + Size: int32(size), + WriteMode: mode, + } + body, err := proto.Marshal(req) + if err != nil { + bmetrics.RecordFailure("http", metricName, u.userClass, 0, err.Error()) + return err + } + + var newDigest string + var newSize int64 + _, err = u.httpProtoCall(ctx, metricName, writeDiskRoute, body, func(respBytes []byte) error { + var resp gluttonpb.WriteDiskResponse + if err := proto.Unmarshal(respBytes, &resp); err != nil { + return fmt.Errorf("unmarshal WriteDiskResponse: %w", err) + } + if resp.GetSize() != size { + return fmt.Errorf("WriteDisk size mismatch: got %d, want %d", resp.GetSize(), size) + } + if len(resp.GetSha256()) == 0 { + return fmt.Errorf("WriteDisk sha256 is empty") + } + newDigest = hex.EncodeToString(resp.GetSha256()) + newSize = resp.GetSize() + return nil + }) + + if err != nil { + return err + } + u.expectedDigest = newDigest + u.expectedSize = newSize + return nil +} + +func (u *durDirUser) readDisk(ctx context.Context, metricName string, readMode gluttonpb.ReadMode) error { + req := &gluttonpb.ReadDiskRequest{ + Key: durDirTestFile, + ReadMode: readMode, + } + body, err := proto.Marshal(req) + if err != nil { + bmetrics.RecordFailure("http", metricName, u.userClass, 0, err.Error()) + return err + } + + expectedDigest := u.expectedDigest + expectedSize := u.expectedSize + + _, err = u.httpProtoCall(ctx, metricName, readDiskRoute, body, func(respBytes []byte) error { + var resp gluttonpb.ReadDiskResponse + if err := proto.Unmarshal(respBytes, &resp); err != nil { + return fmt.Errorf("unmarshal ReadDiskResponse: %w", err) + } + if resp.GetSize() != expectedSize { + return fmt.Errorf("ReadDisk size mismatch: got %d, want %d", resp.GetSize(), expectedSize) + } + respDigest := hex.EncodeToString(resp.GetSha256()) + if respDigest != expectedDigest { + return fmt.Errorf("ReadDisk response sha256 mismatch: got %q, want %q", respDigest, expectedDigest) + } + if readMode == gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY { + return nil + } + h := sha256.Sum256(resp.GetData()) + computedDigest := hex.EncodeToString(h[:]) + if computedDigest != expectedDigest { + return fmt.Errorf("ReadDisk payload sha256 mismatch: computed %q, want %q", computedDigest, expectedDigest) + } + return nil + }) + + return err +} + +// httpProtoCall issues a POST request to route with body and records metrics and traces. +// Metrics record client-perceived latency because the measurement target is Substrate, +// not glutton. +func (u *durDirUser) httpProtoCall(ctx context.Context, metricName, route string, body []byte, validate func([]byte) error) ([]byte, error) { + ctx, span := u.cfg.Tracer.Start(ctx, metricName) + defer span.End() + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, u.cfg.RouterURL+route, bytes.NewReader(body)) + if err != nil { + bmetrics.RecordFailure("http", metricName, u.userClass, 0, err.Error()) + return nil, err + } + httpReq.Host = u.hostHeader + httpReq.Header.Set("Content-Type", "application/x-protobuf") + otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(httpReq.Header)) + + start := time.Now() + resp, err := u.cfg.HTTPClient.Do(httpReq) + clientLatency := time.Since(start) + if err != nil { + bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, err.Error()) + return nil, err + } + defer resp.Body.Close() + + respBody, readErr := io.ReadAll(resp.Body) + if readErr != nil { + bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, readErr.Error()) + return nil, readErr + } + + serverLatency, source := elapsedFromHeader(resp.Header, ateinterceptors.ServerElapsedTrailer, clientLatency) + if source == sourceServer { + span.SetAttributes(attribute.Float64("server.elapsed_ms", msFloat(serverLatency))) + } + + if resp.StatusCode >= 400 { + httpErr := fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) + logSampledTrace(span, metricName, clientLatency, sourceClient, httpErr) + bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, httpErr.Error()) + return nil, httpErr + } + + if validate != nil { + if err := validate(respBody); err != nil { + logSampledTrace(span, metricName, clientLatency, sourceClient, err) + bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, err.Error()) + return nil, err + } + } + + logSampledTrace(span, metricName, clientLatency, sourceClient, nil) + bmetrics.RecordSuccess("http", metricName, u.userClass, clientLatency, int64(len(respBody))) + return respBody, nil +} + +func elapsedFromHeader(h http.Header, key string, fallback time.Duration) (time.Duration, string) { + val := h.Get(key) + if val == "" { + return fallback, sourceClient + } + us, err := strconv.ParseInt(val, 10, 64) + if err != nil { + return fallback, sourceClient + } + return time.Duration(us) * time.Microsecond, sourceServer +} diff --git a/internal/benchmarking/boomer/glutton/durdir_test.go b/internal/benchmarking/boomer/glutton/durdir_test.go new file mode 100644 index 000000000..b99bccbcd --- /dev/null +++ b/internal/benchmarking/boomer/glutton/durdir_test.go @@ -0,0 +1,272 @@ +// Copyright 2026 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 +// +// http://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. + +package glutton + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "reflect" + "regexp" + "slices" + "testing" + + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" + "github.com/agent-substrate/substrate/internal/benchmarking/glutton/fake" + gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton" +) + +func TestDurDirLoopSequence(t *testing.T) { + tests := []struct { + name string + resumeMode string + wantGRPCCall []string + wantHTTPCall []string + }{ + { + name: "explicit resume mode", + resumeMode: dynconfig.ResumeModeExplicit, + wantGRPCCall: []string{"SuspendActor", "ResumeActor"}, + wantHTTPCall: []string{fake.ReadDiskRoute, fake.ReadDiskRoute, fake.WriteDiskRoute}, + }, + { + name: "implicit resume mode", + resumeMode: dynconfig.ResumeModeImplicit, + wantGRPCCall: []string{"SuspendActor"}, // No ResumeActor RPC! + wantHTTPCall: []string{fake.ReadDiskRoute, fake.ReadDiskRoute, fake.WriteDiskRoute}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + srv := &fake.Server{Data: []byte("seq content")} + fakeCtrl := &fakeControlClient{} + cfg := &userclass.Config{ + APIStub: fakeCtrl, + Dyn: dynconfig.NewHolder(dynconfig.Config{ + ResumeMode: tc.resumeMode, + }), + } + du := newTestDurDirUser(t, srv, cfg) + du.expectedDigest = srv.HexDigest() + + dynCfg := cfg.Dyn.Load() + du.step(context.Background(), dynCfg) + + if got := fakeCtrl.recordedCalls(); !reflect.DeepEqual(got, tc.wantGRPCCall) { + t.Errorf("gRPC calls: got %v, want %v", got, tc.wantGRPCCall) + } + if got := srv.RecordedPaths(); !reflect.DeepEqual(got, tc.wantHTTPCall) { + t.Errorf("HTTP calls: got %v, want %v", got, tc.wantHTTPCall) + } + }) + } +} + +func TestDurDirUsesConfiguredFileSize(t *testing.T) { + configuredSize := int64(1048576) // 1 MiB + srv := &fake.Server{Data: make([]byte, configuredSize)} + du := newTestDurDirUser(t, srv, nil) + + if err := du.writeDisk(context.Background(), "TestConfiguredSize", configuredSize, gluttonpb.WriteMode_WRITE_MODE_TRUNCATE); err != nil { + t.Fatalf("writeDisk failed: %v", err) + } + + recorded := srv.RecordedWriteSizes() + if len(recorded) != 1 { + t.Fatalf("recorded write sizes: got %d calls, want 1", len(recorded)) + } + if int64(recorded[0]) != configuredSize { + t.Errorf("WriteDisk received size %d, want %d", recorded[0], configuredSize) + } +} + +func TestDurDirTestFileIsAValidGluttonKey(t *testing.T) { + if !regexp.MustCompile(`^[a-zA-Z0-9_-]+$`).MatchString(durDirTestFile) { + t.Fatalf("durDirTestFile %q would be rejected by glutton", durDirTestFile) + } +} + +func TestDurDirDigestOnlyAcceptsEmptyPayload(t *testing.T) { + srv := &fake.Server{ + Data: make([]byte, 1024), + EmptyPayload: true, + } + du := newTestDurDirUser(t, srv, nil) + du.expectedDigest = srv.HexDigest() + + if err := du.readDisk(context.Background(), t.Name(), gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY); err != nil { + t.Fatalf("expected readDisk to succeed in digest-only mode with empty payload, got: %v", err) + } +} + +func TestDurDirDataModeRejectsEmptyPayload(t *testing.T) { + srv := &fake.Server{ + Data: make([]byte, 1024), + EmptyPayload: true, + } + du := newTestDurDirUser(t, srv, nil) + du.expectedDigest = srv.HexDigest() + + if err := du.readDisk(context.Background(), t.Name(), gluttonpb.ReadMode_READ_MODE_DATA); err == nil { + t.Fatalf("expected readDisk to fail in data mode with empty payload, got nil") + } +} + +func TestDurDirDigestOnlyStillRejectsWrongDigest(t *testing.T) { + wrongHash := sha256.Sum256([]byte("wrong data")) + srv := &fake.Server{ + Data: make([]byte, 1024), + Digest: wrongHash[:], + EmptyPayload: true, + } + du := newTestDurDirUser(t, srv, nil) + h := sha256.Sum256(srv.Data) + du.expectedDigest = hex.EncodeToString(h[:]) + + if err := du.readDisk(context.Background(), t.Name(), gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY); err == nil { + t.Fatalf("expected readDisk to fail in digest-only mode on wrong digest, got nil") + } +} + +func TestDurDirReadModeSentOnWire(t *testing.T) { + srv := &fake.Server{} + du := newTestDurDirUser(t, srv, nil) + du.expectedDigest = srv.HexDigest() + + if err := du.readDisk(context.Background(), t.Name(), gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY); err != nil { + t.Fatalf("readDisk failed: %v", err) + } + + recorded := srv.RecordedReadModes() + if len(recorded) != 1 { + t.Fatalf("recorded read modes: got %d calls, want 1", len(recorded)) + } + if recorded[0] != gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY { + t.Errorf("wire ReadMode: got %v, want %v", recorded[0], gluttonpb.ReadMode_READ_MODE_DIGEST_ONLY) + } +} + +func TestDurDirBootstrapDoesNotBoot(t *testing.T) { + srv := &fake.Server{Data: []byte("data")} + fakeCtrl := &fakeControlClient{} + cfg := newTestConfig(t, srv, &userclass.Config{ + APIStub: fakeCtrl, + Dyn: dynconfig.NewHolder(dynconfig.Config{ + DurDirFileSize: int64(len(srv.Data)), + }), + }) + + rt := &durDirRuntime{cfg: cfg} + _, err := rt.startUser(context.Background(), cfg.Dyn.Load()) + if err != nil { + t.Fatalf("startUser failed: %v", err) + } + + boots := fakeCtrl.recordedBoots() + if len(boots) == 0 { + t.Fatalf("expected ResumeActor to be called during bootstrap, got 0 calls") + } + if boots[0] { + t.Errorf("bootstrap ResumeActor Boot: got %v, want false", boots[0]) + } +} + +func TestDurDirBootstrapUsesConfiguredResumeMode(t *testing.T) { + tests := []struct { + name string + resumeMode string + wantResumeActor bool + }{ + { + name: "explicit resume mode", + resumeMode: dynconfig.ResumeModeExplicit, + wantResumeActor: true, + }, + { + name: "implicit resume mode", + resumeMode: dynconfig.ResumeModeImplicit, + wantResumeActor: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + srv := &fake.Server{Data: []byte("data")} + fakeCtrl := &fakeControlClient{} + cfg := newTestConfig(t, srv, &userclass.Config{ + APIStub: fakeCtrl, + Dyn: dynconfig.NewHolder(dynconfig.Config{ + DurDirFileSize: int64(len(srv.Data)), + ResumeMode: tc.resumeMode, + }), + }) + + rt := &durDirRuntime{cfg: cfg} + _, err := rt.startUser(context.Background(), cfg.Dyn.Load()) + if err != nil { + t.Fatalf("startUser failed: %v", err) + } + + calls := fakeCtrl.recordedCalls() + gotResumeActor := slices.Contains(calls, "ResumeActor") + if gotResumeActor != tc.wantResumeActor { + t.Errorf("ResumeActor in recordedCalls: got %v, want %v (calls = %v)", gotResumeActor, tc.wantResumeActor, calls) + } + }) + } +} + +func TestDurDirBootstrapFailureSuspendsBeforeDelete(t *testing.T) { + srv := &fake.Server{Status: http.StatusInternalServerError} + fakeCtrl := &fakeControlClient{} + cfg := newTestConfig(t, srv, &userclass.Config{ + APIStub: fakeCtrl, + Dyn: dynconfig.NewHolder(dynconfig.Config{ + DurDirFileSize: 1024, + ResumeMode: dynconfig.ResumeModeExplicit, + }), + }) + + rt := &durDirRuntime{cfg: cfg} + _, err := rt.startUser(context.Background(), cfg.Dyn.Load()) + if err == nil { + t.Fatalf("startUser expected error on failing server, got nil") + } + + calls := fakeCtrl.recordedCalls() + if len(calls) < 2 || calls[len(calls)-2] != "SuspendActor" || calls[len(calls)-1] != "DeleteActor" { + t.Errorf("recordedCalls must end with [SuspendActor, DeleteActor], got %v", calls) + } +} + +func TestDurDirShutdownSuspendsBeforeDelete(t *testing.T) { + fakeCtrl := &fakeControlClient{} + cfg := &userclass.Config{ + APIStub: fakeCtrl, + } + du := newTestDurDirUser(t, &fake.Server{}, cfg) + + rt := &durDirRuntime{cfg: du.cfg} + rt.users.Store(goroutineID(), du) + rt.shutdown(context.Background()) + + calls := fakeCtrl.recordedCalls() + if len(calls) < 2 || calls[len(calls)-2] != "SuspendActor" || calls[len(calls)-1] != "DeleteActor" { + t.Errorf("recordedCalls must end with [SuspendActor, DeleteActor], got %v", calls) + } +} diff --git a/internal/benchmarking/boomer/glutton/fixture_test.go b/internal/benchmarking/boomer/glutton/fixture_test.go new file mode 100644 index 000000000..89bb59566 --- /dev/null +++ b/internal/benchmarking/boomer/glutton/fixture_test.go @@ -0,0 +1,118 @@ +// Copyright 2026 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 +// +// http://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. + +package glutton + +import ( + "context" + "sync" + "testing" + + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" + "github.com/agent-substrate/substrate/internal/benchmarking/glutton/fake" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "go.opentelemetry.io/otel" + "google.golang.org/grpc" +) + +type fakeControlClient struct { + ateapipb.ControlClient + mu sync.Mutex + calls []string + resumeBoots []bool +} + +func (f *fakeControlClient) CreateAtespace(ctx context.Context, in *ateapipb.CreateAtespaceRequest, opts ...grpc.CallOption) (*ateapipb.Atespace, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "CreateAtespace") + return &ateapipb.Atespace{}, nil +} + +func (f *fakeControlClient) CreateActor(ctx context.Context, in *ateapipb.CreateActorRequest, opts ...grpc.CallOption) (*ateapipb.Actor, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "CreateActor") + return &ateapipb.Actor{}, nil +} + +func (f *fakeControlClient) ResumeActor(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "ResumeActor") + f.resumeBoots = append(f.resumeBoots, in.GetBoot()) + return &ateapipb.ResumeActorResponse{}, nil +} + +func (f *fakeControlClient) SuspendActor(ctx context.Context, in *ateapipb.SuspendActorRequest, opts ...grpc.CallOption) (*ateapipb.SuspendActorResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "SuspendActor") + return &ateapipb.SuspendActorResponse{}, nil +} + +func (f *fakeControlClient) DeleteActor(ctx context.Context, in *ateapipb.DeleteActorRequest, opts ...grpc.CallOption) (*ateapipb.Actor, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "DeleteActor") + return &ateapipb.Actor{}, nil +} + +func (f *fakeControlClient) recordedCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.calls...) +} + +func (f *fakeControlClient) recordedBoots() []bool { + f.mu.Lock() + defer f.mu.Unlock() + return append([]bool(nil), f.resumeBoots...) +} + +// newTestConfig starts srv, sets HTTPClient and RouterURL, and ensures +// APIStub, Tracer, and Dyn are populated if nil. +func newTestConfig(t *testing.T, srv *fake.Server, cfg *userclass.Config) *userclass.Config { + t.Helper() + ts := srv.Start(t) + if cfg == nil { + cfg = &userclass.Config{} + } + if cfg.APIStub == nil { + cfg.APIStub = &fakeControlClient{} + } + if cfg.Tracer == nil { + cfg.Tracer = otel.Tracer("test") + } + if cfg.Dyn == nil { + cfg.Dyn = dynconfig.NewHolder(dynconfig.Config{}) + } + cfg.HTTPClient = ts.Client() + cfg.RouterURL = ts.URL + return cfg +} + +func newTestDurDirUser(t *testing.T, srv *fake.Server, cfg *userclass.Config) *durDirUser { + t.Helper() + c := newTestConfig(t, srv, cfg) + return &durDirUser{ + cfg: c, + actorName: "duractor", + hostHeader: "duractor.benchmark." + actorDomain, + templateName: defaultDurTemplate, + userClass: durDirUserClass, + expectedSize: int64(len(srv.Data)), + } +} diff --git a/internal/benchmarking/boomer/glutton/lifecycle.go b/internal/benchmarking/boomer/glutton/lifecycle.go index 5cd8a9626..8a04457ca 100644 --- a/internal/benchmarking/boomer/glutton/lifecycle.go +++ b/internal/benchmarking/boomer/glutton/lifecycle.go @@ -32,8 +32,8 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateinterceptors" - "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/google/uuid" @@ -59,30 +59,19 @@ const ( sourceServer = "server" ) -// Config holds the dependencies a glutton task needs. Built once at startup -// and passed to Register. -type Config struct { - // APIStub is the shared gRPC client to ateapi (one connection, all goroutines). - APIStub ateapipb.ControlClient - // HTTPClient is the shared HTTP client for atenet pings. - HTTPClient *http.Client - // RouterURL is the base URL of the atenet router (no trailing slash). - RouterURL string - // Atespace every actor this worker creates lives in. Required; caller - // is responsible for having ensured it exists (see EnsureAtespace). - Atespace string - // Dyn is the runtime-mutable config (wait-time bounds, trace - // probability). Required — every per-iteration read goes through it, - // so tests can mutate it without touching glutton internals. - Dyn *dynconfig.Holder - // Tracer anchors sampled spans; falls back to the otel global if nil. - Tracer trace.Tracer +func init() { + userclass.Add(userclass.Entry{ + Name: "glutton", + LocustFile: "glutton.py", + UserClass: userClass, + Init: initPing, + }) } -// Register creates a runtime tied to cfg and returns a boomer-compatible task +// initPing creates a runtime tied to cfg and returns a boomer-compatible task // function plus a Shutdown hook the caller should run before exit (it // suspend+deletes every actor this worker created). -func Register(cfg *Config) (taskFn func(), shutdown func(context.Context)) { +func initPing(cfg *userclass.Config) (taskFn func(), shutdown func(context.Context)) { if cfg.Tracer == nil { cfg.Tracer = otel.Tracer("substrate-boomer/glutton") } @@ -91,7 +80,7 @@ func Register(cfg *Config) (taskFn func(), shutdown func(context.Context)) { } type taskRuntime struct { - cfg *Config + cfg *userclass.Config users sync.Map // goroutineID → *gluttonUser } @@ -168,7 +157,7 @@ func (r *taskRuntime) dynamicWait() time.Duration { } type gluttonUser struct { - cfg *Config + cfg *userclass.Config actorName string hostHeader string firstResume bool diff --git a/internal/benchmarking/boomer/userclass/config.go b/internal/benchmarking/boomer/userclass/config.go new file mode 100644 index 000000000..e7f7c6007 --- /dev/null +++ b/internal/benchmarking/boomer/userclass/config.go @@ -0,0 +1,43 @@ +// Copyright 2026 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 +// +// http://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. + +package userclass + +import ( + "net/http" + + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "go.opentelemetry.io/otel/trace" +) + +// Config holds the dependencies a user class needs. Built once at startup +// and passed to the entry's Init func. +type Config struct { + // APIStub is the shared gRPC client to ateapi (one connection, all goroutines). + APIStub ateapipb.ControlClient + // HTTPClient is the shared HTTP client for atenet pings. + HTTPClient *http.Client + // RouterURL is the base URL of the atenet router (no trailing slash). + RouterURL string + // Atespace every actor this worker creates lives in. Required; caller + // is responsible for having ensured it exists (see EnsureAtespace). + Atespace string + // Dyn is the runtime-mutable config (wait-time bounds, trace + // probability). Required — every per-iteration read goes through it, + // so tests can mutate it without touching glutton internals. + Dyn *dynconfig.Holder + // Tracer anchors sampled spans; falls back to the otel global if nil. + Tracer trace.Tracer +} diff --git a/internal/benchmarking/boomer/userclass/registry.go b/internal/benchmarking/boomer/userclass/registry.go new file mode 100644 index 000000000..f429c6a9b --- /dev/null +++ b/internal/benchmarking/boomer/userclass/registry.go @@ -0,0 +1,71 @@ +// Copyright 2026 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 +// +// http://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. + +package userclass + +import ( + "context" + "fmt" + "slices" + "sync" +) + +// Entry declares one user class: the flag value that selects it, the Locust +// file and Python class it pairs with, and the func that builds its task. +type Entry struct { + // Name is the --user-class flag value that selects this class. + Name string + // LocustFile is the tests/ the Locust master loads for it. + LocustFile string + // UserClass is the Python class name. It must equal boomer.Task.Name or + // the master's spawn messages never match and no users start. + UserClass string + // Init builds the boomer task func and its shutdown hook. + Init func(*Config) (task func(), shutdown func(context.Context)) +} + +var ( + mu sync.RWMutex + registry = make(map[string]Entry) +) + +// Add registers a user class Entry. It panics if an entry with the same Name is already registered. +func Add(e Entry) { + mu.Lock() + defer mu.Unlock() + if _, exists := registry[e.Name]; exists { + panic(fmt.Sprintf("userclass: duplicate registration for %q", e.Name)) + } + registry[e.Name] = e +} + +// Lookup returns the Entry registered under name, or false if not found. +func Lookup(name string) (Entry, bool) { + mu.RLock() + defer mu.RUnlock() + e, ok := registry[name] + return e, ok +} + +// Names returns a sorted slice of all registered user class names. +func Names() []string { + mu.RLock() + defer mu.RUnlock() + names := make([]string, 0, len(registry)) + for k := range registry { + names = append(names, k) + } + slices.Sort(names) + return names +} diff --git a/internal/benchmarking/boomer/userclass/registry_test.go b/internal/benchmarking/boomer/userclass/registry_test.go new file mode 100644 index 000000000..72144ae7a --- /dev/null +++ b/internal/benchmarking/boomer/userclass/registry_test.go @@ -0,0 +1,58 @@ +// Copyright 2026 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 +// +// http://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. + +package userclass + +import ( + "context" + "testing" +) + +func TestAddRejectsDuplicate(t *testing.T) { + name := "test-dup" + Add(Entry{ + Name: name, + UserClass: "TestDupUser", + Init: func(*Config) (func(), func(context.Context)) { + return nil, nil + }, + }) + defer func() { + mu.Lock() + delete(registry, name) + mu.Unlock() + }() + + defer func() { + r := recover() + if r == nil { + t.Errorf("expected Add with duplicate name %q to panic, but it did not", name) + } + }() + + Add(Entry{ + Name: name, + UserClass: "TestDupUser2", + Init: func(*Config) (func(), func(context.Context)) { + return nil, nil + }, + }) +} + +func TestLookupUnknown(t *testing.T) { + _, ok := Lookup("nonexistent-class-name") + if ok { + t.Errorf("Lookup(nonexistent) got ok=true, want false") + } +} diff --git a/internal/benchmarking/glutton/fake/server.go b/internal/benchmarking/glutton/fake/server.go new file mode 100644 index 000000000..a76f600c8 --- /dev/null +++ b/internal/benchmarking/glutton/fake/server.go @@ -0,0 +1,173 @@ +// Copyright 2026 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 +// +// http://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. + +// Package fake provides an httptest-backed stand-in for a glutton actor. +package fake + +import ( + "crypto/sha256" + "encoding/hex" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/agent-substrate/substrate/internal/ateinterceptors" + gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton" + "google.golang.org/protobuf/proto" +) + +// Routes the fake serves, mirroring glutton's real HTTP mux. Declared here so +// the fake depends on nothing; collapses onto one source when glutton's core moves. +const ( + WriteDiskRoute = "/writedisk" + ReadDiskRoute = "/readdisk" +) + +// Server is an httptest-backed stand-in for a glutton actor holding one file. +// The Data slice is the source of truth: both routes report len(Data) and +// sha256(Data), and /readdisk serves Data as payload. +// Each override field makes the actor lie about exactly one property. +type Server struct { + // Data is the file the actor holds, driving size, digest, and payload. + Data []byte + // Digest overrides the sha256 returned by both routes, leaving size and payload honest. + Digest []byte + // CorruptPayload is served by /readdisk instead of Data, keeping size and digest honest. + CorruptPayload []byte + // EmptyPayload causes /readdisk to omit the Data field entirely (digest-only wire format). + // Silently takes precedence over CorruptPayload if both are set. + EmptyPayload bool + // Status fails every route with this HTTP status code. + Status int + // ElapsedUs sets the x-server-elapsed-us timing header/trailer. + ElapsedUs string + + mu sync.Mutex + paths []string + writeSizes []int32 + readModes []gluttonpb.ReadMode +} + +func (s *Server) reportedDigest() []byte { + if s.Digest != nil { + return s.Digest + } + h := sha256.Sum256(s.Data) + return h[:] +} + +func (s *Server) HexDigest() string { + return hex.EncodeToString(s.reportedDigest()) +} + +func (s *Server) reportedPayload() []byte { + if s.EmptyPayload { + return nil + } + if s.CorruptPayload != nil { + return s.CorruptPayload + } + return s.Data +} + +func (s *Server) RecordedPaths() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.paths...) +} + +func (s *Server) RecordedWriteSizes() []int32 { + s.mu.Lock() + defer s.mu.Unlock() + return append([]int32(nil), s.writeSizes...) +} + +func (s *Server) RecordedReadModes() []gluttonpb.ReadMode { + s.mu.Lock() + defer s.mu.Unlock() + return append([]gluttonpb.ReadMode(nil), s.readModes...) +} + +func (s *Server) Start(t *testing.T) *httptest.Server { + t.Helper() + ts := httptest.NewServer(http.HandlerFunc(s.serve)) + t.Cleanup(ts.Close) + return ts +} + +func (s *Server) serve(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + s.paths = append(s.paths, r.URL.Path) + s.mu.Unlock() + + if s.Status != 0 { + http.Error(w, http.StatusText(s.Status), s.Status) + return + } + + if s.ElapsedUs != "" { + w.Header().Set(ateinterceptors.ServerElapsedTrailer, s.ElapsedUs) + } + w.Header().Set("Content-Type", "application/x-protobuf") + + switch r.URL.Path { + case WriteDiskRoute: + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + var req gluttonpb.WriteDiskRequest + if err := proto.Unmarshal(body, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + s.mu.Lock() + s.writeSizes = append(s.writeSizes, req.GetSize()) + s.mu.Unlock() + + resp, _ := proto.Marshal(&gluttonpb.WriteDiskResponse{ + Size: int64(len(s.Data)), + Sha256: s.reportedDigest(), + }) + _, _ = w.Write(resp) + + case ReadDiskRoute: + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + var req gluttonpb.ReadDiskRequest + if err := proto.Unmarshal(body, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + s.mu.Lock() + s.readModes = append(s.readModes, req.GetReadMode()) + s.mu.Unlock() + + resp, _ := proto.Marshal(&gluttonpb.ReadDiskResponse{ + Size: int64(len(s.Data)), + Sha256: s.reportedDigest(), + Data: s.reportedPayload(), + }) + _, _ = w.Write(resp) + + default: + http.NotFound(w, r) + } +} diff --git a/internal/proto/glutton/glutton.pb.go b/internal/proto/glutton/glutton.pb.go index b54b6663b..8f1827307 100644 --- a/internal/proto/glutton/glutton.pb.go +++ b/internal/proto/glutton/glutton.pb.go @@ -83,6 +83,56 @@ func (WriteMode) EnumDescriptor() ([]byte, []int) { return file_glutton_proto_rawDescGZIP(), []int{0} } +// ReadMode selects how much of the file ReadDisk sends back. +type ReadMode int32 + +const ( + // Return the file's bytes in ReadDiskResponse.data alongside its digest. + ReadMode_READ_MODE_DATA ReadMode = 0 + // Return only size and sha256; ReadDiskResponse.data is empty. The digest + // is the same one READ_MODE_DATA returns, so either mode can verify. + ReadMode_READ_MODE_DIGEST_ONLY ReadMode = 1 +) + +// Enum value maps for ReadMode. +var ( + ReadMode_name = map[int32]string{ + 0: "READ_MODE_DATA", + 1: "READ_MODE_DIGEST_ONLY", + } + ReadMode_value = map[string]int32{ + "READ_MODE_DATA": 0, + "READ_MODE_DIGEST_ONLY": 1, + } +) + +func (x ReadMode) Enum() *ReadMode { + p := new(ReadMode) + *p = x + return p +} + +func (x ReadMode) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (ReadMode) Descriptor() protoreflect.EnumDescriptor { + return file_glutton_proto_enumTypes[1].Descriptor() +} + +func (ReadMode) Type() protoreflect.EnumType { + return &file_glutton_proto_enumTypes[1] +} + +func (x ReadMode) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use ReadMode.Descriptor instead. +func (ReadMode) EnumDescriptor() ([]byte, []int) { + return file_glutton_proto_rawDescGZIP(), []int{1} +} + type WriteRAMRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // name of the array to be written to @@ -244,7 +294,11 @@ func (x *WriteDiskRequest) GetWriteMode() WriteMode { } type WriteDiskResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` + state protoimpl.MessageState `protogen:"open.v1"` + // size of the file after the write + Size int64 `protobuf:"varint,1,opt,name=size,proto3" json:"size,omitempty"` + // sha256 of the whole file after the write + Sha256 []byte `protobuf:"bytes,2,opt,name=sha256,proto3" json:"sha256,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -279,6 +333,136 @@ func (*WriteDiskResponse) Descriptor() ([]byte, []int) { return file_glutton_proto_rawDescGZIP(), []int{3} } +func (x *WriteDiskResponse) GetSize() int64 { + if x != nil { + return x.Size + } + return 0 +} + +func (x *WriteDiskResponse) GetSha256() []byte { + if x != nil { + return x.Sha256 + } + return nil +} + +type ReadDiskRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // name of the file to be read from + Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` + ReadMode ReadMode `protobuf:"varint,2,opt,name=read_mode,json=readMode,proto3,enum=glutton.ReadMode" json:"read_mode,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReadDiskRequest) Reset() { + *x = ReadDiskRequest{} + mi := &file_glutton_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReadDiskRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReadDiskRequest) ProtoMessage() {} + +func (x *ReadDiskRequest) ProtoReflect() protoreflect.Message { + mi := &file_glutton_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReadDiskRequest.ProtoReflect.Descriptor instead. +func (*ReadDiskRequest) Descriptor() ([]byte, []int) { + return file_glutton_proto_rawDescGZIP(), []int{4} +} + +func (x *ReadDiskRequest) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *ReadDiskRequest) GetReadMode() ReadMode { + if x != nil { + return x.ReadMode + } + return ReadMode_READ_MODE_DATA +} + +type ReadDiskResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // total size of bytes read + Size int64 `protobuf:"varint,1,opt,name=size,proto3" json:"size,omitempty"` + // sha256 of bytes read + Sha256 []byte `protobuf:"bytes,2,opt,name=sha256,proto3" json:"sha256,omitempty"` + // data read from the file. Empty under READ_MODE_DIGEST_ONLY. + Data []byte `protobuf:"bytes,3,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReadDiskResponse) Reset() { + *x = ReadDiskResponse{} + mi := &file_glutton_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReadDiskResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReadDiskResponse) ProtoMessage() {} + +func (x *ReadDiskResponse) ProtoReflect() protoreflect.Message { + mi := &file_glutton_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReadDiskResponse.ProtoReflect.Descriptor instead. +func (*ReadDiskResponse) Descriptor() ([]byte, []int) { + return file_glutton_proto_rawDescGZIP(), []int{5} +} + +func (x *ReadDiskResponse) GetSize() int64 { + if x != nil { + return x.Size + } + return 0 +} + +func (x *ReadDiskResponse) GetSha256() []byte { + if x != nil { + return x.Sha256 + } + return nil +} + +func (x *ReadDiskResponse) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + type OpenFDRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // The total number of FDs for the glutton to open @@ -289,7 +473,7 @@ type OpenFDRequest struct { func (x *OpenFDRequest) Reset() { *x = OpenFDRequest{} - mi := &file_glutton_proto_msgTypes[4] + mi := &file_glutton_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -301,7 +485,7 @@ func (x *OpenFDRequest) String() string { func (*OpenFDRequest) ProtoMessage() {} func (x *OpenFDRequest) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[4] + mi := &file_glutton_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -314,7 +498,7 @@ func (x *OpenFDRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use OpenFDRequest.ProtoReflect.Descriptor instead. func (*OpenFDRequest) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{4} + return file_glutton_proto_rawDescGZIP(), []int{6} } func (x *OpenFDRequest) GetCount() int32 { @@ -332,7 +516,7 @@ type OpenFDResponse struct { func (x *OpenFDResponse) Reset() { *x = OpenFDResponse{} - mi := &file_glutton_proto_msgTypes[5] + mi := &file_glutton_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -344,7 +528,7 @@ func (x *OpenFDResponse) String() string { func (*OpenFDResponse) ProtoMessage() {} func (x *OpenFDResponse) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[5] + mi := &file_glutton_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -357,7 +541,7 @@ func (x *OpenFDResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use OpenFDResponse.ProtoReflect.Descriptor instead. func (*OpenFDResponse) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{5} + return file_glutton_proto_rawDescGZIP(), []int{7} } type PingRequest struct { @@ -370,7 +554,7 @@ type PingRequest struct { func (x *PingRequest) Reset() { *x = PingRequest{} - mi := &file_glutton_proto_msgTypes[6] + mi := &file_glutton_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -382,7 +566,7 @@ func (x *PingRequest) String() string { func (*PingRequest) ProtoMessage() {} func (x *PingRequest) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[6] + mi := &file_glutton_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -395,7 +579,7 @@ func (x *PingRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use PingRequest.ProtoReflect.Descriptor instead. func (*PingRequest) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{6} + return file_glutton_proto_rawDescGZIP(), []int{8} } func (x *PingRequest) GetMessage() string { @@ -415,7 +599,7 @@ type PingResponse struct { func (x *PingResponse) Reset() { *x = PingResponse{} - mi := &file_glutton_proto_msgTypes[7] + mi := &file_glutton_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -427,7 +611,7 @@ func (x *PingResponse) String() string { func (*PingResponse) ProtoMessage() {} func (x *PingResponse) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[7] + mi := &file_glutton_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -440,7 +624,7 @@ func (x *PingResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use PingResponse.ProtoReflect.Descriptor instead. func (*PingResponse) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{7} + return file_glutton_proto_rawDescGZIP(), []int{9} } func (x *PingResponse) GetMessage() string { @@ -459,7 +643,7 @@ type GossipRequest struct { func (x *GossipRequest) Reset() { *x = GossipRequest{} - mi := &file_glutton_proto_msgTypes[8] + mi := &file_glutton_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -471,7 +655,7 @@ func (x *GossipRequest) String() string { func (*GossipRequest) ProtoMessage() {} func (x *GossipRequest) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[8] + mi := &file_glutton_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -484,7 +668,7 @@ func (x *GossipRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use GossipRequest.ProtoReflect.Descriptor instead. func (*GossipRequest) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{8} + return file_glutton_proto_rawDescGZIP(), []int{10} } func (x *GossipRequest) GetPeers() []*Peer { @@ -502,7 +686,7 @@ type GossipResponse struct { func (x *GossipResponse) Reset() { *x = GossipResponse{} - mi := &file_glutton_proto_msgTypes[9] + mi := &file_glutton_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -514,7 +698,7 @@ func (x *GossipResponse) String() string { func (*GossipResponse) ProtoMessage() {} func (x *GossipResponse) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[9] + mi := &file_glutton_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -527,7 +711,7 @@ func (x *GossipResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use GossipResponse.ProtoReflect.Descriptor instead. func (*GossipResponse) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{9} + return file_glutton_proto_rawDescGZIP(), []int{11} } type Peer struct { @@ -540,7 +724,7 @@ type Peer struct { func (x *Peer) Reset() { *x = Peer{} - mi := &file_glutton_proto_msgTypes[10] + mi := &file_glutton_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -552,7 +736,7 @@ func (x *Peer) String() string { func (*Peer) ProtoMessage() {} func (x *Peer) ProtoReflect() protoreflect.Message { - mi := &file_glutton_proto_msgTypes[10] + mi := &file_glutton_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -565,7 +749,7 @@ func (x *Peer) ProtoReflect() protoreflect.Message { // Deprecated: Use Peer.ProtoReflect.Descriptor instead. func (*Peer) Descriptor() ([]byte, []int) { - return file_glutton_proto_rawDescGZIP(), []int{10} + return file_glutton_proto_rawDescGZIP(), []int{12} } func (x *Peer) GetHost() string { @@ -597,8 +781,17 @@ const file_glutton_proto_rawDesc = "" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x12\n" + "\x04size\x18\x02 \x01(\x05R\x04size\x121\n" + "\n" + - "write_mode\x18\x03 \x01(\x0e2\x12.glutton.WriteModeR\twriteMode\"\x13\n" + - "\x11WriteDiskResponse\"%\n" + + "write_mode\x18\x03 \x01(\x0e2\x12.glutton.WriteModeR\twriteMode\"?\n" + + "\x11WriteDiskResponse\x12\x12\n" + + "\x04size\x18\x01 \x01(\x03R\x04size\x12\x16\n" + + "\x06sha256\x18\x02 \x01(\fR\x06sha256\"S\n" + + "\x0fReadDiskRequest\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12.\n" + + "\tread_mode\x18\x02 \x01(\x0e2\x11.glutton.ReadModeR\breadMode\"R\n" + + "\x10ReadDiskResponse\x12\x12\n" + + "\x04size\x18\x01 \x01(\x03R\x04size\x12\x16\n" + + "\x06sha256\x18\x02 \x01(\fR\x06sha256\x12\x12\n" + + "\x04data\x18\x03 \x01(\fR\x04data\"%\n" + "\rOpenFDRequest\x12\x14\n" + "\x05count\x18\x01 \x01(\x05R\x05count\"\x10\n" + "\x0eOpenFDResponse\"'\n" + @@ -614,10 +807,14 @@ const file_glutton_proto_rawDesc = "" + "\bdelay_ms\x18\x02 \x01(\x05R\adelayMs*>\n" + "\tWriteMode\x12\x17\n" + "\x13WRITE_MODE_TRUNCATE\x10\x00\x12\x18\n" + - "\x14WRITE_MODE_OVERWRITE\x10\x012\xc3\x02\n" + + "\x14WRITE_MODE_OVERWRITE\x10\x01*9\n" + + "\bReadMode\x12\x12\n" + + "\x0eREAD_MODE_DATA\x10\x00\x12\x19\n" + + "\x15READ_MODE_DIGEST_ONLY\x10\x012\x86\x03\n" + "\aGlutton\x12A\n" + "\bWriteRAM\x12\x18.glutton.WriteRAMRequest\x1a\x19.glutton.WriteRAMResponse\"\x00\x12D\n" + - "\tWriteDisk\x12\x19.glutton.WriteDiskRequest\x1a\x1a.glutton.WriteDiskResponse\"\x00\x12;\n" + + "\tWriteDisk\x12\x19.glutton.WriteDiskRequest\x1a\x1a.glutton.WriteDiskResponse\"\x00\x12A\n" + + "\bReadDisk\x12\x18.glutton.ReadDiskRequest\x1a\x19.glutton.ReadDiskResponse\"\x00\x12;\n" + "\x06OpenFD\x12\x16.glutton.OpenFDRequest\x1a\x17.glutton.OpenFDResponse\"\x00\x125\n" + "\x04Ping\x12\x14.glutton.PingRequest\x1a\x15.glutton.PingResponse\"\x00\x12;\n" + "\x06Gossip\x12\x16.glutton.GossipRequest\x1a\x17.glutton.GossipResponse\"\x00B=Z;github.com/agent-substrate/substrate/internal/proto/gluttonb\x06proto3" @@ -634,41 +831,47 @@ func file_glutton_proto_rawDescGZIP() []byte { return file_glutton_proto_rawDescData } -var file_glutton_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_glutton_proto_msgTypes = make([]protoimpl.MessageInfo, 11) +var file_glutton_proto_enumTypes = make([]protoimpl.EnumInfo, 2) +var file_glutton_proto_msgTypes = make([]protoimpl.MessageInfo, 13) var file_glutton_proto_goTypes = []any{ (WriteMode)(0), // 0: glutton.WriteMode - (*WriteRAMRequest)(nil), // 1: glutton.WriteRAMRequest - (*WriteRAMResponse)(nil), // 2: glutton.WriteRAMResponse - (*WriteDiskRequest)(nil), // 3: glutton.WriteDiskRequest - (*WriteDiskResponse)(nil), // 4: glutton.WriteDiskResponse - (*OpenFDRequest)(nil), // 5: glutton.OpenFDRequest - (*OpenFDResponse)(nil), // 6: glutton.OpenFDResponse - (*PingRequest)(nil), // 7: glutton.PingRequest - (*PingResponse)(nil), // 8: glutton.PingResponse - (*GossipRequest)(nil), // 9: glutton.GossipRequest - (*GossipResponse)(nil), // 10: glutton.GossipResponse - (*Peer)(nil), // 11: glutton.Peer + (ReadMode)(0), // 1: glutton.ReadMode + (*WriteRAMRequest)(nil), // 2: glutton.WriteRAMRequest + (*WriteRAMResponse)(nil), // 3: glutton.WriteRAMResponse + (*WriteDiskRequest)(nil), // 4: glutton.WriteDiskRequest + (*WriteDiskResponse)(nil), // 5: glutton.WriteDiskResponse + (*ReadDiskRequest)(nil), // 6: glutton.ReadDiskRequest + (*ReadDiskResponse)(nil), // 7: glutton.ReadDiskResponse + (*OpenFDRequest)(nil), // 8: glutton.OpenFDRequest + (*OpenFDResponse)(nil), // 9: glutton.OpenFDResponse + (*PingRequest)(nil), // 10: glutton.PingRequest + (*PingResponse)(nil), // 11: glutton.PingResponse + (*GossipRequest)(nil), // 12: glutton.GossipRequest + (*GossipResponse)(nil), // 13: glutton.GossipResponse + (*Peer)(nil), // 14: glutton.Peer } var file_glutton_proto_depIdxs = []int32{ 0, // 0: glutton.WriteRAMRequest.write_mode:type_name -> glutton.WriteMode 0, // 1: glutton.WriteDiskRequest.write_mode:type_name -> glutton.WriteMode - 11, // 2: glutton.GossipRequest.peers:type_name -> glutton.Peer - 1, // 3: glutton.Glutton.WriteRAM:input_type -> glutton.WriteRAMRequest - 3, // 4: glutton.Glutton.WriteDisk:input_type -> glutton.WriteDiskRequest - 5, // 5: glutton.Glutton.OpenFD:input_type -> glutton.OpenFDRequest - 7, // 6: glutton.Glutton.Ping:input_type -> glutton.PingRequest - 9, // 7: glutton.Glutton.Gossip:input_type -> glutton.GossipRequest - 2, // 8: glutton.Glutton.WriteRAM:output_type -> glutton.WriteRAMResponse - 4, // 9: glutton.Glutton.WriteDisk:output_type -> glutton.WriteDiskResponse - 6, // 10: glutton.Glutton.OpenFD:output_type -> glutton.OpenFDResponse - 8, // 11: glutton.Glutton.Ping:output_type -> glutton.PingResponse - 10, // 12: glutton.Glutton.Gossip:output_type -> glutton.GossipResponse - 8, // [8:13] is the sub-list for method output_type - 3, // [3:8] is the sub-list for method input_type - 3, // [3:3] is the sub-list for extension type_name - 3, // [3:3] is the sub-list for extension extendee - 0, // [0:3] is the sub-list for field type_name + 1, // 2: glutton.ReadDiskRequest.read_mode:type_name -> glutton.ReadMode + 14, // 3: glutton.GossipRequest.peers:type_name -> glutton.Peer + 2, // 4: glutton.Glutton.WriteRAM:input_type -> glutton.WriteRAMRequest + 4, // 5: glutton.Glutton.WriteDisk:input_type -> glutton.WriteDiskRequest + 6, // 6: glutton.Glutton.ReadDisk:input_type -> glutton.ReadDiskRequest + 8, // 7: glutton.Glutton.OpenFD:input_type -> glutton.OpenFDRequest + 10, // 8: glutton.Glutton.Ping:input_type -> glutton.PingRequest + 12, // 9: glutton.Glutton.Gossip:input_type -> glutton.GossipRequest + 3, // 10: glutton.Glutton.WriteRAM:output_type -> glutton.WriteRAMResponse + 5, // 11: glutton.Glutton.WriteDisk:output_type -> glutton.WriteDiskResponse + 7, // 12: glutton.Glutton.ReadDisk:output_type -> glutton.ReadDiskResponse + 9, // 13: glutton.Glutton.OpenFD:output_type -> glutton.OpenFDResponse + 11, // 14: glutton.Glutton.Ping:output_type -> glutton.PingResponse + 13, // 15: glutton.Glutton.Gossip:output_type -> glutton.GossipResponse + 10, // [10:16] is the sub-list for method output_type + 4, // [4:10] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name } func init() { file_glutton_proto_init() } @@ -681,8 +884,8 @@ func file_glutton_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_glutton_proto_rawDesc), len(file_glutton_proto_rawDesc)), - NumEnums: 1, - NumMessages: 11, + NumEnums: 2, + NumMessages: 13, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/proto/glutton/glutton.proto b/internal/proto/glutton/glutton.proto index f2b2c91c9..2a32926d6 100644 --- a/internal/proto/glutton/glutton.proto +++ b/internal/proto/glutton/glutton.proto @@ -30,6 +30,9 @@ service Glutton { // written will be random bytes. rpc WriteDisk(WriteDiskRequest) returns (WriteDiskResponse) {} + // Tells glutton to read from disk using the specified mode. + rpc ReadDisk(ReadDiskRequest) returns (ReadDiskResponse) {} + // Tells glutton to make sure it has the specified number of file // descriptors open. It will open or close file descriptors to // hit the desired count (note this count is in addition to the other @@ -52,6 +55,16 @@ enum WriteMode { WRITE_MODE_OVERWRITE = 1; } +// ReadMode selects how much of the file ReadDisk sends back. +enum ReadMode { + // Return the file's bytes in ReadDiskResponse.data alongside its digest. + READ_MODE_DATA = 0; + + // Return only size and sha256; ReadDiskResponse.data is empty. The digest + // is the same one READ_MODE_DATA returns, so either mode can verify. + READ_MODE_DIGEST_ONLY = 1; +} + message WriteRAMRequest { // name of the array to be written to string key = 1; @@ -76,6 +89,29 @@ message WriteDiskRequest { } message WriteDiskResponse { + // size of the file after the write + int64 size = 1; + + // sha256 of the whole file after the write + bytes sha256 = 2; +} + +message ReadDiskRequest { + // name of the file to be read from + string key = 1; + + ReadMode read_mode = 2; +} + +message ReadDiskResponse { + // total size of bytes read + int64 size = 1; + + // sha256 of bytes read + bytes sha256 = 2; + + // data read from the file. Empty under READ_MODE_DIGEST_ONLY. + bytes data = 3; } message OpenFDRequest { diff --git a/internal/proto/glutton/glutton_grpc.pb.go b/internal/proto/glutton/glutton_grpc.pb.go index 232546f89..6d5063570 100644 --- a/internal/proto/glutton/glutton_grpc.pb.go +++ b/internal/proto/glutton/glutton_grpc.pb.go @@ -35,6 +35,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( Glutton_WriteRAM_FullMethodName = "/glutton.Glutton/WriteRAM" Glutton_WriteDisk_FullMethodName = "/glutton.Glutton/WriteDisk" + Glutton_ReadDisk_FullMethodName = "/glutton.Glutton/ReadDisk" Glutton_OpenFD_FullMethodName = "/glutton.Glutton/OpenFD" Glutton_Ping_FullMethodName = "/glutton.Glutton/Ping" Glutton_Gossip_FullMethodName = "/glutton.Glutton/Gossip" @@ -54,6 +55,8 @@ type GluttonClient interface { // Tells glutton to write to disk using the specified mode. Data // written will be random bytes. WriteDisk(ctx context.Context, in *WriteDiskRequest, opts ...grpc.CallOption) (*WriteDiskResponse, error) + // Tells glutton to read from disk using the specified mode. + ReadDisk(ctx context.Context, in *ReadDiskRequest, opts ...grpc.CallOption) (*ReadDiskResponse, error) // Tells glutton to make sure it has the specified number of file // descriptors open. It will open or close file descriptors to // hit the desired count (note this count is in addition to the other @@ -94,6 +97,16 @@ func (c *gluttonClient) WriteDisk(ctx context.Context, in *WriteDiskRequest, opt return out, nil } +func (c *gluttonClient) ReadDisk(ctx context.Context, in *ReadDiskRequest, opts ...grpc.CallOption) (*ReadDiskResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ReadDiskResponse) + err := c.cc.Invoke(ctx, Glutton_ReadDisk_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *gluttonClient) OpenFD(ctx context.Context, in *OpenFDRequest, opts ...grpc.CallOption) (*OpenFDResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(OpenFDResponse) @@ -138,6 +151,8 @@ type GluttonServer interface { // Tells glutton to write to disk using the specified mode. Data // written will be random bytes. WriteDisk(context.Context, *WriteDiskRequest) (*WriteDiskResponse, error) + // Tells glutton to read from disk using the specified mode. + ReadDisk(context.Context, *ReadDiskRequest) (*ReadDiskResponse, error) // Tells glutton to make sure it has the specified number of file // descriptors open. It will open or close file descriptors to // hit the desired count (note this count is in addition to the other @@ -164,6 +179,9 @@ func (UnimplementedGluttonServer) WriteRAM(context.Context, *WriteRAMRequest) (* func (UnimplementedGluttonServer) WriteDisk(context.Context, *WriteDiskRequest) (*WriteDiskResponse, error) { return nil, status.Error(codes.Unimplemented, "method WriteDisk not implemented") } +func (UnimplementedGluttonServer) ReadDisk(context.Context, *ReadDiskRequest) (*ReadDiskResponse, error) { + return nil, status.Error(codes.Unimplemented, "method ReadDisk not implemented") +} func (UnimplementedGluttonServer) OpenFD(context.Context, *OpenFDRequest) (*OpenFDResponse, error) { return nil, status.Error(codes.Unimplemented, "method OpenFD not implemented") } @@ -230,6 +248,24 @@ func _Glutton_WriteDisk_Handler(srv interface{}, ctx context.Context, dec func(i return interceptor(ctx, in, info, handler) } +func _Glutton_ReadDisk_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ReadDiskRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(GluttonServer).ReadDisk(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Glutton_ReadDisk_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(GluttonServer).ReadDisk(ctx, req.(*ReadDiskRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _Glutton_OpenFD_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(OpenFDRequest) if err := dec(in); err != nil { @@ -299,6 +335,10 @@ var Glutton_ServiceDesc = grpc.ServiceDesc{ MethodName: "WriteDisk", Handler: _Glutton_WriteDisk_Handler, }, + { + MethodName: "ReadDisk", + Handler: _Glutton_ReadDisk_Handler, + }, { MethodName: "OpenFD", Handler: _Glutton_OpenFD_Handler,