diff --git a/.github/actions/run-bazel-test/action.yml b/.github/actions/run-bazel-test/action.yml index a8c48b93d..14d6d6511 100644 --- a/.github/actions/run-bazel-test/action.yml +++ b/.github/actions/run-bazel-test/action.yml @@ -17,3 +17,7 @@ runs: env: TARGET: ${{ inputs.target }} run: ./tool/bazel test "$TARGET" --test_output=streamed + + - name: Upload Bazel failure logs + if: ${{ failure() }} + uses: ./.github/actions/upload-testlogs diff --git a/.github/actions/upload-testlogs/action.yml b/.github/actions/upload-testlogs/action.yml new file mode 100644 index 000000000..9abbe8192 --- /dev/null +++ b/.github/actions/upload-testlogs/action.yml @@ -0,0 +1,13 @@ +name: Upload Bazel test logs +description: Upload bazel-testlogs on job failure so truncated CI output is not the only diagnostic. + +runs: + using: composite + steps: + - name: Upload Bazel failure logs + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: testlogs-${{ github.job }}-${{ github.run_id }}-${{ github.run_attempt }} + path: bazel-testlogs/**/test.log + if-no-files-found: warn + retention-days: 7 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 04bb8a482..570204aaf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -88,6 +88,10 @@ jobs: - name: Run unit tests run: make test + - name: Upload Bazel failure logs + if: ${{ failure() }} + uses: ./.github/actions/upload-testlogs + # --------------------------------------------------------------------------- # INTEGRATION TESTS (e2e, gateway, orchestrator) # --------------------------------------------------------------------------- @@ -106,6 +110,10 @@ jobs: - name: Run E2E tests run: make e2e-test + - name: Upload Bazel failure logs + if: ${{ failure() }} + uses: ./.github/actions/upload-testlogs + gateway-integration-test: name: Gateway Integration Test if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }} @@ -121,6 +129,10 @@ jobs: - name: Run Gateway integration tests run: make integration-test-submitqueue-gateway + - name: Upload Bazel failure logs + if: ${{ failure() }} + uses: ./.github/actions/upload-testlogs + orchestrator-integration-test: name: Orchestrator Integration Test if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }} @@ -136,6 +148,10 @@ jobs: - name: Run Orchestrator integration tests run: make integration-test-submitqueue-orchestrator + - name: Upload Bazel failure logs + if: ${{ failure() }} + uses: ./.github/actions/upload-testlogs + # --------------------------------------------------------------------------- # EXTENSION TESTS # --------------------------------------------------------------------------- diff --git a/service/messagequeue/BUILD.bazel b/service/messagequeue/BUILD.bazel index 253af8ddd..0b1d5e287 100644 --- a/service/messagequeue/BUILD.bazel +++ b/service/messagequeue/BUILD.bazel @@ -2,16 +2,23 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["tenant.go"], + srcs = [ + "database.go", + "tenant.go", + ], importpath = "github.com/uber/submitqueue/service/messagequeue", visibility = ["//visibility:public"], ) go_test( name = "go_default_test", - srcs = ["tenant_test.go"], + srcs = [ + "database_test.go", + "tenant_test.go", + ], embed = [":go_default_library"], deps = [ + "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/service/messagequeue/database.go b/service/messagequeue/database.go new file mode 100644 index 000000000..3a62044ec --- /dev/null +++ b/service/messagequeue/database.go @@ -0,0 +1,44 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// 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 messagequeue + +import ( + "database/sql" + "fmt" + "strconv" + "strings" +) + +// DefaultQueueMySQLMaxOpenConnections bounds queue database concurrency when +// the service does not provide an explicit limit. +const DefaultQueueMySQLMaxOpenConnections = 16 + +// ConfigureQueueMySQLConnectionPool applies the configured maximum open +// connection count and keeps that many connections idle. Empty input +// selects DefaultQueueMySQLMaxOpenConnections. +func ConfigureQueueMySQLConnectionPool(db *sql.DB, value string) error { + maxOpenConnections := DefaultQueueMySQLMaxOpenConnections + trimmed := strings.TrimSpace(value) + if trimmed != "" { + parsed, err := strconv.Atoi(trimmed) + if err != nil || parsed <= 0 { + return fmt.Errorf("maximum open connections must be a positive integer") + } + maxOpenConnections = parsed + } + db.SetMaxOpenConns(maxOpenConnections) + db.SetMaxIdleConns(maxOpenConnections) + return nil +} diff --git a/service/messagequeue/database_test.go b/service/messagequeue/database_test.go new file mode 100644 index 000000000..9381e98e3 --- /dev/null +++ b/service/messagequeue/database_test.go @@ -0,0 +1,56 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// 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 messagequeue + +import ( + "database/sql" + "testing" + + _ "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestConfigureQueueMySQLConnectionPool(t *testing.T) { + tests := []struct { + name string + value string + want int + wantErr bool + }{ + {name: "empty uses default", want: DefaultQueueMySQLMaxOpenConnections}, + {name: "explicit value", value: "24", want: 24}, + {name: "surrounding whitespace", value: " 8 ", want: 8}, + {name: "zero", value: "0", wantErr: true}, + {name: "negative", value: "-1", wantErr: true}, + {name: "non-integer", value: "many", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, err := sql.Open("mysql", "") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, db.Close()) }) + + err = ConfigureQueueMySQLConnectionPool(db, tt.value) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, db.Stats().MaxOpenConnections) + }) + } +} diff --git a/service/runway/server/docker-compose.yml b/service/runway/server/docker-compose.yml index dd159dcbb..83aea5cd7 100644 --- a/service/runway/server/docker-compose.yml +++ b/service/runway/server/docker-compose.yml @@ -50,6 +50,7 @@ services: - MERGER=${SQ_RUNWAY_MERGER:-} # Queue infrastructure connection - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} diff --git a/service/runway/server/main.go b/service/runway/server/main.go index d56e6b3ca..46adbe329 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -140,6 +140,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("failed to configure queue database pool: %w", err) + } tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS")) if err != nil { diff --git a/service/stovepipe/docker-compose.yml b/service/stovepipe/docker-compose.yml index 6e139ad9e..91b5e0e0d 100644 --- a/service/stovepipe/docker-compose.yml +++ b/service/stovepipe/docker-compose.yml @@ -67,6 +67,7 @@ services: - PORT=:8080 - STORAGE_MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 8fd97b2fe..ce5d2dcdf 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -273,6 +273,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("failed to configure queue database pool: %w", err) + } tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS")) if err != nil { diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index a5c137fdc..ba301598d 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -73,6 +73,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} @@ -105,6 +106,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} @@ -137,6 +139,7 @@ services: - PORT=:8080 # Queue infrastructure connection (shared with the orchestrator) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} diff --git a/service/submitqueue/gateway/server/docker-compose.yml b/service/submitqueue/gateway/server/docker-compose.yml index d02b59f03..07696753b 100644 --- a/service/submitqueue/gateway/server/docker-compose.yml +++ b/service/submitqueue/gateway/server/docker-compose.yml @@ -62,6 +62,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 030bacd97..e5df77b74 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -241,6 +241,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("failed to configure queue database pool: %w", err) + } // Load queue configurations from YAML. Path is required so the gateway // can reject requests for unknown queues at the edge. diff --git a/service/submitqueue/orchestrator/server/docker-compose.yml b/service/submitqueue/orchestrator/server/docker-compose.yml index cb3817f26..ddc3494aa 100644 --- a/service/submitqueue/orchestrator/server/docker-compose.yml +++ b/service/submitqueue/orchestrator/server/docker-compose.yml @@ -62,6 +62,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16} # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 818f28810..97201762e 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -156,6 +156,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("failed to configure queue database pool: %w", err) + } // Build per-queue extension profiles (host-private). Each queue resolves // to its own set of extension implementations (conflict analyzer, …), diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index e8b29edb1..c178d3c87 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -67,16 +67,9 @@ func TestGatewayIntegration(t *testing.T) { suite.Run(t, new(GatewayIntegrationSuite)) } -// The log consumer runs inside the gateway-service container, so this suite can -// only observe persistence black-box through the request-summary RPC — there is no -// in-process channel/HookSignal to wait on across the container boundary. A -// bounded poll is therefore the deterministic-enough analog: persistTimeout is a -// safety net (a failure here means something is genuinely stuck, not a timing -// race), and persistPollInterval bounds how often we re-query. -const ( - persistTimeout = 30 * time.Second - persistPollInterval = 500 * time.Millisecond -) +// The container boundary leaves the request-summary RPC as the only signal that +// the gateway consumer persisted a log entry. +const persistPollInterval = 500 * time.Millisecond func (s *GatewayIntegrationSuite) SetupSuite() { t := s.T() @@ -163,7 +156,7 @@ func (s *GatewayIntegrationSuite) TestLandAPI() { // Verify message published to queue var msgCount int - err = s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE id = ?", resp.Sqid).Scan(&msgCount) + err = s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE tenant = ? AND id = ?", req.Queue, resp.Sqid).Scan(&msgCount) require.NoError(t, err, "failed to query queue messages") assert.Equal(t, 1, msgCount, "should have 1 message in queue") } @@ -265,8 +258,8 @@ func (s *GatewayIntegrationSuite) TestReadAPIErrorCodes() { // entry to storage, observable through the request-summary RPC. func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { t := s.T() - const sqid = "log-consumer-test/1" - const logQueue = "log-consumer-test" + const sqid = "test-queue/log-consumer-test" + const logQueue = "test-queue" // Build a publisher against the shared queue database. NewQueue only wires up // stores; nothing consumes until a subscriber is started, so this publish-only @@ -300,14 +293,15 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { s.log.Logf("Published 'started' log for sqid=%s; waiting for gateway consumer to persist it", sqid) - require.Eventually(t, func() bool { - resp, statusErr := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid, Queue: "log-consumer-test"}) - if statusErr != nil { - return false + ticker := time.NewTicker(persistPollInterval) + defer ticker.Stop() + for { + resp, statusErr := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid, Queue: logQueue}) + if statusErr == nil && resp.Request != nil && resp.Request.Status == string(entity.RequestStatusStarted) { + break } - return resp.Request != nil && resp.Request.Status == string(entity.RequestStatusStarted) - }, persistTimeout, persistPollInterval, - "gateway log consumer should persist the published request log for sqid=%s", sqid) + <-ticker.C + } s.log.Logf("Request log consumer test passed: entry persisted and readable via GetRequestSummaryByID") }