diff --git a/test/e2e/stovepipe/BUILD.bazel b/test/e2e/stovepipe/BUILD.bazel new file mode 100644 index 00000000..b68903aa --- /dev/null +++ b/test/e2e/stovepipe/BUILD.bazel @@ -0,0 +1,30 @@ +load("@rules_go//go:def.bzl", "go_test") + +go_test( + name = "go_default_test", + srcs = [ + "harness_test.go", + "suite_test.go", + ], + data = [ + "//:MODULE.bazel", + "//:go.mod", + "//platform/extension/messagequeue/mysql/schema", + "//service/stovepipe:docker-compose.yml", + "//stovepipe/extension/storage/mysql/schema", + ], + tags = [ + "e2e", + "external", + "integration", + ], + deps = [ + "//api/stovepipe/protopb:go_default_library", + "//test/testutil:go_default_library", + "@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", + "@com_github_stretchr_testify//suite:go_default_library", + "@org_golang_google_grpc//:go_default_library", + ], +) diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go new file mode 100644 index 00000000..6aa812c9 --- /dev/null +++ b/test/e2e/stovepipe/harness_test.go @@ -0,0 +1,123 @@ +// Copyright (c) 2025 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 e2e_test + +// Reusable e2e helpers so tests read as intent, not plumbing. They drive the +// stack through the real Stovepipe gRPC surface (Ingest) and observe outcomes +// two ways: +// +// - the synchronous side effects of Ingest via raw SQL on the storage DB +// (the request row and its (queue, URI) mapping) and the queue DB (the +// published process message); and +// - the asynchronous completion of the process stage by polling the queue +// backend's per-consumer-group delivery state until the message is acked. +// +// Convergence is bounded by require.Eventually rather than time.Sleep: the +// process consumer runs inside the stovepipe-service container, so there is no +// in-process signal to await; a timeout here means the stage is genuinely stuck, +// not a timing race. + +import ( + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" +) + +// The process consumer's topic and consumer group as wired in +// service/stovepipe/server/main.go (topic name "process", consumer group +// "stovepipe-process"). awaitProcessed reads the queue backend's delivery state +// keyed by this group. +const ( + processTopic = "process" + processConsumerGroup = "stovepipe-process" +) + +// ingest admits a queue's head commit into the pipeline and returns the minted +// request id. +func (s *StovepipeE2ESuite) ingest(queue string) string { + t := s.T() + resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue}) + require.NoError(t, err, "Ingest failed for queue %s", queue) + require.NotEmpty(t, resp.Id, "Ingest returned an empty id for queue %s", queue) + return resp.Id +} + +// requestRowCount returns the number of request rows with the given id (0 or 1). +func (s *StovepipeE2ESuite) requestRowCount(id string) int { + t := s.T() + var count int + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE id = ?", id).Scan(&count), + "failed to count request rows for %s", id) + return count +} + +// uriMapping returns the request id the (queue, URI) mapping points at. +func (s *StovepipeE2ESuite) uriMapping(queue string) string { + t := s.T() + var mappedID string + require.NoError(t, s.db.QueryRow("SELECT request_id FROM request_uri WHERE queue = ?", queue).Scan(&mappedID), + "failed to read request_uri mapping for queue %s", queue) + return mappedID +} + +// publishedMessageCount returns the number of process messages published for the +// given request id (0 or 1). +func (s *StovepipeE2ESuite) publishedMessageCount(id string) int { + t := s.T() + var count int + require.NoError(t, s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE id = ?", id).Scan(&count), + "failed to count queue messages for %s", id) + return count +} + +// awaitProcessed blocks until the process consumer has acked a message on the +// given queue's partition, proving the ingest→process pipeline ran end-to-end. +// The durable signal is the consumer group's acked-offset watermark in +// queue_offsets: it starts at 0 and only advances once a message is acked (see +// offset_store.go). We poll that rather than the message's own delivery-state +// row because the queue GCs acked messages (and their delivery state) from +// queue_messages once the watermark passes them. The queue uses the queue name +// as the message's partition key (see the ingest controller), so the partition +// key here is the queue. +func (s *StovepipeE2ESuite) awaitProcessed(queue string) { + t := s.T() + const query = ` + SELECT offset_acked + FROM queue_offsets + WHERE consumer_group = ? AND topic = ? AND partition_key = ?` + require.Eventually(t, func() bool { + var ackedOffset int64 + err := s.queueDB.QueryRow(query, processConsumerGroup, processTopic, queue).Scan(&ackedOffset) + if err != nil { + // sql.ErrNoRows means the partition offset is not initialized yet. + s.log.Logf("acked offset for queue %s not ready yet: %v", queue, err) + return false + } + s.log.Logf("acked offset for queue %s = %d (want > 0)", queue, ackedOffset) + return ackedOffset > 0 + }, processTimeout, processPollInterval, + "process consumer group %q on topic %q should advance the acked offset for queue %s", + processConsumerGroup, processTopic, queue) +} + +// assertIngestPersisted asserts the synchronous side effects of a successful +// Ingest: the request row, the (queue, URI) mapping pointing at the minted id, +// and exactly one published process message. +func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) { + t := s.T() + assert.Equal(t, 1, s.requestRowCount(id), "request row should be persisted for %s", id) + assert.Equal(t, id, s.uriMapping(queue), "URI mapping should point at the minted request id") + assert.Equal(t, 1, s.publishedMessageCount(id), "should have published one process message for %s", id) +} diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go new file mode 100644 index 00000000..04c75d38 --- /dev/null +++ b/test/e2e/stovepipe/suite_test.go @@ -0,0 +1,157 @@ +// Copyright (c) 2025 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 e2e_test + +// Stovepipe end-to-end tests. +// +// These tests use docker-compose from service/stovepipe/docker-compose.yml, +// which requires a pre-built Linux binary. Run with the make target (builds +// binaries + runs the test): +// +// make e2e-test +// +// or only this package (after building the binary): +// +// bazel test //test/e2e/stovepipe:stovepipe_test +// +// The stack runs the Stovepipe gRPC service plus a storage MySQL (request, +// request_uri) and a queue MySQL (the process stage). Unlike the integration +// suite (test/integration/stovepipe), which asserts only that Ingest *publishes* +// a process message, this suite additionally drives the asynchronous process +// consumer to completion — proving the ingest→process pipeline runs end-to-end. + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + "time" + + _ "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/test/testutil" + "google.golang.org/grpc" +) + +// The process consumer runs inside the stovepipe-service container, so this +// suite can only observe its completion black-box through the queue backend's +// delivery-state table — there is no in-process signal to await across the +// container boundary. A bounded poll is the deterministic-enough analog: +// processTimeout is a safety net (a failure here means the stage is genuinely +// stuck, not a timing race) and processPollInterval bounds re-query frequency. +const ( + processTimeout = 30 * time.Second + processPollInterval = 500 * time.Millisecond +) + +type StovepipeE2ESuite struct { + suite.Suite + ctx context.Context + log *testutil.TestLogger + stack *testutil.ComposeStack + client pb.StovepipeClient + db *sql.DB // storage database (request, request_uri) + queueDB *sql.DB // queue database (process stage) +} + +func TestStovepipeE2E(t *testing.T) { + suite.Run(t, new(StovepipeE2ESuite)) +} + +func (s *StovepipeE2ESuite) SetupSuite() { + t := s.T() + s.ctx = context.Background() + s.log = testutil.NewTestLogger(t) + + s.log.Logf("Starting Stovepipe e2e test suite using docker-compose") + + // Set REPO_ROOT for the docker-compose build context. + repoRoot := testutil.FindRepoRoot(t) + t.Setenv("REPO_ROOT", repoRoot) + + composeFile := filepath.Join(repoRoot, "service/stovepipe/docker-compose.yml") + s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-stovepipe") + + err := s.stack.Up() + require.NoError(t, err, "failed to start compose stack") + + s.db, err = s.stack.ConnectMySQLService("mysql-app") + require.NoError(t, err, "failed to connect to storage MySQL") + + s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue") + require.NoError(t, err, "failed to connect to queue MySQL") + + // Apply schemas after the stack is up; the service connects lazily and the + // consumer retries, so the boot ordering is tolerated. + testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema")) + testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + + var conn *grpc.ClientConn + conn, err = s.stack.ConnectGRPC("stovepipe-service", 8080) + require.NoError(t, err, "failed to connect to stovepipe service") + s.client = pb.NewStovepipeClient(conn) + + s.log.Logf("Stovepipe e2e test suite ready") +} + +func (s *StovepipeE2ESuite) TearDownSuite() { + // Compose stack cleanup is handled automatically by t.Cleanup (registered in + // NewComposeStack). + s.log.Logf("Tearing down Stovepipe e2e test suite") +} + +func (s *StovepipeE2ESuite) TestPing() { + t := s.T() + resp, err := s.client.Ping(s.ctx, &pb.PingRequest{Message: "e2e test"}) + require.NoError(t, err, "Stovepipe Ping failed") + assert.Equal(t, "stovepipe", resp.ServiceName) + assert.NotEmpty(t, resp.Message) + assert.NotZero(t, resp.Timestamp) +} + +// TestIngest_HappyPath_Processes drives a queue's head commit through the whole +// pipeline. Ingest synchronously resolves the head URI via the (fake) +// SourceControl, persists the Request and its (queue, URI) mapping, and publishes +// the request id to the process stage; the process consumer then drains that +// message. This asserts both the synchronous side effects and — the piece the +// integration suite does not cover — that the async process stage acked the +// message. +func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() { + const queue = "monorepo/main" + + id := s.ingest(queue) + s.log.Logf("Ingest succeeded: id=%s; waiting for process stage", id) + + // Synchronous side effects of Ingest. + s.assertIngestPersisted(queue, id) + + // Asynchronous completion: the process consumer acked the message. + s.awaitProcessed(queue) +} + +// TestIngest_Idempotent verifies that re-ingesting the same queue resolves the +// same head URI and dedups to the same request id. +func (s *StovepipeE2ESuite) TestIngest_Idempotent() { + const queue = "monorepo/release" + + id := s.ingest(queue) + s.log.Logf("First ingest: id=%s", id) + + id2 := s.ingest(queue) + assert.Equal(s.T(), id, id2, "re-ingest of the same head should dedup to the same id") +}