dshardorchestrator

README

dshardorchestrator provides a sharding orchestrator for discord bots.

It's purpose is to manage nodes and assign shards to them, aswell as migrating shards between the nodes, which is one method of scaling large discord bots, spreading shards across processes and servers.

A here is a "process", which can be spread over hosts if you want.

I would not recommend using this, as 1. it currently only works against my discordgo fork and 2. its lacking a lot of tests still (although this im improving on)

It's currently used in YAGPDB.

Pitfalls

Currently its somewhat easy to break, if you try to break it that is, im working towards that but yeah, in it's current state i would just not recommend using it, unless you know what you're doing.

Essentials TODO:

• Full upgrade (simple function to migrate all nodes)
• Add in safeguards for doing things like, stopping nodes in the middle of migration.

Later:

• Extended status polling form nodes

Documentation

Index

• Variables
• func ContainsInt(slice []int, i int) bool
• func DecodePayload(evtID EventType, payload []byte) (interface{}, error)
• func EncodeMessage(evtID EventType, data interface{}) ([]byte, error)
• func EncodeMessageRaw(evtID EventType, data []byte) []byte
• func RegisterUserEvent(name string, id EventType, dataType interface{})
• type AllUserDataSentData
• type Conn
  • func ConnFromNetCon(conn net.Conn, logger Logger) *Conn
  • func (c *Conn) Close()
  • func (c *Conn) GetID() string
  • func (c *Conn) Listen()
  • func (c *Conn) Log(level LogLevel, err error, msg string)
  • func (c *Conn) Send(evtID EventType, data interface{}) error
  • func (c *Conn) SendLogErr(evtID EventType, data interface{})
  • func (c *Conn) SendNoLock(data []byte) error
• type EventType
  • func (evt EventType) String() string
• type IdentifiedData
• type IdentifyData
• type LogLevel
• type Logger
• type Message
• type PrepareShardmigrationData
• type ShardInfo
• type ShardMigrationMode
• type StartShardsData
• type StartshardMigrationData
• type StdLogger
  • func (stdl *StdLogger) Log(level LogLevel, message string)
• type StopShardData
• type UnknownEventError
  • func (uee *UnknownEventError) Error() string

Variables

var EventsToStringMap = map[EventType]string{

	1: "Identify",
	2: "Identified",
	3: "Shutdown",

	10: "StartShards",
	11: "StopShard",

	20: "PrepareShardmigration",
	21: "StartShardMigration",
	22: "AllUserdataSent",
}

EventsToStringMap is a mapping of events to their string names

var EvtDataMap = map[EventType]interface{}{
	EvtIdentify:              IdentifyData{},
	EvtIdentified:            IdentifiedData{},
	EvtStartShards:           StartShardsData{},
	EvtStopShard:             StopShardData{},
	EvtPrepareShardmigration: PrepareShardmigrationData{},
	EvtStartShardMigration:   StartshardMigrationData{},
	EvtAllUserdataSent:       AllUserDataSentData{},
}

EvtDataMap is a mapping of events to structs for their data

var StdLogInstance = &StdLogger{Level: LogInfo}

Functions

func ContainsInt

func ContainsInt(slice []int, i int) bool

func DecodePayload

func DecodePayload(evtID EventType, payload []byte) (interface{}, error)

DecodePayload decodes a event payload according to the specific event

func EncodeMessage

func EncodeMessage(evtID EventType, data interface{}) ([]byte, error)

EncodeMessage is the same as EncodeMessageRaw but also encodes the data passed using msgpack

func EncodeMessageRaw

func EncodeMessageRaw(evtID EventType, data []byte) []byte

EncodeMessageRaw encodes the event to the wire format The wire format is pretty basic, first 4 bytes is a uin32 representing what type of event this is next 4 bytes is another uin32 which represents the length of the body next n bytes is the body itself, which can even be empty in some cases

func RegisterUserEvent

func RegisterUserEvent(name string, id EventType, dataType interface{})

RegisterUserEvent registers a new user event to be used in shard migration for example calling this after opening a connection or otherwise concurrently will cause race conditions the reccomended way would be to call this in init()

panics if id is less than 100, as that's reserved id's for inernal use

Types

type AllUserDataSentData

type AllUserDataSentData struct {
	NumEvents int
}

type Conn

type Conn struct {
	ID atomic.Value

	// called on incoming messages
	MessageHandler func(*Message)

	// called when the connection is closed
	ConnClosedHanlder func()
	// contains filtered or unexported fields
}

Conn represents a connection from either node to the orchestrator or the other way around it implements common logic across both sides

func ConnFromNetCon

func ConnFromNetCon(conn net.Conn, logger Logger) *Conn

ConnFromNetCon wraos a Conn around a net.Conn

func (*Conn) Close

func (c *Conn) Close()

func (*Conn) GetID

func (c *Conn) GetID() string

GetID is a simpler helper for retrieving the connection id

func (*Conn) Listen

func (c *Conn) Listen()

Listen starts listening for events on the connection

func (*Conn) Log

func (c *Conn) Log(level LogLevel, err error, msg string)

func (*Conn) Send

func (c *Conn) Send(evtID EventType, data interface{}) error

Send sends the specified message over the connection, marshaling the data using json this locks the writer

func (*Conn) SendLogErr

func (c *Conn) SendLogErr(evtID EventType, data interface{})

Same as Send but logs the error (usefull for launching send in new goroutines)

func (*Conn) SendNoLock

func (c *Conn) SendNoLock(data []byte) error

SendNoLock sends the specified message over the connection, marshaling the data using json This does no locking and the caller is responsible for making sure its not called in multiple goroutines at the same time

type EventType

type EventType uint32

EventType represents a dshardorchestrator protocol event The internal event IDs are hardcoded to preserve compatibility between versions

const (

	// EvtIdentify identify new node connection
	// orchestrator <- node: Identify the new connection, orchestrator responds with a EvtIdentified if successfulll
	EvtIdentify EventType = 1

	// EvtIdentified is a response to EvtIdentify
	// orchestrator -> node: The connection was sucessfully established, now ready for whatever else
	EvtIdentified EventType = 2

	// EvtShutdown is sent to shut down a node completely, exiting
	// orchestrator -> node: shut down the node completely
	EvtShutdown EventType = 3

	// EvtStartShards assigns the following shards to the node, going through the full identify flow
	// orchestrator -> node: assign the shards to this node, respond with the same event when processed
	// orchestrator <- node: sent as a response when the node has been registered, does not need to have fully connected the shard yet, just registered.
	EvtStartShards EventType = 10

	// EvtStopShard is sent to stop the following shard
	// orhcestrator -> node: stop the shard, respond with the same event when done
	// orhcestrator <- node: sent when shard has been stopped
	EvtStopShard EventType = 11

	// EvtPrepareShardmigration is sent from the orchestrator when we should prepare for a shard migration, and also used as a response
	// orchestrator -> origin node: close the gateway connection and respond with a EvtPrepareShardmigration with session ID and sequence number
	// orchestrator <- origin node: send when the origin node has closed the gateway connection, includes sessionID and sequence number for resuming on destination node, the event is forwarded to the destination node
	// orchestrator -> destination node: save the session id and sequence number and prepare for a incoming shard transfe, respond with EvtPrepareShardmigration when ready
	// orchestrator <- destination node: sent as a response when ready for the shard data transfer, followed by EvtStartShardMigration
	EvtPrepareShardmigration EventType = 20

	// EvtStartShardMigration is used when we should start transferring shard data, the flow goes like this:
	// orchestrator -> oirign node: start sending all user data events, should respond with a EvtAllUserdataSent with the total number of user data events sent
	EvtStartShardMigration EventType = 21

	// EvtAllUserdataSent is sent with the total number of user data events sent.
	// UserData events can still be sent after this, the migration is finished when n user data events is received.
	// where n is sent in this event.
	// orchestrator <- origin node: sent at the end or during the shard user data transfer, includes total number of events that will be sent, forwarded to destination node
	// orchestrator -> destination node: the above, directly forwarded to the destination node, when the provided number of user data events has been received, the transfer is complete, the node is responsible for tracking this
	EvtAllUserdataSent EventType = 23

	// EvtShardMigrationDataStartID isn't an event per se, but this marks where user id's start
	// events with higher ID than this are registered and fully handled by implementations of the node interface
	// and will not be decoded or touched by the orchestrator.
	//
	// This can be used to transfer any kind of data during shard migration from the old node to the new node
	// to do that you could register a new event for "Guild" states, and send those over one by one.
	EvtShardMigrationDataStartID EventType = 100
)

func (EventType) String

func (evt EventType) String() string

type IdentifiedData

type IdentifiedData struct {
	TotalShards int
	NodeID      string
}

type IdentifyData

type IdentifyData struct {
	TotalShards   int
	RunningShards []int
	Version       string
	NodeID        string

	// when the logic of the orhcestrator changes in a backwards incompatible way, we reject nodes with mistmatched logic versions
	OrchestratorLogicVersion int
}

type LogLevel

type LogLevel int

const (
	LogError LogLevel = iota
	LogWarning
	LogInfo
	LogDebug
)

type Logger

type Logger interface {
	Log(level LogLevel, message string)
}

type Message

type Message struct {
	EvtID EventType

	// only 1 of RawBody or DecodeBody is present, not both
	RawBody     []byte
	DecodedBody interface{}
}

Message represents a protocol message

type PrepareShardmigrationData

type PrepareShardmigrationData struct {
	// wether this is the node were migrating the shard from
	Origin  bool
	ShardID int

	SessionID        string
	Sequence         int64
	ResumeGatewayUrl string
}

type ShardInfo

type ShardInfo struct {
	ShardID   int
	SessionID string
	Sequence  int64
}

ShardInfo represents basic shard session info

type ShardMigrationMode

type ShardMigrationMode int

const (
	ShardMigrationModeNone ShardMigrationMode = iota
	ShardMigrationModeTo
	ShardMigrationModeFrom
)

type StartShardsData

type StartShardsData struct {
	ShardIDs []int
}

type StartshardMigrationData

type StartshardMigrationData struct {
	ShardID int
}

type StdLogger

type StdLogger struct {
	Level  LogLevel
	Prefix string
}

func (*StdLogger) Log

func (stdl *StdLogger) Log(level LogLevel, message string)

type StopShardData

type StopShardData struct {
	ShardID int
}

type UnknownEventError

type UnknownEventError struct {
	Evt EventType
}

UnknownEventError represents an error for unknown events, this is techincally impossible with protocol versions being enforced, but who knows if you write your own node

func (*UnknownEventError) Error

func (uee *UnknownEventError) Error() string