Method initDistributor

cmd/tempo/app/modules.go:245–292

()

}

func (t *App) initDistributor() (services.Service, error) {
	singleBinary := IsSingleBinary(t.cfg.Target)
	localPushTargets := distributor.LocalPushTargets{}
	var partitionRing ring.PartitionRingReader

	t.cfg.Distributor.PushSpansToKafka = !singleBinary

	if singleBinary {
		localPushTargets.Generator = func(ctx context.Context, req *tempopb.PushSpansRequest) (*tempopb.PushResponse, error) {
			if t.generator == nil {
				return nil, errors.New("metrics-generator not initialized")
			}
			return t.generator.PushSpans(ctx, req)
		}

		localPushTargets.LiveStore = func(ctx context.Context, req *tempopb.PushBytesRequest) (*tempopb.PushResponse, error) {
			if t.liveStore == nil {
				return nil, errors.New("live-store not initialized")
			}
			return t.liveStore.PushBytes(ctx, req)
		}
	} else {
		t.cfg.Distributor.KafkaConfig = t.cfg.Ingest.Kafka
		partitionRing = t.partitionRing
	}

	// todo: make write-path client a module instead of passing the config everywhere
	distributor, err := distributor.New(t.cfg.Distributor,
		localPushTargets,
		partitionRing,
		t.Overrides,
		t.TracesConsumerMiddleware,
		log.Logger, t.cfg.Server.LogLevel, prometheus.DefaultRegisterer)
	if err != nil {
		return nil, fmt.Errorf("failed to create distributor: %w", err)
	}
	t.distributor = distributor

	if distributor.DistributorRing != nil {
		t.Server.HTTPRouter().Handle("/distributor/ring", distributor.DistributorRing)
	}

	if usageHandler := distributor.UsageTrackerHandler(); usageHandler != nil {
		t.Server.HTTPRouter().Handle("/usage_metrics", usageHandler)
	}

	return t.distributor, nil
}

func (t *App) initGenerator() (services.Service, error) {
	t.configureGenerator()