rollup performs stats rollup for sessions that fall within the configured rollup window. For sessions longer than the window, we report them individually.
(now time.Time)
| 160 | // configured rollup window. For sessions longer than the window, |
| 161 | // we report them individually. |
| 162 | func (sc *StatsCollector) rollup(now time.Time) []StatsReport { |
| 163 | sc.mu.Lock() |
| 164 | defer sc.mu.Unlock() |
| 165 | |
| 166 | var report []StatsReport |
| 167 | |
| 168 | for g, group := range sc.groupedStats { |
| 169 | if len(group) == 0 { |
| 170 | // Safety check, this should not happen. |
| 171 | sc.opts.Logger.Error(sc.ctx, "empty stats group", slog.F("group", g)) |
| 172 | delete(sc.groupedStats, g) |
| 173 | continue |
| 174 | } |
| 175 | |
| 176 | var rolledUp *StatsReport |
| 177 | if group[0].rolledUp { |
| 178 | rolledUp = group[0] |
| 179 | group = group[1:] |
| 180 | } else { |
| 181 | rolledUp = &StatsReport{ |
| 182 | UserID: g.UserID, |
| 183 | WorkspaceID: g.WorkspaceID, |
| 184 | AgentID: g.AgentID, |
| 185 | AccessMethod: g.AccessMethod, |
| 186 | SlugOrPort: g.SlugOrPort, |
| 187 | SessionStartedAt: g.StartTimeTrunc, |
| 188 | SessionEndedAt: g.StartTimeTrunc.Add(sc.opts.RollupWindow), |
| 189 | Requests: 0, |
| 190 | rolledUp: true, |
| 191 | } |
| 192 | } |
| 193 | rollupChanged := false |
| 194 | newGroup := []*StatsReport{rolledUp} // Must be first in slice for future iterations (see group[0] above). |
| 195 | for _, stat := range group { |
| 196 | if !stat.SessionEndedAt.IsZero() && stat.SessionEndedAt.Sub(stat.SessionStartedAt) <= sc.opts.RollupWindow { |
| 197 | // This is a short-lived session, roll it up. |
| 198 | if rolledUp.SessionID == uuid.Nil { |
| 199 | rolledUp.SessionID = stat.SessionID // Borrow the first session ID, useful in tests. |
| 200 | } |
| 201 | rolledUp.Requests += stat.Requests |
| 202 | rollupChanged = true |
| 203 | continue |
| 204 | } |
| 205 | if stat.SessionEndedAt.IsZero() && now.Sub(stat.SessionStartedAt) <= sc.opts.RollupWindow { |
| 206 | // This is an incomplete session, wait and see if it'll be rolled up or not. |
| 207 | newGroup = append(newGroup, stat) |
| 208 | continue |
| 209 | } |
| 210 | |
| 211 | // This is a long-lived session, report it individually. |
| 212 | // Make a copy of stat for reporting. |
| 213 | r := *stat |
| 214 | if r.SessionEndedAt.IsZero() { |
| 215 | // Report an end time for incomplete sessions, it will |
| 216 | // be updated later. This ensures that data in the DB |
| 217 | // will have an end time even if the service is stopped. |
| 218 | r.SessionEndedAt = now.UTC() // Use UTC like dbtime.Now(). |
| 219 | } |