diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index ae5c011f..78668eef 100644 --- a/service/stovepipe/README.md +++ b/service/stovepipe/README.md @@ -1,9 +1,11 @@ # Stovepipe Service -Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs the internal pipeline stages as queue consumers: +Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes four RPCs and runs the internal pipeline stages as queue consumers: - **`Ping`** — health check. - **`Ingest`** — resolves a queue's head commit, persists a `Request` (and its head URI) to storage, and publishes the request to the **process** stage. +- **`GetRequestHistoryByID`** — returns the retained request log for one request ID. +- **`GetRequestHistoryByURI`** — returns retained histories selected by an exact commit URI. - **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`). - **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`. - **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`. @@ -70,8 +72,16 @@ Attach with `.vscode/launch.json` (**Debug: attach (dlv in docker)**), then send ```bash # Ingest example grpcurl -plaintext -d '{"queue":"monorepo/main"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/Ingest + +# Retained history by request ID +grpcurl -plaintext -d '{"queue":"monorepo/main","request_id":"request/monorepo/main/1"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByID + +# Retained history by exact commit URI +grpcurl -plaintext -d '{"queue":"monorepo/main","uri":"git://monorepo/main/HEAD"}' localhost:PORT uber.submitqueue.stovepipe.Stovepipe/GetRequestHistoryByURI ``` +History lookup is defined by retained `request_log` rows. A request with no retained rows is not discoverable through these RPCs, even if operational request data still exists. + ### Bazel / Go ```bash diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 509957e2..5612b282 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -79,8 +79,11 @@ go_test( embed = [":go_default_library"], deps = [ "//api/base/hook:go_default_library", + "//api/stovepipe/protopb:go_default_library", "//platform/consumer:go_default_library", + "//stovepipe/controller:go_default_library", "//stovepipe/controller/dlq:go_default_library", + "//stovepipe/entity:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 1e412c4b..eb64d1f1 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -67,8 +67,9 @@ import ( // StovepipeServer wraps the controllers and implements the gRPC service interface. type StovepipeServer struct { pb.UnimplementedStovepipeServer - pingController *controller.PingController - ingestController *controller.IngestController + pingController *controller.PingController + ingestController *controller.IngestController + requestHistoryController controller.RequestHistoryController } // Ping delegates to the controller. @@ -86,6 +87,24 @@ func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*p return mapper.IngestResultToProto(result), nil } +// GetRequestHistoryByID returns retained history for one request ID. +func (s *StovepipeServer) GetRequestHistoryByID(ctx context.Context, req *pb.GetRequestHistoryByIDRequest) (*pb.GetRequestHistoryByIDResponse, error) { + events, err := s.requestHistoryController.GetRequestHistoryByID(ctx, mapper.ProtoToGetRequestHistoryByIDRequest(req)) + if err != nil { + return nil, err + } + return &pb.GetRequestHistoryByIDResponse{Events: mapper.HistoryEventsToProto(events)}, nil +} + +// GetRequestHistoryByURI returns retained histories for one commit URI. +func (s *StovepipeServer) GetRequestHistoryByURI(ctx context.Context, req *pb.GetRequestHistoryByURIRequest) (*pb.GetRequestHistoryByURIResponse, error) { + histories, err := s.requestHistoryController.GetRequestHistoryByURI(ctx, mapper.ProtoToGetRequestHistoryByURIRequest(req)) + if err != nil { + return nil, err + } + return &pb.GetRequestHistoryByURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil +} + // inMemoryCounter is a minimal, process-local counter.Counter used to wire the example // server. It is not durable; a real deployment supplies a persistent implementation // (e.g. platform/extension/counter/mysql). @@ -341,9 +360,11 @@ func run() error { materializer, registry, ) + requestHistoryController := controller.NewRequestHistoryController(logger.Sugar(), scope, storageFty) srv := &StovepipeServer{ - pingController: pingController, - ingestController: ingestController, + pingController: pingController, + ingestController: ingestController, + requestHistoryController: requestHistoryController, } pb.RegisterStovepipeServer(grpcServer, srv) diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 05262ef6..93968ca1 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -16,6 +16,7 @@ package main import ( "context" + "errors" "strings" "testing" @@ -23,11 +24,126 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/stovepipe/controller" "github.com/uber/submitqueue/stovepipe/controller/dlq" + "github.com/uber/submitqueue/stovepipe/entity" "go.uber.org/zap/zaptest" ) +type fakeRequestHistoryController struct { + getByID func(context.Context, entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) + getByURI func(context.Context, entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) +} + +var _ controller.RequestHistoryController = (*fakeRequestHistoryController)(nil) + +func (f *fakeRequestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) { + return f.getByID(ctx, req) +} + +func (f *fakeRequestHistoryController) GetRequestHistoryByURI(ctx context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) { + return f.getByURI(ctx, req) +} + +func TestGetRequestHistoryByID(t *testing.T) { + controllerErr := errors.New("controller failed") + logs := []entity.RequestLog{ + {ID: "occurrence/1", State: entity.RequestStateAccepted, TimestampMs: 1000}, + {ID: "occurrence/2", Event: entity.RequestEventBuildTriggered, TimestampMs: 2000}, + } + tests := []struct { + name string + logs []entity.RequestLog + err error + wantLogs int + }{ + {name: "maps successful result", logs: logs, wantLogs: 2}, + {name: "maps empty result", logs: nil, wantLogs: 0}, + {name: "returns controller error unchanged", err: controllerErr}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotReq entity.GetRequestHistoryByIDRequest + fake := &fakeRequestHistoryController{ + getByID: func(_ context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error) { + gotReq = req + return tt.logs, tt.err + }, + } + srv := &StovepipeServer{requestHistoryController: fake} + + resp, err := srv.GetRequestHistoryByID(context.Background(), &pb.GetRequestHistoryByIDRequest{ + Queue: "monorepo/main", RequestId: "request/1", + }) + + assert.Equal(t, entity.GetRequestHistoryByIDRequest{Queue: "monorepo/main", ID: "request/1"}, gotReq) + if tt.err != nil { + require.ErrorIs(t, err, tt.err) + assert.Nil(t, resp) + return + } + require.NoError(t, err) + require.Len(t, resp.Events, tt.wantLogs) + if tt.wantLogs > 0 { + assert.Equal(t, "accepted", resp.Events[0].GetRequestState()) + assert.Equal(t, "build_triggered", resp.Events[1].GetEvent()) + } + }) + } +} + +func TestGetRequestHistoryByURI(t *testing.T) { + controllerErr := errors.New("controller failed") + histories := []entity.RequestHistory{{ + RequestID: "request/1", + Events: []entity.RequestLog{{ID: "occurrence/1", State: entity.RequestStateAccepted}}, + }} + tests := []struct { + name string + histories []entity.RequestHistory + err error + wantHistories int + }{ + {name: "maps successful result", histories: histories, wantHistories: 1}, + {name: "maps empty result", histories: nil, wantHistories: 0}, + {name: "returns controller error unchanged", err: controllerErr}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotReq entity.GetRequestHistoryByURIRequest + fake := &fakeRequestHistoryController{ + getByURI: func(_ context.Context, req entity.GetRequestHistoryByURIRequest) ([]entity.RequestHistory, error) { + gotReq = req + return tt.histories, tt.err + }, + } + srv := &StovepipeServer{requestHistoryController: fake} + + resp, err := srv.GetRequestHistoryByURI(context.Background(), &pb.GetRequestHistoryByURIRequest{ + Queue: "monorepo/main", Uri: "git://monorepo/abc", + }) + + assert.Equal(t, entity.GetRequestHistoryByURIRequest{Queue: "monorepo/main", URI: "git://monorepo/abc"}, gotReq) + if tt.err != nil { + require.ErrorIs(t, err, tt.err) + assert.Nil(t, resp) + return + } + require.NoError(t, err) + require.Len(t, resp.Histories, tt.wantHistories) + if tt.wantHistories > 0 { + assert.Equal(t, "request/1", resp.Histories[0].RequestId) + require.Len(t, resp.Histories[0].Events, 1) + assert.Equal(t, "accepted", resp.Histories[0].Events[0].GetRequestState()) + } + }) + } +} + // recordingConsumer captures what the host registers instead of subscribing. type recordingConsumer struct { controllers []consumer.Controller diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index 5788bfc4..1a891783 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -153,3 +153,138 @@ func (s *StovepipeIntegrationSuite) TestIngestEmptyQueue() { _, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: ""}) require.Error(t, err, "Ingest with empty queue should fail") } + +func (s *StovepipeIntegrationSuite) TestRequestHistoryAPIs() { + t := s.T() + const ( + queue = "history-api/main" + requestID = "request/history-api/main/7" + uri = "git://history-api/main/abc123" + ) + + _, err := s.db.Exec( + "INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)", + queue, uri, requestID, 1, + ) + require.NoError(t, err) + _, err = s.db.Exec( + `INSERT INTO request_log + (queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata) + VALUES + (?, ?, ?, ?, ?, ?, ?, ?, ?), + (?, ?, ?, ?, ?, ?, ?, ?, ?), + (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + queue, requestID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`, + queue, requestID, "occurrence/002", 2000, "", "build_triggered", 0, "", `{}`, + queue, requestID, "occurrence/003", 2000, "succeeded", "", 3, "build_succeeded", `{}`, + ) + require.NoError(t, err) + + var requestRows int + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE id = ?", requestID).Scan(&requestRows)) + require.Zero(t, requestRows) + + byID, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: requestID}) + require.NoError(t, err) + assertHistoryEvents(t, byID.Events) + + byURI, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: uri}) + require.NoError(t, err) + require.Len(t, byURI.Histories, 1) + assert.Equal(t, requestID, byURI.Histories[0].RequestId) + assertHistoryEvents(t, byURI.Histories[0].Events) +} + +func (s *StovepipeIntegrationSuite) TestRequestHistoryAbsence() { + t := s.T() + const ( + queue = "history-api-absence/main" + mappedID = "request/history-api-absence/main/1" + mappedURI = "git://history-api-absence/main/mapped" + scopedID = "request/history-api-absence/main/2" + scopedURI = "git://history-api-absence/main/scoped" + missingID = "request/history-api-absence/main/missing" + missingURI = "git://history-api-absence/main/missing" + wrongQueue = "history-api-absence/other" + ) + + _, err := s.db.Exec( + "INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?), (?, ?, ?, ?)", + queue, mappedURI, mappedID, 1, queue, scopedURI, scopedID, 1, + ) + require.NoError(t, err) + _, err = s.db.Exec( + `INSERT INTO request_log + (queue, request_id, log_id, timestamp_ms, state, event, request_version, outcome_reason, metadata) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + queue, scopedID, "occurrence/001", 1000, "accepted", "", 1, "", `{}`, + ) + require.NoError(t, err) + + tests := []struct { + name string + call func() error + }{ + { + name: "missing request ID", + call: func() error { + _, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: queue, RequestId: missingID}) + return err + }, + }, + { + name: "missing URI mapping", + call: func() error { + _, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: missingURI}) + return err + }, + }, + { + name: "mapped URI without retained logs", + call: func() error { + _, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: queue, Uri: mappedURI}) + return err + }, + }, + { + name: "request ID in wrong queue", + call: func() error { + _, err := s.client.GetRequestHistoryByID(s.ctx, &pb.GetRequestHistoryByIDRequest{Queue: wrongQueue, RequestId: scopedID}) + return err + }, + }, + { + name: "URI mapping in wrong queue", + call: func() error { + _, err := s.client.GetRequestHistoryByURI(s.ctx, &pb.GetRequestHistoryByURIRequest{Queue: wrongQueue, Uri: scopedURI}) + return err + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Error(t, tt.call()) + }) + } +} + +func assertHistoryEvents(t *testing.T, events []*pb.HistoryEvent) { + t.Helper() + require.Len(t, events, 3) + + assert.Equal(t, "occurrence/001", events[0].EventId) + assert.Equal(t, int64(1000), events[0].TimestampMs) + assert.Equal(t, "accepted", events[0].GetRequestState()) + assert.Empty(t, events[0].GetEvent()) + + assert.Equal(t, "occurrence/002", events[1].EventId) + assert.Equal(t, int64(2000), events[1].TimestampMs) + assert.Equal(t, "build_triggered", events[1].GetEvent()) + assert.Empty(t, events[1].GetRequestState()) + + assert.Equal(t, "occurrence/003", events[2].EventId) + assert.Equal(t, int64(2000), events[2].TimestampMs) + assert.Equal(t, "succeeded", events[2].GetRequestState()) + assert.Equal(t, "build_succeeded", events[2].OutcomeReason) +}