diff --git a/stovepipe/controller/projectanalysis/BUILD.bazel b/stovepipe/controller/projectanalysis/BUILD.bazel index 2c628f061..28f2f0909 100644 --- a/stovepipe/controller/projectanalysis/BUILD.bazel +++ b/stovepipe/controller/projectanalysis/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_library") +load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", @@ -6,9 +6,17 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/projectanalysis", visibility = ["//visibility:public"], deps = [ + "//platform/base/id:go_default_library", + "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", + "//stovepipe/core/loader:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/projectissues:go_default_library", "//stovepipe/core/requestlog:go_default_library", + "//stovepipe/core/validationfact:go_default_library", + "//stovepipe/entity:go_default_library", "//stovepipe/extension/projectissues:go_default_library", "//stovepipe/extension/projectstatus:go_default_library", "//stovepipe/extension/storage:go_default_library", @@ -16,3 +24,31 @@ go_library( "@org_uber_go_zap//:go_default_library", ], ) + +go_test( + name = "go_default_test", + srcs = ["projectanalysis_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/requestlog:go_default_library", + "//stovepipe/core/requestlog/mock:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/projectissues:go_default_library", + "//stovepipe/extension/projectissues/mock:go_default_library", + "//stovepipe/extension/projectstatus:go_default_library", + "//stovepipe/extension/projectstatus/mock:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock: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", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/stovepipe/controller/projectanalysis/README.md b/stovepipe/controller/projectanalysis/README.md new file mode 100644 index 000000000..45e01bc1b --- /dev/null +++ b/stovepipe/controller/projectanalysis/README.md @@ -0,0 +1,12 @@ +# Project analysis + +Project analysis consumes request IDs after terminal build analysis has retained +all final build issues. A missing terminal-analysis completion is an invalid +handoff, not a reason to poll. + +The controller resolves project statuses and records their facts and lookup entries before attributing issues. It materializes both directions of each issue/project association. Each step retains its completion event after its writes; repository recording is queued only after both steps finish. + +Retries adopt immutable facts and repair missing lookup entries. A completed +request skips provider calls and retries only the recording handoff. Empty +results still record completion. Cancelled requests bypass project analysis; +superseded requests require no work. diff --git a/stovepipe/controller/projectanalysis/projectanalysis.go b/stovepipe/controller/projectanalysis/projectanalysis.go index 4e7f1a43b..75742904d 100644 --- a/stovepipe/controller/projectanalysis/projectanalysis.go +++ b/stovepipe/controller/projectanalysis/projectanalysis.go @@ -12,14 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package projectanalysis owns terminal project processing. +// Package projectanalysis records project statuses and issue attribution after build analysis. package projectanalysis import ( + "context" + "errors" + "fmt" + "math" + "strconv" + "time" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/id" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" + "github.com/uber/submitqueue/stovepipe/core/loader" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" projectissuescore "github.com/uber/submitqueue/stovepipe/core/projectissues" "github.com/uber/submitqueue/stovepipe/core/requestlog" + "github.com/uber/submitqueue/stovepipe/core/validationfact" + "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/projectissues" "github.com/uber/submitqueue/stovepipe/extension/projectstatus" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -64,6 +79,194 @@ func NewController( } } +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { + op := metrics.Begin(c.scope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + msg := delivery.Message() + analysis := &stovepipemq.ProjectAnalysis{} + if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil { + return fmt.Errorf("deserialize project analysis: %w", err) + } + if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil { + return fmt.Errorf("invalid project analysis identity: %w", err) + } + if err := id.Validate(analysis.GetId()); err != nil { + return fmt.Errorf("invalid project analysis request ID: %w", err) + } + if msg.PartitionKey != analysis.GetId() { + return fmt.Errorf("project analysis must be partitioned by request ID") + } + store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()}) + if err != nil { + return fmt.Errorf("resolve project analysis storage: %w", err) + } + request, err := loader.ByID(ctx, analysis.GetId(), store.GetRequestStore().Get, "project analysis request") + if err != nil { + return err + } + if request.Queue != analysis.GetQueueName() { + return fmt.Errorf("project analysis request belongs to another queue") + } + switch request.State { + case entity.RequestStateSucceeded, entity.RequestStateFailed: + case entity.RequestStateCancelled: + return publishRecord(ctx, c.registry, request) + case entity.RequestStateSuperseded: + return nil + default: + return fmt.Errorf("project analysis requires a terminal request") + } + if request.TerminalBuildID == "" { + return fmt.Errorf("project analysis requires the request's terminal build") + } + + projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) + history := store.GetRequestLogStore() + _, err = history.Get(ctx, request.ID, projectIssuesRecordedLog.ID) + if err == nil { + return publishRecord(ctx, c.registry, request) + } + if !errors.Is(err, storage.ErrNotFound) { + return fmt.Errorf("read project analysis completion: %w", err) + } + buildAnalysisCompletedLog := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, nil) + if _, err := history.Get(ctx, request.ID, buildAnalysisCompletedLog.ID); err != nil { + return fmt.Errorf("require completed terminal build analysis: %w", err) + } + + if err := c.recordProjectStatuses(ctx, store, request); err != nil { + return err + } + if err := c.recordIssueAttribution(ctx, store, request); err != nil { + return err + } + return publishRecord(ctx, c.registry, request) +} + +func (c *Controller) recordProjectStatuses(ctx context.Context, store storage.Storage, request entity.Request) (retErr error) { + op := metrics.Begin(c.scope, "record_project_statuses", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + resolver, err := c.resolvers.For(projectstatus.ResolverConfig{QueueName: request.Queue}) + if err != nil { + return fmt.Errorf("resolve project status resolver for queue %q: %w", request.Queue, err) + } + projectStatuses, err := resolver.Resolve(ctx, request) + if err != nil { + return fmt.Errorf("resolve project statuses for request %q: %w", request.ID, err) + } + + if err := validateProjectStatuses(request, projectStatuses); err != nil { + return err + } + metrics.NamedHistogram(c.scope, "record_project_statuses", "projects", metrics.CountBuckets, + metrics.TagsFromContext(ctx)..., + ).RecordValue(float64(len(projectStatuses))) + + brokenCount := 0 + for _, status := range projectStatuses { + fact, _, err := validationfact.Persist(ctx, c.logger, c.scope, store.GetValidationFactStore(), entity.ValidationFact{ + URI: request.URI, + Project: status.ProjectInfo.Name, + ProjectType: status.ProjectInfo.ProjectType, + Degree: status.Degree, + RequestID: request.ID, + CreatedAt: time.Now().UnixMilli(), + }) + if err != nil { + return err + } + if !fact.IsGreen() { + brokenCount++ + } + + entry, err := entity.NewProjectFactByDegree(fact) + if err != nil { + return err + } + + if err := store.GetProjectFactByDegreeStore().Create(ctx, entry); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to index project fact for request %q and project %q: %w", request.ID, fact.Project, err) + } + } + projectFactsRecordedLog := requestlog.NewRequestEventLog( + request, + entity.RequestEventProjectFactsRecorded, + "all", + map[string]string{ + requestlog.MetadataKeyProjectFactCount: strconv.Itoa(len(projectStatuses)), + requestlog.MetadataKeyBrokenProjectFactCount: strconv.Itoa(brokenCount), + }, + ) + if err := c.requestLogMaterializer.PersistLog(ctx, store, projectFactsRecordedLog); err != nil { + return fmt.Errorf("failed to record project facts for request %s: %w", request.ID, err) + } + return nil +} + +func (c *Controller) recordIssueAttribution(ctx context.Context, store storage.Storage, request entity.Request) error { + attributer, err := c.attributers.For(projectissues.AttributerConfig{QueueName: request.Queue}) + if err != nil { + return fmt.Errorf("resolve project issue attributer: %w", err) + } + attributions, err := attributer.Attribute(ctx, request) + if err != nil { + return fmt.Errorf("attribute issues for request %q: %w", request.ID, err) + } + for _, attribution := range attributions { + if attribution.TaskID == "" { + return fmt.Errorf("project issue attribution requires a task ID") + } + if attribution.UnitID == "" { + return fmt.Errorf("project issue attribution requires a unit-level issue") + } + for _, project := range attribution.Projects { + entry := entity.ProjectIssueAttribution{TaskID: attribution.TaskID, UnitID: attribution.UnitID, Project: project} + if err := c.projectIssueMaterializer.PersistAttribution(ctx, store, request.ID, entry); err != nil { + return err + } + } + } + projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) + if err := c.requestLogMaterializer.PersistLog(ctx, store, projectIssuesRecordedLog); err != nil { + return fmt.Errorf("record project issue completion: %w", err) + } + return nil +} + +func validateProjectStatuses(request entity.Request, projectStatuses []projectstatus.Result) error { + seen := make(map[string]struct{}, len(projectStatuses)) + for _, status := range projectStatuses { + if status.ProjectInfo.Name == "" { + return fmt.Errorf("project status for request %q has an empty project", request.ID) + } + if _, ok := seen[status.ProjectInfo.Name]; ok { + return fmt.Errorf("project status for request %q contains duplicate project %q", request.ID, status.ProjectInfo.Name) + } + seen[status.ProjectInfo.Name] = struct{}{} + if math.IsNaN(status.Degree) || status.Degree < entity.DegreeGreen || status.Degree > entity.DegreeBroken { + return fmt.Errorf("project status for request %q and project %q has invalid degree %v", request.ID, status.ProjectInfo.Name, status.Degree) + } + } + return nil +} + +func publishRecord(ctx context.Context, registry consumer.TopicRegistry, request entity.Request) error { + payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: request.ID, QueueName: request.Queue}) + if err != nil { + return fmt.Errorf("serialize repository recording request: %w", err) + } + return publish.Message(ctx, registry, stovepipemq.TopicKeyRecord, publish.MessageParams{ + Tenant: request.Queue, + ID: publish.IntentID(request.ID), + Payload: payload, + PartitionKey: request.ID, + }) +} + func (c *Controller) Name() string { return "projectanalysis" } func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } func (c *Controller) ConsumerGroup() string { return c.consumerGroup } + +var _ consumer.Controller = (*Controller)(nil) diff --git a/stovepipe/controller/projectanalysis/projectanalysis_test.go b/stovepipe/controller/projectanalysis/projectanalysis_test.go new file mode 100644 index 000000000..09d9bd060 --- /dev/null +++ b/stovepipe/controller/projectanalysis/projectanalysis_test.go @@ -0,0 +1,827 @@ +// 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 projectanalysis + +import ( + "context" + "errors" + "math" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/platform/metrics" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/projectissues" + projectissuesmock "github.com/uber/submitqueue/stovepipe/extension/projectissues/mock" + "github.com/uber/submitqueue/stovepipe/extension/projectstatus" + projectstatusmock "github.com/uber/submitqueue/stovepipe/extension/projectstatus/mock" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const testQueue = "validation/main" + +type fixture struct { + controller *Controller + ctrl *gomock.Controller + request entity.Request + projectStatuses []projectstatus.Result + attributions []projectissues.Attribution + facts map[string]entity.ValidationFact + indexes map[entity.ProjectFactByDegree]bool + source map[entity.ProjectIssueAttribution]bool + reverse map[entity.ProjectIssueAttribution]bool + logs map[string]entity.RequestLog + failures map[string]error + records []entityqueue.Message + publishAttempts []entityqueue.Message + resolverCalls int + attributerCalls int +} + +func newFixture(t *testing.T) *fixture { + t.Helper() + ctrl := gomock.NewController(t) + f := &fixture{ + ctrl: ctrl, + request: entity.Request{ID: "7", Queue: testQueue, URI: "git://repository/main/commit", State: entity.RequestStateFailed, TerminalBuildID: "build/7"}, + projectStatuses: []projectstatus.Result{ + {ProjectInfo: projectstatus.ProjectInfo{Name: "a", ProjectType: "service"}, Degree: entity.DegreeBroken}, + {ProjectInfo: projectstatus.ProjectInfo{Name: "b", ProjectType: "service"}, Degree: entity.DegreeGreen}, + }, + attributions: []projectissues.Attribution{{TaskID: "test", UnitID: "unit", Projects: []string{"a", "b"}}}, + facts: map[string]entity.ValidationFact{}, + indexes: map[entity.ProjectFactByDegree]bool{}, + source: map[entity.ProjectIssueAttribution]bool{}, + reverse: map[entity.ProjectIssueAttribution]bool{}, + logs: map[string]entity.RequestLog{}, + failures: map[string]error{}, + } + analysis := requestlog.NewRequestEventLog(f.request, entity.RequestEventBuildAnalysisCompleted, f.request.TerminalBuildID, nil) + f.logs[analysis.ID] = analysis + stores := storagemock.NewMockFactory(ctrl) + store := storagemock.NewMockStorage(ctrl) + requests := storagemock.NewMockRequestStore(ctrl) + history := storagemock.NewMockRequestLogStore(ctrl) + facts := storagemock.NewMockValidationFactStore(ctrl) + indexes := storagemock.NewMockProjectFactByDegreeStore(ctrl) + source := storagemock.NewMockRequestProjectsByIssueStore(ctrl) + reverse := storagemock.NewMockRequestIssuesByProjectStore(ctrl) + stores.EXPECT().For(storage.Config{QueueName: testQueue}).DoAndReturn(func(storage.Config) (storage.Storage, error) { + return store, f.fail("storage/for") + }).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetRequestLogStore().Return(history).AnyTimes() + store.EXPECT().GetValidationFactStore().Return(facts).AnyTimes() + store.EXPECT().GetProjectFactByDegreeStore().Return(indexes).AnyTimes() + store.EXPECT().GetRequestProjectsByIssueStore().Return(source).AnyTimes() + store.EXPECT().GetRequestIssuesByProjectStore().Return(reverse).AnyTimes() + requests.EXPECT().Get(gomock.Any(), "7").DoAndReturn(func(context.Context, string) (entity.Request, error) { + return f.request, f.fail("request/get") + }).AnyTimes() + history.EXPECT().Get(gomock.Any(), "7", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, id string) (entity.RequestLog, error) { + operation := "analysis/get" + if id == f.completion().ID { + operation = "completion/get" + } + if err := f.fail(operation); err != nil { + return entity.RequestLog{}, err + } + if log, ok := f.logs[id]; ok { + return log, nil + } + return entity.RequestLog{}, storage.ErrNotFound + }).AnyTimes() + facts.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, fact entity.ValidationFact) error { + if err := f.fail("fact/create"); err != nil { + return err + } + if _, exists := f.facts[fact.Project]; exists { + return storage.ErrAlreadyExists + } + f.facts[fact.Project] = fact + return nil + }).AnyTimes() + facts.EXPECT().Get(gomock.Any(), f.request.URI, gomock.Any()).DoAndReturn(func(_ context.Context, _ string, project string) (entity.ValidationFact, error) { + return f.facts[project], nil + }).AnyTimes() + indexes.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, entry entity.ProjectFactByDegree) error { + if err := f.fail("index/create"); err != nil { + return err + } + if f.indexes[entry] { + return storage.ErrAlreadyExists + } + f.indexes[entry] = true + return nil + }).AnyTimes() + source.EXPECT().Create(gomock.Any(), "7", gomock.Any()).DoAndReturn(f.retainAttribution(f.source, "source/create")).AnyTimes() + reverse.EXPECT().Create(gomock.Any(), "7", gomock.Any()).DoAndReturn(f.retainAttribution(f.reverse, "reverse/create")).AnyTimes() + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + require.Len(t, f.facts, len(f.projectStatuses)) + require.Len(t, f.indexes, len(f.projectStatuses)) + if log.Event == entity.RequestEventProjectIssuesRecorded { + require.Equal(t, f.source, f.reverse) + factsLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectFactsRecorded, "all", nil) + require.Contains(t, f.logs, factsLog.ID) + } + if err := f.fail("log/" + string(log.Event)); err != nil { + return err + } + if _, exists := f.logs[log.ID]; !exists { + f.logs[log.ID] = log + } + return nil + }).AnyTimes() + resolvers := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + resolvers.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).DoAndReturn(func(projectstatus.ResolverConfig) (projectstatus.Resolver, error) { + return resolver, f.fail("resolver/for") + }).AnyTimes() + resolver.EXPECT().Resolve(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request entity.Request) ([]projectstatus.Result, error) { + require.Equal(t, f.request, request) + f.resolverCalls++ + return f.projectStatuses, f.fail("resolve") + }).AnyTimes() + attributers := projectissuesmock.NewMockAttributerFactory(ctrl) + attributer := projectissuesmock.NewMockAttributer(ctrl) + attributers.EXPECT().For(projectissues.AttributerConfig{QueueName: testQueue}).DoAndReturn(func(projectissues.AttributerConfig) (projectissues.Attributer, error) { + return attributer, f.fail("attributer/for") + }).AnyTimes() + attributer.EXPECT().Attribute(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request entity.Request) ([]projectissues.Attribution, error) { + require.Equal(t, f.request, request) + require.Len(t, f.facts, len(f.projectStatuses)) + require.Len(t, f.indexes, len(f.projectStatuses)) + projectFactsRecordedLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectFactsRecorded, "all", nil) + require.Contains(t, f.logs, projectFactsRecordedLog.ID) + f.attributerCalls++ + return f.attributions, f.fail("attribute") + }).AnyTimes() + publisher := queuemock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + if f.request.State != entity.RequestStateCancelled { + require.Contains(t, f.logs, f.completion().ID) + } + f.publishAttempts = append(f.publishAttempts, msg) + if err := f.fail("record/publish"); err != nil { + return err + } + f.records = append(f.records, msg) + return nil + }).AnyTimes() + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: stovepipemq.TopicKeyRecord, Name: "record", Queue: queue}}) + require.NoError(t, err) + f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, stores, requestLogMaterializer, resolvers, attributers, registry, stovepipemq.TopicKeyProjectAnalysis, "stovepipe-projectanalysis") + return f +} + +func (f *fixture) completion() entity.RequestLog { + return requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectIssuesRecorded, f.request.TerminalBuildID, nil) +} + +func (f *fixture) fail(operation string) error { + err := f.failures[operation] + delete(f.failures, operation) + return err +} + +func (f *fixture) retainAttribution(rows map[entity.ProjectIssueAttribution]bool, operation string) func(context.Context, string, entity.ProjectIssueAttribution) error { + return func(_ context.Context, _ string, entry entity.ProjectIssueAttribution) error { + if err := f.fail(operation); err != nil { + return err + } + if rows[entry] { + return storage.ErrAlreadyExists + } + rows[entry] = true + return nil + } +} + +func (f *fixture) message(t *testing.T) entityqueue.Message { + t.Helper() + payload, err := stovepipemq.Marshal(&stovepipemq.ProjectAnalysis{Id: "7", QueueName: testQueue}) + require.NoError(t, err) + return entityqueue.Message{ID: "7", Tenant: testQueue, PartitionKey: "7", Payload: payload} +} + +func (f *fixture) deliver(ctx context.Context, msg entityqueue.Message) error { + delivery := consumermock.NewMockDelivery(f.ctrl) + delivery.EXPECT().Message().Return(msg) + return f.controller.Process(ctx, delivery) +} + +func TestProcessRecordsProjectStatuses(t *testing.T) { + for _, tt := range []struct { + name string + empty bool + unaffected bool + duplicates bool + state entity.RequestState + }{ + {name: "failed build", state: entity.RequestStateFailed}, + {name: "successful build", state: entity.RequestStateSucceeded}, + {name: "empty project statuses", empty: true, state: entity.RequestStateFailed}, + {name: "no affected projects", unaffected: true, state: entity.RequestStateFailed}, + {name: "duplicate associations", duplicates: true, state: entity.RequestStateFailed}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.request.State = tt.state + if tt.empty { + f.projectStatuses, f.attributions = nil, nil + } + if tt.unaffected { + f.attributions[0].Projects = nil + } + if tt.duplicates { + f.attributions[0].Projects = []string{"a", "b", "a"} + } + wantPairs := map[entity.ProjectIssueAttribution]bool{} + for _, attribution := range f.attributions { + for _, project := range attribution.Projects { + wantPairs[entity.ProjectIssueAttribution{TaskID: attribution.TaskID, UnitID: attribution.UnitID, Project: project}] = true + } + } + require.NoError(t, f.deliver(context.Background(), f.message(t))) + require.Len(t, f.facts, len(f.projectStatuses)) + require.Len(t, f.indexes, len(f.projectStatuses)) + require.Equal(t, wantPairs, f.source) + require.Equal(t, f.source, f.reverse) + require.Equal(t, f.completion(), f.logs[f.completion().ID]) + factsLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectFactsRecorded, "all", nil) + expectedCounts := map[string]string{requestlog.MetadataKeyProjectFactCount: "2", requestlog.MetadataKeyBrokenProjectFactCount: "1"} + if tt.empty { + expectedCounts = map[string]string{requestlog.MetadataKeyProjectFactCount: "0", requestlog.MetadataKeyBrokenProjectFactCount: "0"} + } + require.Equal(t, expectedCounts, f.logs[factsLog.ID].Metadata) + require.Len(t, f.records, 1) + record := &stovepipemq.Record{} + require.NoError(t, stovepipemq.Unmarshal(f.records[0].Payload, record)) + require.Equal(t, f.request.ID, record.GetId()) + require.Equal(t, testQueue, record.GetQueueName()) + require.Equal(t, testQueue, f.records[0].Tenant) + require.Equal(t, "7", f.records[0].PartitionKey) + for _, fact := range f.facts { + require.Equal(t, f.request.URI, fact.URI) + require.Equal(t, f.request.ID, fact.RequestID) + require.Equal(t, "service", fact.ProjectType) + require.Positive(t, fact.CreatedAt) + } + }) + } +} + +func TestProcessRetriesPartialWork(t *testing.T) { + for _, operation := range []string{ + "storage/for", "request/get", "completion/get", "analysis/get", "resolver/for", "resolve", "attributer/for", "attribute", + "source/create", "reverse/create", "fact/create", "index/create", "log/project_facts_recorded", "log/project_issues_recorded", "record/publish", + } { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + f.failures[operation] = assert.AnError + msg := f.message(t) + require.ErrorIs(t, f.deliver(context.Background(), msg), assert.AnError) + require.Empty(t, f.records) + if operation != "record/publish" { + require.NotContains(t, f.logs, f.completion().ID) + } + switch operation { + case "resolve": + require.Empty(t, f.source) + require.Empty(t, f.facts) + case "fact/create", "index/create", "log/project_facts_recorded": + require.Zero(t, f.attributerCalls) + require.Empty(t, f.source) + require.Empty(t, f.reverse) + case "attributer/for", "attribute", "source/create", "reverse/create": + require.Len(t, f.facts, 2) + require.Len(t, f.indexes, 2) + projectFactsRecordedLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectFactsRecorded, "all", nil) + require.Contains(t, f.logs, projectFactsRecordedLog.ID) + } + retained := make(map[string]entity.ValidationFact, len(f.facts)) + for project, fact := range f.facts { + retained[project] = fact + } + require.NoError(t, f.deliver(context.Background(), msg)) + require.Len(t, f.facts, 2) + require.Len(t, f.indexes, 2) + require.Len(t, f.source, 2) + require.Equal(t, f.source, f.reverse) + require.Contains(t, f.logs, f.completion().ID) + require.Len(t, f.records, 1) + for project, fact := range retained { + require.Equal(t, fact, f.facts[project]) + } + resolverCalls, attributerCalls := f.resolverCalls, f.attributerCalls + require.NoError(t, f.deliver(context.Background(), msg)) + require.Equal(t, resolverCalls, f.resolverCalls) + require.Equal(t, attributerCalls, f.attributerCalls) + require.Len(t, f.records, 2) + for _, attempt := range f.publishAttempts { + require.Equal(t, f.records[0].ID, attempt.ID) + require.Equal(t, f.records[0].Payload, attempt.Payload) + } + if operation == "record/publish" { + require.Equal(t, 1, f.resolverCalls) + require.Equal(t, 1, f.attributerCalls) + } + }) + } +} + +func TestProcessRequiresFinalAnalysis(t *testing.T) { + for _, abandoned := range []bool{false, true} { + name := "pending analysis" + if abandoned { + name = "abandoned analysis" + } + t.Run(name, func(t *testing.T) { + f := newFixture(t) + f.logs = map[string]entity.RequestLog{} + if abandoned { + log := requestlog.NewRequestEventLog(f.request, entity.RequestEventBuildAnalysisAbandoned, f.request.TerminalBuildID, nil) + f.logs[log.ID] = log + } + require.ErrorIs(t, f.deliver(context.Background(), f.message(t)), storage.ErrNotFound) + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.Empty(t, f.records) + }) + } +} + +func TestProcessRejectsInvalidProviderOutput(t *testing.T) { + for _, tt := range []struct { + name string + invalidStatus bool + attribution projectissues.Attribution + wantFacts int + }{ + {name: "invalid status", invalidStatus: true}, + {name: "missing task", attribution: projectissues.Attribution{UnitID: "unit", Projects: []string{"a"}}, wantFacts: 2}, + {name: "task wide issue", attribution: projectissues.Attribution{TaskID: "task", Projects: []string{"a"}}, wantFacts: 2}, + {name: "missing project", attribution: projectissues.Attribution{TaskID: "task", UnitID: "unit", Projects: []string{""}}, wantFacts: 2}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + if tt.invalidStatus { + f.projectStatuses = []projectstatus.Result{{Degree: entity.DegreeGreen}} + } else { + f.attributions = []projectissues.Attribution{tt.attribution} + } + require.Error(t, f.deliver(context.Background(), f.message(t))) + require.Len(t, f.facts, tt.wantFacts) + require.Len(t, f.indexes, tt.wantFacts) + require.Empty(t, f.source) + require.Empty(t, f.reverse) + require.NotContains(t, f.logs, f.completion().ID) + require.Empty(t, f.records) + }) + } +} + +func TestProcessHandlesNonAnalysisOutcomes(t *testing.T) { + for _, tt := range []struct { + name string + state entity.RequestState + buildID string + wantErr bool + records int + }{ + {name: "cancelled", state: entity.RequestStateCancelled, records: 1}, + {name: "superseded", state: entity.RequestStateSuperseded}, + {name: "not terminal", state: entity.RequestStateProcessing, wantErr: true}, + {name: "no terminal build", state: entity.RequestStateFailed, wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.request.State, f.request.TerminalBuildID = tt.state, tt.buildID + f.logs = map[string]entity.RequestLog{} + err := f.deliver(context.Background(), f.message(t)) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.Empty(t, f.facts) + require.Empty(t, f.source) + require.Len(t, f.records, tt.records) + }) + } +} + +func TestProcessRejectsInvalidIdentity(t *testing.T) { + for _, scenario := range []string{"malformed", "invalid request ID", "tenant mismatch", "partition mismatch", "stored queue mismatch"} { + t.Run(scenario, func(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + switch scenario { + case "malformed": + msg.Payload = []byte("not json") + case "invalid request ID": + var err error + msg.Payload, err = stovepipemq.Marshal(&stovepipemq.ProjectAnalysis{Id: "invalid", QueueName: testQueue}) + require.NoError(t, err) + case "tenant mismatch": + msg.Tenant = "another/queue" + case "partition mismatch": + msg.PartitionKey = "another-request" + case "stored queue mismatch": + f.request.Queue = "another/queue" + } + require.Error(t, f.deliver(context.Background(), msg)) + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.Empty(t, f.records) + }) + } +} + +const projectStatusTestQueue = "monorepo/main" + +var projectStatusTestChangeTime = time.Unix(1_700_000_000, 0).UTC() + +type projectStatusFixture struct { + store *storagemock.MockStorage + factStore *storagemock.MockValidationFactStore + indexStore *storagemock.MockProjectFactByDegreeStore + metricsScope tally.TestScope +} + +func newProjectStatusFixture(ctrl *gomock.Controller) projectStatusFixture { + f := projectStatusFixture{ + store: storagemock.NewMockStorage(ctrl), + factStore: storagemock.NewMockValidationFactStore(ctrl), + indexStore: storagemock.NewMockProjectFactByDegreeStore(ctrl), + metricsScope: tally.NewTestScope("", nil), + } + f.store.EXPECT().GetValidationFactStore().Return(f.factStore).AnyTimes() + f.store.EXPECT().GetProjectFactByDegreeStore().Return(f.indexStore).AnyTimes() + return f +} + +func projectStatusContext() context.Context { + ctx := entityqueue.WithQueueName(context.Background(), projectStatusTestQueue) + return metrics.WithContextTags(ctx, metrics.NewTag("queue", projectStatusTestQueue)) +} + +func projectStatusTestRequest() entity.Request { + return entity.Request{ + ID: "7", Queue: projectStatusTestQueue, URI: "git://remote/monorepo/main/head-sha", + State: entity.RequestStateFailed, Version: 2, TerminalBuildID: "bk-1", + BaseURI: "git://remote/monorepo/main/base-sha", BuildStrategy: entity.BuildStrategyIncrementalSinceGreen, + } +} + +func totalSamples(buckets map[time.Duration]int64) int64 { + var sum int64 + for _, count := range buckets { + sum += count + } + return sum +} + +func TestRecordProjectStatusesPassesRequestBuildContext(t *testing.T) { + for _, tt := range []struct { + name string + buildID string + terminalBuildID string + }{ + {name: "terminal outcome", buildID: "triggered-build", terminalBuildID: "terminal-build"}, + {name: "no terminal outcome", buildID: "triggered-build"}, + {name: "no build"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + m := newProjectStatusFixture(ctrl) + request := projectStatusTestRequest() + request.BuildID = tt.buildID + request.TerminalBuildID = tt.terminalBuildID + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: projectStatusTestQueue}).Return(resolver, nil) + resolver.EXPECT().Resolve(gomock.Any(), request).Return(nil, nil) + + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil) + c := &Controller{ + logger: zap.NewNop().Sugar(), + scope: m.metricsScope.SubScope("projectanalysis_controller"), + resolvers: factory, + requestLogMaterializer: requestLogMaterializer, + } + err := c.recordProjectStatuses(projectStatusContext(), m.store, request) + + require.NoError(t, err) + }) + } +} + +func TestRecordProjectStatusesResolverMetrics(t *testing.T) { + validProjectStatuses := []projectstatus.Result{ + {ProjectInfo: projectstatus.ProjectInfo{Name: "project-a"}, Degree: entity.DegreeGreen}, + {ProjectInfo: projectstatus.ProjectInfo{Name: "project-b"}, Degree: entity.DegreeBroken}, + } + for _, tt := range []struct { + name string + projectStatuses []projectstatus.Result + factoryErr error + resolveErr error + invalid bool + wantResult string + }{ + {name: "projects", projectStatuses: validProjectStatuses, wantResult: "success"}, + {name: "nil project statuses", wantResult: "success"}, + {name: "empty project statuses", projectStatuses: []projectstatus.Result{}, wantResult: "success"}, + {name: "factory error", factoryErr: assert.AnError, wantResult: "error"}, + {name: "factory cancellation", factoryErr: context.Canceled, wantResult: "cancel"}, + {name: "resolver error with partial project statuses", projectStatuses: validProjectStatuses, resolveErr: assert.AnError, wantResult: "error"}, + {name: "resolver cancellation", resolveErr: context.Canceled, wantResult: "cancel"}, + {name: "resolver deadline", resolveErr: context.DeadlineExceeded, wantResult: "error"}, + {name: "invalid project", projectStatuses: []projectstatus.Result{{Degree: entity.DegreeGreen}}, invalid: true, wantResult: "error"}, + {name: "invalid degree", projectStatuses: []projectstatus.Result{{ProjectInfo: projectstatus.ProjectInfo{Name: "project-a"}, Degree: math.NaN()}}, invalid: true, wantResult: "error"}, + {name: "duplicate projects", projectStatuses: []projectstatus.Result{validProjectStatuses[0], validProjectStatuses[0]}, invalid: true, wantResult: "error"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + m := newProjectStatusFixture(ctrl) + request := projectStatusTestRequest() + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: projectStatusTestQueue}).Return(resolver, tt.factoryErr) + if tt.factoryErr == nil { + resolver.EXPECT().Resolve(gomock.Any(), request).Return(tt.projectStatuses, tt.resolveErr) + } + + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + if tt.factoryErr == nil && tt.resolveErr == nil && !tt.invalid { + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).Times(len(tt.projectStatuses)) + m.indexStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).Times(len(tt.projectStatuses)) + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil) + } + c := &Controller{ + logger: zap.NewNop().Sugar(), + scope: m.metricsScope.SubScope("projectanalysis_controller"), + resolvers: factory, + requestLogMaterializer: requestLogMaterializer, + } + err := c.recordProjectStatuses(projectStatusContext(), m.store, request) + + switch { + case tt.factoryErr != nil: + require.ErrorIs(t, err, tt.factoryErr) + case tt.resolveErr != nil: + require.ErrorIs(t, err, tt.resolveErr) + case tt.invalid: + require.Error(t, err) + default: + require.NoError(t, err) + } + assertProjectStatusOperationMetrics(t, m.metricsScope, "record_project_statuses", tt.wantResult) + histograms := m.metricsScope.Snapshot().Histograms() + const projectsKey = "projectanalysis_controller.record_project_statuses.projects+queue=" + projectStatusTestQueue + if err != nil { + assert.NotContains(t, histograms, projectsKey) + return + } + projects, ok := histograms[projectsKey] + require.True(t, ok) + assert.EqualValues(t, 1, projects.Values()[float64(len(tt.projectStatuses))]) + var samples int64 + for _, count := range projects.Values() { + samples += count + } + assert.EqualValues(t, 1, samples) + }) + } +} + +func TestRecordProjectStatusesPersistenceMetrics(t *testing.T) { + for _, tt := range []struct { + name string + empty bool + existing bool + factErr error + indexErr error + completionErr error + wantResult string + }{ + {name: "projects", wantResult: "success"}, + {name: "empty project statuses", empty: true, wantResult: "success"}, + {name: "existing facts and indexes", existing: true, wantResult: "success"}, + {name: "fact write error", factErr: assert.AnError, wantResult: "error"}, + {name: "fact write cancellation", factErr: context.Canceled, wantResult: "cancel"}, + {name: "index write error", indexErr: assert.AnError, wantResult: "error"}, + {name: "index write cancellation", indexErr: context.Canceled, wantResult: "cancel"}, + {name: "completion error", completionErr: assert.AnError, wantResult: "error"}, + {name: "completion cancellation", completionErr: context.Canceled, wantResult: "cancel"}, + {name: "empty completion error", empty: true, completionErr: assert.AnError, wantResult: "error"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + m := newProjectStatusFixture(ctrl) + request := projectStatusTestRequest() + var projectStatuses []projectstatus.Result + if !tt.empty { + projectStatuses = []projectstatus.Result{ + {ProjectInfo: projectstatus.ProjectInfo{Name: "project-a"}, Degree: entity.DegreeGreen}, + {ProjectInfo: projectstatus.ProjectInfo{Name: "project-b"}, Degree: entity.DegreeBroken}, + } + } + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: projectStatusTestQueue}).Return(resolver, nil) + resolver.EXPECT().Resolve(gomock.Any(), request).Return(projectStatuses, nil) + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + c := &Controller{ + logger: zap.NewNop().Sugar(), + scope: m.metricsScope.SubScope("projectanalysis_controller"), + requestLogMaterializer: requestLogMaterializer, + resolvers: factory, + } + for _, status := range projectStatuses { + fact := entity.ValidationFact{URI: request.URI, Project: status.ProjectInfo.Name, Degree: status.Degree, RequestID: request.ID, CreatedAt: projectStatusTestChangeTime.UnixMilli()} + createErr := tt.factErr + if tt.existing { + createErr = storage.ErrAlreadyExists + } + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(createErr) + if tt.factErr != nil { + break + } + indexErr := tt.indexErr + if tt.existing { + m.factStore.EXPECT().Get(gomock.Any(), request.URI, status.ProjectInfo.Name).Return(fact, nil) + indexErr = storage.ErrAlreadyExists + } + entry, err := entity.NewProjectFactByDegree(fact) + require.NoError(t, err) + m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(indexErr) + if tt.indexErr != nil { + break + } + } + if tt.factErr == nil && tt.indexErr == nil { + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventProjectFactsRecorded, log.Event) + return tt.completionErr + }) + } + + err := c.recordProjectStatuses(projectStatusContext(), m.store, request) + + switch { + case tt.factErr != nil: + require.ErrorIs(t, err, tt.factErr) + case tt.indexErr != nil: + require.ErrorIs(t, err, tt.indexErr) + case tt.completionErr != nil: + require.ErrorIs(t, err, tt.completionErr) + default: + require.NoError(t, err) + } + assertProjectStatusOperationMetrics(t, m.metricsScope, "record_project_statuses", tt.wantResult) + }) + } +} + +func TestRecordProjectStatusesRepairsIndexOnRetry(t *testing.T) { + for _, retryErr := range []error{nil, storage.ErrAlreadyExists} { + name := "index repaired" + if retryErr != nil { + name = "index already repaired" + } + t.Run(name, func(t *testing.T) { + ctrl := gomock.NewController(t) + m := newProjectStatusFixture(ctrl) + request := projectStatusTestRequest() + stored := entity.ValidationFact{URI: request.URI, Project: "p", Degree: 0.25, RequestID: request.ID, CreatedAt: 10} + entry, err := entity.NewProjectFactByDegree(stored) + require.NoError(t, err) + writeErr := errors.New("index write unavailable") + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: projectStatusTestQueue}).Return(resolver, nil).Times(2) + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + c := &Controller{ + logger: zap.NewNop().Sugar(), + scope: m.metricsScope.SubScope("projectanalysis_controller"), + requestLogMaterializer: requestLogMaterializer, + resolvers: factory, + } + + gomock.InOrder( + resolver.EXPECT().Resolve(gomock.Any(), request).Return([]projectstatus.Result{{ProjectInfo: projectstatus.ProjectInfo{Name: "p"}, Degree: 0.25}}, nil), + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil), + m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(writeErr), + resolver.EXPECT().Resolve(gomock.Any(), request).Return([]projectstatus.Result{{ProjectInfo: projectstatus.ProjectInfo{Name: "p"}, Degree: 0.75}}, nil), + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists), + m.factStore.EXPECT().Get(gomock.Any(), request.URI, "p").Return(stored, nil), + m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(retryErr), + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventProjectFactsRecorded, log.Event) + return nil + }), + ) + require.ErrorIs(t, c.recordProjectStatuses(projectStatusContext(), m.store, request), writeErr) + require.NoError(t, c.recordProjectStatuses(projectStatusContext(), m.store, request)) + snapshot := m.metricsScope.Snapshot() + assert.EqualValues(t, 2, snapshot.Counters()["projectanalysis_controller.record_project_statuses.start+queue="+projectStatusTestQueue].Value()) + for _, status := range []string{"success", "error"} { + finish, ok := snapshot.Histograms()["projectanalysis_controller.record_project_statuses.finish+queue="+projectStatusTestQueue+",result="+status] + require.True(t, ok) + assert.EqualValues(t, 1, totalSamples(finish.Durations())) + } + }) + } +} + +func TestRecordProjectStatusesCountsPersistedOutcomes(t *testing.T) { + for _, tt := range []struct { + name string + storedDegree, resolvedDegree float64 + broken string + }{ + {name: "new green", storedDegree: entity.DegreeGreen, resolvedDegree: entity.DegreeGreen, broken: "0"}, + {name: "new partially broken", storedDegree: 0.25, resolvedDegree: 0.25, broken: "1"}, + {name: "new fully broken", storedDegree: entity.DegreeBroken, resolvedDegree: entity.DegreeBroken, broken: "1"}, + {name: "adopts existing green", storedDegree: entity.DegreeGreen, resolvedDegree: entity.DegreeBroken, broken: "0"}, + {name: "adopts existing broken", storedDegree: entity.DegreeBroken, resolvedDegree: entity.DegreeGreen, broken: "1"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + m := newProjectStatusFixture(ctrl) + request := projectStatusTestRequest() + stored := entity.ValidationFact{URI: request.URI, Project: "p", Degree: tt.storedDegree, RequestID: request.ID, CreatedAt: 10} + if tt.storedDegree != tt.resolvedDegree { + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.factStore.EXPECT().Get(gomock.Any(), request.URI, "p").Return(stored, nil) + } else { + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + } + m.indexStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + factory := projectstatusmock.NewMockResolverFactory(ctrl) + resolver := projectstatusmock.NewMockResolver(ctrl) + factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: projectStatusTestQueue}).Return(resolver, nil) + resolver.EXPECT().Resolve(gomock.Any(), request).Return([]projectstatus.Result{{ProjectInfo: projectstatus.ProjectInfo{Name: "p"}, Degree: tt.resolvedDegree}}, nil) + requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) + c := &Controller{ + logger: zap.NewNop().Sugar(), + scope: m.metricsScope.SubScope("projectanalysis_controller"), + requestLogMaterializer: requestLogMaterializer, + resolvers: factory, + } + requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, map[string]string{ + requestlog.MetadataKeyProjectFactCount: "1", + requestlog.MetadataKeyBrokenProjectFactCount: tt.broken, + }, log.Metadata) + return nil + }) + require.NoError(t, c.recordProjectStatuses(projectStatusContext(), m.store, request)) + }) + } +} + +func assertProjectStatusOperationMetrics(t *testing.T, scope tally.TestScope, operation, outcome string) { + t.Helper() + snapshot := scope.Snapshot() + start, ok := snapshot.Counters()["projectanalysis_controller."+operation+".start+queue="+projectStatusTestQueue] + require.True(t, ok) + assert.EqualValues(t, 1, start.Value()) + finish, ok := snapshot.Histograms()["projectanalysis_controller."+operation+".finish+queue="+projectStatusTestQueue+",result="+outcome] + require.True(t, ok) + assert.EqualValues(t, 1, totalSamples(finish.Durations())) +}