cluster

Documentation

Index

• type ClusterNode
  • func (cluster *ClusterNode) AuthenticateUser(username string, password string) (string, map[string]interface{}, error)
  • func (cluster *ClusterNode) ConnectToNodes()
  • func (cluster *ClusterNode) CountLog(r io.Reader) int
  • func (cluster *ClusterNode) HandleClientConnection(conn net.Conn, user map[string]interface{})
  • func (cluster *ClusterNode) InsertIntoNode(connection *Connection, insert string, collection string, id string)
  • func (cn *ClusterNode) IsBool(str string) bool
  • func (cn *ClusterNode) IsFloat(str string) bool
  • func (cn *ClusterNode) IsInt(str string) bool
  • func (cn *ClusterNode) IsString(str string) bool
  • func (cluster *ClusterNode) LostReconnect()
  • func (cluster *ClusterNode) NewUser(username, password, permission string) (string, map[string]interface{}, error)
  • func (cluster *ClusterNode) Printl(data string, level string)
  • func (cluster *ClusterNode) QueryNode(n *NodeConnection, body []byte, wg *sync.WaitGroup, mu *sync.RWMutex, ...)
  • func (cluster *ClusterNode) QueryNodes(connection *Connection, body map[string]interface{}) error
  • func (cluster *ClusterNode) QueryNodesRet(body map[string]interface{}) map[string]string
  • func (cluster *ClusterNode) RemoveUser(username string) error
  • func (cluster *ClusterNode) RenewClusterConfig() error
  • func (cluster *ClusterNode) SaveConfig()
  • func (cluster *ClusterNode) SetupClusterConfig() error
  • func (cluster *ClusterNode) SignalListener()
  • func (cluster *ClusterNode) StartTCP_TLS()
  • func (cluster *ClusterNode) ValidatePermission(perm string) bool
• type Config
• type Connection
• type Node
• type NodeConnection
• type NodeReplica

Types

type ClusterNode

type ClusterNode struct {
	TCPAddr         *net.TCPAddr       // TCPAddr represents the address of the clusters TCP end point
	TCPListener     *net.TCPListener   // TCPListener is the cluster TCP network listener.
	Wg              *sync.WaitGroup    // Cluster WaitGroup waits for all goroutines to finish up
	NodeConnections []*NodeConnection  // Configured and forever connected node connections until shutdown.
	SignalChannel   chan os.Signal     // Catch operating system signal
	Config          Config             // Cluster config
	TLSConfig       *tls.Config        // Cluster TLS config if TLS is true
	ContextCancel   context.CancelFunc // For gracefully shutting down
	ConfigMu        *sync.RWMutex      // Cluster config mutex
	Context         context.Context    // Main looped go routine context.  This is for listeners, event loops and so forth
	LogMu           *sync.Mutex        // Log file mutex (only if logging enabled)
	LogFile         *os.File           // Opened log file (only if logging enabled)
	UniquenessMu    *sync.Mutex        // If many connections are inserting the same document there is a chance for the same document so if uniqueness is required there is a lock on insert on the cluster.  The cluster does lock nodes on reads but as this is a concurrent system 2 connections at the same time can cause 2 of the same records without this lock.
}

func (*ClusterNode) AuthenticateUser

func (cluster *ClusterNode) AuthenticateUser(username string, password string) (string, map[string]interface{}, error)

AuthenticateUser checks if a user exists and returns the user

func (*ClusterNode) ConnectToNodes

func (cluster *ClusterNode) ConnectToNodes()

ConnectToNodes connects to configured nodes

func (*ClusterNode) CountLog

func (cluster *ClusterNode) CountLog(r io.Reader) int

CountLog counts amount of lines within log file

func (*ClusterNode) HandleClientConnection

func (cluster *ClusterNode) HandleClientConnection(conn net.Conn, user map[string]interface{})

HandleClientConnection handles tcp/tls client connection

func (*ClusterNode) InsertIntoNode

func (cluster *ClusterNode) InsertIntoNode(connection *Connection, insert string, collection string, id string)

InsertIntoNode selects one node within cluster nodes and inserts json document.

func (*ClusterNode) IsBool

func (cn *ClusterNode) IsBool(str string) bool

IsBool is a provided bool a bool?

func (*ClusterNode) IsFloat

func (cn *ClusterNode) IsFloat(str string) bool

IsFloat is a provided float a float64?

func (*ClusterNode) IsInt

func (cn *ClusterNode) IsInt(str string) bool

IsInt is a provided int an int?

func (*ClusterNode) IsString

func (cn *ClusterNode) IsString(str string) bool

IsString is a provided string a string literal? "hello world" OR 'hello world'

func (*ClusterNode) LostReconnect

func (cluster *ClusterNode) LostReconnect()

func (*ClusterNode) NewUser

func (cluster *ClusterNode) NewUser(username, password, permission string) (string, map[string]interface{}, error)

NewUser creates new database user

func (*ClusterNode) Printl

func (cluster *ClusterNode) Printl(data string, level string)

Printl prints a line to the cluster.log file also will clear at LogMaxLines. Appropriate levels: ERROR, INFO, FATAL, WARN

func (*ClusterNode) QueryNode

func (cluster *ClusterNode) QueryNode(n *NodeConnection, body []byte, wg *sync.WaitGroup, mu *sync.RWMutex, responses *map[string]string, action string)

QueryNode queries a specific node

func (*ClusterNode) QueryNodes

func (cluster *ClusterNode) QueryNodes(connection *Connection, body map[string]interface{}) error

QueryNodes queries all nodes in parallel and gets responses

func (*ClusterNode) QueryNodesRet

func (cluster *ClusterNode) QueryNodesRet(body map[string]interface{}) map[string]string

QueryNodesRet queries all nodes and combines responses

func (*ClusterNode) RemoveUser

func (cluster *ClusterNode) RemoveUser(username string) error

RemoveUser removes a user by username

func (*ClusterNode) RenewClusterConfig

func (cluster *ClusterNode) RenewClusterConfig() error

func (*ClusterNode) SaveConfig

func (cluster *ClusterNode) SaveConfig()

SaveConfig save cluster config such as created users and so forth on shutdown (Don't make changes to .clusterconfig when running as on shutdown changes will get overwritten)

func (*ClusterNode) SetupClusterConfig

func (cluster *ClusterNode) SetupClusterConfig() error

SetupClusterConfig sets up default cluster config i.e .clusterconfig

func (*ClusterNode) SignalListener

func (cluster *ClusterNode) SignalListener()

SignalListener listens for system signals

func (*ClusterNode) StartTCP_TLS

func (cluster *ClusterNode) StartTCP_TLS()

StartTCP_TLS starts listening on tcp/tls on configured host and port

func (*ClusterNode) ValidatePermission

func (cluster *ClusterNode) ValidatePermission(perm string) bool

ValidatePermission validates cluster permissions aka R or RW

type Config

type Config struct {
	Nodes            []Node   `yaml:"nodes"`                         // Node host/ips
	Host             string   `yaml:"host"`                          // Cluster host
	TLSNode          bool     `default:"false" yaml:"tls-node"`      // Connects to nodes with tls.  Nodes MUST be using tls in-order to set this to true.
	TLSCert          string   `yaml:"tls-cert"`                      // Location to TLS cert
	TLSKey           string   `yaml:"tls-key"`                       // Location to TLS key
	TLS              bool     `default:"false" yaml:"tls"`           // TLS on or off ?
	Port             int      `yaml:"port"`                          // Cluster port
	Key              string   `yaml:"key"`                           // Shared key - this key is used to encrypt data on all nodes and to authenticate with a node.
	Users            []string `yaml:"users"`                         // Array of encoded users
	NodeReaderSize   int      `yaml:"node-reader-size"`              // How large of a response buffer can the cluster handle
	LogMaxLines      int      `yaml:"log-max-lines"`                 // At what point to clear logs.  Each log line start's with a [UTC TIME] LOG DATA
	JoinResponses    bool     `default:"true" yaml:"join-responses"` // Joins all nodes results limiting at n
	Logging          bool     `default:"false" yaml:"logging"`       // Log to file ?
	LogQuery         bool     `default:"false" yaml:"log-query"`     // Log incoming queries
	Timezone         string   `default:"Local" yaml:"timezone"`      // i.e America/Chicago default is local system time.  On the cluster we use the Timezone for logging purposes.
	NodeReadDeadline int      `yaml:"node-read-deadline"`            // Amount of seconds to wait for a node or node replica.  Default is 2 seconds
}

Config is the CursusDB cluster config struct

type Connection

type Connection struct {
	Text *textproto.Conn        // Text is used for reading and writing
	Conn net.Conn               // net.Conn is a generic stream-oriented network connection.
	User map[string]interface{} // Authenticated user
}

Connection is the main TCP connection struct for cluster

type Node

type Node struct {
	Host     string        `yaml:"host"` // Cluster node host i.e 0.0.0.0 or cluster0.example.com
	Port     int           `yaml:"port"` // Cluster node port default for a cluster node is 4701
	Replicas []NodeReplica // Cluster node replicas of configured.  If node becomes unavailable where to go to instead.
}

Node is a cluster node

type NodeConnection

type NodeConnection struct {
	Conn       *net.TCPConn    // Net connection
	SecureConn *tls.Conn       // Secure connection with TLS
	Text       *textproto.Conn // For writing and reading
	Mu         *sync.Mutex     // Multiple connections shouldn't hit the same node without the node being locked
	Replica    bool            // is node replica?
	Ok         bool            // Is node ok?
	Node       Node            // The underlaying Node for connection
}

NodeConnection is the cluster connected to a node as a client.

type NodeReplica

type NodeReplica struct {
	Host string `yaml:"host"` // Cluster node replica host i.e 0.0.0.0 or cluster0.example.com
	Port int    `yaml:"port"` // Default cluster node port of 4701 but can be configured
}

NodeReplica is a replica of original node. Used in-case active node is not available