app

package
v0.0.18 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 28, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package log implements the log service

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNodeNotConnected = errors.New("node not connected")
	ErrNodeQueueFull    = errors.New("node outbound queue full")
)

Functions

This section is empty.

Types

type EventService 

type EventService struct {
	Repo     *repository.Repo[*eventsv1.Event]
	Logger   logger.Logger
	Exchange *events.Exchange
	Manager  SessionManager
	// contains filtered or unexported fields
}

func (*EventService) Create 

func (n *EventService) Create(ctx context.Context, event *eventsv1.Event) (*eventsv1.Event, error)

func (*EventService) Delete 

func (n *EventService) Delete(ctx context.Context, id keys.ID) error

func (*EventService) Get 

func (n *EventService) Get(ctx context.Context, id keys.ID) (*eventsv1.Event, error)

func (*EventService) List 

func (n *EventService) List(ctx context.Context, limit int32) ([]*eventsv1.Event, error)

func (*EventService) Publish 

func (n *EventService) Publish(ctx context.Context, ev *eventsv1.Event) (*eventsv1.Event, error)

Publish implements events.EventServiceClient.

func (*EventService) Subscribe 

func (n *EventService) Subscribe(ctx context.Context, in NodeConnectInput) (Session, error)

Subscribe implements events.EventServiceClient.

type FencingToken 

type FencingToken uint64

type HolderID 

type HolderID string

type LeaseGuard 

type LeaseGuard interface {
	IsHolder(ctx context.Context, resourceID ResourceID, holderID HolderID) (bool, error)
	AssertHolder(ctx context.Context, resourceID ResourceID, holderID HolderID) (token FencingToken, err error)
}

type LeaseService 

type LeaseService struct {
	Exchange    *events.Exchange
	Logger      logger.Logger
	LeaseTTL    uint32
	GracePeriod time.Duration
	SigningKey  *ecdsa.PrivateKey
	TokenStore  map[string]string
	Manager     LeaseStore
	// contains filtered or unexported fields
}

func (*LeaseService) Acquire 

func (l *LeaseService) Acquire(ctx context.Context, taskID ResourceID, nodeID HolderID) (*leasesv1.Lease, string, error)

func (*LeaseService) Get 

func (l *LeaseService) Get(ctx context.Context, taskID ResourceID) (*leasesv1.Lease, error)

func (*LeaseService) List 

func (l *LeaseService) List(ctx context.Context) ([]*leasesv1.Lease, error)

func (*LeaseService) Release 

func (l *LeaseService) Release(ctx context.Context, taskID ResourceID, holderID HolderID, token string) error

func (*LeaseService) Renew 

func (l *LeaseService) Renew(ctx context.Context, taskID ResourceID, nodeID HolderID, token string) (*leasesv1.Lease, string, error)

func (*LeaseService) Revoke 

func (l *LeaseService) Revoke(ctx context.Context, taskID ResourceID, holderID HolderID) error

type LeaseStore 

type LeaseStore interface {
	LeaseGuard
	Acquire(ctx context.Context, resource ResourceID, holder HolderID, ttl time.Duration) (*leasesv1.Lease, string, error)
	Renew(ctx context.Context, resource ResourceID, holder HolderID, ttl time.Duration, token string) (*leasesv1.Lease, string, error)
	Release(ctx context.Context, resource ResourceID, holder HolderID, token string) error
	Revoke(ctx context.Context, resource ResourceID, holder HolderID) error
	Get(ctx context.Context, resource ResourceID) (*leasesv1.Lease, error)
	List(ctx context.Context) ([]*leasesv1.Lease, error)
}

type LogService 

type LogService struct {
	Logger      logger.Logger
	Exchange    *events.Exchange
	LogExchange *events.LogExchange
	Manager     SessionManager
	Sender      NodeSender
}

func (*LogService) SendStartLogsCommand 

func (s *LogService) SendStartLogsCommand(ctx context.Context, req *logsv1.TailLogRequest) error

func (*LogService) SendStopLogsCommand 

func (s *LogService) SendStopLogsCommand(ctx context.Context, req *logsv1.TailLogRequest) error

type NodeConnectInput 

type NodeConnectInput struct {
	NodeUID  string
	NodeName string
}

NodeConnectInput represents the identity established for a node stream. NodeUID and NodeName are expected to come from transport metadata.

type NodeSender 

type NodeSender interface {
	SendToNode(ctx context.Context, nodeUID string, ev *eventsv1.Event) error
	IsNodeConnected(nodeUID string) bool
}

type NodeService 

type NodeService struct {
	Repo *repository.Repo[*nodesv1.Node]

	Exchange *events.Exchange
	Logger   logger.Logger
	Manager  SessionManager
	Sender   NodeSender
	// contains filtered or unexported fields
}

func (*NodeService) Condition 

func (l *NodeService) Condition(ctx context.Context, id keys.ID, req *typesv1.ConditionRequest) error

func (*NodeService) Connect 

func (l *NodeService) Connect(ctx context.Context, in NodeConnectInput) (Session, error)

func (*NodeService) Create 

func (l *NodeService) Create(ctx context.Context, node *nodesv1.Node) (*nodesv1.Node, error)

func (*NodeService) Delete 

func (l *NodeService) Delete(ctx context.Context, id keys.ID) error

func (*NodeService) Forget 

func (l *NodeService) Forget(ctx context.Context, id keys.ID) error

func (*NodeService) Get 

func (l *NodeService) Get(ctx context.Context, id keys.ID) (*nodesv1.Node, error)

func (*NodeService) Join 

func (l *NodeService) Join(ctx context.Context, id keys.ID, node *nodesv1.Node) (*nodesv1.Node, error)

func (*NodeService) List 

func (l *NodeService) List(ctx context.Context, limit int32) ([]*nodesv1.Node, error)

func (*NodeService) Patch 

func (l *NodeService) Patch(ctx context.Context, id keys.ID, patch *nodesv1.Node) error

func (*NodeService) Update 

func (l *NodeService) Update(ctx context.Context, id keys.ID, node *nodesv1.Node) error

func (*NodeService) UpdateStatus 

func (l *NodeService) UpdateStatus(ctx context.Context, id keys.ID, st *nodesv1.Status, mask ...string) error

func (*NodeService) Upgrade 

func (l *NodeService) Upgrade(ctx context.Context, req *nodesv1.UpgradeRequest) error

type NodeSessionManager 

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

NodeSessionManager owns node stream sessions and enables targeted delivery.

Intended usage: - transport calls Connect() to open a session for a stream - transport forwards inbound stream messages to Session.Handle() - transport writes outbound messages from Session.Next() to the stream - business logic sends targeted events using SendToNode()

func NewNodeSessionManager 

func NewNodeSessionManager(exchange *events.Exchange, l logger.Logger, opts ...NodeSessionManagerOption) *NodeSessionManager

func (*NodeSessionManager) Connect 

func (m *NodeSessionManager) Connect(ctx context.Context, node *nodesv1.Node, in NodeConnectInput) (Session, error)

func (*NodeSessionManager) Disconnect 

func (m *NodeSessionManager) Disconnect(nodeUID string)

Disconnect closes and removes the session for nodeUID, if present.

func (*NodeSessionManager) Get 

func (m *NodeSessionManager) Get(ctx context.Context, nodeUID string) (*nodesv1.Node, error)

func (*NodeSessionManager) IsNodeConnected 

func (m *NodeSessionManager) IsNodeConnected(nodeUID string) bool

func (*NodeSessionManager) List 

func (m *NodeSessionManager) List(ctx context.Context, limit int32) ([]*nodesv1.Node, error)

func (*NodeSessionManager) SendToNode 

func (m *NodeSessionManager) SendToNode(ctx context.Context, nodeUID string, ev *eventsv1.Event) error

SendToNode enqueues an event for delivery to a single connected node.

It returns ErrNodeNotConnected if the node has no active session. It returns ErrNodeQueueFull if the node session is backpressured.

func (*NodeSessionManager) Set 

func (m *NodeSessionManager) Set(ctx context.Context, nodeUID string, node *nodesv1.Node) error

type NodeSessionManagerOption 

type NodeSessionManagerOption func(*NodeSessionManager)

func WithOutboundBuffer 

func WithOutboundBuffer(size int) NodeSessionManagerOption

type ResourceID 

type ResourceID string

type Scheduler 

type Scheduler interface {
	TaskManager
	Schedule(ctx context.Context, task *tasksv1.Task) (*nodesv1.Node, error)
	Reschedule(ctx context.Context, task *tasksv1.Task, nodeUID string) (*nodesv1.Node, error)
}

type Session 

type Session interface {
	Handle(ctx context.Context, ev *eventsv1.Event) error
	Next(ctx context.Context) (*eventsv1.Event, error)
	Close() error
}

type SessionManager 

type SessionManager interface {
	Connect(ctx context.Context, node *nodesv1.Node, in NodeConnectInput) (Session, error)
	Set(ctx context.Context, nodeUID string, node *nodesv1.Node) error
}

type TaskGetter 

type TaskGetter interface {
	Get(ctx context.Context, id keys.ID) (*tasksv1.Task, error)
	List(ctx context.Context, limit int32) ([]*tasksv1.Task, error)
}

type TaskManager 

type TaskManager interface {
	Start(ctx context.Context, task *tasksv1.Task) (*nodesv1.Node, error)
	Stop(ctx context.Context, task *tasksv1.Task, nodeUID string) error
	Kill(ctx context.Context, task *tasksv1.Task, nodeUID string) error
}

type TaskService 

type TaskService struct {
	Repo *repository.Repo[*tasksv1.Task]

	Exchange *events.Exchange
	Logger   logger.Logger
	Manager  LeaseStore
	// contains filtered or unexported fields
}

func (*TaskService) Condition 

func (l *TaskService) Condition(ctx context.Context, id keys.ID, req *typesv1.ConditionRequest) error

func (*TaskService) Create 

func (l *TaskService) Create(ctx context.Context, task *tasksv1.Task) (*tasksv1.Task, error)

func (*TaskService) Delete 

func (l *TaskService) Delete(ctx context.Context, id keys.ID) error

Delete publishes a delete request and the subscribers are responsible for deleting resources. Once they do, they will update there resource with the status Deleted

func (*TaskService) Get 

func (l *TaskService) Get(ctx context.Context, id keys.ID) (*tasksv1.Task, error)

func (*TaskService) Kill 

func (l *TaskService) Kill(ctx context.Context, id keys.ID) error

func (*TaskService) List 

func (l *TaskService) List(ctx context.Context, limit int32) ([]*tasksv1.Task, error)

func (*TaskService) Patch 

func (l *TaskService) Patch(ctx context.Context, id keys.ID, patch *tasksv1.Task) error

func (*TaskService) Start 

func (l *TaskService) Start(ctx context.Context, id keys.ID) error

func (*TaskService) Stop 

func (l *TaskService) Stop(ctx context.Context, id keys.ID) error

func (*TaskService) Update 

func (l *TaskService) Update(ctx context.Context, id keys.ID, task *tasksv1.Task) error

func (*TaskService) UpdateStatus 

func (l *TaskService) UpdateStatus(ctx context.Context, id keys.ID, st *tasksv1.Status, mask ...string) error

UpdateStatus implements tasks.TaskServiceClient.

type Version 

type Version string
var (
	VersionLeaseV1 Version = "lease/v1"
	VersionEventV1 Version = "event/v1"
)

type VolumeService 

type VolumeService struct {
	Repo *repository.Repo[*volumesv1.Volume]

	Exchange *events.Exchange
	Logger   logger.Logger
	// contains filtered or unexported fields
}

func (*VolumeService) Condition 

func (l *VolumeService) Condition(ctx context.Context, id keys.ID, req *typesv1.ConditionRequest) error

func (*VolumeService) Create 

func (l *VolumeService) Create(ctx context.Context, volume *volumesv1.Volume) (*volumesv1.Volume, error)

func (*VolumeService) Delete 

func (l *VolumeService) Delete(ctx context.Context, id keys.ID) error

Delete publishes a delete request and the subscribers are responsible for deleting resources. Once they do, they will update there resource with the status Deleted

func (*VolumeService) Get 

func (l *VolumeService) Get(ctx context.Context, id keys.ID) (*volumesv1.Volume, error)

func (*VolumeService) List 

func (l *VolumeService) List(ctx context.Context, limit int32) ([]*volumesv1.Volume, error)

func (*VolumeService) Patch 

func (l *VolumeService) Patch(ctx context.Context, id keys.ID, patch *volumesv1.Volume) error

func (*VolumeService) Update 

func (l *VolumeService) Update(ctx context.Context, id keys.ID, volume *volumesv1.Volume) error

func (*VolumeService) UpdateStatus 

func (l *VolumeService) UpdateStatus(ctx context.Context, id keys.ID, st *volumesv1.Status, mask ...string) error

UpdateStatus implements volumes.VolumeServieClient.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.