fix: accumulate agentstats until reported and fix insights DAU offset

mafredri · mafredri · commit a1757f0c6f49 · 2024-12-11T17:31:29.000+02:00
Fixes #15824
diff --git a/agent/stats.go b/agent/stats.go
@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"maps"
 	"sync"
 	"time"
 
@@ -32,7 +33,7 @@ type statsDest interface {
 // statsDest (agent API in prod)
 type statsReporter struct {
 	*sync.Cond
-	networkStats *map[netlogtype.Connection]netlogtype.Counts
+	networkStats map[netlogtype.Connection]netlogtype.Counts
 	unreported   bool
 	lastInterval time.Duration
 
@@ -54,8 +55,18 @@ func (s *statsReporter) callback(_, _ time.Time, virtual, _ map[netlogtype.Conne
 	s.L.Lock()
 	defer s.L.Unlock()
 	s.logger.Debug(context.Background(), "got stats callback")
-	s.networkStats = &virtual
-	s.unreported = true
+	// Accumulate stats have until they've been reported.
+	if s.unreported {
+		if s.networkStats == nil && virtual != nil {
+			s.networkStats = make(map[netlogtype.Connection]netlogtype.Counts)
+		}
+		for k, v := range virtual {
+			s.networkStats[k] = s.networkStats[k].Add(v)
+		}
+	} else {
+		s.networkStats = maps.Clone(virtual)
+		s.unreported = true
+	}
 	s.Broadcast()
 }
 
@@ -96,9 +107,8 @@ func (s *statsReporter) reportLoop(ctx context.Context, dest statsDest) error {
 		if ctxDone {
 			return nil
 		}
-		networkStats := *s.networkStats
 		s.unreported = false
-		if err = s.reportLocked(ctx, dest, networkStats); err != nil {
+		if err = s.reportLocked(ctx, dest, s.networkStats); err != nil {
 			return xerrors.Errorf("report stats: %w", err)
 		}
 	}
diff --git a/coderd/insights.go b/coderd/insights.go
@@ -89,7 +89,7 @@ func (api *API) returnDAUsInternal(rw http.ResponseWriter, r *http.Request, temp
 	}
 	for _, row := range rows {
 		resp.Entries = append(resp.Entries, codersdk.DAUEntry{
-			Date:   row.StartTime.Format(time.DateOnly),
+			Date:   row.StartTime.In(loc).Format(time.DateOnly),
 			Amount: int(row.ActiveUsers),
 		})
 	}
diff --git a/coderd/insights_test.go b/coderd/insights_test.go
@@ -48,16 +48,26 @@ func TestDeploymentInsights(t *testing.T) {
 	db, ps := dbtestutil.NewDB(t, dbtestutil.WithDumpOnFailure())
 	logger := testutil.Logger(t)
 	rollupEvents := make(chan dbrollup.Event)
+	statsInterval := 500 * time.Millisecond
+	batcher, closeBatcher, err := workspacestats.NewBatcher(context.Background(),
+		workspacestats.BatcherWithLogger(logger.Named("batcher").Leveled(slog.LevelDebug)),
+		workspacestats.BatcherWithStore(db),
+		workspacestats.BatcherWithBatchSize(1),
+		workspacestats.BatcherWithInterval(statsInterval),
+	)
+	require.NoError(t, err)
+	defer closeBatcher()
 	client := coderdtest.New(t, &coderdtest.Options{
 		Database:                  db,
 		Pubsub:                    ps,
 		Logger:                    &logger,
 		IncludeProvisionerDaemon:  true,
-		AgentStatsRefreshInterval: time.Millisecond * 100,
+		AgentStatsRefreshInterval: statsInterval,
+		StatsBatcher:              batcher,
 		DatabaseRolluper: dbrollup.New(
 			logger.Named("dbrollup").Leveled(slog.LevelDebug),
 			db,
-			dbrollup.WithInterval(time.Millisecond*100),
+			dbrollup.WithInterval(statsInterval/2),
 			dbrollup.WithEventChannel(rollupEvents),
 		),
 	})
@@ -76,7 +86,7 @@ func TestDeploymentInsights(t *testing.T) {
 	workspace := coderdtest.CreateWorkspace(t, client, template.ID)
 	coderdtest.AwaitWorkspaceBuildJobCompleted(t, client, workspace.LatestBuild.ID)
 
-	ctx := testutil.Context(t, testutil.WaitLong)
+	ctx := testutil.Context(t, testutil.WaitSuperLong)
 
 	// Pre-check, no  permission issues.
 	daus, err := client.DeploymentDAUs(ctx, codersdk.TimezoneOffsetHour(clientTz))
@@ -108,6 +118,13 @@ func TestDeploymentInsights(t *testing.T) {
 	err = sess.Start("cat")
 	require.NoError(t, err)
 
+	select {
+	case <-ctx.Done():
+		require.Fail(t, "timed out waiting for initial rollup event")
+	case ev := <-rollupEvents:
+		require.True(t, ev.Init, "want init event")
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -120,6 +137,7 @@ func TestDeploymentInsights(t *testing.T) {
 		if len(daus.Entries) > 0 && daus.Entries[len(daus.Entries)-1].Amount > 0 {
 			break
 		}
+		t.Logf("waiting for deployment daus to update: %+v", daus)
 	}
 }
 

Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ func (api API) returnDAUsInternal(rw http.ResponseWriter, r http.Request, temp`
`89`	`89`	`}`
`90`	`90`	`for _, row := range rows {`
`91`	`91`	`resp.Entries = append(resp.Entries, codersdk.DAUEntry{`
`92`		`- Date: row.StartTime.Format(time.DateOnly),`
	`92`	`+ Date: row.StartTime.In(loc).Format(time.DateOnly),`
`93`	`93`	`Amount: int(row.ActiveUsers),`
`94`	`94`	`})`
`95`	`95`	`}`