Function NewLifecycler

ring/lifecycler.go:190–250

NewLifecycler creates new Lifecycler. It must be started via StartAsync.

(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringName, ringKey string, flushOnShutdown bool, logger log.Logger, reg prometheus.Registerer)

// NewLifecycler creates new Lifecycler. It must be started via StartAsync.
func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringName, ringKey string, flushOnShutdown bool, logger log.Logger, reg prometheus.Registerer) (*Lifecycler, error) {
	addr, err := GetInstanceAddr(cfg.Addr, cfg.InfNames, logger, cfg.EnableInet6)
	if err != nil {
		return nil, err
	}
	port := GetInstancePort(cfg.Port, cfg.ListenPort)
	codec := GetCodec()
	// Suffix all client names with "-lifecycler" to denote this kv client is used by the lifecycler
	store, err := kv.NewClient(
		cfg.RingConfig.KVStore,
		codec,
		kv.RegistererWithKVName(reg, ringName+"-lifecycler"),
		logger,
	)
	if err != nil {
		return nil, err
	}

	// We do allow a nil FlushTransferer, but to keep the ring logic easier we assume
	// it's always set, so we use a noop FlushTransferer
	if flushTransferer == nil {
		flushTransferer = NewNoopFlushTransferer()
	}

	tokenGenerator := cfg.RingTokenGenerator
	if tokenGenerator == nil {
		tokenGenerator = NewRandomTokenGenerator()
	}

	// We validate cfg before we create a Lifecycler.
	err = cfg.Validate()
	if err != nil {
		return nil, err
	}

	l := &Lifecycler{
		cfg:                   cfg,
		flushTransferer:       flushTransferer,
		KVStore:               store,
		Addr:                  net.JoinHostPort(addr, strconv.Itoa(port)),
		ID:                    cfg.ID,
		RingName:              ringName,
		RingKey:               ringKey,
		flushOnShutdown:       atomic.NewBool(flushOnShutdown),
		unregisterOnShutdown:  atomic.NewBool(cfg.UnregisterOnShutdown),
		clearTokensOnShutdown: atomic.NewBool(false),
		Zone:                  cfg.Zone,
		actorChan:             make(chan func()),
		state:                 PENDING,
		tokenGenerator:        tokenGenerator,
		canJoinTimeout:        5 * time.Minute,
		lifecyclerMetrics:     NewLifecyclerMetrics(ringName, reg),
		logger:                logger,
	}

	l.BasicService = services.
		NewBasicService(nil, l.loop, l.stopping).
		WithName(fmt.Sprintf("%s ring lifecycler", ringName))