From 1f08980becad10dd53503e941f34f3a9dc8072b4 Mon Sep 17 00:00:00 2001 From: raj-prince Date: Thu, 2 Apr 2026 09:18:59 +0000 Subject: [PATCH 1/4] Emulator based test for control client --- internal/storage/storage_handle.go | 3 +- .../control_client_stall_create_40s.yaml | 8 + .../control_client_stall_delete_40s.yaml | 8 + .../configs/control_client_stall_get_40s.yaml | 8 + .../control_client_stall_layout_60s.yaml | 8 + .../control_client_stall_rename_40s.yaml | 8 + .../control_client_stall_test.go | 248 ++++++++++++++++++ .../control_client_stall/setup_test.go | 39 +++ .../emulator_tests/emulator_tests.sh | 24 +- .../grpc_header_validation_test.go | 5 +- .../read_stall/read_stall_test.go | 10 +- .../common_failure_test.go | 9 +- .../write_stall/writes_stall_on_sync_test.go | 16 +- tools/proxy_server/grpc_proxy.go | 28 +- 14 files changed, 404 insertions(+), 18 deletions(-) create mode 100644 tools/integration_tests/emulator_tests/configs/control_client_stall_create_40s.yaml create mode 100644 tools/integration_tests/emulator_tests/configs/control_client_stall_delete_40s.yaml create mode 100644 tools/integration_tests/emulator_tests/configs/control_client_stall_get_40s.yaml create mode 100644 tools/integration_tests/emulator_tests/configs/control_client_stall_layout_60s.yaml create mode 100644 tools/integration_tests/emulator_tests/configs/control_client_stall_rename_40s.yaml create mode 100644 tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go create mode 100644 tools/integration_tests/emulator_tests/control_client_stall/setup_test.go diff --git a/internal/storage/storage_handle.go b/internal/storage/storage_handle.go index 18ee419c79..4502324337 100644 --- a/internal/storage/storage_handle.go +++ b/internal/storage/storage_handle.go @@ -23,6 +23,7 @@ import ( "os" "strconv" "strings" + "time" "cloud.google.com/go/storage" @@ -416,8 +417,6 @@ func NewStorageHandle(ctx context.Context, clientConfig storageutil.StorageClien // Wrap the control client with retry-on-stall logic. // This will retry on only on GetStorageLayout call for all buckets. controlClient = withRetryOnStorageLayout(controlClientWithBillingProject, &clientConfig) - } else { - logger.Infof("Skipping storage control client creation because custom-endpoint %q was passed, which is assumed to be a storage testbench server because of 'localhost' in it.", clientConfig.CustomEndpoint) } sh = &storageClient{ diff --git a/tools/integration_tests/emulator_tests/configs/control_client_stall_create_40s.yaml b/tools/integration_tests/emulator_tests/configs/control_client_stall_create_40s.yaml new file mode 100644 index 0000000000..a6355f10ab --- /dev/null +++ b/tools/integration_tests/emulator_tests/configs/control_client_stall_create_40s.yaml @@ -0,0 +1,8 @@ +proxyType: grpc +targetHost: "localhost:8888" +retryConfig: +- method: CreateFolder + retryInstruction: "stall-for-40s" + retryCount: 1 + skipCount: 2 + diff --git a/tools/integration_tests/emulator_tests/configs/control_client_stall_delete_40s.yaml b/tools/integration_tests/emulator_tests/configs/control_client_stall_delete_40s.yaml new file mode 100644 index 0000000000..77614e7d2c --- /dev/null +++ b/tools/integration_tests/emulator_tests/configs/control_client_stall_delete_40s.yaml @@ -0,0 +1,8 @@ +proxyType: grpc +targetHost: "localhost:8888" +retryConfig: +- method: DeleteFolder + retryInstruction: "stall-for-40s" + retryCount: 1 + skipCount: 0 + diff --git a/tools/integration_tests/emulator_tests/configs/control_client_stall_get_40s.yaml b/tools/integration_tests/emulator_tests/configs/control_client_stall_get_40s.yaml new file mode 100644 index 0000000000..cdc98c9b68 --- /dev/null +++ b/tools/integration_tests/emulator_tests/configs/control_client_stall_get_40s.yaml @@ -0,0 +1,8 @@ +proxyType: grpc +targetHost: "localhost:8888" +retryConfig: +- method: GetFolder + retryInstruction: "stall-for-40s" + retryCount: 1 + skipCount: 0 + diff --git a/tools/integration_tests/emulator_tests/configs/control_client_stall_layout_60s.yaml b/tools/integration_tests/emulator_tests/configs/control_client_stall_layout_60s.yaml new file mode 100644 index 0000000000..51c2a53ed5 --- /dev/null +++ b/tools/integration_tests/emulator_tests/configs/control_client_stall_layout_60s.yaml @@ -0,0 +1,8 @@ +proxyType: grpc +targetHost: "localhost:8888" +retryConfig: +- method: GetStorageLayout + retryInstruction: "stall-for-60s" + retryCount: 1 + skipCount: 0 + diff --git a/tools/integration_tests/emulator_tests/configs/control_client_stall_rename_40s.yaml b/tools/integration_tests/emulator_tests/configs/control_client_stall_rename_40s.yaml new file mode 100644 index 0000000000..2df226b530 --- /dev/null +++ b/tools/integration_tests/emulator_tests/configs/control_client_stall_rename_40s.yaml @@ -0,0 +1,8 @@ +proxyType: grpc +targetHost: "localhost:8888" +retryConfig: +- method: RenameFolder + retryInstruction: "stall-for-40s" + retryCount: 1 + skipCount: 0 + diff --git a/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go b/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go new file mode 100644 index 0000000000..f99a8e8e0c --- /dev/null +++ b/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go @@ -0,0 +1,248 @@ +// 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 control_client_stall + +import ( + "fmt" + "log" + "os" + "path" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + + emulator_tests "github.com/googlecloudplatform/gcsfuse/v3/tools/integration_tests/emulator_tests/util" + "github.com/googlecloudplatform/gcsfuse/v3/tools/integration_tests/util/mounting/static_mounting" + "github.com/googlecloudplatform/gcsfuse/v3/tools/integration_tests/util/setup" +) + +type controlClientStallBase struct { + port int + proxyProcessId int + proxyServerLogFile string + flags []string + configFileName string + maxMountDurationSecs int + testDirPath string + suite.Suite +} + +func (c *controlClientStallBase) SetupTest() { + c.testDirPath = setup.SetupTestDirectory(c.T().Name()) + c.proxyServerLogFile = setup.CreateProxyServerLogFile(c.T()) + var err error + c.port, c.proxyProcessId, err = emulator_tests.StartProxyServer(c.configFileName, c.proxyServerLogFile) + require.NoError(c.T(), err) + + // Add proxy server endpoint and configure gRPC testing + // Copy flags to avoid mutating the original slice across suites + mountFlags := append([]string(nil), c.flags...) + mountFlags = append(mountFlags, fmt.Sprintf("--custom-endpoint=localhost:%d", c.port)) + mountFlags = append(mountFlags, "--anonymous-access") // Required for gRPC localhost endpoint + + startTime := time.Now() + setup.MountGCSFuseWithGivenMountFunc(mountFlags, static_mounting.MountGcsfuseWithStaticMounting) + mountTime := time.Since(startTime) + if c.maxMountDurationSecs != -1 { + assert.True(c.T(), mountTime < time.Duration(c.maxMountDurationSecs)*time.Second, "Mount time %v should be less than %ds", mountTime, c.maxMountDurationSecs) + } +} + +func (c *controlClientStallBase) TearDownTest() { + setup.UnmountGCSFuse(setup.MntDir()) + assert.NoError(c.T(), emulator_tests.KillProxyServerProcess(c.proxyProcessId)) + setup.SaveGCSFuseLogFileInCaseOfFailure(c.T()) + setup.SaveProxyServerLogFileInCaseOfFailure(c.proxyServerLogFile, c.T()) +} + +// --- Test Suites --- + +type createFolderStallSuite struct{ controlClientStallBase } + +// TestCreateFolderStallInducedShouldComplete verifies that creating a folder via +// os.MkdirAll completes successfully even when a stall is induced, +// proving that the experimental retry logic correctly handles the delayed gRPC response. +func (c *createFolderStallSuite) TestCreateFolderStallInducedShouldComplete() { + folderPath := path.Join(c.testDirPath, "stalled_folder_for_create") + + startTime := time.Now() + err := os.MkdirAll(folderPath, 0755) + elapsedTime := time.Since(startTime) + + assert.NoError(c.T(), err) + // Ensure the CreateFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. + assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) + // Validate the directory actually exists + info, err := os.Stat(folderPath) + assert.NoError(c.T(), err) + assert.True(c.T(), info.IsDir()) +} + +type getFolderStallSuite struct{ controlClientStallBase } + +// TestGetFolderStallInducedShouldComplete verifies that stating a pre-created folder via +// os.Stat completes successfully even when a stall is induced, +// proving that the experimental retry logic correctly handles the delayed gRPC response. +func (c *getFolderStallSuite) TestGetFolderStallInducedShouldComplete() { + folderPath := path.Join(c.testDirPath, "stalled_folder_for_get") + err := os.MkdirAll(folderPath, 0755) + assert.NoError(c.T(), err) + + startTime := time.Now() + _, err = os.Stat(folderPath) + elapsedTime := time.Since(startTime) + + assert.NoError(c.T(), err) + // Ensure the GetFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. + assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) +} + +type deleteFolderStallSuite struct{ controlClientStallBase } + +// TestDeleteFolderStallInducedShouldComplete verifies that deleting a pre-created empty folder via +// os.Remove completes successfully even when a stall is induced, +// proving that the experimental retry logic correctly handles the delayed gRPC response. +func (c *deleteFolderStallSuite) TestDeleteFolderStallInducedShouldComplete() { + folderPath := path.Join(c.testDirPath, "stalled_folder_for_delete") + err := os.MkdirAll(folderPath, 0755) + assert.NoError(c.T(), err) + + startTime := time.Now() + err = os.Remove(folderPath) + elapsedTime := time.Since(startTime) + + assert.NoError(c.T(), err) + // Ensure the DeleteFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. + assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) + // Validate the directory actually deleted + _, err = os.Stat(folderPath) + assert.True(c.T(), os.IsNotExist(err)) +} + +type renameFolderStallSuite struct{ controlClientStallBase } + +// TestRenameFolderStallInducedShouldComplete verifies that renaming a pre-created folder via +// os.Rename completes successfully even when a stall is induced, +// proving that the experimental retry logic correctly handles the delayed gRPC response. +func (c *renameFolderStallSuite) TestRenameFolderStallInducedShouldComplete() { + folderPath := path.Join(c.testDirPath, "stalled_folder_for_rename") + err := os.MkdirAll(folderPath, 0755) + assert.NoError(c.T(), err) + destPath := path.Join(c.testDirPath, "stalled_folder_renamed") + + startTime := time.Now() + err = os.Rename(folderPath, destPath) + elapsedTime := time.Since(startTime) + + assert.NoError(c.T(), err) + // Ensure the RenameFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. + assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) + // Validate the directory actually renamed + _, err = os.Stat(destPath) + assert.NoError(c.T(), err) +} + +type getStorageLayoutStallSuite struct{ controlClientStallBase } + +// TestGetStorageLayoutStallInducedShouldComplete verifies that a GCSFuse mount +// completes successfully even when a stall is induced in the very first GetStorageLayout call, +// proving that the experimental retry logic correctly handles the delayed gRPC response. +func (c *getStorageLayoutStallSuite) TestGetStorageLayoutStallInducedShouldComplete() { + // The stall for GetStorageLayout is actually triggered during the bucket mount + // phase within SetupTest() when GCSFuse queries the bucket layout to determine + // if Hierarchical Namespace (HNS) is enabled. + // The fact that this test function is executing means the mount succeeded, + // effectively proving that GetStorageLayout correctly timed out after 30s, + // retried successfully, and allowed the mount to complete! + + // Just perform a basic verification to ensure the mount is functional. + folderPath := path.Join(c.testDirPath, "stalled_folder_for_layout") + err := os.MkdirAll(folderPath, 0755) + assert.NoError(c.T(), err) + // Validate the directory actually exists + info, err := os.Stat(folderPath) + assert.NoError(c.T(), err) + assert.True(c.T(), info.IsDir()) +} + +func TestControlClientStall(t *testing.T) { + // Test matrix + flagsSet := [][]string{ + { + "--client-protocol=grpc", + "--rename-dir-limit=0", // Force use of Control API for folders instead of legacy workarounds + "--experimental-nonrapid-folder-api-stall-retry=true", // Enable the new flag we are testing! + "--metadata-cache-ttl-secs=0", // Disable cache so calls definitely hit the backend + }, + } + + for _, flags := range flagsSet { + suites := []suite.TestingSuite{ + &createFolderStallSuite{ + controlClientStallBase: controlClientStallBase{ + flags: flags, + // Artificially stall the very first CreateFolder call by 40 seconds. + configFileName: "../configs/control_client_stall_create_40s.yaml", + // No check on how long mount takes. + maxMountDurationSecs: -1, + }, + }, + &getFolderStallSuite{ + controlClientStallBase: controlClientStallBase{ + flags: flags, + // Artificially stall the very first GetFolder call by 40 seconds. + configFileName: "../configs/control_client_stall_get_40s.yaml", + // No check on how long mount takes. + maxMountDurationSecs: -1, + }, + }, + &deleteFolderStallSuite{ + controlClientStallBase: controlClientStallBase{ + flags: flags, + // Artificially stall the very first DeleteFolder call by 40 seconds. + configFileName: "../configs/control_client_stall_delete_40s.yaml", + // No check on how long mount takes. + maxMountDurationSecs: -1, + }, + }, + &renameFolderStallSuite{ + controlClientStallBase: controlClientStallBase{ + flags: flags, + // Artificially stall the very first RenameFolder call by 40 seconds. + configFileName: "../configs/control_client_stall_rename_40s.yaml", + // No check on how long mount takes. + maxMountDurationSecs: -1, + }, + }, + &getStorageLayoutStallSuite{ + controlClientStallBase: controlClientStallBase{ + flags: flags, + // Artificially stall the very first GetStorageLayout call by 60 seconds. + configFileName: "../configs/control_client_stall_layout_60s.yaml", + // Ensure that the whole mount operation including a stalled GetStorageLayout call took less than 40 seconds (30 seconds first stalled-call GetStorageLayout timeout + 10 second wiggle-room for 2nd attempt and other steps for mount to go through) because of the internal abortion and retry. + maxMountDurationSecs: 40, + }, + }, + } + + for _, s := range suites { + log.Printf("Running Control Client Stall test suite %T with flags: %s", s, flags) + suite.Run(t, s) + } + } +} diff --git a/tools/integration_tests/emulator_tests/control_client_stall/setup_test.go b/tools/integration_tests/emulator_tests/control_client_stall/setup_test.go new file mode 100644 index 0000000000..1dbbff5f35 --- /dev/null +++ b/tools/integration_tests/emulator_tests/control_client_stall/setup_test.go @@ -0,0 +1,39 @@ +// 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 control_client_stall + +import ( + "log" + "os" + "testing" + + "github.com/googlecloudplatform/gcsfuse/v3/tools/integration_tests/util/setup" +) + +func TestMain(m *testing.M) { + setup.ParseSetUpFlags() + + if setup.MountedDirectory() != "" { + log.Printf("These tests will not run with mounted directory.") + return + } + + // Set up test directory. + setup.SetUpTestDirForTestBucketFlag() + + log.Println("Running control client stall tests...") + successCode := m.Run() + os.Exit(successCode) +} diff --git a/tools/integration_tests/emulator_tests/emulator_tests.sh b/tools/integration_tests/emulator_tests/emulator_tests.sh index 2d79711d0a..a1611648a1 100755 --- a/tools/integration_tests/emulator_tests/emulator_tests.sh +++ b/tools/integration_tests/emulator_tests/emulator_tests.sh @@ -118,8 +118,9 @@ sudo docker pull $DOCKER_IMAGE CONTAINER_ID=$(sudo docker ps -aqf "name=$CONTAINER_NAME") if [[ -n "$CONTAINER_ID" ]]; then log_info "Container with ID:[$CONTAINER_ID] is already running with name:[$CONTAINER_NAME]" - log_info "Stopping container...." - sudo docker stop $CONTAINER_ID + log_info "Stopping and removing container...." + docker stop $CONTAINER_ID || true + docker rm $CONTAINER_ID || true fi wait_for_emulator() { @@ -175,6 +176,17 @@ if ! curl -X POST --data-binary @test.json \ fi rm test.json +# Create an HNS bucket for control client tests +cat << EOF > test_hns.json +{"name":"test-hns-bucket", "hierarchicalNamespace": {"enabled": true}} +EOF +if ! curl -X POST --data-binary @test_hns.json -H "Content-Type: application/json" "$STORAGE_EMULATOR_HOST/storage/v1/b?project=test-project"; then + log_error "Failed to create bucket test-hns-bucket" + exit 1 +fi +rm test_hns.json + + # Start the gRPC server on port 8888. log_info "Starting the gRPC server on port 8888" response=$(curl -w "%{http_code}\n" --retry 5 --retry-max-time 40 -o /dev/null "$STORAGE_EMULATOR_HOST/start_grpc?port=8888") @@ -192,4 +204,10 @@ if [[ -n "$GCSFUSE_PREBUILT_DIR" ]]; then fi # Run all emulator test packages in sequence to avoid high cpu usage. -go test -v -p 1 -timeout 10m ./tools/integration_tests/emulator_tests/... --integrationTest --testbucket=test-bucket "${args[@]}" +TEST_TARGET=${TEST_TARGET:-"./tools/integration_tests/emulator_tests/..."} +# Run all emulator test packages in sequence to avoid high cpu usage. +# Run all other emulator tests with standard bucket +go test -v -p 1 -timeout 10m $(go list ${TEST_TARGET} | grep -v control_client_stall) --integrationTest --testbucket=${TEST_BUCKET:-test-bucket} "${args[@]}" + +# Run control_client_stall with HNS bucket +go test -v -p 1 -timeout 10m ./tools/integration_tests/emulator_tests/control_client_stall/... --integrationTest --testbucket=test-hns-bucket "${args[@]}" diff --git a/tools/integration_tests/emulator_tests/grpc_header_validation/grpc_header_validation_test.go b/tools/integration_tests/emulator_tests/grpc_header_validation/grpc_header_validation_test.go index 5199098949..be892417b7 100644 --- a/tools/integration_tests/emulator_tests/grpc_header_validation/grpc_header_validation_test.go +++ b/tools/integration_tests/emulator_tests/grpc_header_validation/grpc_header_validation_test.go @@ -117,7 +117,10 @@ func TestGRPCHeaderValidation(t *testing.T) { // 2. Bucket named "test-bucket" created in the testbench // 3. Run with: go test -v --integrationTest --testbucket=test-bucket -timeout=5m flagsSet := [][]string{ - {"--client-protocol=grpc"}, + { + "--client-protocol=grpc", + "--enable-hns=false", + }, } for _, flags := range flagsSet { diff --git a/tools/integration_tests/emulator_tests/read_stall/read_stall_test.go b/tools/integration_tests/emulator_tests/read_stall/read_stall_test.go index adf77dadd1..ed8f10da2c 100644 --- a/tools/integration_tests/emulator_tests/read_stall/read_stall_test.go +++ b/tools/integration_tests/emulator_tests/read_stall/read_stall_test.go @@ -93,7 +93,15 @@ func TestReadStall(t *testing.T) { ts := &readStall{} // Define flag set to run the tests. flagsSet := [][]string{ - {"--enable-read-stall-retry=true", "--read-stall-min-req-timeout=" + fmt.Sprintf("%dms", minReqTimeout.Milliseconds()), "--read-stall-initial-req-timeout=" + fmt.Sprintf("%dms", minReqTimeout.Milliseconds())}, + { + "--enable-read-stall-retry=true", + "--read-stall-min-req-timeout=" + fmt.Sprintf("%dms", minReqTimeout.Milliseconds()), + "--read-stall-initial-req-timeout=" + fmt.Sprintf("%dms", minReqTimeout.Milliseconds()), + // Disable HNS to prevent gRPC Control Client initialization. + // Legacy emulator proxy servers run on HTTP/1.1, which causes the + // gRPC HTTP/2 dialer to crash the mount sequence. + "--enable-hns=false", + }, } // Run tests. diff --git a/tools/integration_tests/emulator_tests/streaming_writes_failure/common_failure_test.go b/tools/integration_tests/emulator_tests/streaming_writes_failure/common_failure_test.go index 673bdeb1a0..84d15172f6 100644 --- a/tools/integration_tests/emulator_tests/streaming_writes_failure/common_failure_test.go +++ b/tools/integration_tests/emulator_tests/streaming_writes_failure/common_failure_test.go @@ -61,7 +61,14 @@ type gcsObjectValidator interface { // ////////////////////////////////////////////////////////////////////// func (t *commonFailureTestSuite) SetupSuite() { - t.flags = []string{"--write-block-size-mb=1", "--write-max-blocks-per-file=1"} + t.flags = []string{ + "--write-block-size-mb=1", + "--write-max-blocks-per-file=1", + // Disable HNS to prevent gRPC Control Client initialization. + // Legacy emulator proxy servers run on HTTP/1.1, which causes the + // gRPC HTTP/2 dialer to crash the mount sequence. + "--enable-hns=false", + } // Generate 5 MB random data. var err error t.data, err = operations.GenerateRandomData(5 * operations.MiB) diff --git a/tools/integration_tests/emulator_tests/write_stall/writes_stall_on_sync_test.go b/tools/integration_tests/emulator_tests/write_stall/writes_stall_on_sync_test.go index 08d5c90750..809a8b0fab 100644 --- a/tools/integration_tests/emulator_tests/write_stall/writes_stall_on_sync_test.go +++ b/tools/integration_tests/emulator_tests/write_stall/writes_stall_on_sync_test.go @@ -90,7 +90,13 @@ func TestChunkTransferTimeoutInfinity(t *testing.T) { ts := &chunkTransferTimeoutInfinity{} // Define flag set to run the tests. flagsSet := [][]string{ - {"--chunk-transfer-timeout-secs=0"}, + { + "--chunk-transfer-timeout-secs=0", + // Disable HNS to prevent gRPC Control Client initialization. + // Legacy emulator proxy servers run on HTTP/1.1, which causes the + // gRPC HTTP/2 dialer to crash the mount sequence. + "--enable-hns=false", + }, } // Run tests. @@ -103,8 +109,8 @@ func TestChunkTransferTimeoutInfinity(t *testing.T) { func TestChunkTransferTimeout(t *testing.T) { flagSets := [][]string{ - {}, - {"--chunk-transfer-timeout-secs=5"}, + {"--enable-hns=false"}, + {"--chunk-transfer-timeout-secs=5", "--enable-hns=false"}, } stallScenarios := []struct { @@ -176,13 +182,13 @@ func TestChunkRetryDeadline(t *testing.T) { }{ { name: "StallUnderDeadline_Pass", - flags: []string{"--chunk-transfer-timeout-secs=10", "--chunk-retry-deadline-secs=120"}, + flags: []string{"--chunk-transfer-timeout-secs=10", "--chunk-retry-deadline-secs=120", "--enable-hns=false"}, expectedStall: 40 * time.Second, expectedSuccess: true, }, { name: "StallOverDeadline_Fail", - flags: []string{"--chunk-transfer-timeout-secs=10", "--chunk-retry-deadline-secs=32"}, + flags: []string{"--chunk-transfer-timeout-secs=10", "--chunk-retry-deadline-secs=32", "--enable-hns=false"}, expectedStall: 40 * time.Second, expectedSuccess: false, }, diff --git a/tools/proxy_server/grpc_proxy.go b/tools/proxy_server/grpc_proxy.go index be0998e1c7..5859cff232 100644 --- a/tools/proxy_server/grpc_proxy.go +++ b/tools/proxy_server/grpc_proxy.go @@ -117,12 +117,30 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head // Extract and validate metadata md, ok := metadata.FromIncomingContext(stream.Context()) if ok { - if err := validateGRPCMetadata(md, validations); err != nil { - log.Printf("Metadata validation failed: %v", err) - return err + // Only validate headers for the main Storage API, not Control API + if strings.HasPrefix(fullMethodName, "/google.storage.v2.Storage/") { + if err := validateGRPCMetadata(md, validations); err != nil { + log.Printf("Metadata validation failed: %v", err) + return err + } + } + } else { + md = metadata.New(nil) + if len(validations) > 0 && strings.HasPrefix(fullMethodName, "/google.storage.v2.Storage/") { + log.Println("Warning: No metadata found in request") + } + } + + // Inject testbench instructions via gOpManager + parts := strings.Split(fullMethodName, "/") + methodName := parts[len(parts)-1] + plantOp := gOpManager.retrieveOperation(RequestType(methodName)) + if plantOp != "" { + if *fDebug { + log.Printf("Planting operation: %s for method: %s", plantOp, fullMethodName) } - } else if len(validations) > 0 { - log.Println("Warning: No metadata found in request") + // Inject direct instruction for testbench + md = metadata.Join(md, metadata.Pairs("x-goog-emulator-instructions", plantOp)) } // Forward metadata to target From 06a14827d3d3926ad77bde0925e4157545abbf04 Mon Sep 17 00:00:00 2001 From: raj-prince Date: Tue, 14 Jul 2026 18:06:50 +0000 Subject: [PATCH 2/4] simplifying the code --- internal/storage/storage_handle.go | 4 +- .../control_client_stall_test.go | 169 ++++++++-------- tools/proxy_server/grpc_proxy.go | 187 ++++++++++-------- 3 files changed, 190 insertions(+), 170 deletions(-) diff --git a/internal/storage/storage_handle.go b/internal/storage/storage_handle.go index 4502324337..0d83896195 100644 --- a/internal/storage/storage_handle.go +++ b/internal/storage/storage_handle.go @@ -22,7 +22,6 @@ import ( "net/http" "os" "strconv" - "strings" "time" @@ -390,8 +389,7 @@ func NewStorageHandle(ctx context.Context, clientConfig storageutil.StorageClien var clientOpts []option.ClientOption // Control-client is needed for folder APIs and for getting storage-layout of the bucket. - // GetStorageLayout API is not supported for storage-testbench, which are identified by custom-endpoint containing localhost. - if clientConfig.EnableHNS && !strings.Contains(clientConfig.CustomEndpoint, "localhost") { + if clientConfig.EnableHNS { // For control client, we don't pass billingProject to avoid setting it globally via option.WithQuotaProject. // The wrapper storageControlClientWithBillingProject will manually add it to the context for supported calls. clientOpts, err = createClientOptionForGRPCClient(ctx, &clientConfig, false) diff --git a/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go b/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go index f99a8e8e0c..b4e8b5d922 100644 --- a/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go +++ b/tools/integration_tests/emulator_tests/control_client_stall/control_client_stall_test.go @@ -16,7 +16,6 @@ package control_client_stall import ( "fmt" - "log" "os" "path" "testing" @@ -42,6 +41,8 @@ type controlClientStallBase struct { suite.Suite } +const maxOperationDuration = 32 * time.Second + func (c *controlClientStallBase) SetupTest() { c.testDirPath = setup.SetupTestDirectory(c.T().Name()) c.proxyServerLogFile = setup.CreateProxyServerLogFile(c.T()) @@ -65,11 +66,31 @@ func (c *controlClientStallBase) SetupTest() { func (c *controlClientStallBase) TearDownTest() { setup.UnmountGCSFuse(setup.MntDir()) - assert.NoError(c.T(), emulator_tests.KillProxyServerProcess(c.proxyProcessId)) + if c.proxyProcessId > 0 { + assert.NoError(c.T(), emulator_tests.KillProxyServerProcess(c.proxyProcessId)) + } setup.SaveGCSFuseLogFileInCaseOfFailure(c.T()) setup.SaveProxyServerLogFileInCaseOfFailure(c.proxyServerLogFile, c.T()) } +func (c *controlClientStallBase) assertCompletedWithinThreshold(elapsedTime time.Duration) { + assert.True(c.T(), elapsedTime < maxOperationDuration, "Elapsed time %v should be less than %v", elapsedTime, maxOperationDuration) +} + +func (c *controlClientStallBase) assertDirectoryExists(folderPath string) { + info, err := os.Stat(folderPath) + assert.NoError(c.T(), err) + assert.True(c.T(), info.IsDir()) +} + +func newControlClientStallBase(flags []string, configFileName string, maxMountDurationSecs int) controlClientStallBase { + return controlClientStallBase{ + flags: flags, + configFileName: configFileName, + maxMountDurationSecs: maxMountDurationSecs, + } +} + // --- Test Suites --- type createFolderStallSuite struct{ controlClientStallBase } @@ -85,12 +106,9 @@ func (c *createFolderStallSuite) TestCreateFolderStallInducedShouldComplete() { elapsedTime := time.Since(startTime) assert.NoError(c.T(), err) - // Ensure the CreateFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. - assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) - // Validate the directory actually exists - info, err := os.Stat(folderPath) - assert.NoError(c.T(), err) - assert.True(c.T(), info.IsDir()) + // Ensure the operation completes under 32s (30s timeout + 2s wiggle-room) because of internal abort+retry. + c.assertCompletedWithinThreshold(elapsedTime) + c.assertDirectoryExists(folderPath) } type getFolderStallSuite struct{ controlClientStallBase } @@ -108,8 +126,8 @@ func (c *getFolderStallSuite) TestGetFolderStallInducedShouldComplete() { elapsedTime := time.Since(startTime) assert.NoError(c.T(), err) - // Ensure the GetFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. - assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) + // Ensure the operation completes under 32s (30s timeout + 2s wiggle-room) because of internal abort+retry. + c.assertCompletedWithinThreshold(elapsedTime) } type deleteFolderStallSuite struct{ controlClientStallBase } @@ -127,8 +145,8 @@ func (c *deleteFolderStallSuite) TestDeleteFolderStallInducedShouldComplete() { elapsedTime := time.Since(startTime) assert.NoError(c.T(), err) - // Ensure the DeleteFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. - assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) + // Ensure the operation completes under 32s (30s timeout + 2s wiggle-room) because of internal abort+retry. + c.assertCompletedWithinThreshold(elapsedTime) // Validate the directory actually deleted _, err = os.Stat(folderPath) assert.True(c.T(), os.IsNotExist(err)) @@ -150,11 +168,9 @@ func (c *renameFolderStallSuite) TestRenameFolderStallInducedShouldComplete() { elapsedTime := time.Since(startTime) assert.NoError(c.T(), err) - // Ensure the RenameFolder operation took less than 32 seconds (30 seconds first stalled-call timeout + 2 second wiggle-room) because of the internal abortion and retry. - assert.True(c.T(), elapsedTime < 32*time.Second, "Elapsed time %v should be less than 32s", elapsedTime) - // Validate the directory actually renamed - _, err = os.Stat(destPath) - assert.NoError(c.T(), err) + // Ensure the operation completes under 32s (30s timeout + 2s wiggle-room) because of internal abort+retry. + c.assertCompletedWithinThreshold(elapsedTime) + c.assertDirectoryExists(destPath) } type getStorageLayoutStallSuite struct{ controlClientStallBase } @@ -174,75 +190,66 @@ func (c *getStorageLayoutStallSuite) TestGetStorageLayoutStallInducedShouldCompl folderPath := path.Join(c.testDirPath, "stalled_folder_for_layout") err := os.MkdirAll(folderPath, 0755) assert.NoError(c.T(), err) - // Validate the directory actually exists - info, err := os.Stat(folderPath) - assert.NoError(c.T(), err) - assert.True(c.T(), info.IsDir()) + c.assertDirectoryExists(folderPath) } func TestControlClientStall(t *testing.T) { - // Test matrix - flagsSet := [][]string{ - { - "--client-protocol=grpc", - "--rename-dir-limit=0", // Force use of Control API for folders instead of legacy workarounds - "--experimental-nonrapid-folder-api-stall-retry=true", // Enable the new flag we are testing! - "--metadata-cache-ttl-secs=0", // Disable cache so calls definitely hit the backend + flags := []string{ + "--client-protocol=grpc", + "--rename-dir-limit=0", // Force use of Control API for folders instead of legacy workarounds + "--experimental-nonrapid-folder-api-stall-retry=true", // Enable the new flag we are testing! + "--metadata-cache-ttl-secs=0", // Disable cache so calls definitely hit the backend + } + + suites := []suite.TestingSuite{ + &createFolderStallSuite{ + controlClientStallBase: newControlClientStallBase( + flags, + // Artificially stall the very first CreateFolder call by 40 seconds. + "../configs/control_client_stall_create_40s.yaml", + // No check on how long mount takes. + -1, + ), + }, + &getFolderStallSuite{ + controlClientStallBase: newControlClientStallBase( + flags, + // Artificially stall the very first GetFolder call by 40 seconds. + "../configs/control_client_stall_get_40s.yaml", + // No check on how long mount takes. + -1, + ), + }, + &deleteFolderStallSuite{ + controlClientStallBase: newControlClientStallBase( + flags, + // Artificially stall the very first DeleteFolder call by 40 seconds. + "../configs/control_client_stall_delete_40s.yaml", + // No check on how long mount takes. + -1, + ), + }, + &renameFolderStallSuite{ + controlClientStallBase: newControlClientStallBase( + flags, + // Artificially stall the very first RenameFolder call by 40 seconds. + "../configs/control_client_stall_rename_40s.yaml", + // No check on how long mount takes. + -1, + ), + }, + &getStorageLayoutStallSuite{ + controlClientStallBase: newControlClientStallBase( + flags, + // Artificially stall the very first GetStorageLayout call by 60 seconds. + "../configs/control_client_stall_layout_60s.yaml", + // Ensure that the whole mount operation including a stalled GetStorageLayout call took less than 40 seconds (30 seconds first stalled-call GetStorageLayout timeout + 10 second wiggle-room for 2nd attempt and other steps for mount to go through) because of the internal abortion and retry. + 40, + ), }, } - for _, flags := range flagsSet { - suites := []suite.TestingSuite{ - &createFolderStallSuite{ - controlClientStallBase: controlClientStallBase{ - flags: flags, - // Artificially stall the very first CreateFolder call by 40 seconds. - configFileName: "../configs/control_client_stall_create_40s.yaml", - // No check on how long mount takes. - maxMountDurationSecs: -1, - }, - }, - &getFolderStallSuite{ - controlClientStallBase: controlClientStallBase{ - flags: flags, - // Artificially stall the very first GetFolder call by 40 seconds. - configFileName: "../configs/control_client_stall_get_40s.yaml", - // No check on how long mount takes. - maxMountDurationSecs: -1, - }, - }, - &deleteFolderStallSuite{ - controlClientStallBase: controlClientStallBase{ - flags: flags, - // Artificially stall the very first DeleteFolder call by 40 seconds. - configFileName: "../configs/control_client_stall_delete_40s.yaml", - // No check on how long mount takes. - maxMountDurationSecs: -1, - }, - }, - &renameFolderStallSuite{ - controlClientStallBase: controlClientStallBase{ - flags: flags, - // Artificially stall the very first RenameFolder call by 40 seconds. - configFileName: "../configs/control_client_stall_rename_40s.yaml", - // No check on how long mount takes. - maxMountDurationSecs: -1, - }, - }, - &getStorageLayoutStallSuite{ - controlClientStallBase: controlClientStallBase{ - flags: flags, - // Artificially stall the very first GetStorageLayout call by 60 seconds. - configFileName: "../configs/control_client_stall_layout_60s.yaml", - // Ensure that the whole mount operation including a stalled GetStorageLayout call took less than 40 seconds (30 seconds first stalled-call GetStorageLayout timeout + 10 second wiggle-room for 2nd attempt and other steps for mount to go through) because of the internal abortion and retry. - maxMountDurationSecs: 40, - }, - }, - } - - for _, s := range suites { - log.Printf("Running Control Client Stall test suite %T with flags: %s", s, flags) - suite.Run(t, s) - } + for _, s := range suites { + suite.Run(t, s) } } diff --git a/tools/proxy_server/grpc_proxy.go b/tools/proxy_server/grpc_proxy.go index 5859cff232..ada0b3a7c3 100644 --- a/tools/proxy_server/grpc_proxy.go +++ b/tools/proxy_server/grpc_proxy.go @@ -15,6 +15,7 @@ package main import ( + "context" "fmt" "io" "log" @@ -57,6 +58,8 @@ func init() { encoding.RegisterCodec(rawCodec{}) } +const storageServicePrefix = "/google.storage.v2.Storage/" + // validateGRPCMetadata validates gRPC metadata based on configured header validations func validateGRPCMetadata(md metadata.MD, validations []HeaderValidation) error { for _, validation := range validations { @@ -92,6 +95,98 @@ func validateGRPCMetadata(md metadata.MD, validations []HeaderValidation) error return nil } +func shouldValidateMetadata(fullMethodName string) bool { + return strings.HasPrefix(fullMethodName, storageServicePrefix) +} + +func methodNameFromFullMethod(fullMethodName string) string { + idx := strings.LastIndex(fullMethodName, "/") + if idx < 0 || idx == len(fullMethodName)-1 { + return fullMethodName + } + return fullMethodName[idx+1:] +} + +func buildForwardContext(ctx context.Context, fullMethodName string, validations []HeaderValidation) (context.Context, error) { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + md = metadata.New(nil) + if len(validations) > 0 && shouldValidateMetadata(fullMethodName) { + log.Println("Warning: No metadata found in request") + } + } + + if shouldValidateMetadata(fullMethodName) { + if err := validateGRPCMetadata(md, validations); err != nil { + log.Printf("Metadata validation failed: %v", err) + return nil, err + } + } + + plantOp := gOpManager.retrieveOperation(RequestType(methodNameFromFullMethod(fullMethodName))) + if plantOp != "" { + if *fDebug { + log.Printf("Planting operation: %s for method: %s", plantOp, fullMethodName) + } + md = metadata.Join(md, metadata.Pairs("x-goog-emulator-instructions", plantOp)) + } + + return metadata.NewOutgoingContext(ctx, md), nil +} + +func proxyClientToTarget(stream grpc.ServerStream, clientStream grpc.ClientStream) error { + for { + req := new([]byte) + if err := stream.RecvMsg(req); err != nil { + if err == io.EOF { + return clientStream.CloseSend() + } + return fmt.Errorf("receiving from client: %w", err) + } + if err := clientStream.SendMsg(req); err != nil { + return fmt.Errorf("sending to server: %w", err) + } + } +} + +func proxyTargetToClient(clientStream grpc.ClientStream, stream grpc.ServerStream) error { + for { + resp := new([]byte) + if err := clientStream.RecvMsg(resp); err != nil { + if err == io.EOF { + return nil + } + return fmt.Errorf("receiving from server: %w", err) + } + if err := stream.SendMsg(resp); err != nil { + return fmt.Errorf("sending to client: %w", err) + } + } +} + +func waitForProxyCompletion(clientToServer, serverToClient <-chan error) error { + var err1, err2 error + for i := 0; i < 2; i++ { + select { + case err := <-clientToServer: + if err != nil { + log.Printf("Client to server error: %v", err) + err1 = err + } + case err := <-serverToClient: + if err != nil { + log.Printf("Server to client error: %v", err) + err2 = err + } + } + } + + if err1 != nil { + return err1 + } + return err2 +} + // startGRPCProxy creates a transparent gRPC proxy that validates metadata and forwards all requests to the target func startGRPCProxy(listener net.Listener, targetHost string, validations []HeaderValidation) error { // Create connection to target for forwarding with raw codec to avoid unmarshaling @@ -114,38 +209,11 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head // Log the gRPC call log.Printf("=== Proxying Call: %s ===", fullMethodName) - // Extract and validate metadata - md, ok := metadata.FromIncomingContext(stream.Context()) - if ok { - // Only validate headers for the main Storage API, not Control API - if strings.HasPrefix(fullMethodName, "/google.storage.v2.Storage/") { - if err := validateGRPCMetadata(md, validations); err != nil { - log.Printf("Metadata validation failed: %v", err) - return err - } - } - } else { - md = metadata.New(nil) - if len(validations) > 0 && strings.HasPrefix(fullMethodName, "/google.storage.v2.Storage/") { - log.Println("Warning: No metadata found in request") - } - } - - // Inject testbench instructions via gOpManager - parts := strings.Split(fullMethodName, "/") - methodName := parts[len(parts)-1] - plantOp := gOpManager.retrieveOperation(RequestType(methodName)) - if plantOp != "" { - if *fDebug { - log.Printf("Planting operation: %s for method: %s", plantOp, fullMethodName) - } - // Inject direct instruction for testbench - md = metadata.Join(md, metadata.Pairs("x-goog-emulator-instructions", plantOp)) + forwardCtx, err := buildForwardContext(stream.Context(), fullMethodName, validations) + if err != nil { + return err } - // Forward metadata to target - forwardCtx := metadata.NewOutgoingContext(stream.Context(), md) - // Invoke the method on the target clientStream, err := targetConn.NewStream(forwardCtx, &grpc.StreamDesc{ StreamName: fullMethodName, @@ -161,68 +229,15 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head clientToServer := make(chan error, 1) serverToClient := make(chan error, 1) - // Receive from client, send to target go func() { - for { - req := new([]byte) - if err := stream.RecvMsg(req); err != nil { - if err == io.EOF { - err = clientStream.CloseSend() - clientToServer <- err - return - } - clientToServer <- fmt.Errorf("receiving from client: %w", err) - return - } - if err := clientStream.SendMsg(req); err != nil { - clientToServer <- fmt.Errorf("sending to server: %w", err) - return - } - } + clientToServer <- proxyClientToTarget(stream, clientStream) }() - // Receive from target, send to client go func() { - for { - resp := new([]byte) - if err := clientStream.RecvMsg(resp); err != nil { - if err == io.EOF { - serverToClient <- nil - return - } - serverToClient <- fmt.Errorf("receiving from server: %w", err) - return - } - if err := stream.SendMsg(resp); err != nil { - serverToClient <- fmt.Errorf("sending to client: %w", err) - return - } - } + serverToClient <- proxyTargetToClient(clientStream, stream) }() - // Wait for BOTH directions to complete - // This is important for both unary and streaming RPCs - var err1, err2 error - for i := 0; i < 2; i++ { - select { - case err := <-clientToServer: - if err != nil { - log.Printf("Client to server error: %v", err) - err1 = err - } - case err := <-serverToClient: - if err != nil { - log.Printf("Server to client error: %v", err) - err2 = err - } - } - } - - // Return first error encountered, or nil if both succeeded - if err1 != nil { - return err1 - } - return err2 + return waitForProxyCompletion(clientToServer, serverToClient) } // Create gRPC server with unknown service handler and raw codec From 6ebab09887cb7dcd1f439610ace596e9c66137e0 Mon Sep 17 00:00:00 2001 From: raj-prince Date: Tue, 14 Jul 2026 18:14:58 +0000 Subject: [PATCH 3/4] minor change --- tools/integration_tests/emulator_tests/emulator_tests.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tools/integration_tests/emulator_tests/emulator_tests.sh b/tools/integration_tests/emulator_tests/emulator_tests.sh index a1611648a1..6feca2987a 100755 --- a/tools/integration_tests/emulator_tests/emulator_tests.sh +++ b/tools/integration_tests/emulator_tests/emulator_tests.sh @@ -119,8 +119,8 @@ CONTAINER_ID=$(sudo docker ps -aqf "name=$CONTAINER_NAME") if [[ -n "$CONTAINER_ID" ]]; then log_info "Container with ID:[$CONTAINER_ID] is already running with name:[$CONTAINER_NAME]" log_info "Stopping and removing container...." - docker stop $CONTAINER_ID || true - docker rm $CONTAINER_ID || true + sudo docker stop $CONTAINER_ID || true + sudo docker rm $CONTAINER_ID || true fi wait_for_emulator() { From b31f841b3ee199110746f28c64aba104c2e856ae Mon Sep 17 00:00:00 2001 From: raj-prince Date: Wed, 15 Jul 2026 04:45:56 +0000 Subject: [PATCH 4/4] minor refactoring in grpc proxy code --- tools/proxy_server/grpc_proxy.go | 58 +++++++++++++++++--------------- 1 file changed, 30 insertions(+), 28 deletions(-) diff --git a/tools/proxy_server/grpc_proxy.go b/tools/proxy_server/grpc_proxy.go index ada0b3a7c3..a6a1e2de7c 100644 --- a/tools/proxy_server/grpc_proxy.go +++ b/tools/proxy_server/grpc_proxy.go @@ -134,32 +134,34 @@ func buildForwardContext(ctx context.Context, fullMethodName string, validations return metadata.NewOutgoingContext(ctx, md), nil } -func proxyClientToTarget(stream grpc.ServerStream, clientStream grpc.ClientStream) error { +// proxyGCSFuseToEmulator forwards request messages from gcsfuse to the emulated storage server. +func proxyGCSFuseToEmulator(gcsfuseToProxyServerStream grpc.ServerStream, proxyToEmulatorClientStream grpc.ClientStream) error { for { req := new([]byte) - if err := stream.RecvMsg(req); err != nil { + if err := gcsfuseToProxyServerStream.RecvMsg(req); err != nil { if err == io.EOF { - return clientStream.CloseSend() + return proxyToEmulatorClientStream.CloseSend() } - return fmt.Errorf("receiving from client: %w", err) + return fmt.Errorf("receiving from gcsfuse: %w", err) } - if err := clientStream.SendMsg(req); err != nil { - return fmt.Errorf("sending to server: %w", err) + if err := proxyToEmulatorClientStream.SendMsg(req); err != nil { + return fmt.Errorf("sending to emulated storage server: %w", err) } } } -func proxyTargetToClient(clientStream grpc.ClientStream, stream grpc.ServerStream) error { +// proxyEmulatorToGCSFuse forwards response messages from the emulated storage server back to gcsfuse. +func proxyEmulatorToGCSFuse(proxyToEmulatorClientStream grpc.ClientStream, gcsfuseToProxyServerStream grpc.ServerStream) error { for { resp := new([]byte) - if err := clientStream.RecvMsg(resp); err != nil { + if err := proxyToEmulatorClientStream.RecvMsg(resp); err != nil { if err == io.EOF { return nil } - return fmt.Errorf("receiving from server: %w", err) + return fmt.Errorf("receiving from emulated storage server: %w", err) } - if err := stream.SendMsg(resp); err != nil { - return fmt.Errorf("sending to client: %w", err) + if err := gcsfuseToProxyServerStream.SendMsg(resp); err != nil { + return fmt.Errorf("sending to gcsfuse: %w", err) } } } @@ -170,12 +172,12 @@ func waitForProxyCompletion(clientToServer, serverToClient <-chan error) error { select { case err := <-clientToServer: if err != nil { - log.Printf("Client to server error: %v", err) + log.Printf("client to server error: %v", err) err1 = err } case err := <-serverToClient: if err != nil { - log.Printf("Server to client error: %v", err) + log.Printf("server to client error: %v", err) err2 = err } } @@ -187,21 +189,21 @@ func waitForProxyCompletion(clientToServer, serverToClient <-chan error) error { return err2 } -// startGRPCProxy creates a transparent gRPC proxy that validates metadata and forwards all requests to the target -func startGRPCProxy(listener net.Listener, targetHost string, validations []HeaderValidation) error { - // Create connection to target for forwarding with raw codec to avoid unmarshaling - targetConn, err := grpc.NewClient( - targetHost, +// startGRPCProxy creates a transparent gRPC proxy that validates metadata and forwards all requests to the emulated storage server. +func startGRPCProxy(listener net.Listener, emulatedStorageHost string, validations []HeaderValidation) error { + // Create connection to emulated storage server for forwarding with raw codec to avoid unmarshaling. + emulatedStorageConn, err := grpc.NewClient( + emulatedStorageHost, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.ForceCodec(rawCodec{})), ) if err != nil { - return fmt.Errorf("failed to connect to target %s: %w", targetHost, err) + return fmt.Errorf("failed to connect to emulated storage server %s: %w", emulatedStorageHost, err) } // Create unknown service handler that validates metadata and forwards all calls - unknownHandler := func(srv any, stream grpc.ServerStream) error { - fullMethodName, ok := grpc.MethodFromServerStream(stream) + unknownHandler := func(srv any, gcsfuseToProxyServerStream grpc.ServerStream) error { + fullMethodName, ok := grpc.MethodFromServerStream(gcsfuseToProxyServerStream) if !ok { return status.Errorf(codes.Internal, "failed to get method name") } @@ -209,19 +211,19 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head // Log the gRPC call log.Printf("=== Proxying Call: %s ===", fullMethodName) - forwardCtx, err := buildForwardContext(stream.Context(), fullMethodName, validations) + forwardCtx, err := buildForwardContext(gcsfuseToProxyServerStream.Context(), fullMethodName, validations) if err != nil { return err } - // Invoke the method on the target - clientStream, err := targetConn.NewStream(forwardCtx, &grpc.StreamDesc{ + // Invoke the same method on the emulated storage server. + proxyToEmulatorClientStream, err := emulatedStorageConn.NewStream(forwardCtx, &grpc.StreamDesc{ StreamName: fullMethodName, ServerStreams: true, ClientStreams: true, }, fullMethodName, grpc.ForceCodec(rawCodec{})) if err != nil { - return status.Errorf(codes.Internal, "failed to create target stream: %v", err) + return status.Errorf(codes.Internal, "failed to create emulated storage server stream: %v", err) } // Proxy messages bidirectionally using raw bytes @@ -230,11 +232,11 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head serverToClient := make(chan error, 1) go func() { - clientToServer <- proxyClientToTarget(stream, clientStream) + clientToServer <- proxyGCSFuseToEmulator(gcsfuseToProxyServerStream, proxyToEmulatorClientStream) }() go func() { - serverToClient <- proxyTargetToClient(clientStream, stream) + serverToClient <- proxyEmulatorToGCSFuse(proxyToEmulatorClientStream, gcsfuseToProxyServerStream) }() return waitForProxyCompletion(clientToServer, serverToClient) @@ -247,7 +249,7 @@ func startGRPCProxy(listener net.Listener, targetHost string, validations []Head } grpcServer := grpc.NewServer(opts...) - log.Printf("gRPC proxy server ready to accept connections, forwarding to %s", targetHost) + log.Printf("gRPC proxy server ready to accept connections, forwarding to emulated storage server %s", emulatedStorageHost) return grpcServer.Serve(listener) }