SendTraces queues up traces to be sent to the metrics-generators
(ctx context.Context, tenantID string, keys []uint32, traces []*rebatchedTrace)
| 79 | |
| 80 | // SendTraces queues up traces to be sent to the metrics-generators |
| 81 | func (f *generatorForwarder) SendTraces(ctx context.Context, tenantID string, keys []uint32, traces []*rebatchedTrace) { |
| 82 | select { |
| 83 | case <-f.shutdown: |
| 84 | return |
| 85 | default: |
| 86 | } |
| 87 | |
| 88 | q := f.getOrCreateQueue(tenantID) |
| 89 | err := q.Push(ctx, &request{tenantID: tenantID, keys: keys, traces: traces, noGenerateMetrics: generator.ExtractNoGenerateMetrics(ctx)}) |
| 90 | if err != nil { |
| 91 | _ = level.Error(f.logger).Log("msg", "failed to push traces to queue", "tenant", tenantID, "err", err) |
| 92 | metricForwarderPushesFailures.WithLabelValues(tenantID).Inc() |
| 93 | } |
| 94 | |
| 95 | metricForwarderPushes.WithLabelValues(tenantID).Inc() |
| 96 | } |
| 97 | |
| 98 | // getQueueConfig returns queueSize and workerCount for the given tenant |
| 99 | func (f *generatorForwarder) getQueueConfig(tenantID string) (queueSize, workerCount int) { |