coordinator

Jan 30, 2014

Changes in this version

type ClusterConsensus

+ ForceLogCompaction func() error

type Coordinator

+ ForceCompaction func(user common.User) error

type CoordinatorImpl

+ func (self *CoordinatorImpl) ForceCompaction(user common.User) error

type RaftServer

+ func (s *RaftServer) ForceLogCompaction() error

Jan 17, 2014

Changes in this version

+ const HOST_ID_OFFSET

+ const IS_CONNECTED

+ const IS_RECONNECTING

+ const KILOBYTE

+ const MAX_REQUEST_SIZE

+ const MAX_REQUEST_TIME

+ const MAX_RESPONSE_SIZE

+ const MAX_SIZE

+ const MEGABYTE

+ const POINT_SEQUENCE_NUMBER_KEY

+ const RECONNECT_RETRY_WAIT

+ const REQUEST_RETRY_ATTEMPTS

+ var BARRIER_TIME_MAX int64 = math.MaxInt64

+ var BARRIER_TIME_MIN int64 = math.MinInt64

+ var VALID_NAMES *regexp.Regexp

+ type ChangeDbUserPassword struct

+ Database string

+ Hash string

+ Username string

+ func NewChangeDbUserPasswordCommand(db, username, hash string) *ChangeDbUserPassword

+ func (c *ChangeDbUserPassword) Apply(server raft.Server) (interface{}, error)

+ func (c *ChangeDbUserPassword) CommandName() string

type ClusterConfiguration

+ ClusterVersion uint32

+ func (self *ClusterConfiguration) ChangeDbUserPassword(db, username, hash string) error

+ func (self *ClusterConfiguration) CreateContinuousQuery(db string, query string) error

+ func (self *ClusterConfiguration) DeleteContinuousQuery(db string, id uint32) error

+ func (self *ClusterConfiguration) GetContinuousQueries(db string) []*ContinuousQuery

+ func (self *ClusterConfiguration) GetDatabaseReplicationFactor(name string) uint8

+ func (self *ClusterConfiguration) GetOwnerIdByLocation(location *int) *uint32

+ func (self *ClusterConfiguration) GetReplicas(server *ClusterServer, database *string) (*ClusterServer, []*ClusterServer)

+ func (self *ClusterConfiguration) GetReplicationFactor(database *string) uint8

+ func (self *ClusterConfiguration) GetRingFilterFunction(database string, countOfServersToInclude uint32) func(database, series *string, time *int64) bool

+ func (self *ClusterConfiguration) GetServerById(id *uint32) *ClusterServer

+ func (self *ClusterConfiguration) GetServerByProtobufConnectionString(connectionString string) *ClusterServer

+ func (self *ClusterConfiguration) GetServerIndexByLocation(location *int) int

+ func (self *ClusterConfiguration) GetServersByIndexAndReplicationFactor(database *string, index *int) (*ClusterServer, []*ClusterServer)

+ func (self *ClusterConfiguration) GetServersByRingLocation(database *string, location *int) []*ClusterServer

+ func (self *ClusterConfiguration) GetServersToMakeQueryTo(database *string) (servers []*serverToQuery, replicationFactor uint32)

+ func (self *ClusterConfiguration) IsSingleServer() bool

+ func (self *ClusterConfiguration) Recovery(b []byte) error

+ func (self *ClusterConfiguration) Save() ([]byte, error)

+ func (self *ClusterConfiguration) Servers() []*ClusterServer

+ func (self *ClusterConfiguration) SetActive()

+ func (self *ClusterConfiguration) SetContinuousQueryTimestamp(timestamp time.Time) error

+ func (self *ClusterConfiguration) WaitForLocalServerLoaded()

type ClusterConsensus

+ AssignEngineAndCoordinator func(engine queryRunner, coordinator *CoordinatorImpl) error

+ ChangeDbUserPassword func(db, username string, hash []byte) error

+ CreateContinuousQuery func(db string, query string) error

+ DeleteContinuousQuery func(db string, id uint32) error

type ClusterServer

+ ProtobufConnectionString string

+ func (self *ClusterServer) Connect()

+ func (self *ClusterServer) MakeRequest(request *protocol.Request, responseStream chan *protocol.Response) error

type CommonUser

+ CacheKey string

+ type ContinuousQuery struct

+ Id uint32

+ Query string

type Coordinator

+ CreateContinuousQuery func(user common.User, db string, query string) error

+ DeleteContinuousQuery func(user common.User, db string, id uint32) error

+ DeleteSeriesData func(user common.User, db string, query *parser.DeleteQuery, localOnly bool) error

+ DropSeries func(user common.User, db, series string) error

+ GetLastSequenceNumber func(replicationFactor uint8, ownerServerId, originatingServerId uint32) (uint64, error)

+ ListContinuousQueries func(user common.User, db string) ([]*protocol.Series, error)

+ ListSeries func(user common.User, database string) ([]*protocol.Series, error)

+ ReplayReplication func(request *protocol.Request, replicationFactor *uint8, owningServerId *uint32, ...)

+ ReplicateDelete func(request *protocol.Request) error

+ ReplicateWrite func(request *protocol.Request) error

type CoordinatorImpl

+ func (self *CoordinatorImpl) CommitSeriesData(db string, series *protocol.Series) error

+ func (self *CoordinatorImpl) ConnectToProtobufServers(localConnectionString string) error

+ func (self *CoordinatorImpl) CreateContinuousQuery(user common.User, db string, query string) error

+ func (self *CoordinatorImpl) DeleteContinuousQuery(user common.User, db string, id uint32) error

+ func (self *CoordinatorImpl) DeleteSeriesData(user common.User, db string, query *parser.DeleteQuery, localOnly bool) error

+ func (self *CoordinatorImpl) DropSeries(user common.User, db, series string) error

+ func (self *CoordinatorImpl) GetLastSequenceNumber(replicationFactor uint8, originatingServer, owningServer uint32) (uint64, error)

+ func (self *CoordinatorImpl) ListContinuousQueries(user common.User, db string) ([]*protocol.Series, error)

+ func (self *CoordinatorImpl) ListSeries(user common.User, database string) ([]*protocol.Series, error)

+ func (self *CoordinatorImpl) ProcessContinuousQueries(db string, series *protocol.Series)

+ func (self *CoordinatorImpl) ReplayReplication(request *protocol.Request, replicationFactor *uint8, owningServerId *uint32, ...)

+ func (self *CoordinatorImpl) ReplicateDelete(request *protocol.Request) error

+ func (self *CoordinatorImpl) ReplicateWrite(request *protocol.Request) error

+ func (self *CoordinatorImpl) SyncLogIteration()

+ func (self *CoordinatorImpl) SyncLogs()

+ type CreateContinuousQueryCommand struct

+ Database string

+ Query string

+ func NewCreateContinuousQueryCommand(database string, query string) *CreateContinuousQueryCommand

+ func (c *CreateContinuousQueryCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *CreateContinuousQueryCommand) CommandName() string

type CreateDatabaseCommand

+ ReplicationFactor uint8

+ type Database struct

+ Name string

+ ReplicationFactor uint8

+ type DeleteContinuousQueryCommand struct

+ Database string

+ Id uint32

+ func NewDeleteContinuousQueryCommand(database string, id uint32) *DeleteContinuousQueryCommand

+ func (c *DeleteContinuousQueryCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *DeleteContinuousQueryCommand) CommandName() string

+ type InfluxJoinCommand struct

+ ConnectionString string

+ Name string

+ ProtobufConnectionString string

+ func (c *InfluxJoinCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *InfluxJoinCommand) CommandName() string

+ func (c *InfluxJoinCommand) NodeName() string

+ type NextPoint struct

+ type ProtobufClient struct

+ func NewProtobufClient(hostAndPort string) *ProtobufClient

+ func (self *ProtobufClient) Close()

+ func (self *ProtobufClient) MakeRequest(request *protocol.Request, responseStream chan *protocol.Response) error

+ type ProtobufRequestHandler struct

+ func NewProtobufRequestHandler(db datastore.Datastore, coordinator Coordinator, ...) *ProtobufRequestHandler

+ func (self *ProtobufRequestHandler) HandleRequest(request *protocol.Request, conn net.Conn) error

+ func (self *ProtobufRequestHandler) WriteResponse(conn net.Conn, response *protocol.Response) error

+ type ProtobufServer struct

+ func NewProtobufServer(port string, requestHandler RequestHandler) *ProtobufServer

+ func (self *ProtobufServer) Close()

+ func (self *ProtobufServer) ListenAndServe()

type RaftServer

+ func (s *RaftServer) AssignEngineAndCoordinator(engine queryRunner, coordinator *CoordinatorImpl) error

+ func (s *RaftServer) ChangeDbUserPassword(db, username string, hash []byte) error

+ func (s *RaftServer) CompactLog()

+ func (s *RaftServer) CreateContinuousQuery(db string, query string) error

+ func (s *RaftServer) DeleteContinuousQuery(db string, id uint32) error

+ func (s *RaftServer) SetContinuousQueryTimestamp(timestamp time.Time) error

+ type RequestHandler interface

+ HandleRequest func(request *protocol.Request, conn net.Conn) error

+ type SetContinuousQueryTimestampCommand struct

+ Timestamp time.Time

+ func NewSetContinuousQueryTimestampCommand(timestamp time.Time) *SetContinuousQueryTimestampCommand

+ func (c *SetContinuousQueryTimestampCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *SetContinuousQueryTimestampCommand) CommandName() string

Nov 6, 2013

Changes in this version

+ const DEFAULT_ROOT_PWD

Nov 5, 2013

Changes in this version

+ type ClusterConsensus interface

+ ActivateServer func(server *ClusterServer) error

+ AddServer func(server *ClusterServer, insertIndex int) error

+ CreateDatabase func(name string) error

+ CreateRootUser func() error

+ DropDatabase func(name string) error

+ MovePotentialServer func(server *ClusterServer, insertIndex int) error

+ ReplaceServer func(oldServer *ClusterServer, replacement *ClusterServer) error

+ SaveClusterAdminUser func(u *clusterAdmin) error

+ SaveDbUser func(user *dbUser) error

type RaftServer

+ func (s *RaftServer) ActivateServer(server *ClusterServer) error

+ func (s *RaftServer) AddServer(server *ClusterServer, insertIndex int) error

+ func (s *RaftServer) MovePotentialServer(server *ClusterServer, insertIndex int) error

+ func (s *RaftServer) ReplaceServer(oldServer *ClusterServer, replacement *ClusterServer) error

Nov 1, 2013

Changes in this version

+ type AddPotentialServerCommand struct

+ Server *ClusterServer

+ func NewAddPotentialServerCommand(s *ClusterServer) *AddPotentialServerCommand

+ func (c *AddPotentialServerCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *AddPotentialServerCommand) CommandName() string

type ClusterConfiguration

+ func (self *ClusterConfiguration) GetServerByRaftName(name string) *ClusterServer

+ func (self *ClusterConfiguration) IsActive() bool

+ func (self *ClusterConfiguration) UpdateServerState(serverId uint32, state ServerState) error

type ClusterServer

+ RaftConnectionString string

type RaftServer

+ func (s *RaftServer) ActivateCluster() error

+ func (s *RaftServer) ClusterServer() *ClusterServer

+ func (s *RaftServer) Serve(l net.Listener, potentialLeaders []string, retryUntilJoin bool) error

type ServerState

+ const Potential

+ const SendingRingData

+ type UpdateServerStateCommand struct

+ ServerId uint32

+ State ServerState

+ func NewUpdateServerStateCommand(serverId uint32, state ServerState) *UpdateServerStateCommand

+ func (c *UpdateServerStateCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *UpdateServerStateCommand) CommandName() string

Oct 28, 2013

Changes in this version

+ const NUMBER_OF_RING_LOCATIONS

type ClusterConfiguration

+ func (self *ClusterConfiguration) AddPotentialServer(server *ClusterServer)

+ func (self *ClusterConfiguration) DropDatabase(name string) error

+ func (self *ClusterConfiguration) GetClusterAdmin(username string) *clusterAdmin

+ func (self *ClusterConfiguration) GetClusterAdmins() (names []string)

+ func (self *ClusterConfiguration) GetDbUser(db, username string) *dbUser

+ func (self *ClusterConfiguration) GetDbUsers(db string) (names []string)

+ func (self *ClusterConfiguration) MoveRebalancingToActive()

+ func (self *ClusterConfiguration) RebalanceBasedOnPotentialConfig()

+ func (self *ClusterConfiguration) UpdatePotentialServerOrder(serverIds []uint32)

+ type ClusterServer struct

+ Id uint32

+ RaftName string

+ State ServerState

type Coordinator

+ DropDatabase func(user common.User, db string) error

+ ListDatabases func(user common.User) ([]string, error)

type CoordinatorImpl

+ func (self *CoordinatorImpl) DropDatabase(user common.User, db string) error

+ func (self *CoordinatorImpl) ListClusterAdmins(requester common.User) ([]string, error)

+ func (self *CoordinatorImpl) ListDatabases(user common.User) ([]string, error)

+ func (self *CoordinatorImpl) ListDbUsers(requester common.User, db string) ([]string, error)

+ type DropDatabaseCommand struct

+ Name string

+ func NewDropDatabaseCommand(name string) *DropDatabaseCommand

+ func (c *DropDatabaseCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *DropDatabaseCommand) CommandName() string

type Matcher

+ IsRegex bool

+ Name string

type RaftServer

+ func (s *RaftServer) DropDatabase(name string) error

+ type ServerState int

+ const DeletingOldData

+ const LoadingRingData

+ const Running

type UserManager

+ ListClusterAdmins func(requester common.User) ([]string, error)

+ ListDbUsers func(requester common.User, db string) ([]string, error)

Oct 24, 2013

Changes in this version

+ type AddApiKeyCommand struct

+ ApiKey string

+ Database string

+ KeyType ApiKeyType

+ func NewAddApikeyCommand(db, key string, keyType ApiKeyType) *AddApiKeyCommand

+ func (c *AddApiKeyCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *AddApiKeyCommand) CommandName() string

+ type AddServerToLocationCommand struct

+ Host string

+ Location int64

+ func NewAddServerToLocationCommand(host string, location int64) *AddServerToLocationCommand

+ func (c *AddServerToLocationCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *AddServerToLocationCommand) CommandName() string

+ type ApiKeyType int

+ const ReadKey

+ const WriteKey

+ type ClusterConfiguration struct

+ MaxRingLocation int64

+ ReadApiKeys map[string]bool

+ RingLocationToServers map[int64][]string

+ WriteApiKeys map[string]bool

+ func NewClusterConfiguration(maxRingLocation int64) *ClusterConfiguration

+ func (self *ClusterConfiguration) AddApiKey(database, key string, apiKeyType ApiKeyType)

+ func (self *ClusterConfiguration) AddRingLocationToServer(hostnameAndPort string, ringLocation int64)

+ func (self *ClusterConfiguration) CreateDatabase(name string) error

+ func (self *ClusterConfiguration) CurrentDatabaseId() string

+ func (self *ClusterConfiguration) DeleteApiKey(database, key string)

+ func (self *ClusterConfiguration) GetDatabases() map[string]bool

+ func (self *ClusterConfiguration) IsValidReadKey(database, key string) bool

+ func (self *ClusterConfiguration) IsValidWriteKey(database, key string) bool

+ func (self *ClusterConfiguration) NextDatabaseId() string

+ func (self *ClusterConfiguration) RemoveRingLocationFromServer(hostnameAndPort string, ringLocation int64)

+ func (self *ClusterConfiguration) SaveClusterAdmin(u *clusterAdmin)

+ func (self *ClusterConfiguration) SaveDbUser(u *dbUser)

+ type CommonUser struct

+ Hash string

+ IsUserDeleted bool

+ Name string

+ func (self *CommonUser) GetDb() string

+ func (self *CommonUser) GetName() string

+ func (self *CommonUser) HasReadAccess(name string) bool

+ func (self *CommonUser) HasWriteAccess(name string) bool

+ func (self *CommonUser) IsClusterAdmin() bool

+ func (self *CommonUser) IsDbAdmin(db string) bool

+ func (self *CommonUser) IsDeleted() bool

+ type Coordinator interface

+ CreateDatabase func(db, initialApiKey, requestingApiKey string) error

+ DistributeQuery func(db string, query *parser.Query, yield func(*protocol.Series) error) error

+ WriteSeriesData func(db string, series *protocol.Series) error

+ type CoordinatorImpl struct

+ func NewCoordinatorImpl(datastore datastore.Datastore, raftServer *RaftServer, ...) *CoordinatorImpl

+ func (self *CoordinatorImpl) AuthenticateClusterAdmin(username, password string) (User, error)

+ func (self *CoordinatorImpl) AuthenticateDbUser(db, username, password string) (User, error)

+ func (self *CoordinatorImpl) ChangeClusterAdminPassword(requester User, username, password string) error

+ func (self *CoordinatorImpl) ChangeDbUserPassword(requester User, db, username, password string) error

+ func (self *CoordinatorImpl) CreateClusterAdminUser(requester User, username string) error

+ func (self *CoordinatorImpl) CreateDatabase(db, initialApiKey, requestingApiKey string) error

+ func (self *CoordinatorImpl) CreateDbUser(requester User, db, username string) error

+ func (self *CoordinatorImpl) DeleteClusterAdminUser(requester User, username string) error

+ func (self *CoordinatorImpl) DeleteDbUser(requester User, db, username string) error

+ func (self *CoordinatorImpl) DistributeQuery(db string, query *parser.Query, yield func(*protocol.Series) error) error

+ func (self *CoordinatorImpl) SetDbAdmin(requester User, db, username string, isAdmin bool) error

+ func (self *CoordinatorImpl) WriteSeriesData(db string, series *protocol.Series) error

+ type CreateDatabaseCommand struct

+ Name string

+ func NewCreateDatabaseCommand(name string) *CreateDatabaseCommand

+ func (c *CreateDatabaseCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *CreateDatabaseCommand) CommandName() string

+ type Matcher struct

+ func (self *Matcher) Matches(name string) bool

+ type NextDatabaseIdCommand struct

+ LastId int

+ func NewNextDatabaseIdCommand(lastId int) *NextDatabaseIdCommand

+ func (c *NextDatabaseIdCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *NextDatabaseIdCommand) CommandName() string

+ type RaftServer struct

+ func NewRaftServer(path string, host string, port int, clusterConfig *ClusterConfiguration) *RaftServer

+ func (s *RaftServer) AddReadApiKey(db, key string) error

+ func (s *RaftServer) AddServerToLocation(host string, location int64) error

+ func (s *RaftServer) AddWriteApiKey(db, key string) error

+ func (s *RaftServer) CreateDatabase(name string) error

+ func (s *RaftServer) CreateRootUser() error

+ func (s *RaftServer) GetNextDatabaseId() (string, error)

+ func (s *RaftServer) HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request))

+ func (s *RaftServer) Join(leader string) error

+ func (s *RaftServer) ListenAndServe(potentialLeaders []string, retryUntilJoin bool) error

+ func (s *RaftServer) RemoveApiKey(db, key string) error

+ func (s *RaftServer) RemoveServerFromLocation(host string, location int64) error

+ func (s *RaftServer) SaveClusterAdminUser(u *clusterAdmin) error

+ func (s *RaftServer) SaveDbUser(u *dbUser) error

+ func (self *RaftServer) Close()

+ type RemoveApiKeyCommand struct

+ ApiKey string

+ Database string

+ func NewRemoveApiKeyCommand(db, key string) *RemoveApiKeyCommand

+ func (c *RemoveApiKeyCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *RemoveApiKeyCommand) CommandName() string

+ type RemoveServerFromLocationCommand struct

+ Host string

+ Location int64

+ func NewRemoveServerFromLocationCommand(host string, location int64) *RemoveServerFromLocationCommand

+ func (c *RemoveServerFromLocationCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *RemoveServerFromLocationCommand) CommandName() string

+ type SaveClusterAdminCommand struct

+ User *clusterAdmin

+ func NewSaveClusterAdminCommand(u *clusterAdmin) *SaveClusterAdminCommand

+ func (c *SaveClusterAdminCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *SaveClusterAdminCommand) CommandName() string

+ type SaveDbUserCommand struct

+ User *dbUser

+ func NewSaveDbUserCommand(u *dbUser) *SaveDbUserCommand

+ func (c *SaveDbUserCommand) Apply(server raft.Server) (interface{}, error)

+ func (c *SaveDbUserCommand) CommandName() string

+ type User interface

+ GetDb func() string

+ GetName func() string

+ HasReadAccess func(name string) bool

+ HasWriteAccess func(name string) bool

+ IsClusterAdmin func() bool

+ IsDbAdmin func(db string) bool

+ IsDeleted func() bool

+ type UserManager interface

+ AuthenticateClusterAdmin func(username, password string) (User, error)

+ AuthenticateDbUser func(db, username, password string) (User, error)

+ ChangeClusterAdminPassword func(requester User, username, password string) error

+ ChangeDbUserPassword func(requester User, db, username, password string) error

+ CreateClusterAdminUser func(request User, username string) error

+ CreateDbUser func(request User, db, username string) error

+ DeleteClusterAdminUser func(requester User, username string) error

+ DeleteDbUser func(requester User, db, username string) error

+ SetDbAdmin func(requester User, db, username string, isAdmin bool) error