ensign

Documentation

Overview

Package ensign implements the Ensign single node server.

Index

• Constants
• type PubSub
  • func NewPubSub() (ps *PubSub)
  • func (ps *PubSub) Finish(id uuid.UUID)
  • func (ps *PubSub) Publish(event *api.EventWrapper) rlid.RLID
  • func (ps *PubSub) Subscribe() (id uuid.UUID, c buffer.Channel)
• type PublisherHandler
  • func (p PublisherHandler) Ack(eventID []byte, committed time.Time) error
  • func (p PublisherHandler) Nack(eventID []byte, code api.Nack_Code, err string) error
• type Server
  • func New(conf config.Config) (s *Server, err error)
  • func (s *Server) CreateTopic(ctx context.Context, in *api.Topic) (_ *api.Topic, err error)
  • func (s *Server) DeleteTopic(ctx context.Context, in *api.TopicMod) (out *api.TopicTombstone, err error)
  • func (s *Server) Info(ctx context.Context, in *api.InfoRequest) (out *api.ProjectInfo, err error)
  • func (s *Server) ListTopics(ctx context.Context, in *api.PageInfo) (out *api.TopicsPage, err error)
  • func (s *Server) Publish(stream api.Ensign_PublishServer) (err error)
  • func (s *Server) RetrieveTopic(ctx context.Context, in *api.Topic) (out *api.Topic, err error)
  • func (s *Server) Run(sock net.Listener)
  • func (s *Server) Serve() (err error)
  • func (s *Server) Shutdown() (err error)
  • func (s *Server) Status(ctx context.Context, in *api.HealthCheck) (out *api.ServiceState, err error)
  • func (s *Server) StoreMock() *mock.Store
  • func (s *Server) StreamInterceptors() []grpc.StreamServerInterceptor
  • func (s *Server) Subscribe(stream api.Ensign_SubscribeServer) (err error)
  • func (s *Server) TopicExists(ctx context.Context, in *api.TopicName) (out *api.TopicExistsInfo, err error)
  • func (s *Server) TopicNames(ctx context.Context, in *api.PageInfo) (out *api.TopicNamesPage, err error)
  • func (s *Server) UnaryInterceptors() []grpc.UnaryServerInterceptor
  • func (s *Server) WaitForQuarterdeck() (err error)
• type StreamHandler
  • func (s *StreamHandler) AllowedTopics() (group *topics.NameGroup, err error)
  • func (s *StreamHandler) Authorize(permission string) (_ *tokens.Claims, err error)
  • func (s *StreamHandler) ProjectID() (ulid.ULID, error)
• type SubscriberHandler
  • func (s SubscriberHandler) Send(event *api.EventWrapper) error

Constants

const BufferSize = 16384

const EventMaxDataSize int = 5.243e6

Cannot publish events > 5MiB long

Types

type PubSub

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

PubSub is a simple dispatcher that has a publish queue that fans in events from different publisher streams and assigns event ids then fans the events out to send them to one or more subscriber streams with an outgoing buffer. Backpressure is applied to the publisher streams when the buffers get full.

func NewPubSub

func NewPubSub() (ps *PubSub)

func (*PubSub) Finish

func (ps *PubSub) Finish(id uuid.UUID)

Finish closes a subscribe channel and removes it so that the PubSub no longer sends.

func (*PubSub) Publish

func (ps *PubSub) Publish(event *api.EventWrapper) rlid.RLID

Publish puts an event on the input queue and waits until the event is handled and an ID is assigned then returns the ID of the event to the caller.

func (*PubSub) Subscribe

func (ps *PubSub) Subscribe() (id uuid.UUID, c buffer.Channel)

Subscribe creates returns a channel that the caller can use to fetch events off of the in memory queue from. It also returns an ID so that the caller can close and cleanup the channel when it is done listenting for events.

type PublisherHandler

type PublisherHandler struct {
	StreamHandler
	// contains filtered or unexported fields
}

func (PublisherHandler) Ack

func (p PublisherHandler) Ack(eventID []byte, committed time.Time) error

func (PublisherHandler) Nack

func (p PublisherHandler) Nack(eventID []byte, code api.Nack_Code, err string) error

type Server

type Server struct {
	health.ProbeServer
	api.UnimplementedEnsignServer
	// contains filtered or unexported fields
}

An Ensign server implements the Ensign service as defined by the wire protocol.

func New

func New(conf config.Config) (s *Server, err error)

New creates a new ensign server with the given configuration. Most server setup is conducted in this method including setting up logging, connecting to databases, etc. If this method succeeds without an error, the server is ready to be served, but it will not listen to or handle requests until the Serve() method is called.

func (*Server) CreateTopic

func (s *Server) CreateTopic(ctx context.Context, in *api.Topic) (_ *api.Topic, err error)

CreateTopic is a user-facing request to create a Topic. Ensign first verifies that the topic is eligible to be created, then stores the topic in a pending state to disk and returns success to the user. Afterwards, Ensign sends a notification to the placement service in order to figure out where the topic should be assigned so that it can start receiving events.

func (*Server) DeleteTopic

func (s *Server) DeleteTopic(ctx context.Context, in *api.TopicMod) (out *api.TopicTombstone, err error)

DeleteTopic is a user-facing request to modify a topic and either archive it, which will make it read-only permanently or to destroy it, which will also have the effect of removing all of the data in the topic. This method is a stateful method, e.g. the topic will be updated to the current status then the Ensign placement server will take action from there.

func (*Server) Info

func (s *Server) Info(ctx context.Context, in *api.InfoRequest) (out *api.ProjectInfo, err error)

The Info RPC returns a summary of the current state of the project retrieved from the claims of the request. This RPC requires the ReadTopics permission in order to return any information. The status request can be filtered by a list of topics to specify exactly what statistics are returned.

func (*Server) ListTopics

func (s *Server) ListTopics(ctx context.Context, in *api.PageInfo) (out *api.TopicsPage, err error)

ListTopics associated with the project ID in the claims of the request. This unary request is paginated to prevent a huge amount of data transfer.

func (*Server) Publish

func (s *Server) Publish(stream api.Ensign_PublishServer) (err error)

Publish implements the streaming endpoint for the API.

func (*Server) RetrieveTopic

func (s *Server) RetrieveTopic(ctx context.Context, in *api.Topic) (out *api.Topic, err error)

RetrieveTopic is a user-face request to fetch a single Topic and is typically used for existence checks; e.g. does this topic exist or not. The user only has to specify a TopicID in the request and then a complete topic is returned. If the topic is not found a status error with codes.NotFound is returned.

func (*Server) Run

func (s *Server) Run(sock net.Listener)

Run the gRPC server on the specified socket. This method can be used to serve TCP requests or to connect to a bufconn for testing purposes. This method blocks while the server is running so it should be run in a go routine.

func (*Server) Serve

func (s *Server) Serve() (err error)

Serve RPC requests on the bind address specified in the configuration.

func (*Server) Shutdown

func (s *Server) Shutdown() (err error)

Shutdown stops the ensign server and all long running processes gracefully. May return a multierror if there were multiple problems during shutdown but it will attempt to close all open services and processes.

func (*Server) Status

func (s *Server) Status(ctx context.Context, in *api.HealthCheck) (out *api.ServiceState, err error)

Status implements a simple heartbeat mechanism for checking on the state of the Ensign server and making sure that the node is up and responding.

func (*Server) StoreMock

func (s *Server) StoreMock() *mock.Store

StoreMock returns the underlying store for testing purposes. Can only be accessed in storage testing mode otherwise the method panics.

func (*Server) StreamInterceptors

func (s *Server) StreamInterceptors() []grpc.StreamServerInterceptor

Prepares the interceptors (middleware) for the unary RPC endpoints of the server. The first interceptor will be the outer most, while the last interceptor will be the inner most wrapper around the real call. All stream interceptors returned by this method should be chained using grpc.ChainStreamInterceptor().

func (*Server) Subscribe

func (s *Server) Subscribe(stream api.Ensign_SubscribeServer) (err error)

Subscribe implements the streaming endpoint for the API

func (*Server) TopicExists

func (s *Server) TopicExists(ctx context.Context, in *api.TopicName) (out *api.TopicExistsInfo, err error)

TopicExists does a quick check to see if the topic ID or name exists in the project and returns a simple yes or no bool with the original query. If both ID and name are specified then this method checks if a topic with the specified name has the specified ID (e.g. it is a more strict existence check). The claims must have any of the topics:read, publisher, or subscriber permissions in order to access this endpoint.

func (*Server) TopicNames

func (s *Server) TopicNames(ctx context.Context, in *api.PageInfo) (out *api.TopicNamesPage, err error)

TopicNames returns a paginated response that maps topic names to IDs for all topics in the project. The claims must have any of the topics:read, publisher, or subscriber permissions in order to access this endpoint.

func (*Server) UnaryInterceptors

func (s *Server) UnaryInterceptors() []grpc.UnaryServerInterceptor

Prepares the interceptors (middleware) for the unary RPC endpoings of the server. The first interceptor will be the outer most, while the last interceptor will be the inner most wrapper around the real call. All unary interceptors returned by this method should be chained using grpc.ChainUnaryInterceptor().

func (*Server) WaitForQuarterdeck

func (s *Server) WaitForQuarterdeck() (err error)

Creates a quarterdeck client connected to the same host as the Auth KeysURL and waits until Quarterdeck returns a healthy response or the exponential timeout backoff limit is reached before returning.

type StreamHandler

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

StreamHandler provides some common functionality to both the Publisher and Subscriber stream handlers, for example providing authentication and collecting allowed topics.

func (*StreamHandler) AllowedTopics

func (s *StreamHandler) AllowedTopics() (group *topics.NameGroup, err error)

AllowedTopics returns a set of topic IDs and hashed topic names that are allowed to be accessed by the given claims. This set can be filtered to further restrict the stream based on user input. A specialized data structure is used to make it easy to perform content filtering based on name and ID.

func (*StreamHandler) Authorize

func (s *StreamHandler) Authorize(permission string) (_ *tokens.Claims, err error)

Authorize fetches the claims from the stream context, returning an error if the user is not authorized. The claims are cached on the stream handler and returned without error after the first time they are correctly fetched. Fetching claims requires a permission (e.g. subscribe or publish). If not authorized a status error is returned. Authorize MUST be called before projectID or projectTopics is called.

func (*StreamHandler) ProjectID

func (s *StreamHandler) ProjectID() (ulid.ULID, error)

Returns the ProjectID associated with the claims. Authorize must be called first or this method will error. Returns a status error if no project ID is on the claims. The projectID is cached on the stream handler and will be returned without error.

type SubscriberHandler

type SubscriberHandler struct {
	StreamHandler
	// contains filtered or unexported fields
}

func (SubscriberHandler) Send

func (s SubscriberHandler) Send(event *api.EventWrapper) error