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
}
updateTab reads the latest grid state for the tab and summarizes it.
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
}