| 279 | } |
| 280 | |
| 281 | func (ma *MetricsAggregator) Run(ctx context.Context) func() { |
| 282 | ctx, cancelFunc := context.WithCancel(ctx) |
| 283 | done := make(chan struct{}) |
| 284 | |
| 285 | cleanupTicker := time.NewTicker(ma.metricsCleanupInterval) |
| 286 | pproflabel.Go(ctx, pproflabel.Service(pproflabel.ServiceAgentMetricAggregator), func(ctx context.Context) { |
| 287 | defer close(done) |
| 288 | defer cleanupTicker.Stop() |
| 289 | |
| 290 | for { |
| 291 | select { |
| 292 | case req := <-ma.updateCh: |
| 293 | ma.log.Debug(ctx, "update metrics") |
| 294 | |
| 295 | timer := prometheus.NewTimer(ma.updateHistogram) |
| 296 | for _, m := range req.metrics { |
| 297 | key := hashKey(&req, m) |
| 298 | |
| 299 | if val, ok := ma.store[key]; ok { |
| 300 | val.Stats_Metric.Value = m.Value |
| 301 | val.expiryDate = req.timestamp.Add(ma.metricsCleanupInterval) |
| 302 | ma.store[key] = val |
| 303 | } else { |
| 304 | ma.store[key] = annotatedMetric{ |
| 305 | Stats_Metric: m, |
| 306 | username: req.username, |
| 307 | workspaceName: req.workspaceName, |
| 308 | agentName: req.agentName, |
| 309 | templateName: req.templateName, |
| 310 | expiryDate: req.timestamp.Add(ma.metricsCleanupInterval), |
| 311 | } |
| 312 | } |
| 313 | } |
| 314 | timer.ObserveDuration() |
| 315 | |
| 316 | ma.storeSizeGauge.Set(float64(len(ma.store))) |
| 317 | case outputCh := <-ma.collectCh: |
| 318 | ma.log.Debug(ctx, "collect metrics") |
| 319 | |
| 320 | var input []annotatedMetric |
| 321 | output := make([]prometheus.Metric, 0, len(ma.store)) |
| 322 | |
| 323 | if len(ma.aggregateByLabels) == 0 { |
| 324 | ma.aggregateByLabels = agentmetrics.LabelAll |
| 325 | } |
| 326 | |
| 327 | // If custom aggregation labels have not been chosen, generate Prometheus metrics without any pre-aggregation. |
| 328 | // This results in higher cardinality, but may be desirable in larger deployments. |
| 329 | // |
| 330 | // Default behavior. |
| 331 | if len(ma.aggregateByLabels) == len(agentmetrics.LabelAll) { |
| 332 | for _, m := range ma.store { |
| 333 | // Aggregate by all available metrics. |
| 334 | m.aggregateByLabels = defaultAgentMetricsLabels |
| 335 | input = append(input, m) |
| 336 | } |
| 337 | } else { |
| 338 | // However, if custom aggregations have been chosen, we need to aggregate the values from the annotated |