870
}
871
872
>
var totalLag int64
db.go
873
>
var oldestTime time.Time
874
>
counts := make(map[int32]int64)
875
>
for _, s := range db.subqueues {
876
>
counts[s.Key.Priority] += s.ApproximateBacklogCount
877
>
oldestTime = minNonZeroTime(oldestTime, s.oldestTime)
878
>
// note: this metric is only an estimation for the lag.
879
>
// taskID in DB may not be continuous, especially when task list ownership changes.
880
>
if s.FairMaxReadLevel != nil && s.FairAckLevel != nil {
881
// TODO(fairness): this is not a good estimate of anything, we should probably just
882
// get rid of this metric.
883
totalLag += s.FairMaxReadLevel.TaskId - s.FairAckLevel.TaskId
885
totalLag += s.maxReadLevel - s.AckLevel
886
}
887
}
888
889
>
backlogCountGauge := metrics.ApproximateBacklogCount
db.go
890
>
backlogAgeGauge := metrics.ApproximateBacklogAgeSeconds
891
>
if attributionEnabled {
892
>
backlogCountGauge = metrics.PhysicalApproximateBacklogCount
893
>
backlogAgeGauge = metrics.PhysicalApproximateBacklogAgeSeconds
894
>
}
895
896
>
for priority, count := range counts {
db.go
897
>
backlogCountGauge.With(db.metricsHandler).Record(float64(count), metrics.MatchingTaskPriorityTag(priority))
898
>
}
899
>
if oldestTime.IsZero() {
900
backlogAgeGauge.With(db.metricsHandler).Record(0)
902
backlogAgeGauge.With(db.metricsHandler).Record(time.Since(oldestTime).Seconds())
903
}
904
>
metrics.TaskLagPerTaskQueueGauge.With(db.metricsHandler).Record(float64(totalLag))
db.go
905
}
906