lxd/cluster/raft.go
func raftInstanceInit(
	db *db.Node, node *db.RaftNode, cert *shared.CertInfo, latency float64) (*raftInstance, error) {
	// FIXME: should be a parameter
	timeout := 5 * time.Second

	raftLogger := raftLogger()

	// Raft config.
	config := raftConfig(latency)
	config.Logger = raftLogger
	config.LocalID = raft.ServerID(strconv.Itoa(int(node.ID)))

	// Raft transport
	var handler *rafthttp.Handler
	var membershipChanger func(*raft.Raft)
	var layer *rafthttp.Layer
	var transport raft.Transport
	addr := node.Address
	if addr == "" {
		// This should normally be used only for testing as it can
		// cause split-brian, but since we are not exposing raft to the
		// network at all it's safe to do so. When this node gets
		// exposed to the network and assigned an address, we need to
		// restart raft anyways.
		config.StartAsLeader = true
		transport = raftMemoryTransport()
	} else {
		dial, err := raftDial(cert)
		if err != nil {
			return nil, err
		}

		transport, handler, layer, err = raftNetworkTransport(db, addr, log.New(&raftLogWriter{}, "", 0), timeout, dial)
		if err != nil {
			return nil, err
		}
		membershipChanger = func(raft *raft.Raft) {
			raftmembership.HandleChangeRequests(raft, handler.Requests())
		}
	}

	err := raft.ValidateConfig(config)
	if err != nil {
		return nil, errors.Wrap(err, "invalid raft configuration")
	}

	// Rename legacy data directory if needed.
	dir := filepath.Join(db.Dir(), "global")
	legacyDir := filepath.Join(db.Dir(), "..", "raft")
	if shared.PathExists(legacyDir) {
		if shared.PathExists(dir) {
			return nil, fmt.Errorf("both legacy and new global database directories exist")
		}
		logger.Info("Renaming global database directory from raft/ to database/global/")
		err := os.Rename(legacyDir, dir)
		if err != nil {
			return nil, errors.Wrap(err, "failed to rename legacy global database directory")
		}
	}

	// Data directory
	if !shared.PathExists(dir) {
		err := os.Mkdir(dir, 0750)
		if err != nil {
			return nil, err
		}
	}

	// Raft logs store
	logs, err := raftboltdb.New(raftboltdb.Options{
		Path:        filepath.Join(dir, "logs.db"),
		BoltOptions: &bolt.Options{Timeout: timeout},
	})
	if err != nil {
		return nil, errors.Wrap(err, "failed to create bolt store for raft logs")
	}

	// Raft snapshot store (don't log snapshots since we take them frequently)
	snaps, err := raft.NewFileSnapshotStoreWithLogger(dir, 2, log.New(ioutil.Discard, "", 0))
	if err != nil {
		return nil, errors.Wrap(err, "failed to create file snapshot store")
	}

	// If we are the initial node, we use the last index persisted in the
	// logs store and other checks to determine if we have ever
	// bootstrapped the cluster, and if not we do so (see raft.HasExistingState).
	if node.ID == 1 {
		err := raftMaybeBootstrap(config, logs, snaps, transport)
		if err != nil {
			return nil, errors.Wrap(err, "failed to boostrap cluster")
		}
	}

	// The dqlite registry and FSM.
	registry := dqlite.NewRegistry(strconv.Itoa(serial))
	serial++
	fsm := dqlite.NewFSM(registry)

	// The actual raft instance.
	raft, err := raft.NewRaft(config, fsm, logs, logs, snaps, transport)
	if err != nil {
		logs.Close()
		return nil, errors.Wrap(err, "failed to start raft")
	}

	if membershipChanger != nil {
		// Process Raft connections over HTTP. This goroutine will
		// terminate when instance.handler.Close() is called, which
		// happens indirectly when the raft instance is shutdown in
		// instance.Shutdown(), and the associated transport is closed.
		go membershipChanger(raft)
	}

	instance := &raftInstance{
		layer:             layer,
		handler:           raftHandler(cert, handler),
		membershipChanger: membershipChanger,
		logs:              logs,
		registry:          registry,
		fsm:               fsm,
		raft:              raft,
	}

	return instance, nil
}
func raftInstanceInit(
	db *db.Node, node *db.RaftNode, cert *shared.CertInfo, latency float64) (*raftInstance, error) {
	// FIXME: should be a parameter
	timeout := 5 * time.Second

	raftLogger := raftLogger()

	// Raft config.
	config := raftConfig(latency)
	config.Logger = raftLogger
	config.LocalID = raft.ServerID(strconv.Itoa(int(node.ID)))

	// Raft transport
	var handler *rafthttp.Handler
	var membershipChanger func(*raft.Raft)
	var layer *rafthttp.Layer
	var transport raft.Transport
	addr := node.Address
	if addr == "" {
		// This should normally be used only for testing as it can
		// cause split-brian, but since we are not exposing raft to the
		// network at all it's safe to do so. When this node gets
		// exposed to the network and assigned an address, we need to
		// restart raft anyways.
		config.StartAsLeader = true
		transport = raftMemoryTransport()
	} else {
		dial, err := raftDial(cert)
		if err != nil {
			return nil, err
		}

		transport, handler, layer, err = raftNetworkTransport(db, addr, log.New(&raftLogWriter{}, "", 0), timeout, dial)
		if err != nil {
			return nil, err
		}
		membershipChanger = func(raft *raft.Raft) {
			raftmembership.HandleChangeRequests(raft, handler.Requests())
		}
	}

	err := raft.ValidateConfig(config)
	if err != nil {
		return nil, errors.Wrap(err, "invalid raft configuration")
	}

	// Rename legacy data directory if needed.
	dir := filepath.Join(db.Dir(), "global")
	legacyDir := filepath.Join(db.Dir(), "..", "raft")
	if shared.PathExists(legacyDir) {
		if shared.PathExists(dir) {
			return nil, fmt.Errorf("both legacy and new global database directories exist")
		}
		logger.Info("Renaming global database directory from raft/ to database/global/")
		err := os.Rename(legacyDir, dir)
		if err != nil {
			return nil, errors.Wrap(err, "failed to rename legacy global database directory")
		}
	}

	// Data directory
	if !shared.PathExists(dir) {
		err := os.Mkdir(dir, 0750)
		if err != nil {
			return nil, err
		}
	}

	// Raft logs store
	logs, err := raftboltdb.New(raftboltdb.Options{
		Path:        filepath.Join(dir, "logs.db"),
		BoltOptions: &bolt.Options{Timeout: timeout},
	})
	if err != nil {
		return nil, errors.Wrap(err, "failed to create bolt store for raft logs")
	}

	// Raft snapshot store (don't log snapshots since we take them frequently)
	snaps, err := raft.NewFileSnapshotStoreWithLogger(dir, 2, log.New(ioutil.Discard, "", 0))
	if err != nil {
		return nil, errors.Wrap(err, "failed to create file snapshot store")
	}

	// If we are the initial node, we use the last index persisted in the
	// logs store and other checks to determine if we have ever
	// bootstrapped the cluster, and if not we do so (see raft.HasExistingState).
	if node.ID == 1 {
		err := raftMaybeBootstrap(config, logs, snaps, transport)
		if err != nil {
			return nil, errors.Wrap(err, "failed to boostrap cluster")
		}
	}

	// The dqlite registry and FSM.
	registry := dqlite.NewRegistry(strconv.Itoa(serial))
	serial++
	fsm := dqlite.NewFSM(registry)

	// The actual raft instance.
	raft, err := raft.NewRaft(config, fsm, logs, logs, snaps, transport)
	if err != nil {
		logs.Close()
		return nil, errors.Wrap(err, "failed to start raft")
	}

	if membershipChanger != nil {
		// Process Raft connections over HTTP. This goroutine will
		// terminate when instance.handler.Close() is called, which
		// happens indirectly when the raft instance is shutdown in
		// instance.Shutdown(), and the associated transport is closed.
		go membershipChanger(raft)
	}

	instance := &raftInstance{
		layer:             layer,
		handler:           raftHandler(cert, handler),
		membershipChanger: membershipChanger,
		logs:              logs,
		registry:          registry,
		fsm:               fsm,
		raft:              raft,
	}

	return instance, nil
}
func raftInstanceInit ( db * db . Node , node * db . RaftNode , cert * shared . CertInfo , latency float64 ) ( * raftInstance , error ) { // FIXME: should be a parameter timeout := 5 * time . Second raftLogger := raftLogger ( ) // Raft config. config := raftConfig ( latency ) config . Logger = raftLogger config . LocalID = raft . ServerID ( strconv . Itoa ( int ( node . ID ) ) ) // Raft transport var handler * rafthttp . Handler var membershipChanger func ( * raft . Raft ) var layer * rafthttp . Layer var transport raft . Transport addr := node . Address if addr == " " { // This should normally be used only for testing as it can // cause split-brian, but since we are not exposing raft to the // network at all it's safe to do so. When this node gets // exposed to the network and assigned an address, we need to // restart raft anyways. config . StartAsLeader = true transport = raftMemoryTransport ( ) } else { dial , err := raftDial ( cert ) if err != nil { return nil , err } transport , handler , layer , err = raftNetworkTransport ( db , addr , log . New ( & raftLogWriter { } , " " , 0 ) , timeout , dial ) if err != nil { return nil , err } membershipChanger = func ( raft * raft . Raft ) { raftmembership . HandleChangeRequests ( raft , handler . Requests ( ) ) } } err := raft . ValidateConfig ( config ) if err != nil { return nil , errors . Wrap ( err , " " ) } // Rename legacy data directory if needed. dir := filepath . Join ( db . Dir ( ) , " " ) legacyDir := filepath . Join ( db . Dir ( ) , " " , " " ) if shared . PathExists ( legacyDir ) { if shared . PathExists ( dir ) { return nil , fmt . Errorf ( " " ) } logger . Info ( " " ) err := os . Rename ( legacyDir , dir ) if err != nil { return nil , errors . Wrap ( err , " " ) } } // Data directory if ! shared . PathExists ( dir ) { err := os . Mkdir ( dir , 0750 ) if err != nil { return nil , err } } // Raft logs store logs , err := raftboltdb . New ( raftboltdb . Options { Path : filepath . Join ( dir , " " ) , BoltOptions : & bolt . Options { Timeout : timeout } , } ) if err != nil { return nil , errors . Wrap ( err , " " ) } // Raft snapshot store (don't log snapshots since we take them frequently) snaps , err := raft . NewFileSnapshotStoreWithLogger ( dir , 2 , log . New ( ioutil . Discard , " " , 0 ) ) if err != nil { return nil , errors . Wrap ( err , " " ) } // If we are the initial node, we use the last index persisted in the // logs store and other checks to determine if we have ever // bootstrapped the cluster, and if not we do so (see raft.HasExistingState). if node . ID == 1 { err := raftMaybeBootstrap ( config , logs , snaps , transport ) if err != nil { return nil , errors . Wrap ( err , " " ) } } // The dqlite registry and FSM. registry := dqlite . NewRegistry ( strconv . Itoa ( serial ) ) serial ++ fsm := dqlite . NewFSM ( registry ) // The actual raft instance. raft , err := raft . NewRaft ( config , fsm , logs , logs , snaps , transport ) if err != nil { logs . Close ( ) return nil , errors . Wrap ( err , " " ) } if membershipChanger != nil { // Process Raft connections over HTTP. This goroutine will // terminate when instance.handler.Close() is called, which // happens indirectly when the raft instance is shutdown in // instance.Shutdown(), and the associated transport is closed. go membershipChanger ( raft ) } instance := & raftInstance { layer : layer , handler : raftHandler ( cert , handler ) , membershipChanger : membershipChanger , logs : logs , registry : registry , fsm : fsm , raft : raft , } return instance , nil }
// Create a new raftFactory, instantiating all needed raft dependencies.
Create a new raftFactory instantiating all needed raft dependencies .