hasql

package module
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 10, 2023 License: Apache-2.0 Imports: 10 Imported by: 4

README

hasql

PkgGoDev GoDoc tests lint Go Report Card codecov

hasql provides simple and reliable way to access high-availability database setups with multiple hosts.

Status

hasql is production-ready and is actively used inside Yandex' production environment.

Prerequisites

Installation

With Go module support, simply add the following import

import "golang.yandex/hasql"

to your code, and then go [build|run|test] will automatically fetch the necessary dependencies.

Otherwise, to install the hasql package, run the following command:

$ go get -u golang.yandex/hasql

How does it work

hasql operates using standard database/sql connection pool objects. User creates *sql.DB objects for each node of database cluster and passes them to constructor. Library keeps up to date information on state of each node by 'pinging' them periodically. User is provided with a set of interfaces to retrieve *sql.DB object suitable for required operation.

dbFoo, _ := sql.Open("pgx", "host=foo")
dbBar, _ := sql.Open("pgx", "host=bar")
cl, err := hasql.NewCluster(
    []hasql.Node{hasql.NewNode("foo", dbFoo), hasql.NewNode("bar", dbBar) },
    checkers.PostgreSQL,
)
if err != nil { ... }

node := cl.Primary()
if node == nil {
    err := cl.Err() // most recent errors for all nodes in the cluster
}

// Do anything you like
fmt.Println("Node address", node.Addr)

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err = node.DB().PingContext(ctx); err != nil { ... }

hasql does not configure provided connection pools in any way. It is user's job to set them up properly. Library does handle their lifetime though - pools are closed when Cluster object is closed.

Supported criteria

Alive primary|Alive standby|Any alive node, or none otherwise

node := c.Primary()
if node == nil { ... }

Alive primary|Alive standby, or any alive node, or none otherwise

node := c.PreferPrimary()
if node == nil { ... }
Ways of accessing nodes

Any of currently alive nodes satisfying criteria, or none otherwise

node := c.Primary()
if node == nil { ... }

Any of currently alive nodes satisfying criteria, or wait for one to become alive

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
node, err := c.WaitForPrimary(ctx)
if err == nil { ... }
Node pickers

When user asks Cluster object for a node a random one from a list of suitable nodes is returned. User can override this behavior by providing a custom node picker.

Library provides a couple of predefined pickers. For example if user wants 'closest' node (with lowest latency) PickNodeClosest picker should be used.

cl, err := hasql.NewCluster(
    []hasql.Node{hasql.NewNode("foo", dbFoo), hasql.NewNode("bar", dbBar) },
    checkers.PostgreSQL,
    hasql.WithNodePicker(hasql.PickNodeClosest()),
)
if err != nil { ... }

Supported databases

Since library works over standard database/sql it supports any database that has a database/sql driver. All it requires is a database-specific checker function that can tell if node is primary or standby.

Check out golang.yandex/hasql/checkers package for more information.

Caveats

Node's state is transient at any given time. If Primary() returns a node it does not mean that node is still primary when you execute statement on it. All it means is that it was primary when it was last checked. Nodes can change their state at a whim or even go offline and hasql can't control it in any manner.

This is one of the reasons why nodes do not expose their perceived state to user.

Extensions

Instrumentation

You can add instrumentation via Tracer object similar to httptrace in standard library.

sqlx

hasql can operate over database/sql pools wrapped with sqlx. It works the same as with standard library but requires user to import golang.yandex/hasql/sqlx instead.

Refer to golang.yandex/hasql/sqlx package for more information.

Documentation

Index

Examples

Constants

View Source 
const (
	DefaultUpdateInterval = time.Second * 5
	DefaultUpdateTimeout  = time.Second
)

Default values for cluster config

Variables

This section is empty.

Functions

This section is empty.

Types

type AliveNodes 

type AliveNodes struct {
	Alive     []Node
	Primaries []Node
	Standbys  []Node
}

AliveNodes of Cluster

type Cluster 

type Cluster struct {
	// contains filtered or unexported fields
}

Cluster consists of number of 'nodes' of a single SQL database. Background goroutine periodically checks nodes and updates their status.

func NewCluster 

func NewCluster(nodes []Node, checker NodeChecker, opts ...ClusterOption) (*Cluster, error)

NewCluster constructs cluster object representing a single 'cluster' of SQL database. Close function must be called when cluster is not needed anymore.

Example 
package main

import (
	"context"
	"database/sql"
	"time"

	"golang.yandex/hasql"
	"golang.yandex/hasql/checkers"
)

func main() {
	// cluster hosts
	hosts := []struct {
		Addr       string
		Connstring string
	}{
		{
			Addr:       "host1.example.com",
			Connstring: "host=host1.example.com",
		},
		{
			Addr:       "host2.example.com",
			Connstring: "host=host2.example.com",
		},
		{
			Addr:       "host3.example.com",
			Connstring: "host=host3.example.com",
		},
	}

	// Construct cluster nodes
	nodes := make([]hasql.Node, 0, len(hosts))
	for _, host := range hosts {
		// Create database pools for each node
		db, err := sql.Open("pgx", host.Connstring)
		if err != nil {
			panic(err)
		}
		nodes = append(nodes, hasql.NewNode(host.Addr, db))
	}

	// Use options to fine-tune cluster behavior
	opts := []hasql.ClusterOption{
		hasql.WithUpdateInterval(2 * time.Second),        // set custom update interval
		hasql.WithNodePicker(hasql.PickNodeRoundRobin()), // set desired nodes selection algorithm
	}

	// Create cluster handler
	c, err := hasql.NewCluster(nodes, checkers.PostgreSQL, opts...)
	if err != nil {
		panic(err)
	}
	defer func() { _ = c.Close() }() // close cluster when it is not needed

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Wait for current primary
	node, err := c.WaitForPrimary(ctx)
	if err != nil {
		panic(err)
	}

	// Wait for any alive standby
	node, err = c.WaitForStandby(ctx)
	if err != nil {
		panic(err)
	}

	// Wait for any alive node
	node, err = c.WaitForAlive(ctx)
	if err != nil {
		panic(err)
	}
	// Wait for secondary node if possible, primary otherwise
	node, err = c.WaitForNode(ctx, hasql.PreferStandby)
	if err != nil {
		panic(err)
	}

	// Retrieve current primary
	node = c.Primary()
	if node == nil {
		panic("no primary")
	}
	// Retrieve any alive standby
	node = c.Standby()
	if node == nil {
		panic("no standby")
	}
	// Retrieve any alive node
	node = c.Alive()
	if node == nil {
		panic("everything is dead")
	}

	// Retrieve primary node if possible, secondary otherwise
	node = c.Node(hasql.PreferPrimary)
	if node == nil {
		panic("no primary nor secondary")
	}

	// Retrieve secondary node if possible, primary otherwise
	node = c.Node(hasql.PreferStandby)
	if node == nil {
		panic("no primary nor secondary")
	}

	// Do something on retrieved node
	if err = node.DB().PingContext(ctx); err != nil {
		panic(err)
	}
}

Output:

func (*Cluster) Alive 

func (cl *Cluster) Alive() Node

Alive returns node that is considered alive

func (*Cluster) Close 

func (cl *Cluster) Close() error

Close databases and stop node updates.

func (*Cluster) Err added in v1.1.1 

func (cl *Cluster) Err() error

Err returns the combined error including most recent errors for all nodes. This error is CollectedErrors or nil.

func (*Cluster) Node 

func (cl *Cluster) Node(criteria NodeStateCriteria) Node

Node returns cluster node with specified status.

func (*Cluster) Nodes 

func (cl *Cluster) Nodes() []Node

Nodes returns list of all nodes

func (*Cluster) Primary 

func (cl *Cluster) Primary() Node

Primary returns first available node that is considered alive and is primary (able to execute write operations)

func (*Cluster) PrimaryPreferred 

func (cl *Cluster) PrimaryPreferred() Node

PrimaryPreferred returns primary node if possible, standby otherwise

func (*Cluster) Standby 

func (cl *Cluster) Standby() Node

Standby returns node that is considered alive and is standby (unable to execute write operations)

func (*Cluster) StandbyPreferred 

func (cl *Cluster) StandbyPreferred() Node

StandbyPreferred returns standby node if possible, primary otherwise

func (*Cluster) WaitForAlive 

func (cl *Cluster) WaitForAlive(ctx context.Context) (Node, error)

WaitForAlive node to appear or until context is canceled

func (*Cluster) WaitForNode 

func (cl *Cluster) WaitForNode(ctx context.Context, criteria NodeStateCriteria) (Node, error)

WaitForNode with specified status to appear or until context is canceled

func (*Cluster) WaitForPrimary 

func (cl *Cluster) WaitForPrimary(ctx context.Context) (Node, error)

WaitForPrimary node to appear or until context is canceled

func (*Cluster) WaitForPrimaryPreferred 

func (cl *Cluster) WaitForPrimaryPreferred(ctx context.Context) (Node, error)

WaitForPrimaryPreferred node to appear or until context is canceled

func (*Cluster) WaitForStandby 

func (cl *Cluster) WaitForStandby(ctx context.Context) (Node, error)

WaitForStandby node to appear or until context is canceled

func (*Cluster) WaitForStandbyPreferred 

func (cl *Cluster) WaitForStandbyPreferred(ctx context.Context) (Node, error)

WaitForStandbyPreferred node to appear or until context is canceled

type ClusterOption 

type ClusterOption func(*Cluster)

ClusterOption is a functional option type for Cluster constructor

func WithNodePicker 

func WithNodePicker(picker NodePicker) ClusterOption

WithNodePicker sets algorithm for node selection (e.g. random, round robin etc)

func WithTracer 

func WithTracer(tracer Tracer) ClusterOption

WithTracer sets tracer for actions happening in the background

func WithUpdateInterval 

func WithUpdateInterval(d time.Duration) ClusterOption

WithUpdateInterval sets interval between cluster node updates

func WithUpdateTimeout 

func WithUpdateTimeout(d time.Duration) ClusterOption

WithUpdateTimeout sets ping timeout for update of each node in cluster

type CollectedErrors added in v1.1.1 

type CollectedErrors struct {
	Errors []NodeError
}

CollectedErrors are errors collected when checking node statuses

func (*CollectedErrors) Error added in v1.1.1 

func (e *CollectedErrors) Error() string

type Node 

type Node interface {
	fmt.Stringer

	Addr() string
	DB() *sql.DB
}

Node of single cluster

func NewNode 

func NewNode(addr string, db *sql.DB) Node

NewNode constructs node from database/sql DB

type NodeChecker 

type NodeChecker func(ctx context.Context, db *sql.DB) (bool, error)

NodeChecker is a signature for functions that check if specific node is alive and is primary. Returns true for primary and false if not. If error is returned, node is considered dead. Check function can be used to perform a query returning single boolean value that signals if node is primary or not.

type NodeError added in v1.1.1 

type NodeError struct {
	Addr       string
	Err        error
	OccurredAt time.Time
}

NodeError is error that background goroutine got while check given node

func (*NodeError) Error added in v1.1.1 

func (e *NodeError) Error() string

type NodePicker 

type NodePicker func(nodes []Node) Node

NodePicker is a signature for functions that determine how to pick single node from set of nodes. Nodes passed to the picker function are sorted according to latency (from lowest to greatest).

func PickNodeClosest added in v1.0.0 

func PickNodeClosest() NodePicker

PickNodeClosest returns node with least latency

func PickNodeRandom 

func PickNodeRandom() NodePicker

PickNodeRandom returns random node from nodes set

func PickNodeRoundRobin 

func PickNodeRoundRobin() NodePicker

PickNodeRoundRobin returns next node based on Round Robin algorithm

type NodeStateCriteria 

type NodeStateCriteria int

NodeStateCriteria for choosing a node 

const (
	// Alive for choosing any alive node
	Alive NodeStateCriteria = iota + 1
	// Primary for choosing primary node
	Primary
	// Standby for choosing standby node
	Standby
	// PreferPrimary for choosing primary or any alive node
	PreferPrimary
	// PreferStandby for choosing standby or any alive node
	PreferStandby
)

type Tracer 

type Tracer struct {
	// UpdateNodes is called when before updating nodes status.
	UpdateNodes func()
	// UpdatedNodes is called after all nodes are updated. The nodes is a list of currently alive nodes.
	UpdatedNodes func(nodes AliveNodes)
	// NodeDead is called when it is determined that specified node is dead.
	NodeDead func(node Node, err error)
	// NodeAlive is called when it is determined that specified node is alive.
	NodeAlive func(node Node)
	// NotifiedWaiters is called when all callers of 'WaitFor*' functions have been notified.
	NotifiedWaiters func()
}

Tracer is a set of hooks to run at various stages of background nodes status update. Any particular hook may be nil. Functions may be called concurrently from different goroutines.

Example 
package main

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"golang.yandex/hasql"
	"golang.yandex/hasql/checkers"
)

func main() {
	const hostname = "host=host1.example.com"
	db, err := sql.Open("pgx", "host="+hostname)
	if err != nil {
		panic(err)
	}

	nodes := []hasql.Node{hasql.NewNode(hostname, db)}

	tracer := hasql.Tracer{
		UpdateNodes: func() {
			fmt.Println("Started updating nodes")
		},
		UpdatedNodes: func(nodes hasql.AliveNodes) {
			fmt.Printf("Finished updating nodes: %+v\n", nodes)
		},
		NodeDead: func(node hasql.Node, err error) {
			fmt.Printf("Node %q is dead: %s", node, err)
		},
		NodeAlive: func(node hasql.Node) {
			fmt.Printf("Node %q is alive", node)
		},
		NotifiedWaiters: func() {
			fmt.Println("Notified all waiters")
		},
	}

	c, err := hasql.NewCluster(nodes, checkers.PostgreSQL, hasql.WithTracer(tracer))
	if err != nil {
		panic(err)
	}
	defer func() { _ = c.Close() }()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	_, err = c.WaitForPrimary(ctx)
	if err != nil {
		panic(err)
	}
}

Output:

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.