Documentation
¶
Overview ¶
Package grpcoperator serves v1.OperatorService, the engine API for operators that run outside the engine process (kind = GRPC in v1_operator). It owns operator and worker registration and the Listen stream, and delegates task-event and durable-task RPCs to the dispatcher.
Index ¶
- Constants
- type OperatorService
- type OperatorServiceImpl
- func (s *OperatorServiceImpl) Cleanup() error
- func (s *OperatorServiceImpl) DurableTask(stream v1contracts.OperatorService_DurableTaskServer) error
- func (s *OperatorServiceImpl) Listen(stream v1contracts.OperatorService_ListenServer) (err error)
- func (s *OperatorServiceImpl) Register(ctx context.Context, req *v1contracts.OperatorRegisterRequest) (*v1contracts.OperatorRegisterResponse, error)
- func (s *OperatorServiceImpl) SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
- type OperatorServiceOpt
- func WithAnalytics(a analytics.Analytics) OperatorServiceOpt
- func WithDispatcher(d *dispatcher.DispatcherImpl) OperatorServiceOpt
- func WithLogger(l *zerolog.Logger) OperatorServiceOpt
- func WithMaxActionsPerOperator(n int64) OperatorServiceOpt
- func WithMaxListenStreamsPerOperator(n int) OperatorServiceOpt
- func WithRepository(r repository.Repository) OperatorServiceOpt
- func WithValidator(v validator.Validator) OperatorServiceOpt
- type OperatorServiceOpts
Constants ¶
const ( // DefaultMaxListenStreamsPerOperator caps the Listen streams one operator may hold open on // this engine replica at once. Every stream is a live worker with its own session and // action set, so the cap bounds what one tenant token can allocate here. DefaultMaxListenStreamsPerOperator = 100 // DefaultMaxActionsPerOperator caps the action links held across all workers of one // operator. Deltas that would exceed it are refused with ResourceExhausted. DefaultMaxActionsPerOperator = 1_000_000 )
const (
// MaxActionsPerDelta caps the ids (adds plus removes) one OperatorActionsDelta may carry.
MaxActionsPerDelta = 1000
)
const OperatorIdMetadataKey = "hatchet-operator-id"
OperatorIdMetadataKey is the incoming gRPC metadata key that carries the operator id on every RPC after Register (Listen, SendStepActionEvent, DurableTask).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type OperatorService ¶
type OperatorService interface {
v1contracts.OperatorServiceServer
Cleanup() error
}
type OperatorServiceImpl ¶
type OperatorServiceImpl struct {
v1contracts.UnimplementedOperatorServiceServer
// contains filtered or unexported fields
}
func New ¶
func New(fs ...OperatorServiceOpt) (*OperatorServiceImpl, error)
func (*OperatorServiceImpl) Cleanup ¶
func (s *OperatorServiceImpl) Cleanup() error
Cleanup stops the operator cache's expiry goroutine.
func (*OperatorServiceImpl) DurableTask ¶
func (s *OperatorServiceImpl) DurableTask(stream v1contracts.OperatorService_DurableTaskServer) error
DurableTask is the durable task event stream. The operator is authorized once when the stream opens, and the worker named by the first register_worker message is checked for ownership before the dispatcher's V1 service sees it; the dispatcher then owns the stream. The SDK's durable listener registers exactly once per stream (it opens a new stream to register again), so a second register_worker is refused as a protocol error rather than re-checked: the stream stays bound to the worker it was authorized for.
func (*OperatorServiceImpl) Listen ¶
func (s *OperatorServiceImpl) Listen(stream v1contracts.OperatorService_ListenServer) (err error)
Listen activates a registered worker for the lifetime of the stream and fans assigned actions out to it. The dispatcher session owns the send side of the stream: actions go out through its fan-out and delta acks go out through the session handle, so the two never overlap. This goroutine consumes heartbeats and action deltas.
func (*OperatorServiceImpl) Register ¶
func (s *OperatorServiceImpl) Register(ctx context.Context, req *v1contracts.OperatorRegisterRequest) (*v1contracts.OperatorRegisterResponse, error)
Register upserts the operator by name and creates the worker for this connection, or resumes the worker named by req.WorkerId when it still belongs to the operator. The worker starts with no actions: the client adds them on the Listen stream.
func (*OperatorServiceImpl) SendStepActionEvent ¶
func (s *OperatorServiceImpl) SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
SendStepActionEvent reports task progress for an action delivered on a Listen stream. The event's worker must belong to the calling operator; the dispatcher then handles it exactly like an SDK worker's event.
type OperatorServiceOpt ¶
type OperatorServiceOpt func(*OperatorServiceOpts)
func WithAnalytics ¶
func WithAnalytics(a analytics.Analytics) OperatorServiceOpt
func WithDispatcher ¶
func WithDispatcher(d *dispatcher.DispatcherImpl) OperatorServiceOpt
func WithLogger ¶
func WithLogger(l *zerolog.Logger) OperatorServiceOpt
func WithMaxActionsPerOperator ¶
func WithMaxActionsPerOperator(n int64) OperatorServiceOpt
WithMaxActionsPerOperator caps the action links held across all workers of one operator; zero disables the cap.
func WithMaxListenStreamsPerOperator ¶
func WithMaxListenStreamsPerOperator(n int) OperatorServiceOpt
WithMaxListenStreamsPerOperator caps the Listen streams one operator may hold open on this replica; zero disables the cap.
func WithRepository ¶
func WithRepository(r repository.Repository) OperatorServiceOpt
func WithValidator ¶
func WithValidator(v validator.Validator) OperatorServiceOpt
type OperatorServiceOpts ¶
type OperatorServiceOpts struct {
// contains filtered or unexported fields
}
Source Files