(ctx context.Context, u *userTSDB, metrics *ingesterMetrics)
| 245 | } |
| 246 | |
| 247 | func (m *labelSetCounter) UpdateMetric(ctx context.Context, u *userTSDB, metrics *ingesterMetrics) error { |
| 248 | currentLbsLimitHash := map[uint64]validation.LimitsPerLabelSet{} |
| 249 | limits := m.limiter.limits.LimitsPerLabelSet(u.userID) |
| 250 | for _, l := range limits { |
| 251 | currentLbsLimitHash[l.Hash] = l |
| 252 | } |
| 253 | |
| 254 | nonDefaultPartitionChanged := false |
| 255 | for i := range numMetricCounterShards { |
| 256 | s := m.shards[i] |
| 257 | s.RLock() |
| 258 | for h, entry := range s.valuesCounter { |
| 259 | lbls := entry.labels.String() |
| 260 | // This limit no longer exists |
| 261 | if _, ok := currentLbsLimitHash[h]; !ok { |
| 262 | metrics.usagePerLabelSet.DeleteLabelValues(u.userID, "max_series", lbls) |
| 263 | metrics.limitsPerLabelSet.DeleteLabelValues(u.userID, "max_series", lbls) |
| 264 | if entry.labels.Len() > 0 { |
| 265 | nonDefaultPartitionChanged = true |
| 266 | } |
| 267 | continue |
| 268 | } |
| 269 | // Delay exposing default partition metrics from current label limits as if |
| 270 | // another label set is added or removed then we need to backfill default partition again. |
| 271 | if entry.labels.Len() > 0 { |
| 272 | metrics.usagePerLabelSet.WithLabelValues(u.userID, "max_series", lbls).Set(float64(entry.count)) |
| 273 | metrics.limitsPerLabelSet.WithLabelValues(u.userID, "max_series", lbls).Set(float64(currentLbsLimitHash[h].Limits.MaxSeries)) |
| 274 | delete(currentLbsLimitHash, h) |
| 275 | } |
| 276 | } |
| 277 | s.RUnlock() |
| 278 | } |
| 279 | |
| 280 | // Check if we need to backfill default partition. We don't need to backfill when any condition meet: |
| 281 | // 1. Default partition doesn't exist. |
| 282 | // 2. No new partition added and no old partition removed. |
| 283 | if !nonDefaultPartitionChanged { |
| 284 | for _, l := range currentLbsLimitHash { |
| 285 | if l.LabelSet.Len() > 0 { |
| 286 | nonDefaultPartitionChanged = true |
| 287 | break |
| 288 | } |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | // Backfill all limits that are not being tracked yet |
| 293 | for _, l := range currentLbsLimitHash { |
| 294 | s := m.shards[util.HashFP(model.Fingerprint(l.Hash))%numMetricCounterShards] |
| 295 | force := false |
| 296 | // Force backfill to make sure we update the counter for the default partition |
| 297 | // when other limits got added or removed. If no partition is changed then we |
| 298 | // can use the value in the counter. |
| 299 | if l.LabelSet.Len() == 0 && nonDefaultPartitionChanged { |
| 300 | force = true |
| 301 | } |
| 302 | count, err := m.backFillLimit(ctx, u, force, limits, l, s) |
| 303 | if err != nil { |
| 304 | return err |
no test coverage detected