| package web |
| |
| import ( |
| "bytes" |
| "context" |
| "crypto/md5" |
| "encoding/json" |
| "fmt" |
| "image" |
| "image/png" |
| "net/http" |
| "net/url" |
| "path" |
| "sort" |
| "strconv" |
| "strings" |
| "sync" |
| "time" |
| |
| "github.com/gorilla/mux" |
| lru "github.com/hashicorp/golang-lru" |
| "github.com/jackc/pgx/v4" |
| "github.com/jackc/pgx/v4/pgxpool" |
| "go.opencensus.io/trace" |
| "golang.org/x/sync/errgroup" |
| "golang.org/x/time/rate" |
| |
| "go.skia.org/infra/go/httputils" |
| "go.skia.org/infra/go/human" |
| "go.skia.org/infra/go/login" |
| "go.skia.org/infra/go/metrics2" |
| "go.skia.org/infra/go/now" |
| "go.skia.org/infra/go/paramtools" |
| "go.skia.org/infra/go/skerr" |
| "go.skia.org/infra/go/sklog" |
| "go.skia.org/infra/go/util" |
| "go.skia.org/infra/golden/go/baseline" |
| "go.skia.org/infra/golden/go/clstore" |
| "go.skia.org/infra/golden/go/diff" |
| "go.skia.org/infra/golden/go/expectations" |
| "go.skia.org/infra/golden/go/ignore" |
| "go.skia.org/infra/golden/go/indexer" |
| "go.skia.org/infra/golden/go/search" |
| search_fe "go.skia.org/infra/golden/go/search/frontend" |
| "go.skia.org/infra/golden/go/search/query" |
| "go.skia.org/infra/golden/go/search2" |
| "go.skia.org/infra/golden/go/sql" |
| "go.skia.org/infra/golden/go/status" |
| "go.skia.org/infra/golden/go/storage" |
| "go.skia.org/infra/golden/go/tilesource" |
| "go.skia.org/infra/golden/go/tiling" |
| "go.skia.org/infra/golden/go/tjstore" |
| "go.skia.org/infra/golden/go/types" |
| "go.skia.org/infra/golden/go/validation" |
| "go.skia.org/infra/golden/go/web/frontend" |
| ) |
| |
| const ( |
| // pageSize is the default page size used for pagination. |
| pageSize = 20 |
| |
| // maxPageSize is the maximum page size used for pagination. |
| maxPageSize = 100 |
| |
| // These params limit how anonymous (not logged-in) users can hit various endpoints. |
| // We have two buckets of requests - cheap and expensive. Expensive stuff hits a database |
| // or similar, where as cheap stuff is cached. These limits are shared by *all* endpoints |
| // in a given bucket. See skbug.com/9476 for more. |
| maxAnonQPSExpensive = rate.Limit(0.01) |
| maxAnonBurstExpensive = 50 |
| maxAnonQPSCheap = rate.Limit(5.0) |
| maxAnonBurstCheap = 50 |
| // Special settings for RPCs serving the gerrit plugin. See skbug.com/10768 for more. |
| maxAnonQPSGerritPlugin = rate.Limit(200.0) |
| maxAnonBurstGerritPlugin = 1000 |
| |
| changelistSummaryCacheSize = 10000 |
| |
| // RPCCallCounterMetric is the metric that should be used when counting how many times a given |
| // RPC route is called from clients. |
| RPCCallCounterMetric = "gold_rpc_call_counter" |
| ) |
| |
| type validateFields int |
| |
| const ( |
| // FullFrontEnd means all fields should be set |
| FullFrontEnd validateFields = iota |
| // BaselineSubset means just the fields needed for Baseline Server should be set. |
| BaselineSubset |
| ) |
| |
| // HandlersConfig holds the environment needed by the various http handler functions. |
| type HandlersConfig struct { |
| Baseliner baseline.BaselineFetcher |
| DB *pgxpool.Pool |
| ExpectationsStore expectations.Store |
| GCSClient storage.GCSClient |
| IgnoreStore ignore.Store |
| Indexer indexer.IndexSource |
| ReviewSystems []clstore.ReviewSystem |
| SearchAPI search.SearchAPI |
| Search2API search2.API |
| StatusWatcher *status.StatusWatcher |
| TileSource tilesource.TileSource |
| TryJobStore tjstore.Store |
| } |
| |
| // Handlers represents all the handlers (e.g. JSON endpoints) of Gold. |
| // It should be created by clients using NewHandlers. |
| type Handlers struct { |
| HandlersConfig |
| |
| anonymousExpensiveQuota *rate.Limiter |
| anonymousCheapQuota *rate.Limiter |
| anonymousGerritQuota *rate.Limiter |
| |
| clSummaryCache *lru.Cache |
| |
| // These can be set for unit tests to simplify the testing. |
| testingAuthAs string |
| } |
| |
| // NewHandlers returns a new instance of Handlers. |
| func NewHandlers(conf HandlersConfig, val validateFields) (*Handlers, error) { |
| // These fields are required by all types. |
| if conf.Baseliner == nil { |
| return nil, skerr.Fmt("Baseliner cannot be nil") |
| } |
| if len(conf.ReviewSystems) == 0 { |
| return nil, skerr.Fmt("ReviewSystems cannot be empty") |
| } |
| if conf.GCSClient == nil { |
| return nil, skerr.Fmt("GCSClient cannot be nil") |
| } |
| |
| if val == FullFrontEnd { |
| if conf.DB == nil { |
| return nil, skerr.Fmt("DB cannot be nil") |
| } |
| if conf.ExpectationsStore == nil { |
| return nil, skerr.Fmt("ExpectationsStore cannot be nil") |
| } |
| if conf.IgnoreStore == nil { |
| return nil, skerr.Fmt("IgnoreStore cannot be nil") |
| } |
| if conf.Indexer == nil { |
| return nil, skerr.Fmt("Indexer cannot be nil") |
| } |
| if conf.SearchAPI == nil { |
| return nil, skerr.Fmt("SearchAPI cannot be nil") |
| } |
| if conf.Search2API == nil { |
| return nil, skerr.Fmt("Search2API cannot be nil") |
| } |
| if conf.StatusWatcher == nil { |
| return nil, skerr.Fmt("StatusWatcher cannot be nil") |
| } |
| if conf.TileSource == nil { |
| return nil, skerr.Fmt("TileSource cannot be nil") |
| } |
| if conf.TryJobStore == nil { |
| return nil, skerr.Fmt("TryJobStore cannot be nil") |
| } |
| } |
| |
| clcache, err := lru.New(changelistSummaryCacheSize) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| |
| return &Handlers{ |
| HandlersConfig: conf, |
| anonymousExpensiveQuota: rate.NewLimiter(maxAnonQPSExpensive, maxAnonBurstExpensive), |
| anonymousCheapQuota: rate.NewLimiter(maxAnonQPSCheap, maxAnonBurstCheap), |
| anonymousGerritQuota: rate.NewLimiter(maxAnonQPSGerritPlugin, maxAnonBurstGerritPlugin), |
| clSummaryCache: clcache, |
| testingAuthAs: "", // Just to be explicit that we do *not* bypass Auth. |
| }, nil |
| } |
| |
| // limitForAnonUsers blocks using the configured rate.Limiter for expensive queries. |
| func (wh *Handlers) limitForAnonUsers(r *http.Request) error { |
| if login.LoggedInAs(r) != "" { |
| return nil |
| } |
| return wh.anonymousExpensiveQuota.Wait(r.Context()) |
| } |
| |
| // cheapLimitForAnonUsers blocks using the configured rate.Limiter for cheap queries. |
| func (wh *Handlers) cheapLimitForAnonUsers(r *http.Request) error { |
| if login.LoggedInAs(r) != "" { |
| return nil |
| } |
| return wh.anonymousCheapQuota.Wait(r.Context()) |
| } |
| |
| // cheapLimitForGerritPlugin blocks using the configured rate.Limiter for queries for the |
| // Gerrit Plugin. |
| func (wh *Handlers) cheapLimitForGerritPlugin(r *http.Request) error { |
| if login.LoggedInAs(r) != "" { |
| return nil |
| } |
| return wh.anonymousGerritQuota.Wait(r.Context()) |
| } |
| |
| // TODO(stephana): once the byBlameHandler is removed, refactor this to |
| // remove the redundant types ByBlameEntry and ByBlame. |
| |
| // ByBlameHandler returns a json object with the digests to be triaged grouped by blamelist. |
| func (wh *Handlers) ByBlameHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| // Extract the corpus from the query parameters. |
| corpus := "" |
| if v := r.FormValue("query"); v != "" { |
| if qp, err := url.ParseQuery(v); err != nil { |
| httputils.ReportError(w, err, "invalid input", http.StatusBadRequest) |
| return |
| } else if corpus = qp.Get(types.CorpusField); corpus == "" { |
| // If no corpus specified report an error. |
| http.Error(w, "did not receive value for corpus", http.StatusBadRequest) |
| return |
| } |
| } |
| |
| blameEntries, err := wh.computeByBlame(r.Context(), corpus) |
| if err != nil { |
| httputils.ReportError(w, err, "could not compute blames", http.StatusInternalServerError) |
| return |
| } |
| |
| // Wrap the result in an object because we don't want to return a JSON array. |
| sendJSONResponse(w, frontend.ByBlameResponse{Data: blameEntries}) |
| } |
| |
| // computeByBlame creates several ByBlameEntry structs based on the state |
| // of HEAD and returns them in a slice, for use by the frontend. |
| func (wh *Handlers) computeByBlame(ctx context.Context, corpus string) ([]frontend.ByBlameEntry, error) { |
| idx := wh.Indexer.GetIndex() |
| // At this point query contains at least a corpus. |
| untriagedSummaries, err := idx.SummarizeByGrouping(ctx, corpus, nil, types.ExcludeIgnoredTraces, true) |
| if err != nil { |
| return nil, skerr.Wrapf(err, "could not get summaries for corpus %q", corpus) |
| } |
| commits := idx.Tile().DataCommits() |
| |
| // This is a very simple grouping of digests, for every digest we look up the |
| // blame list for that digest and then use the concatenated git hashes as a |
| // group id. All of the digests are then grouped by their group id. |
| |
| // Collects a ByBlame for each untriaged digest, keyed by group id. |
| grouped := map[string][]frontend.ByBlame{} |
| |
| // The Commit info for each group id. |
| commitinfo := map[string][]tiling.Commit{} |
| // map [groupid] [test] TestRollup |
| rollups := map[string]map[types.TestName]frontend.TestRollup{} |
| |
| for _, s := range untriagedSummaries { |
| test := s.Name |
| for _, d := range s.UntHashes { |
| dist := idx.GetBlame(test, d, commits) |
| if dist.IsEmpty() { |
| // Should only happen if the index isn't quite ready being prepared. |
| // Since we wait until the index is created before exposing the web |
| // server, this should never happen. |
| sklog.Warningf("empty blame for %s %s", test, d) |
| continue |
| } |
| groupid := strings.Join(lookUpCommits(dist.Freq, commits), ":") |
| // Only fill in commitinfo for each groupid only once. |
| if _, ok := commitinfo[groupid]; !ok { |
| var blameCommits []tiling.Commit |
| for _, index := range dist.Freq { |
| blameCommits = append(blameCommits, commits[index]) |
| } |
| sort.Slice(blameCommits, func(i, j int) bool { |
| return blameCommits[i].CommitTime.After(blameCommits[j].CommitTime) |
| }) |
| commitinfo[groupid] = blameCommits |
| } |
| // Construct a ByBlame and add it to grouped. |
| value := frontend.ByBlame{ |
| Test: test, |
| Digest: d, |
| Blame: dist, |
| CommitIndices: dist.Freq, |
| } |
| if _, ok := grouped[groupid]; !ok { |
| grouped[groupid] = []frontend.ByBlame{value} |
| } else { |
| grouped[groupid] = append(grouped[groupid], value) |
| } |
| if _, ok := rollups[groupid]; !ok { |
| rollups[groupid] = map[types.TestName]frontend.TestRollup{} |
| } |
| // Calculate the rollups. |
| r, ok := rollups[groupid][test] |
| if !ok { |
| r = frontend.TestRollup{ |
| Test: test, |
| Num: 0, |
| SampleDigest: d, |
| } |
| } |
| r.Num += 1 |
| rollups[groupid][test] = r |
| } |
| } |
| |
| // Assemble the response. |
| blameEntries := make([]frontend.ByBlameEntry, 0, len(grouped)) |
| for groupid, byBlames := range grouped { |
| rollup := rollups[groupid] |
| nTests := len(rollup) |
| var affectedTests []frontend.TestRollup |
| |
| // Only include the affected tests if there are no more than 10 of them. |
| if nTests <= 10 { |
| affectedTests = make([]frontend.TestRollup, 0, nTests) |
| for _, testInfo := range rollup { |
| affectedTests = append(affectedTests, testInfo) |
| } |
| sort.Slice(affectedTests, func(i, j int) bool { |
| // Put the highest amount of digests first |
| return affectedTests[i].Num > affectedTests[j].Num || |
| // Break ties based on test name (for determinism). |
| (affectedTests[i].Num == affectedTests[j].Num && affectedTests[i].Test < affectedTests[j].Test) |
| }) |
| } |
| |
| blameEntries = append(blameEntries, frontend.ByBlameEntry{ |
| GroupID: groupid, |
| NDigests: len(byBlames), |
| NTests: nTests, |
| AffectedTests: affectedTests, |
| Commits: frontend.FromTilingCommits(commitinfo[groupid]), |
| }) |
| } |
| sort.Slice(blameEntries, func(i, j int) bool { |
| return blameEntries[i].NDigests > blameEntries[j].NDigests || |
| // For test determinism, use GroupID as a tie-breaker |
| (blameEntries[i].NDigests == blameEntries[j].NDigests && blameEntries[i].GroupID < blameEntries[j].GroupID) |
| }) |
| |
| return blameEntries, nil |
| } |
| |
| // lookUpCommits returns the commit hashes for the commit indices in 'freq'. |
| func lookUpCommits(freq []int, commits []tiling.Commit) []string { |
| var ret []string |
| for _, index := range freq { |
| ret = append(ret, commits[index].Hash) |
| } |
| return ret |
| } |
| |
| // ChangelistsHandler returns the list of code_review.Changelists that have |
| // uploaded results to Gold (via TryJobs). |
| func (wh *Handlers) ChangelistsHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| values := r.URL.Query() |
| offset, size, err := httputils.PaginationParams(values, 0, pageSize, maxPageSize) |
| if err != nil { |
| httputils.ReportError(w, err, "Invalid pagination params.", http.StatusInternalServerError) |
| return |
| } |
| |
| _, activeOnly := values["active"] |
| cls, pagination, err := wh.getIngestedChangelists(r.Context(), offset, size, activeOnly) |
| |
| if err != nil { |
| httputils.ReportError(w, err, "Retrieving changelists results failed.", http.StatusInternalServerError) |
| return |
| } |
| |
| sendResponseWithPagination(w, cls, pagination) |
| } |
| |
| // getIngestedChangelists performs the core of the logic for ChangelistsHandler, |
| // by fetching N Changelists given an offset. |
| func (wh *Handlers) getIngestedChangelists(ctx context.Context, offset, size int, activeOnly bool) ([]frontend.Changelist, *httputils.ResponsePagination, error) { |
| so := clstore.SearchOptions{ |
| StartIdx: offset, |
| Limit: size, |
| } |
| if activeOnly { |
| so.OpenCLsOnly = true |
| } |
| |
| grandTotal := 0 |
| var retCls []frontend.Changelist |
| for _, system := range wh.ReviewSystems { |
| cls, total, err := system.Store.GetChangelists(ctx, so) |
| if err != nil { |
| return nil, nil, skerr.Wrapf(err, "fetching Changelists from [%d:%d)", offset, offset+size) |
| } |
| |
| for _, cl := range cls { |
| retCls = append(retCls, frontend.ConvertChangelist(cl, system.ID, system.URLTemplate)) |
| } |
| if grandTotal == clstore.CountMany || total == clstore.CountMany { |
| grandTotal = clstore.CountMany |
| } else { |
| grandTotal += total |
| } |
| } |
| |
| pagination := &httputils.ResponsePagination{ |
| Offset: offset, |
| Size: size, |
| Total: grandTotal, |
| } |
| return retCls, pagination, nil |
| } |
| |
| // PatchsetsAndTryjobsForCL returns a summary of the data we have collected |
| // for a given Changelist, specifically any TryJobs that have uploaded data |
| // to Gold belonging to various patchsets in it. |
| func (wh *Handlers) PatchsetsAndTryjobsForCL(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| clID, ok := mux.Vars(r)["id"] |
| if !ok { |
| http.Error(w, "Must specify 'id' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| crs, ok := mux.Vars(r)["system"] |
| if !ok { |
| http.Error(w, "Must specify 'system' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| system, ok := wh.getCodeReviewSystem(crs) |
| if !ok { |
| http.Error(w, "Invalid Code Review System", http.StatusBadRequest) |
| return |
| } |
| |
| rv, err := wh.getCLSummary(r.Context(), system, clID) |
| if err != nil { |
| httputils.ReportError(w, err, "could not retrieve data for the specified CL.", http.StatusInternalServerError) |
| return |
| } |
| sendJSONResponse(w, rv) |
| } |
| |
| // A list of CI systems we support. So far, the mapping of task ID to link is project agnostic. If |
| // that stops being the case, then we'll need to supply this mapping on a per-instance basis. |
| var cisTemplates = map[string]string{ |
| "cirrus": "https://cirrus-ci.com/task/%s", |
| "buildbucket": "https://cr-buildbucket.appspot.com/build/%s", |
| } |
| |
| // getCLSummary does a bulk of the work for PatchsetsAndTryjobsForCL, specifically |
| // fetching the Changelist and Patchsets from clstore and any associated TryJobs from |
| // the tjstore. |
| func (wh *Handlers) getCLSummary(ctx context.Context, system clstore.ReviewSystem, clID string) (frontend.ChangelistSummary, error) { |
| cl, err := system.Store.GetChangelist(ctx, clID) |
| if err != nil { |
| return frontend.ChangelistSummary{}, skerr.Wrapf(err, "getting CL %s", clID) |
| } |
| |
| // We know xps is sorted by order, if it is non-nil |
| xps, err := system.Store.GetPatchsets(ctx, clID) |
| if err != nil { |
| return frontend.ChangelistSummary{}, skerr.Wrapf(err, "getting Patchsets for CL %s", clID) |
| } |
| |
| var patchsets []frontend.Patchset |
| maxOrder := 0 |
| |
| // TODO(kjlubick): maybe fetch these in parallel (with errgroup) |
| for _, ps := range xps { |
| if ps.Order > maxOrder { |
| maxOrder = ps.Order |
| } |
| psID := tjstore.CombinedPSID{ |
| CL: clID, |
| CRS: system.ID, |
| PS: ps.SystemID, |
| } |
| xtj, err := wh.TryJobStore.GetTryJobs(ctx, psID) |
| if err != nil { |
| return frontend.ChangelistSummary{}, skerr.Wrapf(err, "getting TryJobs for CL %s - PS %s", clID, ps.SystemID) |
| } |
| var tryjobs []frontend.TryJob |
| for _, tj := range xtj { |
| templ := cisTemplates[tj.System] |
| tryjobs = append(tryjobs, frontend.ConvertTryJob(tj, templ)) |
| } |
| |
| patchsets = append(patchsets, frontend.Patchset{ |
| SystemID: ps.SystemID, |
| Order: ps.Order, |
| TryJobs: tryjobs, |
| }) |
| } |
| |
| return frontend.ChangelistSummary{ |
| CL: frontend.ConvertChangelist(cl, system.ID, system.URLTemplate), |
| Patchsets: patchsets, |
| NumTotalPatchsets: maxOrder, |
| }, nil |
| } |
| |
| // ChangelistUntriagedHandler writes out a list of untriaged digests uploaded by this CL that |
| // are not on master already and are not ignored. |
| func (wh *Handlers) ChangelistUntriagedHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForGerritPlugin(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| requestVars := mux.Vars(r) |
| clID, ok := requestVars["id"] |
| if !ok { |
| http.Error(w, "Must specify 'id' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| psID, ok := requestVars["patchset"] |
| if !ok { |
| http.Error(w, "Must specify 'patchset' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| crs, ok := requestVars["system"] |
| if !ok { |
| http.Error(w, "Must specify 'system' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| |
| id := tjstore.CombinedPSID{ |
| CL: clID, |
| CRS: crs, |
| PS: psID, |
| } |
| dl, err := wh.SearchAPI.UntriagedUnignoredTryJobExclusiveDigests(r.Context(), id) |
| if err != nil { |
| sklog.Warningf("Could not get untriaged digests for %v - possibly this CL/PS has none or is too old to be indexed: %s", id, err) |
| // Errors can trip up the Gerrit Plugin (at least until skbug/10706 is resolved). |
| sendJSONResponse(w, search_fe.UntriagedDigestList{TS: time.Now()}) |
| return |
| } |
| sendJSONResponse(w, dl) |
| } |
| |
| // SearchHandler is the endpoint for all searches, including accessing |
| // results that belong to a tryjob. It times out after 3 minutes, to prevent outstanding requests |
| // from growing unbounded. |
| func (wh *Handlers) SearchHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| q, ok := parseSearchQuery(w, r) |
| if !ok { |
| return |
| } |
| ctx, cancel := context.WithTimeout(r.Context(), 3*time.Minute) |
| defer cancel() |
| ctx, span := trace.StartSpan(ctx, "SearchHandler_sql") |
| defer span.End() |
| |
| searchResponse, err := wh.SearchAPI.Search(ctx, q) |
| if err != nil { |
| httputils.ReportError(w, err, "Search for digests failed.", http.StatusInternalServerError) |
| return |
| } |
| sendJSONResponse(w, searchResponse) |
| } |
| |
| // parseSearchQuery extracts the search query from request. |
| func parseSearchQuery(w http.ResponseWriter, r *http.Request) (*query.Search, bool) { |
| q := query.Search{Limit: 50} |
| if err := query.ParseSearch(r, &q); err != nil { |
| httputils.ReportError(w, err, "Search for digests failed.", http.StatusInternalServerError) |
| return nil, false |
| } |
| return &q, true |
| } |
| |
| // DetailsHandler returns the details about a single digest. |
| func (wh *Handlers) DetailsHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| // Extract: test, digest, issue |
| if err := r.ParseForm(); err != nil { |
| httputils.ReportError(w, err, "Failed to parse form values", http.StatusInternalServerError) |
| return |
| } |
| test := r.Form.Get("test") |
| digest := r.Form.Get("digest") |
| if test == "" || !validation.IsValidDigest(digest) { |
| http.Error(w, "Some query parameters are wrong or missing", http.StatusBadRequest) |
| return |
| } |
| clID := r.Form.Get("changelist_id") |
| crs := r.Form.Get("crs") |
| if clID != "" { |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid Code Review System; did you include crs?", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| ret, err := wh.SearchAPI.GetDigestDetails(r.Context(), types.TestName(test), types.Digest(digest), clID, crs) |
| if err != nil { |
| httputils.ReportError(w, err, "Unable to get digest details.", http.StatusInternalServerError) |
| return |
| } |
| sendJSONResponse(w, ret) |
| } |
| |
| // DiffHandler returns difference between two digests. |
| func (wh *Handlers) DiffHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| // Extract: test, left, right where left and right are digests. |
| if err := r.ParseForm(); err != nil { |
| httputils.ReportError(w, err, "Failed to parse form values", http.StatusInternalServerError) |
| return |
| } |
| // TODO(kjlubick) require corpus |
| test := r.Form.Get("test") |
| left := r.Form.Get("left") |
| right := r.Form.Get("right") |
| if test == "" || !validation.IsValidDigest(left) || !validation.IsValidDigest(right) { |
| sklog.Debugf("Bad query params: %q %q %q", test, left, right) |
| http.Error(w, "invalid query params", http.StatusBadRequest) |
| return |
| } |
| clID := r.Form.Get("changelist_id") |
| crs := r.Form.Get("crs") |
| if clID != "" { |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid Code Review System; did you include crs?", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| ret, err := wh.SearchAPI.DiffDigests(r.Context(), types.TestName(test), types.Digest(left), types.Digest(right), clID, crs) |
| if err != nil { |
| httputils.ReportError(w, err, "Unable to compare digests", http.StatusInternalServerError) |
| return |
| } |
| |
| sendJSONResponse(w, ret) |
| } |
| |
| // ListIgnoreRules returns the current ignore rules in JSON format. |
| func (wh *Handlers) ListIgnoreRules(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| |
| _, includeCounts := r.URL.Query()["counts"] |
| // Counting can be expensive, since it goes through every trace. |
| if includeCounts { |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| } else if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| ignores, err := wh.getIgnores(r.Context(), includeCounts) |
| if err != nil { |
| httputils.ReportError(w, err, "Failed to retrieve ignore rules, there may be none.", http.StatusInternalServerError) |
| return |
| } |
| |
| sendJSONResponse(w, ignores) |
| } |
| |
| // getIgnores fetches the ignores from the store and optionally counts how many |
| // times they are applied. |
| func (wh *Handlers) getIgnores(ctx context.Context, withCounts bool) ([]*frontend.IgnoreRule, error) { |
| rules, err := wh.IgnoreStore.List(ctx) |
| if err != nil { |
| return nil, skerr.Wrapf(err, "fetching ignores from store") |
| } |
| |
| // We want to make a slice of pointers because addIgnoreCounts will add the counts in-place. |
| ret := make([]*frontend.IgnoreRule, 0, len(rules)) |
| for _, r := range rules { |
| fr, err := frontend.ConvertIgnoreRule(r) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| ret = append(ret, &fr) |
| } |
| |
| if withCounts { |
| // addIgnoreCounts updates the values of ret directly |
| if err := wh.addIgnoreCounts(ctx, ret); err != nil { |
| return nil, skerr.Wrapf(err, "adding ignore counts to %d rules", len(ret)) |
| } |
| } |
| |
| return ret, nil |
| } |
| |
| // addIgnoreCounts goes through the whole tile and counts how many traces each of the rules |
| // applies to. This uses the most recent index, so there may be some discrepancies in the counts |
| // if a new rule has been added since the last index was computed. |
| func (wh *Handlers) addIgnoreCounts(ctx context.Context, rules []*frontend.IgnoreRule) error { |
| defer metrics2.FuncTimer().Stop() |
| sklog.Debugf("adding counts to %d rules", len(rules)) |
| |
| exp, err := wh.ExpectationsStore.Get(ctx) |
| if err != nil { |
| return skerr.Wrap(err) |
| } |
| // Go through every trace and look for only those that are ignored. Then, count how many |
| // rules apply to a given ignored trace. |
| idx := wh.Indexer.GetIndex() |
| nonIgnoredTraces := idx.DigestCountsByTrace(types.ExcludeIgnoredTraces) |
| traces := idx.SlicedTraces(types.IncludeIgnoredTraces, nil) |
| const numShards = 32 |
| chunkSize := len(traces) / numShards |
| // Very small shards are likely not worth the overhead. |
| if chunkSize < 50 { |
| chunkSize = 50 |
| } |
| // This mutex protects the passed in rules array and allows the final step of each |
| // of the goroutines below to be done safely in parallel to add each shard's results |
| // to the total. |
| var mutex sync.Mutex |
| err = util.ChunkIterParallel(ctx, len(traces), chunkSize, func(ctx context.Context, start, stop int) error { |
| type counts struct { |
| Count int |
| UntriagedCount int |
| ExclusiveCount int |
| ExclusiveUntriagedCount int |
| } |
| ruleCounts := make([]counts, len(rules)) |
| for _, tp := range traces[start:stop] { |
| if err := ctx.Err(); err != nil { |
| return skerr.Wrap(err) |
| } |
| id, tr := tp.ID, tp.Trace |
| if _, ok := nonIgnoredTraces[id]; ok { |
| // This wasn't ignored, so we can skip having to count it |
| continue |
| } |
| idxMatched := -1 |
| untIdxMatched := -1 |
| numMatched := 0 |
| untMatched := 0 |
| for i, r := range rules { |
| if tr.Matches(r.ParsedQuery) { |
| numMatched++ |
| ruleCounts[i].Count++ |
| idxMatched = i |
| |
| // Check to see if the digest is untriaged at head |
| if d := tr.AtHead(); d != tiling.MissingDigest && exp.Classification(tr.TestName(), d) == expectations.Untriaged { |
| ruleCounts[i].UntriagedCount++ |
| untMatched++ |
| untIdxMatched = i |
| } |
| } |
| } |
| // Check for any exclusive matches |
| if numMatched == 1 { |
| ruleCounts[idxMatched].ExclusiveCount++ |
| } |
| if untMatched == 1 { |
| ruleCounts[untIdxMatched].ExclusiveUntriagedCount++ |
| } |
| } |
| mutex.Lock() |
| defer mutex.Unlock() |
| for i, r := range rules { |
| r.Count += ruleCounts[i].Count |
| r.UntriagedCount += ruleCounts[i].UntriagedCount |
| r.ExclusiveCount += ruleCounts[i].ExclusiveCount |
| r.ExclusiveUntriagedCount += ruleCounts[i].ExclusiveUntriagedCount |
| } |
| return nil |
| }) |
| return skerr.Wrap(err) |
| } |
| |
| // UpdateIgnoreRule updates an existing ignores rule. |
| func (wh *Handlers) UpdateIgnoreRule(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| user := wh.loggedInAs(r) |
| if user == "" { |
| http.Error(w, "You must be logged in to update an ignore rule.", http.StatusUnauthorized) |
| return |
| } |
| id := mux.Vars(r)["id"] |
| if id == "" { |
| http.Error(w, "ID must be non-empty.", http.StatusBadRequest) |
| return |
| } |
| expiresInterval, irb, err := getValidatedIgnoreRule(r) |
| if err != nil { |
| httputils.ReportError(w, err, "invalid ignore rule input", http.StatusBadRequest) |
| return |
| } |
| ts := now.Now(r.Context()) |
| ignoreRule := ignore.NewRule(user, ts.Add(expiresInterval), irb.Filter, irb.Note) |
| ignoreRule.ID = id |
| if err := wh.IgnoreStore.Update(r.Context(), ignoreRule); err != nil { |
| httputils.ReportError(w, err, "Unable to update ignore rule", http.StatusInternalServerError) |
| return |
| } |
| |
| sklog.Infof("Successfully updated ignore with id %s", id) |
| sendJSONResponse(w, map[string]string{"updated": "true"}) |
| } |
| |
| // getValidatedIgnoreRule parses the JSON from the given request into an IgnoreRuleBody. As a |
| // convenience, the duration as a time.Duration is returned. |
| func getValidatedIgnoreRule(r *http.Request) (time.Duration, frontend.IgnoreRuleBody, error) { |
| irb := frontend.IgnoreRuleBody{} |
| if err := parseJSON(r, &irb); err != nil { |
| return 0, irb, skerr.Wrapf(err, "reading request JSON") |
| } |
| if irb.Filter == "" { |
| return 0, irb, skerr.Fmt("must supply a filter") |
| } |
| // If a user accidentally includes a huge amount of text, we'd like to catch that here. |
| if len(irb.Filter) >= 10*1024 { |
| return 0, irb, skerr.Fmt("Filter must be < 10 KB") |
| } |
| if len(irb.Note) >= 1024 { |
| return 0, irb, skerr.Fmt("Note must be < 1 KB") |
| } |
| d, err := human.ParseDuration(irb.Duration) |
| if err != nil { |
| return 0, irb, skerr.Wrapf(err, "invalid duration") |
| } |
| return d, irb, nil |
| } |
| |
| // DeleteIgnoreRule deletes an existing ignores rule. |
| func (wh *Handlers) DeleteIgnoreRule(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| user := wh.loggedInAs(r) |
| if user == "" { |
| http.Error(w, "You must be logged in to delete an ignore rule", http.StatusUnauthorized) |
| return |
| } |
| id := mux.Vars(r)["id"] |
| if id == "" { |
| http.Error(w, "ID must be non-empty.", http.StatusBadRequest) |
| return |
| } |
| |
| if err := wh.IgnoreStore.Delete(r.Context(), id); err != nil { |
| httputils.ReportError(w, err, "Unable to delete ignore rule", http.StatusInternalServerError) |
| return |
| } |
| sklog.Infof("Successfully deleted ignore with id %s", id) |
| sendJSONResponse(w, map[string]string{"deleted": "true"}) |
| } |
| |
| // AddIgnoreRule is for adding a new ignore rule. |
| func (wh *Handlers) AddIgnoreRule(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| user := wh.loggedInAs(r) |
| if user == "" { |
| http.Error(w, "You must be logged in to add an ignore rule", http.StatusUnauthorized) |
| return |
| } |
| |
| expiresInterval, irb, err := getValidatedIgnoreRule(r) |
| if err != nil { |
| httputils.ReportError(w, err, "invalid ignore rule input", http.StatusBadRequest) |
| return |
| } |
| ts := now.Now(r.Context()) |
| ignoreRule := ignore.NewRule(user, ts.Add(expiresInterval), irb.Filter, irb.Note) |
| if err := wh.IgnoreStore.Create(r.Context(), ignoreRule); err != nil { |
| httputils.ReportError(w, err, "Failed to create ignore rule", http.StatusInternalServerError) |
| return |
| } |
| |
| sklog.Infof("Successfully added ignore from %s", user) |
| sendJSONResponse(w, map[string]string{"added": "true"}) |
| } |
| |
| // TriageHandler handles a request to change the triage status of one or more |
| // digests of one test. |
| // |
| // It accepts a POST'd JSON serialization of TriageRequest and updates |
| // the expectations. |
| func (wh *Handlers) TriageHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| user := login.LoggedInAs(r) |
| if user == "" { |
| http.Error(w, "You must be logged in to triage.", http.StatusUnauthorized) |
| return |
| } |
| |
| req := frontend.TriageRequest{} |
| if err := parseJSON(r, &req); err != nil { |
| httputils.ReportError(w, err, "Failed to parse JSON request.", http.StatusBadRequest) |
| return |
| } |
| sklog.Infof("Triage request: %#v", req) |
| |
| if err := wh.triage(r.Context(), user, req); err != nil { |
| httputils.ReportError(w, err, "Could not triage", http.StatusInternalServerError) |
| return |
| } |
| // Nothing to return, so just set 200 |
| w.WriteHeader(http.StatusOK) |
| } |
| |
| // triage processes the given TriageRequest. |
| func (wh *Handlers) triage(ctx context.Context, user string, req frontend.TriageRequest) error { |
| // TODO(kjlubick) remove the legacy check for "0" when the frontend no longer sends it. |
| if req.ChangelistID != "" && req.ChangelistID != "0" { |
| if req.CodeReviewSystem == "" { |
| // TODO(kjlubick) remove this default after the search page is converted to lit-html. |
| req.CodeReviewSystem = wh.ReviewSystems[0].ID |
| } |
| if _, ok := wh.getCodeReviewSystem(req.CodeReviewSystem); !ok { |
| return skerr.Fmt("Unknown Code Review System; did you remember to include crs?") |
| } |
| } else { |
| req.CodeReviewSystem = "" |
| } |
| |
| // Build the expectations change request from the list of digests passed in. |
| tc := make([]expectations.Delta, 0, len(req.TestDigestStatus)) |
| for test, digests := range req.TestDigestStatus { |
| for d, label := range digests { |
| if label == "" { |
| // Empty string means the frontend didn't have a closest digest to use when making a |
| // "bulk triage to the closest digest" request. It's easier to catch this on the server |
| // side than make the JS check for empty string and mutate the POST body. |
| continue |
| } |
| if !expectations.ValidLabel(label) { |
| return skerr.Fmt("invalid label %q in triage request", label) |
| } |
| tc = append(tc, expectations.Delta{ |
| Grouping: test, |
| Digest: d, |
| Label: label, |
| }) |
| } |
| } |
| |
| // Use the expectations store for the master branch, unless an issue was given |
| // in the request, then get the expectations store for the issue. |
| expStore := wh.ExpectationsStore |
| // TODO(kjlubick) remove the legacy check here after the frontend bakes in. |
| if req.ChangelistID != "" && req.ChangelistID != "0" { |
| expStore = wh.ExpectationsStore.ForChangelist(req.ChangelistID, req.CodeReviewSystem) |
| } |
| |
| // If set, use the image matching algorithm's name as the author of this change. |
| if req.ImageMatchingAlgorithm != "" { |
| user = req.ImageMatchingAlgorithm |
| } |
| |
| // Add the change. |
| if err := expStore.AddChange(ctx, tc, user); err != nil { |
| return skerr.Wrapf(err, "Failed to store the updated expectations.") |
| } |
| return nil |
| } |
| |
| // StatusHandler returns the current status of with respect to HEAD. |
| func (wh *Handlers) StatusHandler(w http.ResponseWriter, _ *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| |
| // This should be an incredibly cheap call and therefore does not count against any quota. |
| sendJSONResponse(w, wh.StatusWatcher.GetStatus()) |
| } |
| |
| // ClusterDiffHandler calculates the NxN diffs of all the digests that match |
| // the incoming query and returns the data in a format appropriate for |
| // handling in d3. |
| func (wh *Handlers) ClusterDiffHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| // Extract the test name as we only allow clustering within a test. |
| q := query.Search{Limit: 50} |
| if err := query.ParseSearch(r, &q); err != nil { |
| httputils.ReportError(w, err, "Unable to parse query parameter.", http.StatusBadRequest) |
| return |
| } |
| testNames := q.TraceValues[types.PrimaryKeyField] |
| if len(testNames) == 0 { |
| http.Error(w, "No test name provided.", http.StatusBadRequest) |
| return |
| } |
| testName := testNames[0] |
| ctx := r.Context() |
| ctx, span := trace.StartSpan(ctx, "ClusterDiff_sql") |
| defer span.End() |
| |
| idx := wh.Indexer.GetIndex() |
| searchResponse, err := wh.SearchAPI.Search(ctx, &q) |
| if err != nil { |
| httputils.ReportError(w, err, "Search for digests failed.", http.StatusInternalServerError) |
| return |
| } |
| |
| // TODO(kjlubick): Check if we need to sort these |
| // Sort the digests so they are displayed with untriaged last, which means |
| // they will be displayed 'on top', because in SVG document order is z-order. |
| |
| digests := types.DigestSlice{} |
| for _, digest := range searchResponse.Results { |
| digests = append(digests, digest.Digest) |
| } |
| |
| digestIndex := map[types.Digest]int{} |
| for i, d := range digests { |
| digestIndex[d] = i |
| } |
| |
| d3 := ClusterDiffResult{ |
| Test: types.TestName(testName), |
| Nodes: []Node{}, |
| Links: []Link{}, |
| ParamsetByDigest: map[types.Digest]paramtools.ParamSet{}, |
| ParamsetsUnion: paramtools.ParamSet{}, |
| } |
| for i, d := range searchResponse.Results { |
| d3.Nodes = append(d3.Nodes, Node{ |
| Name: d.Digest, |
| Status: d.Status, |
| }) |
| remaining := digests[i:] |
| links, err := wh.getLinksBetween(r.Context(), d.Digest, remaining) |
| if err != nil { |
| httputils.ReportError(w, err, "could not compute diff metrics", http.StatusInternalServerError) |
| return |
| } |
| for otherDigest, distance := range links { |
| d3.Links = append(d3.Links, Link{ |
| Source: digestIndex[d.Digest], |
| Target: digestIndex[otherDigest], |
| Value: distance, |
| }) |
| } |
| d3.ParamsetByDigest[d.Digest] = idx.GetParamsetSummary(d.Test, d.Digest, types.ExcludeIgnoredTraces) |
| for _, p := range d3.ParamsetByDigest[d.Digest] { |
| sort.Strings(p) |
| } |
| d3.ParamsetsUnion.AddParamSet(d3.ParamsetByDigest[d.Digest]) |
| } |
| |
| for _, p := range d3.ParamsetsUnion { |
| sort.Strings(p) |
| } |
| |
| sendJSONResponse(w, d3) |
| } |
| |
| // getLinksBetween queries the SQL DB for the PercentPixelsDiff between the left digest and |
| // the right digests. It returns them in a map. |
| func (wh *Handlers) getLinksBetween(ctx context.Context, left types.Digest, right types.DigestSlice) (map[types.Digest]float32, error) { |
| ctx, span := trace.StartSpan(ctx, "getLinksBetween") |
| span.AddAttributes(trace.Int64Attribute("num_right", int64(len(right)))) |
| defer span.End() |
| const statement = ` |
| SELECT encode(right_digest, 'hex'), percent_pixels_diff FROM DiffMetrics |
| AS OF SYSTEM TIME '-0.1s' |
| WHERE left_digest = $1 AND right_digest IN ` |
| arguments := make([]interface{}, 0, len(right)+1) |
| lb, err := sql.DigestToBytes(left) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| arguments = append(arguments, lb) |
| for _, r := range right { |
| rb, err := sql.DigestToBytes(r) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| arguments = append(arguments, rb) |
| } |
| vp := sql.ValuesPlaceholders(len(arguments), 1) |
| rows, err := wh.DB.Query(ctx, statement+vp, arguments...) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| defer rows.Close() |
| rv := map[types.Digest]float32{} |
| for rows.Next() { |
| var rightD types.Digest |
| var linkDistance float32 |
| if err := rows.Scan(&rightD, &linkDistance); err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| rv[rightD] = linkDistance |
| } |
| return rv, nil |
| } |
| |
| // Node represents a single node in a d3 diagram. Used in ClusterDiffResult. |
| type Node struct { |
| Name types.Digest `json:"name"` |
| Status expectations.Label `json:"status"` |
| } |
| |
| // Link represents a link between d3 nodes, used in ClusterDiffResult. |
| type Link struct { |
| Source int `json:"source"` |
| Target int `json:"target"` |
| Value float32 `json:"value"` |
| } |
| |
| // ClusterDiffResult contains the result of comparing all digests within a test. |
| // It is structured to be easy to render by the D3.js. |
| type ClusterDiffResult struct { |
| Nodes []Node `json:"nodes"` |
| Links []Link `json:"links"` |
| |
| Test types.TestName `json:"test"` |
| ParamsetByDigest map[types.Digest]paramtools.ParamSet `json:"paramsetByDigest"` |
| ParamsetsUnion paramtools.ParamSet `json:"paramsetsUnion"` |
| } |
| |
| // ListTestsHandler returns a summary of the digests seen for a given test. |
| func (wh *Handlers) ListTestsHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| // Inputs: (head, ignored, corpus, keys) |
| q, err := frontend.ParseListTestsQuery(r) |
| if err != nil { |
| httputils.ReportError(w, err, "Failed to parse form data.", http.StatusBadRequest) |
| return |
| } |
| |
| idx := wh.Indexer.GetIndex() |
| summaries, err := idx.SummarizeByGrouping(r.Context(), q.Corpus, q.TraceValues, q.IgnoreState, q.OnlyIncludeDigestsProducedAtHead) |
| if err != nil { |
| httputils.ReportError(w, err, "Could not compute query.", http.StatusInternalServerError) |
| return |
| } |
| // We explicitly want a zero-length slice instead of a nil slice because the latter serializes |
| // to JSON as null instead of [] |
| tests := make([]frontend.TestSummary, 0, len(summaries)) |
| for _, s := range summaries { |
| if s != nil { |
| tests = append(tests, frontend.TestSummary{ |
| Name: s.Name, |
| PositiveDigests: s.Pos, |
| NegativeDigests: s.Neg, |
| UntriagedDigests: s.Untriaged, |
| }) |
| } |
| } |
| // For determinism, sort by test name. The client will have the power to sort these differently. |
| sort.Slice(tests, func(i, j int) bool { |
| return tests[i].Name < tests[j].Name |
| }) |
| // Outputs: []frontend.TestSummary |
| // Frontend will have option to hide tests with no digests. |
| sendJSONResponse(w, tests) |
| } |
| |
| // TriageLogHandler returns the entries in the triagelog paginated |
| // in reverse chronological order. |
| func (wh *Handlers) TriageLogHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| // Get the pagination params. |
| q := r.URL.Query() |
| offset, size, err := httputils.PaginationParams(q, 0, pageSize, maxPageSize) |
| if err != nil { |
| httputils.ReportError(w, err, "Invalid Pagination params", http.StatusBadRequest) |
| return |
| } |
| |
| clID := q.Get("changelist_id") |
| crs := q.Get("crs") |
| if clID != "" { |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid Code Review System; did you include crs?", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| details := q.Get("details") == "true" |
| logEntries, total, err := wh.getTriageLog(r.Context(), crs, clID, offset, size, details) |
| |
| if err != nil { |
| httputils.ReportError(w, err, "Unable to retrieve triage logs", http.StatusInternalServerError) |
| return |
| } |
| |
| pagination := &httputils.ResponsePagination{ |
| Offset: offset, |
| Size: size, |
| Total: total, |
| } |
| |
| sendResponseWithPagination(w, logEntries, pagination) |
| } |
| |
| // getTriageLog does the actual work of the TriageLogHandler, but is easier to test. |
| func (wh *Handlers) getTriageLog(ctx context.Context, crs, changelistID string, offset, size int, withDetails bool) ([]frontend.TriageLogEntry, int, error) { |
| expStore := wh.ExpectationsStore |
| // TODO(kjlubick) remove this legacy handler |
| if changelistID != "" && changelistID != "0" { |
| expStore = wh.ExpectationsStore.ForChangelist(changelistID, crs) |
| } |
| entries, total, err := expStore.QueryLog(ctx, offset, size, withDetails) |
| if err != nil { |
| return nil, -1, skerr.Wrap(err) |
| } |
| logEntries := make([]frontend.TriageLogEntry, 0, len(entries)) |
| for _, e := range entries { |
| logEntries = append(logEntries, frontend.ConvertLogEntry(e)) |
| } |
| return logEntries, total, nil |
| } |
| |
| // TriageUndoHandler performs an "undo" for a given change id. |
| // The change id's are returned in the result of jsonTriageLogHandler. |
| // It accepts one query parameter 'id' which is the id if the change |
| // that should be reversed. |
| // If successful it returns the same result as a call to jsonTriageLogHandler |
| // to reflect the changed triagelog. |
| // TODO(kjlubick): This does not properly handle undoing of ChangelistExpectations. |
| func (wh *Handlers) TriageUndoHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| // Get the user and make sure they are logged in. |
| user := login.LoggedInAs(r) |
| if user == "" { |
| http.Error(w, "You must be logged in to change expectations", http.StatusUnauthorized) |
| return |
| } |
| |
| // Extract the id to undo. |
| changeID := r.URL.Query().Get("id") |
| |
| // Do the undo procedure. |
| if err := wh.ExpectationsStore.UndoChange(r.Context(), changeID, user); err != nil { |
| httputils.ReportError(w, err, "Unable to undo.", http.StatusInternalServerError) |
| return |
| } |
| |
| // Send the same response as a query for the first page. |
| wh.TriageLogHandler(w, r) |
| } |
| |
| // ParamsHandler returns the union of all parameters. |
| func (wh *Handlers) ParamsHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| if err := r.ParseForm(); err != nil { |
| httputils.ReportError(w, err, "Invalid form headers", http.StatusBadRequest) |
| return |
| } |
| clID := r.Form.Get("changelist_id") |
| crs := r.Form.Get("crs") |
| if clID != "" { |
| if crs == "" { |
| // TODO(kjlubick) remove this default after the search page is converted to lit-html. |
| crs = wh.ReviewSystems[0].ID |
| } |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid Code Review System; did you include crs?", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| if clID != "" { |
| clIdx := wh.Indexer.GetIndexForCL(crs, clID) |
| if clIdx != nil { |
| sendJSONResponse(w, clIdx.ParamSet) |
| return |
| } |
| // Fallback to master branch |
| } |
| |
| tile := wh.Indexer.GetIndex().Tile().GetTile(types.IncludeIgnoredTraces) |
| sendJSONResponse(w, tile.ParamSet) |
| } |
| |
| // CommitsHandler returns the commits from the most recent tile. |
| func (wh *Handlers) CommitsHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| cpxTile := wh.TileSource.GetTile() |
| if cpxTile == nil { |
| httputils.ReportError(w, nil, "Not loaded yet - try back later", http.StatusInternalServerError) |
| return |
| } |
| w.Header().Set("Content-Type", "application/json") |
| if err := json.NewEncoder(w).Encode(frontend.FromTilingCommits(cpxTile.DataCommits())); err != nil { |
| sklog.Errorf("Failed to write or encode result: %s", err) |
| } |
| } |
| |
| // TextKnownHashesProxy returns known hashes that have been written to GCS in the background |
| // Each line contains a single digest for an image. Bots will then only upload images which |
| // have a hash not found on this list, avoiding significant amounts of unnecessary uploads. |
| func (wh *Handlers) TextKnownHashesProxy(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| // No limit for anon users - this is an endpoint backed up by baseline servers, and |
| // should be able to handle a large load. |
| |
| w.Header().Set("Content-Type", "text/plain") |
| if err := wh.GCSClient.LoadKnownDigests(r.Context(), w); err != nil { |
| sklog.Errorf("Failed to copy the known hashes from GCS: %s", err) |
| return |
| } |
| } |
| |
| // BaselineHandlerV1 differs from BaselineHandlerV2 in that the "primary" field in the JSON response |
| // is named "master_str". |
| // |
| // TODO(lovisolo): Remove this after all clients have been migrated to the V2 of this RPC. |
| func (wh *Handlers) BaselineHandlerV1(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| // No limit for anon users - this is an endpoint backed up by baseline servers, and |
| // should be able to handle a large load. |
| |
| // Track usage of the legacy /json/expectations/commit/{commit_hash} route. |
| if _, ok := mux.Vars(r)["commit_hash"]; ok { |
| metrics2.GetCounter("gold_baselinehandler_route_legacy").Inc(1) |
| } else { |
| metrics2.GetCounter("gold_baselinehandler_route_new").Inc(1) |
| } |
| |
| q := r.URL.Query() |
| clID := q.Get("issue") |
| issueOnly := q.Get("issueOnly") == "true" |
| crs := q.Get("crs") |
| |
| if clID != "" { |
| if crs == "" { |
| // TODO(kjlubick) remove this default after the search page is converted to lit-html. |
| crs = wh.ReviewSystems[0].ID |
| } |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid CRS provided.", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| bl, err := wh.Baseliner.FetchBaseline(r.Context(), clID, crs, issueOnly) |
| if err != nil { |
| httputils.ReportError(w, err, "Fetching baselines failed.", http.StatusInternalServerError) |
| return |
| } |
| bl.Expectations = expectations.Baseline{} |
| sendJSONResponse(w, bl) |
| } |
| |
| // BaselineHandlerV2 returns a JSON representation of that baseline including |
| // baselines for a options issue. It can respond to requests like these: |
| // |
| // /json/expectations |
| // /json/expectations?issue=123456 |
| // /json/expectations?issue=123456&issueOnly=true |
| // |
| // The "issue" parameter indicates the changelist ID for which we would like to |
| // retrieve the baseline. In that case the returned options will be a blend of |
| // the master baseline and the baseline defined for the changelist (usually |
| // based on tryjob results). |
| // |
| // Parameter "issueOnly" is for debugging purposes only. |
| func (wh *Handlers) BaselineHandlerV2(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| // No limit for anon users - this is an endpoint backed up by baseline servers, and |
| // should be able to handle a large load. |
| |
| q := r.URL.Query() |
| clID := q.Get("issue") |
| issueOnly := q.Get("issueOnly") == "true" |
| crs := q.Get("crs") |
| |
| if clID != "" { |
| if crs == "" { |
| // TODO(kjlubick) remove this default after the search page is converted to lit-html. |
| crs = wh.ReviewSystems[0].ID |
| } |
| if _, ok := wh.getCodeReviewSystem(crs); !ok { |
| http.Error(w, "Invalid CRS provided.", http.StatusBadRequest) |
| return |
| } |
| } else { |
| crs = "" |
| } |
| |
| bl, err := wh.Baseliner.FetchBaseline(r.Context(), clID, crs, issueOnly) |
| if err != nil { |
| httputils.ReportError(w, err, "Fetching baselines failed.", http.StatusInternalServerError) |
| return |
| } |
| |
| // TODO(lovisolo): Delete after the ExpectationsMasterStr field has been removed. |
| bl.DeprecatedExpectations = expectations.Baseline{} |
| |
| sendJSONResponse(w, bl) |
| } |
| |
| // MakeResourceHandler creates a static file handler that sets a caching policy. |
| func MakeResourceHandler(resourceDir string) func(http.ResponseWriter, *http.Request) { |
| fileServer := http.FileServer(http.Dir(resourceDir)) |
| return func(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| // No limit for anon users - this should be fast enough to handle a large load. |
| w.Header().Add("Cache-Control", "max-age=300") |
| fileServer.ServeHTTP(w, r) |
| } |
| } |
| |
| // DigestListHandler returns a list of digests for a given test. This is used by goldctl's |
| // local diff tech. |
| func (wh *Handlers) DigestListHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| if err := r.ParseForm(); err != nil { |
| httputils.ReportError(w, err, "Failed to parse form values", http.StatusInternalServerError) |
| return |
| } |
| |
| test := r.Form.Get("test") |
| corpus := r.Form.Get("corpus") |
| if test == "" || corpus == "" { |
| http.Error(w, "You must include 'test' and 'corpus'", http.StatusBadRequest) |
| return |
| } |
| |
| out := wh.getDigestsResponse(test, corpus) |
| sendJSONResponse(w, out) |
| } |
| |
| // getDigestsResponse returns the digests belonging to the given test (and eventually corpus). |
| func (wh *Handlers) getDigestsResponse(test, corpus string) frontend.DigestListResponse { |
| // TODO(kjlubick): Grouping by only test is something we should avoid. We should |
| // at least group by test and corpus, but maybe something more robust depending |
| // on the instance (e.g. Skia might want to group by colorspace) |
| idx := wh.Indexer.GetIndex() |
| dc := idx.DigestCountsByTest(types.IncludeIgnoredTraces) |
| |
| var xd []types.Digest |
| for d := range dc[types.TestName(test)] { |
| xd = append(xd, d) |
| } |
| |
| // Sort alphabetically for determinism |
| sort.Slice(xd, func(i, j int) bool { |
| return xd[i] < xd[j] |
| }) |
| |
| return frontend.DigestListResponse{ |
| Digests: xd, |
| } |
| } |
| |
| // Whoami returns the email address of the user or service account used to authenticate the |
| // request. For debugging purposes only. |
| func (wh *Handlers) Whoami(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| user := wh.loggedInAs(r) |
| sendJSONResponse(w, map[string]string{"whoami": user}) |
| } |
| |
| func (wh *Handlers) LatestPositiveDigestHandler(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| |
| traceId, ok := mux.Vars(r)["traceId"] |
| if !ok { |
| http.Error(w, "Must specify traceId.", http.StatusBadRequest) |
| return |
| } |
| |
| digest, err := wh.Indexer.GetIndex().MostRecentPositiveDigest(r.Context(), tiling.TraceID(traceId)) |
| if err != nil { |
| httputils.ReportError(w, err, "Could not retrieve most recent positive digest.", http.StatusInternalServerError) |
| return |
| } |
| |
| sendJSONResponse(w, frontend.MostRecentPositiveDigestResponse{Digest: digest}) |
| } |
| |
| // GetPerTraceDigestsByTestName returns the digests in the current trace for the given test name |
| // and corpus, grouped by trace ID. |
| func (wh *Handlers) GetPerTraceDigestsByTestName(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.limitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| } |
| |
| corpus, ok := mux.Vars(r)["corpus"] |
| if !ok { |
| http.Error(w, "Must specify corpus.", http.StatusBadRequest) |
| return |
| } |
| |
| testName, ok := mux.Vars(r)["testName"] |
| if !ok { |
| http.Error(w, "Must specify testName.", http.StatusBadRequest) |
| return |
| } |
| |
| digestsByTraceId := frontend.GetPerTraceDigestsByTestNameResponse{} |
| |
| // Iterate over all traces in the current tile for the given test name. |
| tracesById := wh.Indexer.GetIndex().SlicedTraces(types.IncludeIgnoredTraces, map[string][]string{ |
| types.CorpusField: {corpus}, |
| types.PrimaryKeyField: {testName}, |
| }) |
| for _, tracePair := range tracesById { |
| // Populate map with the trace's digests. |
| digestsByTraceId[tracePair.ID] = tracePair.Trace.Digests |
| } |
| |
| sendJSONResponse(w, digestsByTraceId) |
| } |
| |
| const maxFlakyTraces = 10000 // We don't want to return a slice longer than this because it could |
| // end up with a result that is too big. 10k * ~200 bytes per trace means this return size will be |
| // <= 2MB. |
| |
| // GetFlakyTracesData returns all traces with a number of unique digests (in the current sliding |
| // window of commits) greater than or equal to a certain threshold. |
| func (wh *Handlers) GetFlakyTracesData(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| } |
| |
| minUniqueDigests := 10 |
| minUniqueDigestsStr, ok := mux.Vars(r)["minUniqueDigests"] |
| if ok { |
| var err error |
| minUniqueDigests, err = strconv.Atoi(minUniqueDigestsStr) |
| if err != nil { |
| httputils.ReportError(w, err, "invalid value for minUniqueDigests", http.StatusBadRequest) |
| return |
| } |
| } |
| |
| idx := wh.Indexer.GetIndex() |
| counts := idx.DigestCountsByTrace(types.IncludeIgnoredTraces) |
| |
| flakyData := frontend.FlakyTracesDataResponse{ |
| TileSize: len(idx.Tile().DataCommits()), |
| TotalTraces: len(counts), |
| } |
| |
| for traceID, dc := range counts { |
| if len(dc) >= minUniqueDigests { |
| flakyData.Traces = append(flakyData.Traces, frontend.FlakyTrace{ |
| ID: traceID, |
| UniqueDigests: len(dc), |
| }) |
| } |
| } |
| flakyData.TotalFlakyTraces = len(flakyData.Traces) |
| |
| // Sort the flakiest traces first. |
| sort.Slice(flakyData.Traces, func(i, j int) bool { |
| if flakyData.Traces[i].UniqueDigests == flakyData.Traces[j].UniqueDigests { |
| return flakyData.Traces[i].ID < flakyData.Traces[j].ID |
| } |
| return flakyData.Traces[i].UniqueDigests > flakyData.Traces[j].UniqueDigests |
| }) |
| |
| // Limit the number of traces to maxFlakyTraces, if needed. |
| if len(flakyData.Traces) > maxFlakyTraces { |
| flakyData.Traces = flakyData.Traces[:maxFlakyTraces] |
| } |
| |
| sendJSONResponse(w, flakyData) |
| } |
| |
| // ChangelistSearchRedirect redirects the user to a search page showing the search results |
| // for a given CL. It will do a quick scan of the untriaged digests - if it finds some, it will |
| // include the corpus containing some of those untriaged digests in the search query so the user |
| // will see results (instead of getting directed to a corpus with no results). |
| func (wh *Handlers) ChangelistSearchRedirect(w http.ResponseWriter, r *http.Request) { |
| defer metrics2.FuncTimer().Stop() |
| if err := wh.cheapLimitForAnonUsers(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| } |
| |
| requestVars := mux.Vars(r) |
| crs, ok := requestVars["system"] |
| if !ok { |
| http.Error(w, "Must specify 'system' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| clID, ok := requestVars["id"] |
| if !ok { |
| http.Error(w, "Must specify 'id' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| system, ok := wh.getCodeReviewSystem(crs) |
| if !ok { |
| http.Error(w, "Invalid Code Review System", http.StatusBadRequest) |
| return |
| } |
| |
| baseURL := fmt.Sprintf("/search?issue=%s&crs=%s", clID, system.ID) |
| |
| clIdx := wh.Indexer.GetIndexForCL(system.ID, clID) |
| if clIdx == nil { |
| // Not cached, so we can't cheaply determine the corpus to include |
| if _, err := system.Store.GetChangelist(r.Context(), clID); err != nil { |
| http.NotFound(w, r) |
| return |
| } |
| http.Redirect(w, r, baseURL, http.StatusTemporaryRedirect) |
| return |
| } |
| |
| digestList, err := wh.SearchAPI.UntriagedUnignoredTryJobExclusiveDigests(r.Context(), clIdx.LatestPatchset) |
| if err != nil { |
| sklog.Errorf("Could not find corpus to redirect to for CL %s: %s", clID, err) |
| http.Redirect(w, r, baseURL, http.StatusTemporaryRedirect) |
| return |
| } |
| if len(digestList.Corpora) == 0 { |
| http.Redirect(w, r, baseURL, http.StatusTemporaryRedirect) |
| return |
| } |
| |
| withCorpus := baseURL + "&corpus=" + digestList.Corpora[0] |
| sklog.Debugf("Redirecting to %s", withCorpus) |
| http.Redirect(w, r, withCorpus, http.StatusTemporaryRedirect) |
| } |
| |
| func (wh *Handlers) loggedInAs(r *http.Request) string { |
| if wh.testingAuthAs != "" { |
| return wh.testingAuthAs |
| } |
| return login.LoggedInAs(r) |
| } |
| |
| func (wh *Handlers) getCodeReviewSystem(crs string) (clstore.ReviewSystem, bool) { |
| var system clstore.ReviewSystem |
| found := false |
| for _, rs := range wh.ReviewSystems { |
| if rs.ID == crs { |
| system = rs |
| found = true |
| } |
| } |
| return system, found |
| } |
| |
| const ( |
| validDigestLength = 2 * md5.Size |
| dotPNG = ".png" |
| ) |
| |
| // ImageHandler returns either a single image or a diff between two images identified by their |
| // respective digests. |
| func (wh *Handlers) ImageHandler(w http.ResponseWriter, r *http.Request) { |
| // No rate limit, as this should be quite fast. |
| _, imgFile := path.Split(r.URL.Path) |
| // Get the file that was requested and verify that it's a valid PNG file. |
| if !strings.HasSuffix(imgFile, dotPNG) { |
| noCacheNotFound(w, r) |
| return |
| } |
| |
| // Trim the image extension to get the image or diff ID. |
| imgID := imgFile[:len(imgFile)-len(dotPNG)] |
| // Cache images for 12 hours. |
| w.Header().Set("Cache-Control", "public, max-age=43200") |
| if len(imgID) == validDigestLength { |
| // Example request: |
| // https://skia-infra-gold.skia.org/img/images/8588cad6f3821b948468df35b67778ef.png |
| wh.serveImageWithDigest(w, r, types.Digest(imgID)) |
| } else if len(imgID) == validDigestLength*2+1 { |
| // Example request: |
| // https://skia-infra-gold.skia.org/img/diffs/81c4d3a64cf32143ff6c1fbf4cbbec2d-d20731492287002a3f046eae4bd4ce7d.png |
| left := types.Digest(imgID[:validDigestLength]) |
| // + 1 for the dash |
| right := types.Digest(imgID[validDigestLength+1:]) |
| wh.serveImageDiff(w, r, left, right) |
| } else { |
| noCacheNotFound(w, r) |
| return |
| } |
| } |
| |
| // serveImageWithDigest downloads the image from GCS and returns it. If there is an error, a 404 |
| // or 500 error is returned, as appropriate. |
| func (wh *Handlers) serveImageWithDigest(w http.ResponseWriter, r *http.Request, digest types.Digest) { |
| ctx, span := trace.StartSpan(r.Context(), "frontend_serveImageWithDigest") |
| defer span.End() |
| // Go's image package has no color profile support and we convert to 8-bit NRGBA to diff, |
| // but our source images may have embedded color profiles and be up to 16-bit. So we must |
| // at least take care to serve the original .pngs unaltered. |
| b, err := wh.GCSClient.GetImage(ctx, digest) |
| if err != nil { |
| sklog.Warningf("Could not get image with digest %s: %s", digest, err) |
| noCacheNotFound(w, r) |
| return |
| } |
| if _, err := w.Write(b); err != nil { |
| httputils.ReportError(w, err, "Could not load image. Try again later.", http.StatusInternalServerError) |
| return |
| } |
| } |
| |
| // serveImageDiff downloads the left and right images, computes the diff between them, encodes |
| // the diff as a PNG image and writes it to the provided ResponseWriter. If there is an error, it |
| // returns a 404 or 500 error as appropriate. |
| func (wh *Handlers) serveImageDiff(w http.ResponseWriter, r *http.Request, left types.Digest, right types.Digest) { |
| ctx, span := trace.StartSpan(r.Context(), "frontend_serveImageDiff") |
| defer span.End() |
| // TODO(lovisolo): Diff in NRGBA64? |
| // TODO(lovisolo): Make sure each pair of images is in the same color space before diffing? |
| // (They probably are today but it'd be a good correctness check to make sure.) |
| eg, eCtx := errgroup.WithContext(ctx) |
| var leftImg *image.NRGBA |
| var rightImg *image.NRGBA |
| eg.Go(func() error { |
| b, err := wh.GCSClient.GetImage(eCtx, left) |
| if err != nil { |
| return skerr.Wrap(err) |
| } |
| leftImg, err = decode(b) |
| return skerr.Wrap(err) |
| }) |
| eg.Go(func() error { |
| b, err := wh.GCSClient.GetImage(eCtx, right) |
| if err != nil { |
| return skerr.Wrap(err) |
| } |
| rightImg, err = decode(b) |
| return skerr.Wrap(err) |
| }) |
| if err := eg.Wait(); err != nil { |
| sklog.Warningf("Could not get diff for images %q and %q: %s", left, right, err) |
| noCacheNotFound(w, r) |
| return |
| } |
| // Compute the diff image. |
| _, diffImg := diff.PixelDiff(leftImg, rightImg) |
| |
| // Write output image to the http.ResponseWriter. Content-Type is set automatically |
| // based on the first 512 bytes of written data. See docs for ResponseWriter.Write() |
| // for details. |
| // |
| // The encoding step below does not take color profiles into account. This is fine since |
| // both the left and right images used to compute the diff are in the same color space, |
| // and also because the resulting diff image is just a visual approximation of the |
| // differences between the left and right images. |
| if err := encodeImg(w, diffImg); err != nil { |
| httputils.ReportError(w, err, "could not serve diff image", http.StatusInternalServerError) |
| return |
| } |
| } |
| |
| // decode decodes the provided bytes as a PNG and returns them as an *image.NRGBA. |
| func decode(b []byte) (*image.NRGBA, error) { |
| im, err := png.Decode(bytes.NewReader(b)) |
| if err != nil { |
| return nil, skerr.Wrap(err) |
| } |
| return diff.GetNRGBA(im), nil |
| } |
| |
| // noCacheNotFound disables caching and returns a 404. |
| func noCacheNotFound(w http.ResponseWriter, r *http.Request) { |
| w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate") |
| http.NotFound(w, r) |
| } |
| |
| // ChangelistSummaryHandler returns a summary of the new and untriaged digests produced by this |
| // CL across all Patchsets. |
| func (wh *Handlers) ChangelistSummaryHandler(w http.ResponseWriter, r *http.Request) { |
| ctx, span := trace.StartSpan(r.Context(), "web_ChangelistSummaryHandler") |
| defer span.End() |
| if err := wh.cheapLimitForGerritPlugin(r); err != nil { |
| httputils.ReportError(w, err, "Try again later", http.StatusInternalServerError) |
| return |
| } |
| clID, ok := mux.Vars(r)["id"] |
| if !ok { |
| http.Error(w, "Must specify 'id' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| crs, ok := mux.Vars(r)["system"] |
| if !ok { |
| http.Error(w, "Must specify 'system' of Changelist.", http.StatusBadRequest) |
| return |
| } |
| system, ok := wh.getCodeReviewSystem(crs) |
| if !ok { |
| http.Error(w, "Invalid Code Review System", http.StatusBadRequest) |
| return |
| } |
| |
| qCLID := sql.Qualify(system.ID, clID) |
| sum, err := wh.getCLSummary2(ctx, qCLID) |
| if err != nil { |
| httputils.ReportError(w, err, "Could not get summary", http.StatusInternalServerError) |
| return |
| } |
| rv := frontend.ConvertChangelistSummaryResponseV1(sum) |
| sendJSONResponse(w, rv) |
| } |
| |
| // getCLSummary2 fetches, caches, and returns the summary for a given CL. If the result has already |
| // been cached and the cached value is still valid, it will return that cached value. |
| func (wh *Handlers) getCLSummary2(ctx context.Context, qCLID string) (search2.NewAndUntriagedSummary, error) { |
| ts, err := wh.Search2API.ChangelistLastUpdated(ctx, qCLID) |
| if err != nil { |
| return search2.NewAndUntriagedSummary{}, skerr.Wrap(err) |
| } |
| |
| cached, ok := wh.clSummaryCache.Get(qCLID) |
| if ok { |
| sum, ok := cached.(search2.NewAndUntriagedSummary) |
| if ok { |
| if ts.Before(sum.LastUpdated) || sum.LastUpdated.Equal(ts) { |
| return sum, nil |
| } |
| } |
| // Invalid or old cache entry - must overwrite. |
| } |
| |
| sum, err := wh.Search2API.NewAndUntriagedSummaryForCL(ctx, qCLID) |
| if err != nil { |
| return search2.NewAndUntriagedSummary{}, skerr.Wrap(err) |
| } |
| wh.clSummaryCache.Add(qCLID, sum) |
| return sum, nil |
| } |
| |
| // StartCacheWarming starts a go routine to warm the CL Summary cache. This way, most summaries are |
| // responsive, even on big instances. |
| func (wh *Handlers) StartCacheWarming(ctx context.Context) { |
| // We warm every CL that was open and produced data in the last 5 days. After the first cycle, |
| // we will incrementally update the cache. |
| lastCheck := now.Now(ctx).Add(-5 * 24 * time.Hour) |
| go util.RepeatCtx(ctx, time.Minute, func(ctx context.Context) { |
| ctx, span := trace.StartSpan(ctx, "web_warmCacheCycle", trace.WithSampler(trace.AlwaysSample())) |
| defer span.End() |
| newTS := now.Now(ctx) |
| rows, err := wh.DB.Query(ctx, ` |
| SELECT changelist_id FROM Changelists AS OF SYSTEM TIME '-0.1s' |
| WHERE status = 'open' and last_ingested_data > $1 |
| ORDER BY last_ingested_data DESC`, lastCheck) |
| if err != nil { |
| if err == pgx.ErrNoRows { |
| sklog.Infof("No CLS updated since %s", lastCheck) |
| lastCheck = newTS |
| return |
| } |
| sklog.Errorf("Could not fetch updated CLs to warm cache: %s", err) |
| return |
| } |
| defer rows.Close() |
| var qualifiedIDS []string |
| for rows.Next() { |
| var qID string |
| if err := rows.Scan(&qID); err != nil { |
| sklog.Errorf("Could not scan: %s", err) |
| } |
| qualifiedIDS = append(qualifiedIDS, qID) |
| } |
| sklog.Infof("Warming cache for %d CLs", len(qualifiedIDS)) |
| span.AddAttributes(trace.Int64Attribute("num_cls", int64(len(qualifiedIDS)))) |
| // warm cache 3 at a time. |
| _ = util.ChunkIterParallel(ctx, len(qualifiedIDS), len(qualifiedIDS)/3+1, func(ctx context.Context, startIdx int, endIdx int) error { |
| if err := ctx.Err(); err != nil { |
| return nil |
| } |
| for _, qCLID := range qualifiedIDS[startIdx:endIdx] { |
| _, err := wh.getCLSummary2(ctx, qCLID) |
| if err != nil { |
| sklog.Warningf("Ignoring error while warming CL Cache for %s: %s", qCLID, err) |
| } |
| } |
| return nil |
| }) |
| lastCheck = newTS |
| sklog.Infof("Done warming cache") |
| }) |
| } |