Function New

modules/generator/processor/servicegraphs/servicegraphs.go:107–160

(cfg Config, tenant string, reg registry.Registry, logger log.Logger, filteredSpansCounter, invalidUTF8Counter prometheus.Counter)

}

func New(cfg Config, tenant string, reg registry.Registry, logger log.Logger, filteredSpansCounter, invalidUTF8Counter prometheus.Counter) (gen.Processor, error) {
	if cfg.EnableVirtualNodeLabel {
		cfg.Dimensions = append(cfg.Dimensions, virtualNodeLabel)
	}

	sanitizeCache := reclaimable.New(validation.SanitizeLabelName, 10000)
	filter, err := spanfilter.NewSpanFilter(cfg.FilterPolicies)
	if err != nil {
		return nil, err
	}

	p := &Processor{
		Cfg:      cfg,
		registry: reg,
		closeCh:  make(chan struct{}, 1),

		serviceGraphRequestTotal:                           reg.NewCounter(metricRequestTotal),
		serviceGraphRequestFailedTotal:                     reg.NewCounter(metricRequestFailedTotal),
		serviceGraphRequestServerSecondsHistogram:          reg.NewHistogram(metricRequestServerSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
		serviceGraphRequestClientSecondsHistogram:          reg.NewHistogram(metricRequestClientSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
		serviceGraphRequestMessagingSystemSecondsHistogram: reg.NewHistogram(metricRequestMessagingSystemSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
		sanitizeCache: sanitizeCache,
		filter:        filter,

		filteredSpansCounter:                filteredSpansCounter,
		metricDroppedSpans:                  metricDroppedSpans.WithLabelValues(tenant),
		metricDroppedEdges:                  metricDroppedEdges.WithLabelValues(tenant),
		metricDroppedSpanSideCacheOverflows: metricDroppedSpanSideCacheOverflows.WithLabelValues(tenant),
		metricTotalEdges:                    metricTotalEdges.WithLabelValues(tenant),
		metricExpiredEdges:                  metricExpiredEdges.WithLabelValues(tenant),
		invalidUTF8Counter:                  invalidUTF8Counter,
		logger:                              log.With(logger, "component", "service-graphs"),
	}

	p.store = store.NewStore(cfg.Wait, cfg.MaxItems, p.onComplete, p.onExpire, p.metricDroppedSpanSideCacheOverflows)

	expirationTicker := time.NewTicker(2 * time.Second)
	for i := 0; i < cfg.Workers; i++ {
		go func() {
			for {
				select {
				// Periodically clean expired edges from the store
				case <-expirationTicker.C:
					p.store.Expire()

				case <-p.closeCh:
					return
				}
			}
		}()
	}

	return p, nil
}

func (p *Processor) Name() string {
	return gen.ServiceGraphsName