syndicate

Documentation

Index

• Constants
• Variables
• func ExtractCapacity(path string, disks []*pb.Disk) uint32
• func FatalIf(err error, msg string)
• func Filter(vs []string, f func(string) bool) []string
• func ParseManagedNodeAddress(addr string, port int) (string, error)
• type Config
• type ManagedNode
  • func NewManagedNode(o *ManagedNodeOpts) (ManagedNode, error)
• type ManagedNodeOpts
• type MockOpt
  • func WithGetBuilderFn(l func(path string) (*ring.Builder, error)) MockOpt
  • func WithRingBuilderBytesLoader(l func(path string) ([]byte, error)) MockOpt
  • func WithRingBuilderPersister(p func(c *RingChange, renameMaster bool) (error, error)) MockOpt
• type RingChange
• type RingSlave
• type RingSubscribers
• type Server
  • func NewServer(cfg *Config, servicename string, opts ...MockOpt) (*Server, error)
  • func (s *Server) AddNode(c context.Context, e *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) GetGlobalConfig(c context.Context, n *pb.EmptyMsg) (*pb.RingConf, error)
  • func (s *Server) GetNodeConfig(c context.Context, n *pb.Node) (*pb.RingConf, error)
  • func (s *Server) GetRing(c context.Context, e *pb.EmptyMsg) (*pb.Ring, error)
  • func (s *Server) GetRingStream(req *pb.SubscriberID, stream pb.Syndicate_GetRingStreamServer) error
  • func (s *Server) GetVersion(c context.Context, n *pb.EmptyMsg) (*pb.RingStatus, error)
  • func (s *Server) ModNode(c context.Context, n *pb.ModifyMsg) (*pb.RingStatus, error)
  • func (s *Server) NotifyNodes()
  • func (s *Server) RegisterNode(c context.Context, r *pb.RegisterRequest) (*pb.NodeConfig, error)
  • func (s *Server) RegisterSlave(slave *RingSlave) error
  • func (s *Server) RemoveNode(c context.Context, n *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) ReplaceAddresses(c context.Context, n *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) ReplaceTiers(c context.Context, n *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) RingChangeManager()
  • func (s *Server) SearchNodes(c context.Context, n *pb.Node) (*pb.SearchResult, error)
  • func (s *Server) SetActive(c context.Context, n *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) SetCapacity(c context.Context, n *pb.Node) (*pb.RingStatus, error)
  • func (s *Server) SetConf(c context.Context, conf *pb.Conf) (*pb.RingStatus, error)

Constants

const (
	DefaultPort        = 8443                        //The default port to use for the main backend service
	DefaultCmdCtrlPort = 4443                        //The default port to use for cmdctrl (address0)
	DefaultMsgRingPort = 8001                        //The default port the TCPMsgRing should use (address1)
	DefaultStorePort   = 6379                        //The default port the Store's should use (address2)
	DefaultRingDir     = "/etc/syndicate/ring"       //The default directory where to store the rings
	DefaultCertFile    = "/etc/syndicate/server.crt" //The default SSL Cert
	DefaultCertKey     = "/etc/syndicate/server.key" //The default SSL Key
)

Variables

var (
	DefaultNetFilter  = []string{"10.0.0.0/8", "192.168.0.0/16"} //Default the netfilters to private networks
	DefaultTierFilter = []string{".*"}                           //Default to ...anything

	InvalidTiers = errors.New("Tier0 already present in ring")
	InvalidAddrs = errors.New("No valid addresses provided")
)

var (
	DEFAULT_CTX_TIMEOUT = 10 * time.Second
)

Functions

func ExtractCapacity

func ExtractCapacity(path string, disks []*pb.Disk) uint32

func FatalIf

func FatalIf(err error, msg string)

FatalIf is just a lazy log/panic on error func

func Filter

func Filter(vs []string, f func(string) bool) []string

func ParseManagedNodeAddress

func ParseManagedNodeAddress(addr string, port int) (string, error)

Types

type Config

type Config struct {
	Master           bool
	Debug            bool
	Slaves           []string
	NetFilter        []string
	TierFilter       []string
	Port             int
	MsgRingPort      int
	CmdCtrlPort      int
	CmdCtrlIndex     int
	StorePort        int
	RingDir          string
	CertFile         string
	KeyFile          string
	WeightAssignment string
}

Config options for syndicate manager

type ManagedNode

type ManagedNode interface {
	ConnWaitForStateChange(context.Context, time.Duration, grpc.ConnectivityState) (grpc.ConnectivityState, error)
	ConnState() (grpc.ConnectivityState, error)
	Connect() error
	Disconnect() error
	Ping() (bool, string, error)
	Stop() error
	RingUpdate(*[]byte, int64) (bool, error)
	Lock()
	Unlock()
	RLock()
	RUnlock()
	Address() string
}

func NewManagedNode

func NewManagedNode(o *ManagedNodeOpts) (ManagedNode, error)

type ManagedNodeOpts

type ManagedNodeOpts struct {
	Address  string
	GrpcOpts []grpc.DialOption
}

type MockOpt

type MockOpt func(*Server)

MockOpt is just used for testing

func WithGetBuilderFn

func WithGetBuilderFn(l func(path string) (*ring.Builder, error)) MockOpt

WithGetBuilderFn is used for testing/mocking

func WithRingBuilderBytesLoader

func WithRingBuilderBytesLoader(l func(path string) ([]byte, error)) MockOpt

WithRingBuilderBytesLoader is used for testing/mocking

func WithRingBuilderPersister

func WithRingBuilderPersister(p func(c *RingChange, renameMaster bool) (error, error)) MockOpt

WithRingBuilderPersister is used for testing/mocking

type RingChange

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

type RingSlave

type RingSlave struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

type RingSubscribers

type RingSubscribers struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

type Server

type Server struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Server is the syndicate manager instance

func NewServer

func NewServer(cfg *Config, servicename string, opts ...MockOpt) (*Server, error)

NewServer returns a new instance of an up and running syndicate mangement node

func (*Server) AddNode

func (s *Server) AddNode(c context.Context, e *pb.Node) (*pb.RingStatus, error)

AddNode not currently used by anything. TODO: Need field/value error checks

func (*Server) GetGlobalConfig

func (s *Server) GetGlobalConfig(c context.Context, n *pb.EmptyMsg) (*pb.RingConf, error)

GetGlobalConfig retrieves the current global config []bytes present in the ring

func (*Server) GetNodeConfig

func (s *Server) GetNodeConfig(c context.Context, n *pb.Node) (*pb.RingConf, error)

GetNodeConfig retrieves a specific nodes ring config []bytes or an error if the node is not found.

func (*Server) GetRing

func (s *Server) GetRing(c context.Context, e *pb.EmptyMsg) (*pb.Ring, error)

GetRing returns the current ring bytes and version

func (*Server) GetRingStream

func (s *Server) GetRingStream(req *pb.SubscriberID, stream pb.Syndicate_GetRingStreamServer) error

GetRingStream return a stream of rings as they become available

func (*Server) GetVersion

func (s *Server) GetVersion(c context.Context, n *pb.EmptyMsg) (*pb.RingStatus, error)

func (*Server) ModNode

func (s *Server) ModNode(c context.Context, n *pb.ModifyMsg) (*pb.RingStatus, error)

func (*Server) NotifyNodes

func (s *Server) NotifyNodes()

NotifyNodes is called when a ring change occur's and just drops a change message on the changeChan for the RingChangeManager.

func (*Server) RegisterNode

func (s *Server) RegisterNode(c context.Context, r *pb.RegisterRequest) (*pb.NodeConfig, error)

func (*Server) RegisterSlave

func (s *Server) RegisterSlave(slave *RingSlave) error

func (*Server) RemoveNode

func (s *Server) RemoveNode(c context.Context, n *pb.Node) (*pb.RingStatus, error)

RemoveNode removes a node given node to the ring. If any errors are encountered the ring change is discarded. The response RingStatus message should only have True Status if the ring change succeeded. The active Ring Version at the end of the call is always returned.

func (*Server) ReplaceAddresses

func (s *Server) ReplaceAddresses(c context.Context, n *pb.Node) (*pb.RingStatus, error)

ReplaceAddresses explicitly sets a node to the provided addresses. NO validation is performed on the addresses provided and the address is NOT checked against the NetFilter list. The only check performed is to verify that the address(s) are not in use on another ring entry.

func (*Server) ReplaceTiers

func (s *Server) ReplaceTiers(c context.Context, n *pb.Node) (*pb.RingStatus, error)

ReplaceTiers explicitly sets a node to the provided tiers. NO validation is performed on the tiers provided and the address is NOT checked against the TierFilter list.

func (*Server) RingChangeManager

func (s *Server) RingChangeManager()

RingChangeManager gets ring change messages from the change chan and handles notifying all managed nodes.

func (*Server) SearchNodes

func (s *Server) SearchNodes(c context.Context, n *pb.Node) (*pb.SearchResult, error)

SearchNodes uses the ring's node Filter() method return all nodes matching the provided filters. The filter options are currently limited too: id~=, meta~=, tier~=, address~=

func (*Server) SetActive

func (s *Server) SetActive(c context.Context, n *pb.Node) (*pb.RingStatus, error)

func (*Server) SetCapacity

func (s *Server) SetCapacity(c context.Context, n *pb.Node) (*pb.RingStatus, error)

func (*Server) SetConf

func (s *Server) SetConf(c context.Context, conf *pb.Conf) (*pb.RingStatus, error)

SetConf sets the Ring global config to the provided bytes. If any errors are encountered the ring change is discarded. The response RingStatus message should only have True Status if the ring change succeeded. The active Ring Version at the end of the call is always returned.