master

Documentation

Index

• Variables
• type Server
  • func New(cfg config.MasterRuntimeProvider, logger *zap.Logger) (*Server, error)
  • func (s *Server) Done() <-chan struct{}
  • func (s *Server) GetEmbeddedAgentStore() *cache.Store
  • func (s *Server) InitAdminUser(username, password string) error
  • func (s *Server) ListenAddress() (string, bool)
  • func (s *Server) ResourceCountsForTest() app.ResourceCounts
  • func (s *Server) Run() error
  • func (s *Server) SetChannelMasterListen(addr string)
  • func (s *Server) SetupEmbeddedAgentForTest(listenAddr string) error
  • func (s *Server) Shutdown(ctx context.Context) error

Variables

var (
	ErrAlreadyRunning = errors.New("master server: already running")
)

Types

type Server

type Server struct {
	Cfg     *config.MasterRuntimeConfig
	Logger  *zap.Logger
	DB      *gorm.DB
	LogDB   *gorm.DB
	Bus     app.EventBus
	Router  *gin.Engine
	Version atomic.Int64

	Hub            *msync.Hub
	RelayHub       *mastertunnel.Hub
	InstanceID     string
	Signer         *masteragentauth.Signer
	Connections    *connectivity.Service
	ProbeScheduler *connectivity.Scheduler
	Operations     *masteroperations.Service
	Listener       net.Listener

	MetricsListener net.Listener

	App                     app.Application
	MetricsRegistry         *prometheus.Registry
	RelayMetrics            *pkgmetrics.AgentRelayMetrics
	GenericAPIMetrics       *agentgenericapi.APIMetrics
	StatsCache              *dao.StatsCache
	ModelPerformanceCache   *apimodelmarketplace.GlobalModelPerformanceCache
	ModelMarketplaceHandler *apimodelmarketplace.Handler
	LogDeliveryWorker       *masterlogqueue.LogDeliveryWorker
	APIUsageQueue           *masterapiusage.APIUsageQueue
	APIUsageWorker          *masterapiusage.APIUsageWorker
	HistoryBackfillWorker   *masterhistorybackfill.Worker

	// Heartbeat captures agent last_seen in memory and periodically flushes
	// to DB; also serves freshness reads for API enrichment. Started in Run
	// and stopped (force-flushed) in Shutdown.
	Heartbeat *msync.HeartbeatTracker

	// RebuildRunner schedules async log-owned daily billing rebuilds.
	// Submitted jobs run as background goroutines (one per Submit);
	// the gc loop spawns inside NewRebuildRunner. Stopped in Shutdown
	// before Heartbeat shutdown.
	RebuildRunner *billing.RebuildRunner

	// DailyBillingBackfill discovers request-log history and submits the
	// versioned daily-only rebuild after RebuildRunner starts.
	DailyBillingBackfill *billing.DailyBillingBackfill

	// BillingLogRetention removes billing facts older than the live retention
	// setting. It starts only after Run commits and stops before database close.
	BillingLogRetention *billing.LogRetentionWorker

	// LimitEvaluator periodically evaluates per-channel usage limits,
	// toggling Status + LimitState. Stopped in Shutdown.
	LimitEvaluator *billing.LimitEvaluator

	// BYOKProvider 是 BYOK cipher 的注入点。private_channel.Handler
	// 通过它获取 *Cipher，避免污染 app.Application 顶层接口。
	BYOKProvider byokcrypto.Provider
	// contains filtered or unexported fields
}

func New

func New(cfg config.MasterRuntimeProvider, logger *zap.Logger) (*Server, error)

func (*Server) Done

func (s *Server) Done() <-chan struct{}

func (*Server) GetEmbeddedAgentStore

func (s *Server) GetEmbeddedAgentStore() *cache.Store

GetEmbeddedAgentStore returns the embedded agent's cache store. Tests use this to wait for cache sync barriers (e.g. polling until __system_test__ token is visible to the relay's auth middleware).

Returns nil if embedded agent has not been set up yet.

func (*Server) InitAdminUser

func (s *Server) InitAdminUser(username, password string) error

func (*Server) ListenAddress

func (s *Server) ListenAddress() (string, bool)

ListenAddress returns a snapshot of the published listener address.

func (*Server) ResourceCountsForTest

func (s *Server) ResourceCountsForTest() app.ResourceCounts

func (*Server) Run

func (s *Server) Run() error

func (*Server) SetChannelMasterListen

func (s *Server) SetChannelMasterListen(addr string)

SetChannelMasterListen overrides the channel handler's MasterListen after master.New() has run. Used by tests that bind a real listener (which yields the actual port) only after server construction.

func (*Server) SetupEmbeddedAgentForTest

func (s *Server) SetupEmbeddedAgentForTest(listenAddr string) error

SetupEmbeddedAgentForTest mounts the embedded agent relay routes on the master router using the given listen address. This is the test-only escape hatch that replicates the production path in Run() without requiring a real net.Listener. Call it after httptest.NewServer so you have the actual port.

func (*Server) Shutdown

func (s *Server) Shutdown(ctx context.Context) error