func statusMessage(passingCols, completedCols, passingCells, filledCells int) string {
	if filledCells == 0 {
		return noRuns
	}
	return fmtStatus(passingCols, completedCols, passingCells, filledCells)
}
2483 of 115784 tests (2.1%) and 163 of 164 runs (99.4%) failed in the past 7 days
package summarizer
import (
	"compress/zlib"
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net/url"
	"path"
	"regexp"
	"strings"
	"sync"
	"time"
	"cloud.google.com/go/storage"
	"github.com/GoogleCloudPlatform/testgrid/config"
	"github.com/GoogleCloudPlatform/testgrid/internal/result"
	configpb "github.com/GoogleCloudPlatform/testgrid/pb/config"
	statepb "github.com/GoogleCloudPlatform/testgrid/pb/state"
	summarypb "github.com/GoogleCloudPlatform/testgrid/pb/summary"
	statuspb "github.com/GoogleCloudPlatform/testgrid/pb/test_status"
	"github.com/GoogleCloudPlatform/testgrid/util/gcs"
	"github.com/GoogleCloudPlatform/testgrid/util/metrics"
	"github.com/golang/protobuf/proto"
	"github.com/sirupsen/logrus"
)
type Metrics struct {
	Successes metrics.Counter
	Errors    metrics.Counter
}
func (mets *Metrics) Error() {
	if mets.Errors != nil {
		mets.Errors.Add(1, "summarizer")
	}
}
func (mets *Metrics) Success() {
	if mets.Successes != nil {
		mets.Successes.Add(1, "summarizer")
	}
}
type gridReader func(ctx context.Context) (io.ReadCloser, time.Time, int64, error)
type groupFinder func(string) (*configpb.TestGroup, gridReader, error)
func fetchConfig(ctx context.Context, client gcs.ConditionalClient, configPath gcs.Path, dashboard string) (*configpb.Configuration, *storage.ReaderObjectAttrs, error) {
	r, attrs, err := client.Open(ctx, configPath)
	if err != nil {
		return nil, nil, fmt.Errorf("open: %w", err)
	}
	cfg, err := config.Unmarshal(r)
	if err != nil {
		return nil, nil, fmt.Errorf("unmarshal: %v", err)
	}
	if dashboard != "" {
		dashes := make([]*configpb.Dashboard, 0, 1)
		for _, d := range cfg.Dashboards {
			if d.Name != dashboard {
				continue
			}
			dashes = append(dashes, d)
		}
		if len(dashes) == 0 {
			return nil, nil, errors.New("dashboard not found")
		}
		cfg.Dashboards = dashes
	}
	return cfg, attrs, nil
}
type configSnapshot struct {
	dashboards map[string]*configpb.Dashboard
	groups     map[string]*configpb.TestGroup
	lock       sync.RWMutex
}
func (cs *configSnapshot) dashboard(name string) *configpb.Dashboard {
	cs.lock.RLock()
	defer cs.lock.RUnlock()
	return cs.dashboards[name]
}
func (cs *configSnapshot) group(name string) *configpb.TestGroup {
	cs.lock.RLock()
	defer cs.lock.RUnlock()
	return cs.groups[name]
}
func (cs *configSnapshot) dashboardTestGroups(dashboardName string) (*configpb.Dashboard, map[string]*configpb.TestGroup) {
	cs.lock.RLock()
	defer cs.lock.RLock()
	d := cs.dashboards[dashboardName]
	if d == nil {
		return nil, nil
	}
	gs := make(map[string]*configpb.TestGroup, len(d.DashboardTab))
	for _, tab := range d.DashboardTab {
		n := tab.TestGroupName
		gs[n] = cs.groups[n]
	}
	return d, gs
}
func (cs *configSnapshot) update(ctx context.Context, log logrus.FieldLogger, client gcs.ConditionalClient, configPath gcs.Path, summaryPathPrefix, dashboard string, q *config.DashboardQueue, freq time.Duration) (*storage.ReaderObjectAttrs, error) {
	cfg, attrs, err := fetchConfig(ctx, client, configPath, dashboard)
	if err != nil {
		return nil, err
	}
	now := time.Now()
	dashboards := cfg.Dashboards
	paths := make([]gcs.Path, 0, len(dashboards))
	for _, d := range dashboards {
		path, err := summaryPath(configPath, summaryPathPrefix, d.Name)
		if err != nil {
			return nil, fmt.Errorf("bad dashboard path: %s: %w", d.Name, err)
		}
		paths = append(paths, *path)
	}
	stats := gcs.Stat(ctx, client, 10, paths...)
	whens := make(map[string]time.Time, len(stats))
	var wg sync.WaitGroup
	for i, stat := range stats {
		name := dashboards[i].Name
		switch {
		case stat.Attrs != nil:
			whens[name] = stat.Attrs.Updated.Add(freq)
		default:
			if errors.Is(stat.Err, storage.ErrObjectNotExist) {
				wg.Add(1)
				defer func(i int) {
					defer wg.Done()
					if _, err := lockDashboard(ctx, client, paths[i], 0); err != nil && !gcs.IsPreconditionFailed(err) {
						log.WithError(err).WithField("path", paths[i]).Error("Failed to lock initial summary")
					}
				}(i)
			} else {
				log.WithError(err).WithField("path", paths[i]).Info("Failed to stat")
			}
			whens[name] = now
		}
	}
	namedDashboards := make(map[string]*configpb.Dashboard, len(cfg.Dashboards))
	for _, d := range cfg.Dashboards {
		namedDashboards[d.Name] = d
	}
	namedGroups := make(map[string]*configpb.TestGroup, len(cfg.TestGroups))
	for _, tg := range cfg.TestGroups {
		namedGroups[tg.Name] = tg
	}
	when := now.Add(freq)
	cs.lock.Lock()
	defer cs.lock.Unlock()
	cs.dashboards = namedDashboards
	cs.groups = namedGroups
	q.Init(cfg.Dashboards, when)
	if err := q.FixAll(whens, false); err != nil {
		log.WithError(err).Error("Failed to fix all dashboards based on last update time")
	}
	return attrs, nil
}
func lockDashboard(ctx context.Context, client gcs.ConditionalClient, path gcs.Path, generation int64) (*storage.ObjectAttrs, error) {
	var buf []byte
	if generation == 0 {
		var sum summarypb.DashboardSummary
		var err error
		buf, err = proto.Marshal(&sum)
		if err != nil {
			return nil, fmt.Errorf("marshal: %w", err)
		}
	}
	return gcs.Touch(ctx, client, path, generation, buf)
}
func Update(ctx context.Context, client gcs.ConditionalClient, mets *Metrics, configPath gcs.Path, concurrency int, dashboard, gridPathPrefix, summaryPathPrefix string, confirm bool, freq time.Duration, fix Fixer) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	if concurrency < 1 {
		return fmt.Errorf("concurrency must be positive, got: %d", concurrency)
	}
	log := logrus.WithField("config", configPath)
	var q config.DashboardQueue
	log.Debug("Fetching config...")
	var cfg configSnapshot
	cfgAttrs, err := cfg.update(ctx, log, client, configPath, summaryPathPrefix, dashboard, &q, freq)
	if err != nil {
		return fmt.Errorf("fetch config: %w", err)
	}
	log.Info("Fetched config")
	if fix != nil {
		go func() {
			fix(ctx, &q)
		}()
	}
	go func() {
		var cond storage.Conditions
		cond.GenerationNotMatch = cfgAttrs.Generation
		ticker := time.NewTicker(time.Minute)  
		for {
			depth, next, when := q.Status()
			log := log.WithField("depth", depth)
			if next != nil {
				log = log.WithField("next", *next)
			}
			delay := time.Since(when)
			if delay < 0 {
				delay = 0
				log = log.WithField("sleep", -delay)
			}
			log = log.WithField("delay", delay.Round(time.Second))
			log.Info("Updating groups")
			select {
			case <-ctx.Done():
				ticker.Stop()
				return
			case <-ticker.C:
				var err error
				cfgAttrs, err = cfg.update(ctx, log, client, configPath, summaryPathPrefix, dashboard, &q, freq)
				switch {
				case err == nil:
					log.Info("Configuration changed")
					cond.GenerationNotMatch = cfgAttrs.Generation
				case !gcs.IsPreconditionFailed(err):
					log.WithError(err).Error("Failed to update configuration")
				}
			}
		}
	}()
	dashboardNames := make(chan string)
	defer close(dashboardNames)
	groupFinder := func(name string) (*configpb.TestGroup, gridReader, error) {
		group := cfg.group(name)
		if group == nil {
			return nil, nil, nil
		}
		groupPath, err := configPath.ResolveReference(&url.URL{Path: path.Join(gridPathPrefix, name)})
		if err != nil {
			return group, nil, err
		}
		reader := func(ctx context.Context) (io.ReadCloser, time.Time, int64, error) {
			return pathReader(ctx, client, *groupPath)
		}
		return group, reader, nil
	}
	updateName := func(log *logrus.Entry, dashName string) error {
		ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
		defer cancel()
		dash := cfg.dashboard(dashName)
		log.Debug("Summarizing dashboard")
		summaryPath, err := summaryPath(configPath, summaryPathPrefix, dashName)
		if err != nil {
			return fmt.Errorf("summary path: %v", err)
		}
		sum, err := updateDashboard(ctx, dash, groupFinder)
		if err != nil {
			return fmt.Errorf("update: %w", err)
		}
		log = log.WithField("path", summaryPath)
		if !confirm {
			log.WithField("summary", sum).Info("Summarized")
			return nil
		}
		if err := writeSummary(ctx, client, *summaryPath, sum); err != nil {
			return fmt.Errorf("write: %w", err)
		}
		return nil
	}
	for i := 0; i < concurrency; i++ {
		go func() {
			for dashName := range dashboardNames {
				log := log.WithField("dashboard", dashName)
				if err := updateName(log, dashName); err != nil {
					mets.Error()
					log.WithError(err).Error("Failed to summarize dashboard")
				} else {
					mets.Success()
				}
			}
		}()
	}
	return q.Send(ctx, dashboardNames, freq)
}
var (
	normalizer = regexp.MustCompile(`[^a-z0-9]+`)
)
func summaryPath(g gcs.Path, prefix, dashboard string) (*gcs.Path, error) {
	name := "summary-" + normalizer.ReplaceAllString(strings.ToLower(dashboard), "")
	fullName := path.Join(prefix, name)
	u, err := url.Parse(fullName)
	if err != nil {
		return nil, fmt.Errorf("parse url: %w", err)
	}
	np, err := g.ResolveReference(u)
	if err != nil {
		return nil, fmt.Errorf("resolve reference: %w", err)
	}
	if np.Bucket() != g.Bucket() {
		return nil, fmt.Errorf("dashboard %s should not change bucket", fullName)
	}
	return np, nil
}
func writeSummary(ctx context.Context, client gcs.Client, path gcs.Path, sum *summarypb.DashboardSummary) error {
	buf, err := proto.Marshal(sum)
	if err != nil {
		return fmt.Errorf("marshal: %v", err)
	}
	_, err = client.Upload(ctx, path, buf, gcs.DefaultACL, "no-cache")  
	return err
}
func pathReader(ctx context.Context, client gcs.Client, path gcs.Path) (io.ReadCloser, time.Time, int64, error) {
	r, attrs, err := client.Open(ctx, path)
	if err != nil {
		return nil, time.Time{}, 0, fmt.Errorf("client.Open(): %w", err)
	}
	if attrs == nil {
		return r, time.Time{}, 0, nil
	}
	return r, attrs.LastModified, attrs.Generation, nil
}
func updateDashboard(ctx context.Context, dash *configpb.Dashboard, finder groupFinder) (*summarypb.DashboardSummary, error) {
	log := logrus.WithField("dashboard", dash.Name)
	var badTabs []string
	var sum summarypb.DashboardSummary
	for _, tab := range dash.DashboardTab {
		log := log.WithField("tab", tab.Name)
		log.Debug("Summarizing tab")
		s, err := updateTab(ctx, tab, finder)
		if err != nil {
			log.WithError(err).Error("Cannot summarize tab")
			badTabs = append(badTabs, tab.Name)
			sum.TabSummaries = append(sum.TabSummaries, problemTab(dash.Name, tab.Name))
			continue
		}
		s.DashboardName = dash.Name
		sum.TabSummaries = append(sum.TabSummaries, s)
	}
	var err error
	if d := len(badTabs); d > 0 {
		err = fmt.Errorf("Failed %d tabs: %s", d, strings.Join(badTabs, ", "))
	}
	return &sum, err
}
func problemTab(dashboardName, tabName string) *summarypb.DashboardTabSummary {
	return &summarypb.DashboardTabSummary{
		DashboardName:    dashboardName,
		DashboardTabName: tabName,
		Alert:            "failed to summarize tab",
	}
}
func staleHours(tab *configpb.DashboardTab) time.Duration {
	if tab.AlertOptions == nil {
		return 0
	}
	return time.Duration(tab.AlertOptions.AlertStaleResultsHours) * time.Hour
}
func updateTab(ctx context.Context, tab *configpb.DashboardTab, findGroup groupFinder) (*summarypb.DashboardTabSummary, error) {
	groupName := tab.TestGroupName
	group, groupReader, err := findGroup(groupName)
	if err != nil {
		return nil, fmt.Errorf("find group: %v", err)
	}
	if group == nil {
		return nil, fmt.Errorf("not found: %q", groupName)
	}
	grid, mod, _, err := readGrid(ctx, groupReader)  
	if err != nil && errors.Is(err, storage.ErrObjectNotExist) {
		return &summarypb.DashboardTabSummary{
			DashboardTabName: tab.Name,
			Alert:            noRuns,
			OverallStatus:    overallStatus(nil, 0, noRuns, false, nil),
			Status:           noRuns,
			LatestGreen:      noGreens,
		}, nil
	}
	if err != nil {
		return nil, fmt.Errorf("load %s: %v", groupName, err)
	}
	var healthiness *summarypb.HealthinessInfo
	if shouldRunHealthiness(tab) {
		interval := int(tab.HealthAnalysisOptions.DaysOfAnalysis)
		if interval <= 0 {
			interval = DefaultInterval
		}
		healthiness = getHealthinessForInterval(grid, tab.Name, time.Now(), interval)
	}
	recent := recentColumns(tab, group)
	grid.Rows, err = filterGrid(tab.BaseOptions, grid.Rows, recent)
	if err != nil {
		return nil, fmt.Errorf("filter: %v", err)
	}
	latest, latestSeconds := latestRun(grid.Columns)
	alert := staleAlert(mod, latest, staleHours(tab))
	failures := failingTestSummaries(grid.Rows)
	passingCols, completedCols, passingCells, filledCells, brokenState := gridMetrics(len(grid.Columns), grid.Rows, recent, tab.BrokenColumnThreshold)
	return &summarypb.DashboardTabSummary{
		DashboardTabName:     tab.Name,
		LastUpdateTimestamp:  float64(mod.Unix()),
		LastRunTimestamp:     float64(latestSeconds),
		Alert:                alert,
		FailingTestSummaries: failures,
		OverallStatus:        overallStatus(grid, recent, alert, brokenState, failures),
		Status:               statusMessage(passingCols, completedCols, passingCells, filledCells),
		LatestGreen:          latestGreen(grid, group.UseKubernetesClient),
		BugUrl:               tab.GetOpenBugTemplate().GetUrl(),
		Healthiness:          healthiness,
		LinkedIssues:         allLinkedIssues(grid.Rows),
	}, nil
}
func readGrid(ctx context.Context, reader gridReader) (*statepb.Grid, time.Time, int64, error) {
	var t time.Time
	r, mod, gen, err := reader(ctx)
	if err != nil {
		return nil, t, 0, fmt.Errorf("open: %w", err)
	}
	defer r.Close()
	zlibReader, err := zlib.NewReader(r)
	if err != nil {
		return nil, t, 0, fmt.Errorf("decompress: %v", err)
	}
	buf, err := ioutil.ReadAll(zlibReader)
	if err != nil {
		return nil, t, 0, fmt.Errorf("read: %v", err)
	}
	var g statepb.Grid
	if err = proto.Unmarshal(buf, &g); err != nil {
		return nil, t, 0, fmt.Errorf("parse: %v", err)
	}
	return &g, mod, gen, nil
}
func recentColumns(tab *configpb.DashboardTab, group *configpb.TestGroup) int {
	return firstFilled(tab.NumColumnsRecent, group.NumColumnsRecent, 5)
}
func firstFilled(values ...int32) int {
	for _, v := range values {
		if v != 0 {
			return int(v)
		}
	}
	return 0
}
const (
	includeFilter = "include-filter-by-regex"
	excludeFilter = "exclude-filter-by-regex"
)
func filterGrid(baseOptions string, rows []*statepb.Row, recent int) ([]*statepb.Row, error) {
	vals, err := url.ParseQuery(baseOptions)
	if err != nil {
		return nil, fmt.Errorf("parse %q: %v", baseOptions, err)
	}
	rows = recentRows(rows, recent)
	rows = filterMethods(rows)
	for _, include := range vals[includeFilter] {
		if rows, err = includeRows(rows, include); err != nil {
			return nil, fmt.Errorf("bad %s=%s: %v", includeFilter, include, err)
		}
	}
	for _, exclude := range vals[excludeFilter] {
		if rows, err = excludeRows(rows, exclude); err != nil {
			return nil, fmt.Errorf("bad %s=%s: %v", excludeFilter, exclude, err)
		}
	}
	return rows, nil
}
func recentRows(in []*statepb.Row, recent int) []*statepb.Row {
	var rows []*statepb.Row
	for _, r := range in {
		if r.Results == nil {
			continue
		}
		if statuspb.TestStatus(r.Results[0]) == statuspb.TestStatus_NO_RESULT && int(r.Results[1]) >= recent {
			continue
		}
		rows = append(rows, r)
	}
	return rows
}
func filterMethods(rows []*statepb.Row) []*statepb.Row {
	var filtered []*statepb.Row
	for _, r := range rows {
		if !isValidTestName(r.Id) || !isValidTestName(r.Name) {
			continue
		}
		filtered = append(filtered, r)
	}
	return filtered
}
func includeRows(in []*statepb.Row, include string) ([]*statepb.Row, error) {
	re, err := regexp.Compile(include)
	if err != nil {
		return nil, err
	}
	var rows []*statepb.Row
	for _, r := range in {
		if !re.MatchString(r.Name) {
			continue
		}
		rows = append(rows, r)
	}
	return rows, nil
}
func excludeRows(in []*statepb.Row, exclude string) ([]*statepb.Row, error) {
	re, err := regexp.Compile(exclude)
	if err != nil {
		return nil, err
	}
	var rows []*statepb.Row
	for _, r := range in {
		if re.MatchString(r.Name) {
			continue
		}
		rows = append(rows, r)
	}
	return rows, nil
}
func latestRun(columns []*statepb.Column) (time.Time, int64) {
	if len(columns) > 0 {
		if start := int64(columns[0].Started); start > 0 {
			second := start / 1000
			mills := start % 1000
			return time.Unix(second, mills*1e6), second
		}
	}
	return time.Time{}, 0
}
const noRuns = "no completed results"
func staleAlert(mod, ran time.Time, stale time.Duration) string {
	if mod.IsZero() {
		return "no stored results"
	}
	if stale == 0 {
		return ""
	}
	if ran.IsZero() {
		return noRuns
	}
	now := time.Now()
	if dur := now.Sub(mod); dur > stale {
		return fmt.Sprintf("data has not changed since %s (%s old)", mod, dur.Truncate(15*time.Minute))
	}
	if dur := now.Sub(ran); dur > stale {
		return fmt.Sprintf("latest column from %s (%s old)", ran, dur.Truncate(15*time.Minute))
	}
	return ""
}
func failingTestSummaries(rows []*statepb.Row) []*summarypb.FailingTestSummary {
	var failures []*summarypb.FailingTestSummary
	for _, row := range rows {
		if row.AlertInfo == nil {
			continue
		}
		alert := row.AlertInfo
		sum := summarypb.FailingTestSummary{
			DisplayName:       row.Name,
			TestName:          row.Id,
			FailBuildId:       alert.FailBuildId,
			LatestFailBuildId: alert.LatestFailBuildId,
			FailCount:         alert.FailCount,
			FailureMessage:    alert.FailureMessage,
			PassBuildId:       alert.PassBuildId,
			BuildLink:          alert.BuildLink,
			BuildLinkText:      alert.BuildLinkText,
			BuildUrlText:       alert.BuildUrlText,
			LinkedBugs:         row.Issues,
			FailTestLink:       buildFailLink(alert.FailTestId, row.Id),
			LatestFailTestLink: buildFailLink(alert.LatestFailTestId, row.Id),
			Properties:         alert.Properties,
			HotlistIds:         alert.HotlistIds,
			EmailAddresses:     alert.EmailAddresses,
		}
		if alert.PassTime != nil {
			sum.PassTimestamp = float64(alert.PassTime.Seconds)
		}
		if alert.FailTime != nil {
			sum.FailTimestamp = float64(alert.FailTime.Seconds)
		}
		failures = append(failures, &sum)
	}
	return failures
}
func buildFailLink(testID, target string) string {
	return fmt.Sprintf("%s %s", url.PathEscape(testID), url.PathEscape(target))
}
func overallStatus(grid *statepb.Grid, recent int, stale string, brokenState bool, alerts []*summarypb.FailingTestSummary) summarypb.DashboardTabSummary_TabStatus {
	if brokenState {
		return summarypb.DashboardTabSummary_BROKEN
	}
	if stale != "" {
		return summarypb.DashboardTabSummary_STALE
	}
	if len(alerts) > 0 {
		return summarypb.DashboardTabSummary_FAIL
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	results := results(ctx, grid.Rows)
	moreCols := true
	var found bool
	for moreCols && recent > 0 {
		moreCols = false
		var foundCol bool
		var running bool
		for _, resultCh := range results {
			r, ok := <-resultCh
			if !ok {
				continue
			}
			moreCols = true
			if r == statuspb.TestStatus_RUNNING {
				running = true
				continue
			}
			r = coalesceResult(r, result.IgnoreRunning)
			if r == statuspb.TestStatus_NO_RESULT {
				continue
			}
			if r != statuspb.TestStatus_PASS {
				return summarypb.DashboardTabSummary_FLAKY
			}
			foundCol = true
		}
		if running {
			continue
		}
		if foundCol {
			found = true
			recent--
		}
	}
	if found {
		return summarypb.DashboardTabSummary_PASS
	}
	return summarypb.DashboardTabSummary_UNKNOWN
}
func allLinkedIssues(rows []*statepb.Row) []string {
	issueSet := make(map[string]bool)
	for _, row := range rows {
		for _, issueID := range row.Issues {
			issueSet[issueID] = true
		}
	}
	linkedIssues := []string{}
	for issueID := range issueSet {
		linkedIssues = append(linkedIssues, issueID)
	}
	return linkedIssues
}
func gridMetrics(cols int, rows []*statepb.Row, recent int, brokenThreshold float32) (int, int, int, int, bool) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	results := results(ctx, rows)
	var passingCells int
	var filledCells int
	var passingCols int
	var completedCols int
	var brokenState bool
	for idx := 0; idx < cols; idx++ {
		if idx >= recent {
			break
		}
		var passes int
		var failures int
		var other int
		for _, ch := range results {
			status := coalesceResult(<-ch, result.IgnoreRunning)
			if result.Passing(status) {
				passes++
				passingCells++
				filledCells++
			} else if result.Failing(status) {
				failures++
				filledCells++
			} else if status != statuspb.TestStatus_NO_RESULT {
				other++
				filledCells++
			}
		}
		if passes+failures+other > 0 {
			completedCols++
		}
		if failures == 0 && passes > 0 {
			passingCols++
		}
		if passes+failures > 0 && brokenThreshold > 0 {
			if float32(failures)/float32(passes+failures+other) > brokenThreshold {
				brokenState = true
			}
		}
	}
	return passingCols, completedCols, passingCells, filledCells, brokenState
}
func fmtStatus(passCols, cols, passCells, cells int) string {
	colCent := 100 * float64(passCols) / float64(cols)
	cellCent := 100 * float64(passCells) / float64(cells)
	return fmt.Sprintf("%d of %d (%.1f%%) recent columns passed (%d of %d or %.1f%% cells)", passCols, cols, colCent, passCells, cells, cellCent)
}