lxd/db/db.go
func OpenCluster(name string, store dqlite.ServerStore, address, dir string, timeout time.Duration, options ...dqlite.DriverOption) (*Cluster, error) {
	db, err := cluster.Open(name, store, options...)
	if err != nil {
		return nil, errors.Wrap(err, "failed to open database")
	}

	// Test that the cluster database is operational. We wait up to the
	// given timeout , in case there's no quorum of nodes online yet.
	timer := time.After(timeout)
	for i := 0; ; i++ {
		// Log initial attempts at debug level, but use warn
		// level after the 5'th attempt (about 10 seconds).
		// After the 15'th attempt (about 30 seconds), log
		// only one attempt every 5.
		logPriority := 1 // 0 is discard, 1 is Debug, 2 is Warn
		if i > 5 {
			logPriority = 2
			if i > 15 && !((i % 5) == 0) {
				logPriority = 0
			}
		}

		err = db.Ping()
		if err == nil {
			break
		}

		cause := errors.Cause(err)
		if cause != dqlite.ErrNoAvailableLeader {
			return nil, err
		}

		switch logPriority {
		case 1:
			logger.Debugf("Failed connecting to global database (attempt %d): %v", i, err)
		case 2:
			logger.Warnf("Failed connecting to global database (attempt %d): %v", i, err)
		}

		time.Sleep(2 * time.Second)
		select {
		case <-timer:
			return nil, fmt.Errorf("failed to connect to cluster database")
		default:
		}
	}

	nodesVersionsMatch, err := cluster.EnsureSchema(db, address, dir)
	if err != nil {
		return nil, errors.Wrap(err, "failed to ensure schema")
	}

	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)

	if !nodesVersionsMatch {
		cluster := &Cluster{
			db:    db,
			stmts: map[int]*sql.Stmt{},
		}

		return cluster, ErrSomeNodesAreBehind
	}

	stmts, err := cluster.PrepareStmts(db)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to prepare statements")
	}

	cluster := &Cluster{
		db:    db,
		stmts: stmts,
	}

	// Figure out the ID of this node.
	err = cluster.Transaction(func(tx *ClusterTx) error {
		nodes, err := tx.Nodes()
		if err != nil {
			return errors.Wrap(err, "failed to fetch nodes")
		}
		if len(nodes) == 1 && nodes[0].Address == "0.0.0.0" {
			// We're not clustered
			cluster.NodeID(1)
			return nil
		}
		for _, node := range nodes {
			if node.Address == address {
				cluster.nodeID = node.ID
				return nil
			}
		}
		return fmt.Errorf("no node registered with address %s", address)
	})
	if err != nil {
		return nil, err
	}

	return cluster, err
}
func OpenCluster(name string, store dqlite.ServerStore, address, dir string, timeout time.Duration, options ...dqlite.DriverOption) (*Cluster, error) {
	db, err := cluster.Open(name, store, options...)
	if err != nil {
		return nil, errors.Wrap(err, "failed to open database")
	}

	// Test that the cluster database is operational. We wait up to the
	// given timeout , in case there's no quorum of nodes online yet.
	timer := time.After(timeout)
	for i := 0; ; i++ {
		// Log initial attempts at debug level, but use warn
		// level after the 5'th attempt (about 10 seconds).
		// After the 15'th attempt (about 30 seconds), log
		// only one attempt every 5.
		logPriority := 1 // 0 is discard, 1 is Debug, 2 is Warn
		if i > 5 {
			logPriority = 2
			if i > 15 && !((i % 5) == 0) {
				logPriority = 0
			}
		}

		err = db.Ping()
		if err == nil {
			break
		}

		cause := errors.Cause(err)
		if cause != dqlite.ErrNoAvailableLeader {
			return nil, err
		}

		switch logPriority {
		case 1:
			logger.Debugf("Failed connecting to global database (attempt %d): %v", i, err)
		case 2:
			logger.Warnf("Failed connecting to global database (attempt %d): %v", i, err)
		}

		time.Sleep(2 * time.Second)
		select {
		case <-timer:
			return nil, fmt.Errorf("failed to connect to cluster database")
		default:
		}
	}

	nodesVersionsMatch, err := cluster.EnsureSchema(db, address, dir)
	if err != nil {
		return nil, errors.Wrap(err, "failed to ensure schema")
	}

	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)

	if !nodesVersionsMatch {
		cluster := &Cluster{
			db:    db,
			stmts: map[int]*sql.Stmt{},
		}

		return cluster, ErrSomeNodesAreBehind
	}

	stmts, err := cluster.PrepareStmts(db)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to prepare statements")
	}

	cluster := &Cluster{
		db:    db,
		stmts: stmts,
	}

	// Figure out the ID of this node.
	err = cluster.Transaction(func(tx *ClusterTx) error {
		nodes, err := tx.Nodes()
		if err != nil {
			return errors.Wrap(err, "failed to fetch nodes")
		}
		if len(nodes) == 1 && nodes[0].Address == "0.0.0.0" {
			// We're not clustered
			cluster.NodeID(1)
			return nil
		}
		for _, node := range nodes {
			if node.Address == address {
				cluster.nodeID = node.ID
				return nil
			}
		}
		return fmt.Errorf("no node registered with address %s", address)
	})
	if err != nil {
		return nil, err
	}

	return cluster, err
}
// OpenCluster creates a new Cluster object for interacting with the dqlite
// database.
//
// - name: Basename of the database file holding the data. Typically "db.bin".
// - dialer: Function used to connect to the dqlite backend via gRPC SQL.
// - address: Network address of this node (or empty string).
// - dir: Base LXD database directory (e.g. /var/lib/lxd/database)
//
// The address and api parameters will be used to determine if the cluster
// database matches our version, and possibly trigger a schema update. If the
// schema update can't be performed right now, because some nodes are still
// behind, an Upgrading error is returned.