broker

Documentation

Overview

Package broker implements the broker runtime and protocol.JournalServer APIs (Read, Append, Replicate, List, Apply). Its `pipeline` type manages the coordination of write transactions, and `resolver` the mapping of journal names to Routes of responsible brokers. `replica` is a top-level collection of runtime state and maintenance tasks associated with the processing of a journal. gRPC proxy support is also implemented by this package.

Index

• Variables
• func JournalIsConsistent(item allocator.Item, _ keyspace.KeyValue, assignments keyspace.KeyValues) bool
• func JournalRouteMatchesAssignments(rt pb.Route, assignments keyspace.KeyValues) bool
• func NewKeySpace(prefix string) *keyspace.KeySpace
• func SetSharedPersister(p *fragment.Persister)
• type Service
  • func NewService(state *allocator.State, jc pb.JournalClient, etcd *clientv3.Client) *Service
  • func (svc *Service) Append(stream pb.Journal_AppendServer) (err error)
  • func (svc *Service) Apply(ctx context.Context, req *pb.ApplyRequest) (resp *pb.ApplyResponse, err error)
  • func (svc *Service) IsNoopRouter() bool
  • func (svc *Service) List(ctx context.Context, req *pb.ListRequest) (resp *pb.ListResponse, err error)
  • func (svc *Service) ListFragments(ctx context.Context, req *pb.FragmentsRequest) (resp *pb.FragmentsResponse, err error)
  • func (svc *Service) QueueTasks(tasks *task.Group, server *server.Server, finishFn func())
  • func (svc *Service) Read(req *pb.ReadRequest, stream pb.Journal_ReadServer) (err error)
  • func (svc *Service) Replicate(stream pb.Journal_ReplicateServer) (err error)
  • func (svc *Service) Route(ctx context.Context, item string) pb.Route
  • func (svc *Service) UpdateRoute(string, *pb.Route)

Variables

var (
	// MaxAppendRate is the maximum rate at which any journal may be appended to,
	// in bytes per second. It upper-bounds the MaxAppendRate of any particular
	// JournalSpec. If zero, there is no maximum rate.
	MaxAppendRate int64 = 0 // No limit.
	// MinAppendRate is the minimum rate at which any Append RPC client may
	// stream content chunks, in bytes per second. Client RPCs that are unable
	// to sustain this flow rate in any given second are aborted, allowing other
	// blocked RPCs to proceed. MinAppendRate provides baseline protection to
	// limit the impact of slow or faulted clients over the pipeline, which is
	// an exclusively owned and highly contended resource.
	MinAppendRate int64 = 1 << 16 // 64K per second.
	// ErrFlowControlUnderflow is returned if an Append RPC was terminated due to
	// flow control policing. Specifically, the client failed to sustain the
	// MinAppendRate when sending content chunks of the stream.
	ErrFlowControlUnderflow = errors.New(
		"client stream didn't sustain the minimum append data rate")
)

Functions

func JournalIsConsistent

func JournalIsConsistent(item allocator.Item, _ keyspace.KeyValue, assignments keyspace.KeyValues) bool

JournalIsConsistent returns true if all allocator.Assignments of the JournalSpec identified by Item advertise the same Route, denoting that all replicas of the journal have synchronized.

func JournalRouteMatchesAssignments

func JournalRouteMatchesAssignments(rt pb.Route, assignments keyspace.KeyValues) bool

JournalRouteMatchesAssignments returns true iff the Route is equivalent to the Route marshaled with each of the journal's |assignments|.

func NewKeySpace

func NewKeySpace(prefix string) *keyspace.KeySpace

NewKeySpace returns a KeySpace suitable for use with an Allocator. It decodes allocator Items as JournalSpec messages, Members as BrokerSpecs, and Assignments as Routes.

func SetSharedPersister

func SetSharedPersister(p *fragment.Persister)

SetSharedPersister sets the Persister instance used by the `broker` package.

Types

type Service

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

Service is the top-level runtime concern of a Gazette Broker process. It drives local journal handling in response to allocator.State, powers journal resolution, and is also an implementation of protocol.JournalServer.

func NewService

func NewService(state *allocator.State, jc pb.JournalClient, etcd *clientv3.Client) *Service

NewService constructs a new broker Service, driven by allocator.State.

func (*Service) Append

func (svc *Service) Append(stream pb.Journal_AppendServer) (err error)

Append dispatches the JournalServer.Append API.

func (*Service) Apply

func (svc *Service) Apply(ctx context.Context, req *pb.ApplyRequest) (resp *pb.ApplyResponse, err error)

Apply dispatches the JournalServer.Apply API.

func (*Service) IsNoopRouter

func (svc *Service) IsNoopRouter() bool

IsNoopRouter returns false.

func (*Service) List

func (svc *Service) List(ctx context.Context, req *pb.ListRequest) (resp *pb.ListResponse, err error)

List dispatches the JournalServer.List API.

func (*Service) ListFragments

func (svc *Service) ListFragments(ctx context.Context, req *pb.FragmentsRequest) (resp *pb.FragmentsResponse, err error)

ListFragments dispatches the JournalServer.ListFragments API.

func (*Service) QueueTasks

func (svc *Service) QueueTasks(tasks *task.Group, server *server.Server, finishFn func())

QueueTasks of the Service to watch its KeySpace and serve local replicas.

func (*Service) Read

func (svc *Service) Read(req *pb.ReadRequest, stream pb.Journal_ReadServer) (err error)

Read dispatches the JournalServer.Read API.

func (*Service) Replicate

func (svc *Service) Replicate(stream pb.Journal_ReplicateServer) (err error)

Replicate dispatches the JournalServer.Replicate API.

func (*Service) Route

func (svc *Service) Route(ctx context.Context, item string) pb.Route

Route an item using the Service resolver. Route implements the protocol.DispatchRouter interface, and enables usages of protocol.WithDispatchItemRoute (eg, `client` & `http_gateway` packages) to resolve items via the Service resolver.

func (*Service) UpdateRoute

func (svc *Service) UpdateRoute(string, *pb.Route)

UpdateRoute is a no-op implementation of protocol.DispatchRouter.