services

package
v0.11.6 Latest Latest
Warning

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

 Go to latest Published: Sep 12, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package services holds the Connect-RPC handler implementations for AgentService (admin <-> agent) and ControlService (UI <-> admin).

Both services depend on the same shared state: nodes.Registry, routing.EventBus, routing.Replay, routing.SnapshotRouter. The State struct in this package wires them together.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func PushAggregate 

func PushAggregate(e *nodes.Entry, bus *routing.EventBus)

PushAggregate ships the current aggregate demand to a single agent. server.New wires it as State.OnAgentSubMode so a (re)connecting agent starts shipping immediately when UI streams are already open — without it, an agent that restarts mid-stream stays silent until some UI reopens its subscription.

Types

type AgentService 

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

AgentService implements adminv1connect.AgentServiceHandler.

func NewAgentService 

func NewAgentService(state *State) *AgentService

NewAgentService constructs the handler.

func (*AgentService) Stream 

func (s *AgentService) Stream(ctx context.Context, stream *connect.BidiStream[adminv1.Frame, adminv1.Frame]) error

Stream is the bidi entry point for an agent. The handler:

  • reads the first frame, expects NodeRegistration, registers the agent in the nodes.Registry,
  • spawns a writer goroutine that drains entry.Send into the stream,
  • dispatches subsequent inbound frames (events, heartbeats, snapshot responses) until the stream ends.

type ControlService 

type ControlService struct {

	// EventChannelSize is the per-StreamEvents subscription buffer.
	EventChannelSize int

	// SnapshotTimeout caps how long GetSnapshot waits for the agent.
	SnapshotTimeout time.Duration

	// MaxStreamsPerOperator caps the StreamEvents subscriptions one
	// identity may hold at once; the next open is refused with
	// ResourceExhausted. Every open used to fan the whole ring out and
	// push a new aggregate filter to every agent, with no ceiling per
	// operator (OR-26). Zero means the default, 8.
	MaxStreamsPerOperator int

	// MaxReplayEvents caps how many ring-buffer events include_recent
	// replays into a fresh stream. Zero means the default, 500.
	MaxReplayEvents int
	// contains filtered or unexported fields
}

ControlService implements adminv1connect.ControlServiceHandler.

func NewControlService 

func NewControlService(state *State, eventChannelSize int, snapshotTimeout time.Duration) *ControlService

NewControlService constructs the handler.

func (*ControlService) GetSelf 

func (s *ControlService) GetSelf(ctx context.Context, _ *connect.Request[adminv1.GetSelfRequest]) (*connect.Response[adminv1.SelfInfo], error)

GetSelf echoes the authenticated caller's identity (resolved by the UI auth chain, carried in the request context) plus the server's build version, so the UI can show who the operator is audited as and which server they hit.

func (*ControlService) GetSnapshot 

func (s *ControlService) GetSnapshot(ctx context.Context, req *connect.Request[adminv1.GetSnapshotRequest]) (*connect.Response[adminv1.Snapshot], error)

GetSnapshot routes a snapshot request to the right agent and waits for the matching SnapshotResponse over its bidi stream.

func (*ControlService) ListNodes 

func (s *ControlService) ListNodes(ctx context.Context, req *connect.Request[adminv1.ListNodesRequest]) (*connect.Response[adminv1.ListNodesResponse], error)

ListNodes returns a stable view of every connected agent.

func (*ControlService) StreamEvents 

func (s *ControlService) StreamEvents(ctx context.Context, req *connect.Request[adminv1.StreamEventsRequest], stream *connect.ServerStream[adminv1.Event]) error

StreamEvents subscribes to live events. The handler:

  • registers a new EventBus subscription with the requested filter,
  • rebuilds and pushes the agent-side aggregate filter to every connected agent (so they switch on/refine ingress as needed),
  • optionally replays the recent ring buffer first when include_recent is set,
  • drains the subscription channel into the response stream until ctx is cancelled.

type DataStudioService 

type DataStudioService struct {
	Timeout time.Duration
	// contains filtered or unexported fields
}

DataStudioService implements adminv1connect.DataStudioServiceHandler.

Every method routes the call to a connected agent that knows the requested model, sends a DataStudioRequest down its bidi stream, and blocks on the matching DataStudioResponse for at most Timeout.

func NewDataStudioService 

func NewDataStudioService(state *State, timeout time.Duration, allowedModels []string) *DataStudioService

NewDataStudioService constructs the handler. timeout <= 0 defaults to 10s. allowedModels is the per-model mutation allowlist (see Config.DataStudioAllowedModels): empty refuses every mutation, the single entry "*" allows them on every model.

func (*DataStudioService) BulkAction 

func (s *DataStudioService) BulkAction(ctx context.Context, req *connect.Request[adminv1.BulkActionRequest]) (*connect.Response[adminv1.BulkActionResponse], error)

func (*DataStudioService) CreateRecord 

func (s *DataStudioService) CreateRecord(ctx context.Context, req *connect.Request[adminv1.CreateRecordRequest]) (*connect.Response[adminv1.Record], error)

func (*DataStudioService) DeleteRecord 

func (s *DataStudioService) DeleteRecord(ctx context.Context, req *connect.Request[adminv1.DeleteRecordRequest]) (*connect.Response[adminv1.DeleteRecordResponse], error)

func (*DataStudioService) GetRecord 

func (s *DataStudioService) GetRecord(_ context.Context, req *connect.Request[adminv1.GetRecordRequest]) (*connect.Response[adminv1.Record], error)

func (*DataStudioService) GetSchema 

func (s *DataStudioService) GetSchema(_ context.Context, req *connect.Request[adminv1.GetSchemaRequest]) (*connect.Response[adminv1.ModelSchema], error)

func (*DataStudioService) ListModels 

func (s *DataStudioService) ListModels(_ context.Context, req *connect.Request[adminv1.ListModelsRequest]) (*connect.Response[adminv1.ListModelsResponse], error)

ListModels: returns the union of every connected agent's registered model set. When include_counts is true the call is routed to ONE agent (the first connected one) so the counts are coherent.

func (*DataStudioService) ListRecords 

func (s *DataStudioService) ListRecords(_ context.Context, req *connect.Request[adminv1.ListRecordsRequest]) (*connect.Response[adminv1.PaginatedRecords], error)

func (*DataStudioService) UpdateRecord 

func (s *DataStudioService) UpdateRecord(ctx context.Context, req *connect.Request[adminv1.UpdateRecordRequest]) (*connect.Response[adminv1.Record], error)

type ManageService 

type ManageService struct {
	Timeout time.Duration
	// contains filtered or unexported fields
}

ManageService implements adminv1connect.ManageServiceHandler — the UI-facing surface for the Access control and Audit log screens.

GetRbac routes to a connected agent (the application's authorizer is the source of truth); ListAudit reads the server's own fleet-plane audit ring.

func NewManageService 

func NewManageService(state *State, timeout time.Duration) *ManageService

NewManageService constructs the handler. timeout <= 0 defaults to 10s.

func (*ManageService) GetRbac 

func (s *ManageService) GetRbac(_ context.Context, req *connect.Request[adminv1.GetRbacRequest]) (*connect.Response[adminv1.GetRbacResponse], error)

GetRbac snapshots the Casbin roles/policies of one connected agent. An empty node_id picks the first connected agent (single-node fleets need no selector; multi-node fleets usually share one policy store).

func (*ManageService) ListAudit 

func (s *ManageService) ListAudit(_ context.Context, req *connect.Request[adminv1.ListAuditRequest]) (*connect.Response[adminv1.ListAuditResponse], error)

ListAudit returns the newest fleet-plane audit entries, newest first.

type State 

type State struct {
	Nodes          *nodes.Registry
	EventBus       *routing.EventBus
	Replay         *routing.Replay
	Snapshots      *routing.SnapshotRouter
	DataStudio     *routing.DataStudioRouter
	Rbac           *routing.RbacRouter
	Audit          *routing.AuditRing
	Logger         *slog.Logger
	SendChanBuffer int
	OnAgentSubMode func(*nodes.Entry, *routing.EventBus) // hook called whenever bus demand changes
	HeartbeatGrace time.Duration                         // tolerance window for stale heartbeat reports
}

State groups the shared dependencies the two service handlers need. One State per server lifetime.

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.