grpc

package
v4.2.1 Latest Latest
Warning

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

 Go to latest Published: Oct 12, 2026 License: MIT Imports: 57 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	AgentProtocolVersion = agentcontrol.ProtocolV1
)
View Source 
const NodeLogService_ReportLogs_FullMethodName = "/v2board.NodeLogService/ReportLogs"

Variables

View Source 
var NodeLogService_ServiceDesc = grpc.ServiceDesc{
	ServiceName: "v2board.NodeLogService",
	HandlerType: (*NodeLogServiceServer)(nil),
	Methods: []grpc.MethodDesc{
		{
			MethodName: "ReportLogs",
			Handler:    _NodeLogService_ReportLogs_Handler,
		},
	},
	Metadata: "api/grpc/v2board.proto",
}

Functions

func AuthInterceptor 

func AuthInterceptor(apiToken, jwtSecret string) grpc.UnaryServerInterceptor

AuthInterceptor 认证拦截器: 先校验节点自身 x-api-key, 否则回退到全局 API Token / JWT 两种认证方式 (管理端/兼容旧调用)

func AuthInterceptorWithAgents added in v4.1.0 

func AuthInterceptorWithAgents(apiToken, jwtSecret string, agents *AgentAuthenticator) grpc.UnaryServerInterceptor

AuthInterceptorWithAgents is AuthInterceptor on a listener that accepts agent client certificates. The Agent services authenticate in their handlers. On the v2board services a client certificate, when presented, must be a proxy node's agent certificate and authenticates the call; the request's node_id must name that node. Without a certificate the legacy credentials apply, whatever agent_control.mtls says: third-party node software keeps using them.

func GetPeerAddr 

func GetPeerAddr(ctx context.Context) string

GetPeerAddr 从上下文获取客户端地址

func GetUserAdminFromContext 

func GetUserAdminFromContext(ctx context.Context) (bool, bool)

GetUserAdminFromContext 从上下文获取用户管理员状态

func GetUserEmailFromContext 

func GetUserEmailFromContext(ctx context.Context) (string, bool)

GetUserEmailFromContext 从上下文获取用户邮箱

func GetUserIDFromContext 

func GetUserIDFromContext(ctx context.Context) (uint, bool)

GetUserIDFromContext 从上下文获取用户ID

func LoggingInterceptor 

func LoggingInterceptor() grpc.UnaryServerInterceptor

LoggingInterceptor 日志拦截器

func RegisterNodeLogServiceServer 

func RegisterNodeLogServiceServer(s grpc.ServiceRegistrar, srv NodeLogServiceServer)

func SetUserAdminToContext 

func SetUserAdminToContext(ctx context.Context, isAdmin bool) context.Context

SetUserAdminToContext 设置用户管理员状态到上下文

func SetUserEmailToContext 

func SetUserEmailToContext(ctx context.Context, email string) context.Context

SetUserEmailToContext 设置用户邮箱到上下文

func SetUserIDToContext 

func SetUserIDToContext(ctx context.Context, userID uint) context.Context

SetUserIDToContext 设置用户ID到上下文

func StreamAuthInterceptor 

func StreamAuthInterceptor(apiToken, jwtSecret string) grpc.StreamServerInterceptor

StreamAuthInterceptor 流式认证拦截器: 先校验节点自身 x-api-key, 否则回退到 全局 API Token / JWT 两种认证方式 (管理端/兼容旧调用)

func StreamAuthInterceptorWithAgents added in v4.1.0 

func StreamAuthInterceptorWithAgents(apiToken, jwtSecret string, agents *AgentAuthenticator) grpc.StreamServerInterceptor

StreamAuthInterceptorWithAgents is StreamAuthInterceptor on a listener that accepts agent client certificates; see AuthInterceptorWithAgents. Every message of a certificate-authenticated v2board stream must carry the certificate's node_id.

func StreamLoggingInterceptor 

func StreamLoggingInterceptor() grpc.StreamServerInterceptor

StreamLoggingInterceptor 流式日志拦截器

func WriteAgentConfigPrometheus added in v4.1.0 

func WriteAgentConfigPrometheus(body *strings.Builder)

WriteAgentConfigPrometheus renders the configuration push counters and the lagging-node gauge in the Prometheus text format.

func WriteAgentMaintenancePrometheus added in v4.2.0 

func WriteAgentMaintenancePrometheus(body *strings.Builder)

WriteAgentMaintenancePrometheus renders the maintenance event counter in the Prometheus text format.

func WriteAgentPackageReportPrometheus added in v4.2.0 

func WriteAgentPackageReportPrometheus(body *strings.Builder)

WriteAgentPackageReportPrometheus renders the package report counters in the Prometheus text format.

func WriteAgentUsersPrometheus added in v4.1.0 

func WriteAgentUsersPrometheus(body *strings.Builder)

WriteAgentUsersPrometheus renders the user delta counters and gauges of the Agent Control stream in the Prometheus text format.

Types

type AgentArtifactsGRPCServer added in v4.2.0 

type AgentArtifactsGRPCServer struct {
	agentv1pb.UnimplementedAgentArtifactsServer
	// contains filtered or unexported fields
}

AgentArtifactsGRPCServer serves AgentArtifacts on the agent listener.

func NewAgentArtifactsGRPCServer added in v4.2.0 

func NewAgentArtifactsGRPCServer(auth *AgentAuthenticator) *AgentArtifactsGRPCServer

NewAgentArtifactsGRPCServer returns the artifact service of auth's agent PKI. Without one, every call is refused with agent_cert_invalid.

func (*AgentArtifactsGRPCServer) DownloadPluginArtifact added in v4.2.0 

func (s *AgentArtifactsGRPCServer) DownloadPluginArtifact(request *agentv1pb.DownloadPluginArtifactRequest, stream agentv1pb.AgentArtifacts_DownloadPluginArtifactServer) error

DownloadPluginArtifact streams the artifact of a release assigned to the certificate's node, in chunks of at most artifactChunkSize bytes; the first carries the release.

func (*AgentArtifactsGRPCServer) GetPluginManifest added in v4.2.0 

func (s *AgentArtifactsGRPCServer) GetPluginManifest(ctx context.Context, request *agentv1pb.GetPluginManifestRequest) (*agentv1pb.GetPluginManifestResponse, error)

GetPluginManifest returns the canonical manifest of a release assigned to the certificate's node.

type AgentAuthenticator added in v4.1.0 

type AgentAuthenticator struct {
	// PKI verifies client certificates and serves AgentEnrollment; nil when
	// the built-in module PKI is off.
	PKI *agentpki.Service
	// Mode is agent_control.mtls; empty means the configuration default
	// (required from 4.2). A nil authenticator is optional.
	Mode string
	// Sunset is agent_control.legacy_sunset, announced with the deprecation
	// header; zero announces none.
	Sunset time.Time
}

AgentAuthenticator authenticates AnixOps Agents on the node-facing listener: by client certificate (agent PKI) or by the legacy node credential, as agent_control.mtls allows. The zero value and nil accept legacy credentials only and refuse client certificates, which nothing could verify.

type AgentControlConnection 

type AgentControlConnection struct {
	NodeID       uint32
	SessionID    string
	AgentVersion string
	InstanceID   string
	Capabilities []*agentv1pb.Capability
	ConnectedAt  time.Time
	LastSeen     time.Time
	DesiredRev   uint64
	ObservedRev  uint64
	// ServerCapabilities is what the HelloAck advertised to this agent
	// (serverCapabilities); Diagnostics is whether it advertised diag.v1,
	// for NO-8's vantage selection.
	ServerCapabilities []*agentv1pb.Capability
	Diagnostics        bool
	// Identity is the authenticated identity: the agent's SPIFFE ID, or
	// agentstreams.IdentityAPIKey for a node key or forward token.
	Identity string
	// contains filtered or unexported fields
}

AgentControlConnection is the current bidirectional control stream for a node.

type AgentControlGRPCServer 

type AgentControlGRPCServer struct {
	agentv1pb.UnimplementedAgentControlServiceServer
	// contains filtered or unexported fields
}

AgentControlGRPCServer implements the Agent-first control stream.

func NewAgentControlGRPCServer 

func NewAgentControlGRPCServer(manager *AgentControlManager) *AgentControlGRPCServer

func (*AgentControlGRPCServer) ControlStream 

func (s *AgentControlGRPCServer) ControlStream(stream agentv1pb.AgentControlService_ControlStreamServer) error

ControlStream serves one agent's control stream. The stream's node comes from the client certificate (proxy or forward node) or from the legacy node API key (proxy nodes); every envelope's node_id must name it. A refusal ends the stream with its error code in the x-anix-error-code trailer (agent_refusals.go).

func (*AgentControlGRPCServer) WithAuthenticator added in v4.1.0 

func (s *AgentControlGRPCServer) WithAuthenticator(auth *AgentAuthenticator) *AgentControlGRPCServer

WithAuthenticator sets how agents authenticate: by client certificate and, unless agent_control.mtls is required, by node API key. Without it only node API keys authenticate.

type AgentControlManager 

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

AgentControlManager owns live streams and correlates desired operations with ACKs.

func GetAgentControlManager 

func GetAgentControlManager() *AgentControlManager

func GetForwardAgentControlManager added in v4.1.0 

func GetForwardAgentControlManager() *AgentControlManager

GetForwardAgentControlManager returns the streams of forward-node agents. Forward and proxy node ids overlap, so they have managers of their own.

func NewAgentControlManager 

func NewAgentControlManager() *AgentControlManager

func (*AgentControlManager) AddObservedStateHandler 

func (m *AgentControlManager) AddObservedStateHandler(handler ObservedStateHandler)

func (*AgentControlManager) CancelOperation 

func (m *AgentControlManager) CancelOperation(ctx context.Context, nodeID uint32, operationID string, revision uint64) error

CancelOperation sends an out-of-band cancellation for an operation already present in desired state. It intentionally does not allocate another revision or desired record; the Agent reports the original operation's terminal observation after cancelling its execution context.

func (*AgentControlManager) Connection 

func (m *AgentControlManager) Connection(nodeID uint32) (AgentControlSnapshot, bool)

func (*AgentControlManager) DesiredRevision 

func (m *AgentControlManager) DesiredRevision(nodeID uint32) uint64

func (*AgentControlManager) DispatchOperation 

func (m *AgentControlManager) DispatchOperation(ctx context.Context, nodeID uint32, operation *agentv1pb.DesiredOperation) (*agentv1pb.OperationAck, error)

DispatchOperation pushes a desired operation and waits only for receipt ACK. Runtime completion is reported separately through ObservedState.

func (*AgentControlManager) ObservedState 

func (m *AgentControlManager) ObservedState(nodeID uint32) (*agentv1pb.ObservedState, bool)

func (*AgentControlManager) UseRevisionStore added in v4.2.0 

func (m *AgentControlManager) UseRevisionStore(store NodeRevisionStore)

UseRevisionStore attaches the durable revision allocator. Only the proxy node manager takes one: forward node ids overlap proxy node ids and have no cursor of their own.

type AgentControlSnapshot 

type AgentControlSnapshot struct {
	NodeID       uint32    `json:"node_id"`
	SessionID    string    `json:"session_id"`
	AgentVersion string    `json:"agent_version"`
	InstanceID   string    `json:"instance_id"`
	Capabilities []string  `json:"capabilities"`
	ConnectedAt  time.Time `json:"connected_at"`
	LastSeen     time.Time `json:"last_seen"`
	DesiredRev   uint64    `json:"desired_revision"`
	ObservedRev  uint64    `json:"observed_revision"`
	// ServerCapabilities are the data-plane features the HelloAck advertised
	// to the agent (config.v1, users.v1, reports.v1); one is in use only when
	// Capabilities lists it too. Diagnostics is whether the agent advertised
	// diag.v1, so a diagnostic may run from the node's vantage.
	ServerCapabilities []string `json:"server_capabilities"`
	Diagnostics        bool     `json:"diagnostics"`
	// NegotiatedCapabilities are the data-plane capabilities in use on the
	// session, listed by both the Hello and the HelloAck (name.version).
	NegotiatedCapabilities []string `json:"negotiated_capabilities"`
	// Authentication is how the session authenticated: "mtls" (an agent
	// client certificate) or "api-key" (the node API key or forward token).
	// Transport is the transport inventory's name for it (mtls-stream,
	// apikey-stream) and Identity the SPIFFE ID or "api-key".
	Authentication string `json:"authentication"`
	Transport      string `json:"transport"`
	Identity       string `json:"identity"`
	// Certificate is the client certificate of an mtls session (serial,
	// expiry, URI SAN); nil for api-key.
	Certificate *agentstreams.SessionCertificate `json:"certificate"`
	// AgentMetrics are the Agent's own health metrics from its latest
	// heartbeat that carried any (agent_control_*, agent_identity_*,
	// agent_dataplane_*), reported at AgentMetricsAt; nil before one.
	AgentMetrics   map[string]float64 `json:"agent_metrics"`
	AgentMetricsAt *time.Time         `json:"agent_metrics_at"`
}

AgentControlSnapshot is a read-only view used by control-plane services.

type AgentEnrollmentGRPCServer added in v4.1.0 

type AgentEnrollmentGRPCServer struct {
	agentv1pb.UnimplementedAgentEnrollmentServer
	// contains filtered or unexported fields
}

AgentEnrollmentGRPCServer serves AgentEnrollment on the agent listener.

func NewAgentEnrollmentGRPCServer added in v4.1.0 

func NewAgentEnrollmentGRPCServer(auth *AgentAuthenticator) *AgentEnrollmentGRPCServer

NewAgentEnrollmentGRPCServer returns the enrollment service of auth's agent PKI. Without one, every call answers FailedPrecondition.

func (*AgentEnrollmentGRPCServer) Enroll added in v4.1.0 

func (s *AgentEnrollmentGRPCServer) Enroll(ctx context.Context, request *agentv1pb.EnrollAgentRequest) (*agentv1pb.EnrollAgentResponse, error)

Enroll issues a node's first certificate for a bootstrap credential: a one-time enrollment credential, or the node credential in metadata (refused when agent_control.mtls is required).

func (*AgentEnrollmentGRPCServer) GetLinkTrustBundle added in v4.2.0 

func (s *AgentEnrollmentGRPCServer) GetLinkTrustBundle(ctx context.Context, request *agentv1pb.GetLinkTrustBundleRequest) (*agentv1pb.GetLinkTrustBundleResponse, error)

GetLinkTrustBundle returns the forward link CAs to an agent that holds a certificate.

func (*AgentEnrollmentGRPCServer) GetTrustBundle added in v4.1.0 

func (s *AgentEnrollmentGRPCServer) GetTrustBundle(ctx context.Context, request *agentv1pb.GetAgentTrustBundleRequest) (*agentv1pb.GetAgentTrustBundleResponse, error)

GetTrustBundle returns the CAs of agent certificates to an agent that holds one.

func (*AgentEnrollmentGRPCServer) IssueLinkCertificate added in v4.2.0 

func (s *AgentEnrollmentGRPCServer) IssueLinkCertificate(ctx context.Context, request *agentv1pb.IssueLinkCertificateRequest) (*agentv1pb.IssueLinkCertificateResponse, error)

IssueLinkCertificate issues the certificate's node a forward link certificate.

func (*AgentEnrollmentGRPCServer) Renew added in v4.1.0 

func (s *AgentEnrollmentGRPCServer) Renew(ctx context.Context, request *agentv1pb.RenewAgentCertificateRequest) (*agentv1pb.RenewAgentCertificateResponse, error)

Renew issues a new certificate to the holder of a valid one.

type AgentStreams added in v4.1.0 

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

AgentStreams is the kernel's Agent Control streams by node kind (agentstreams.Streams): proxy nodes stream in one manager and forward nodes in another, since their ids overlap. KernelNodeOps dispatches node operations through it (node-ops-service.md section 3.8).

func GetAgentStreams added in v4.1.0 

func GetAgentStreams() *AgentStreams

GetAgentStreams returns the streams of the process's managers.

func NewAgentStreams added in v4.1.0 

func NewAgentStreams(proxy, forward *AgentControlManager) *AgentStreams

NewAgentStreams returns the streams over the two managers.

func (*AgentStreams) Cancel added in v4.1.0 

func (s *AgentStreams) Cancel(ctx context.Context, node agentcontrol.AgentNode, operationID string, revision uint64) error

Cancel asks the agent to stop an operation it holds.

func (*AgentStreams) ConfigNegotiated added in v4.1.0 

func (s *AgentStreams) ConfigNegotiated(node agentcontrol.AgentNode) bool

ConfigNegotiated reports whether the node's current session negotiated config.v1 (agentstreams.ConfigStreams).

func (*AgentStreams) Dispatch added in v4.1.0 

func (s *AgentStreams) Dispatch(ctx context.Context, node agentcontrol.AgentNode, operation *agentv1pb.DesiredOperation) (*agentv1pb.OperationAck, error)

Dispatch sends operation on the node's stream and waits for the acknowledgement. Its errors wrap the agentstreams sentinels and keep the manager's text.

func (*AgentStreams) Observed added in v4.1.0 

func (s *AgentStreams) Observed(node agentcontrol.AgentNode) (*agentv1pb.ObservedState, bool)

Observed returns the last observed state the node's agent reported.

func (*AgentStreams) OnConfigStatus added in v4.1.0 

func (s *AgentStreams) OnConfigStatus(handler agentstreams.ConfigStatusHandler)

OnConfigStatus registers handler with both managers (agentstreams.ConfigStreams).

func (*AgentStreams) OnObserved added in v4.1.0 

func (s *AgentStreams) OnObserved(handler agentstreams.ObservedHandler)

OnObserved registers handler with both managers.

func (*AgentStreams) PushConfig added in v4.1.0 

func (s *AgentStreams) PushConfig(ctx context.Context, node agentcontrol.AgentNode, snapshot *agentv1pb.ConfigSnapshot) (string, bool, error)

PushConfig sends snapshot on the node's session (agentstreams.ConfigStreams).

func (*AgentStreams) Session added in v4.1.0 

func (s *AgentStreams) Session(node agentcontrol.AgentNode) (agentstreams.Session, bool)

Session returns the node's current stream session.

func (*AgentStreams) Sessions added in v4.1.0 

func (s *AgentStreams) Sessions() []agentstreams.Session

Sessions lists every stream session of both kinds.

func (*AgentStreams) StatusConnection added in v4.2.0 

func (s *AgentStreams) StatusConnection(node agentcontrol.AgentNode) (any, bool)

StatusConnection is the node's session as the administrator's Agent Control status shows it (agentstreams.StatusConnections): its manager's snapshot.

type DesiredOperationDispatcher 

type DesiredOperationDispatcher interface {
	DispatchOperation(context.Context, uint32, *agentv1pb.DesiredOperation) (*agentv1pb.OperationAck, error)
	CancelOperation(context.Context, uint32, string, uint64) error
}

DesiredOperationDispatcher lets durable runtime workers push desired state without depending on the gRPC stream implementation.

type HealthGRPCServer 

type HealthGRPCServer struct {
	pb.UnimplementedHealthServiceServer
}

HealthGRPCServer 健康检查 gRPC 服务实现

func NewHealthGRPCServer 

func NewHealthGRPCServer() *HealthGRPCServer

NewHealthGRPCServer 创建健康检查 gRPC 服务

func (*HealthGRPCServer) Check 

func (s *HealthGRPCServer) Check(ctx context.Context, req *pb.HealthCheckRequest) (*pb.HealthCheckResponse, error)

Check 简单健康检查

func (*HealthGRPCServer) Watch 

func (s *HealthGRPCServer) Watch(stream pb.HealthService_WatchServer) error

Watch 双向流：持续健康检查

type KernelOperationBridge 

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

KernelOperationBridge joins durable Kernel operations to the ephemeral Agent stream. The database stays authoritative: the stream supplies only the active session and delivery transport.

func NewKernelOperationBridge 

func NewKernelOperationBridge(db *gorm.DB, stream kernelOperationStream) (*KernelOperationBridge, error)

func (*KernelOperationBridge) Run added in v4.1.0 

func (b *KernelOperationBridge) Run(ctx context.Context, interval time.Duration, reportError func(error))

Run performs durable dispatch every interval and blocks until ctx is cancelled. Callers supply logging so this package remains usable by tests and non-server control processes. ctx must not be nil.

func (*KernelOperationBridge) RunOnce 

func (b *KernelOperationBridge) RunOnce(ctx context.Context) (int, error)

RunOnce dispatches all pending node operations and recovers dispatch attempts whose Control process stopped before an ACK could be persisted.

func (*KernelOperationBridge) Start 

func (b *KernelOperationBridge) Start(ctx context.Context, interval time.Duration, reportError func(error))

Start runs Run in a new goroutine.

type NodeConnection 

type NodeConnection struct {
	NodeID     uint32
	LastSeen   time.Time
	RemoteAddr string
	Connection any // 可以是具体的流对象
}

NodeConnection 节点连接信息

type NodeConnectionManager 

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

NodeConnectionManager 节点连接管理器

func GetConnectionManager 

func GetConnectionManager() *NodeConnectionManager

GetConnectionManager 获取连接管理器

func NewNodeConnectionManager 

func NewNodeConnectionManager() *NodeConnectionManager

NewNodeConnectionManager 创建连接管理器

func (*NodeConnectionManager) IsConfigChanged 

func (m *NodeConnectionManager) IsConfigChanged(nodeID uint32, currentVer int64) bool

IsConfigChanged 检查配置是否变更

func (*NodeConnectionManager) Register 

func (m *NodeConnectionManager) Register(nodeID uint32, addr string)

Register 注册节点连接

func (*NodeConnectionManager) SetNodeConfigVersion 

func (m *NodeConnectionManager) SetNodeConfigVersion(nodeID uint32, version int64)

SetNodeConfigVersion 记录节点已推送的配置版本

func (*NodeConnectionManager) Unregister 

func (m *NodeConnectionManager) Unregister(nodeID uint32)

Unregister 注销节点连接

func (*NodeConnectionManager) UpdateLastSeen 

func (m *NodeConnectionManager) UpdateLastSeen(nodeID uint32)

UpdateLastSeen 更新最后活跃时间

type NodeGRPCServer 

type NodeGRPCServer struct {
	pb.UnimplementedNodeServiceServer
	// contains filtered or unexported fields
}

NodeGRPCServer 节点 gRPC 服务实现

func NewNodeGRPCServer 

func NewNodeGRPCServer() *NodeGRPCServer

NewNodeGRPCServer 创建节点 gRPC 服务

func (*NodeGRPCServer) GetConfig 

func (s *NodeGRPCServer) GetConfig(ctx context.Context, req *pb.NodeConfigRequest) (*pb.NodeConfigResponse, error)

GetConfig 获取节点配置

func (*NodeGRPCServer) Register 

func (s *NodeGRPCServer) Register(ctx context.Context, req *pb.NodeRegisterRequest) (*pb.NodeRegisterResponse, error)

Register 节点注册

func (*NodeGRPCServer) ReportStatus 

func (s *NodeGRPCServer) ReportStatus(ctx context.Context, req *pb.NodeStatusRequest) (*pb.StatusResponse, error)

ReportStatus 上报节点状态

func (*NodeGRPCServer) StatusStream 

func (s *NodeGRPCServer) StatusStream(stream pb.NodeService_StatusStreamServer) error

StatusStream 双向流：节点状态实时通信

type NodeLogGRPCServer 

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

func NewNodeLogGRPCServer 

func NewNodeLogGRPCServer() *NodeLogGRPCServer

func (*NodeLogGRPCServer) ReportLogs 

func (s *NodeLogGRPCServer) ReportLogs(ctx context.Context, req *dynamicpb.Message) (*pb.StatusResponse, error)

type NodeLogServiceServer 

type NodeLogServiceServer interface {
	ReportLogs(context.Context, *dynamicpb.Message) (*pb.StatusResponse, error)
}

type NodeRevisionStore added in v4.2.0 

type NodeRevisionStore interface {
	// AllocateRevision reserves the next revision of nodeID, above floor.
	// ok is false when no durable store is available; the manager then
	// falls back to its in-memory counter.
	AllocateRevision(ctx context.Context, nodeID uint32, floor uint64) (revision uint64, ok bool, err error)
	// RaiseRevision raises the stored revision of nodeID to at least
	// revision (the revision a connecting Agent reports).
	RaiseRevision(ctx context.Context, nodeID uint32, revision uint64) error
}

NodeRevisionStore is the durable per-node allocator of Agent operation revisions (v3_kernel_node_operation_revision). Durable plugin operations take their revisions from it when they are created; with a store attached, AgentControlManager allocates the revisions of one-off stream operations from it too, so one node's revisions are strictly increasing across both kinds. The Agent supersedes any operation at or below its observed revision, and the manager refuses a revision that is not newer than the last one sent.

func NewDatabaseRevisionStore added in v4.2.0 

func NewDatabaseRevisionStore(db func() *gorm.DB) NodeRevisionStore

NewDatabaseRevisionStore returns the NodeRevisionStore over the database that db resolves at each call.

type ObservedStateHandler 

type ObservedStateHandler func(nodeID uint32, observed *agentv1pb.ObservedState)

ObservedStateHandler adapts terminal Agent state back into durable job state.

type Server 

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

Server gRPC 服务器

func NewServer 

func NewServer(cfg *ServerConfig) *Server

NewServer 创建 gRPC 服务器

func (*Server) Addr added in v4.1.0 

func (s *Server) Addr() string

Addr returns the address the started listener is bound to, for a configured port of 0.

func (*Server) AgentPKI added in v4.1.0 

func (s *Server) AgentPKI() *agentpki.Service

AgentPKI returns the agent PKI the listener verifies client certificates with, nil when the built-in module PKI is off.

func (*Server) GetAgentControlManager 

func (s *Server) GetAgentControlManager() *AgentControlManager

GetAgentControlManager returns the Agent-first desired/observed connection manager.

func (*Server) Start 

func (s *Server) Start() error

Start 启动服务器

func (*Server) Stop 

func (s *Server) Stop()

Stop 停止服务器

type ServerConfig 

type ServerConfig struct {
	// 监听地址
	Host string
	// 监听端口
	Port int
	// API Token (可选，用于认证)
	APIToken string
	// JWT Secret (可选，用于 JWT 认证)
	JWTSecret string
	// TLS 证书/私钥路径 (可选，都为空时使用明文)
	TLSCertFile string
	TLSKeyFile  string
	// Keepalive 时间
	KeepaliveTime time.Duration
	// Keepalive 超时
	KeepaliveTimeout time.Duration
	// 最大连接空闲时间
	MaxConnectionIdle time.Duration
	// 最大连接年龄
	MaxConnectionAge time.Duration
	// AgentPKI verifies agent client certificates and serves
	// AgentEnrollment; nil when the built-in module PKI is off. Client
	// certificates need TLS (TLSCertFile).
	AgentPKI *agentpki.Service
	// AgentMTLS is agent_control.mtls: off, optional, preferred or
	// required (the configuration default from 4.2).
	AgentMTLS string
	// AgentLegacySunset is agent_control.legacy_sunset, announced to legacy
	// agents in preferred mode; zero announces none.
	AgentLegacySunset time.Time
}

ServerConfig gRPC 服务器配置

func DefaultServerConfig 

func DefaultServerConfig() *ServerConfig

DefaultServerConfig 默认配置

type TrafficGRPCServer 

type TrafficGRPCServer struct {
	pb.UnimplementedTrafficServiceServer
	// contains filtered or unexported fields
}

TrafficGRPCServer 流量 gRPC 服务实现

func NewTrafficGRPCServer 

func NewTrafficGRPCServer() *TrafficGRPCServer

NewTrafficGRPCServer 创建流量 gRPC 服务

func (*TrafficGRPCServer) OnlineStream 

func (s *TrafficGRPCServer) OnlineStream(stream pb.TrafficService_OnlineStreamServer) error

OnlineStream 双向流：实时在线状态

func (*TrafficGRPCServer) ReportOnline 

func (s *TrafficGRPCServer) ReportOnline(ctx context.Context, req *pb.OnlineReportRequest) (*pb.StatusResponse, error)

ReportOnline 上报在线状态

func (*TrafficGRPCServer) ReportTraffic 

func (s *TrafficGRPCServer) ReportTraffic(ctx context.Context, req *pb.TrafficReportRequest) (*pb.TrafficReportResponse, error)

ReportTraffic 批量上报流量

func (*TrafficGRPCServer) TrafficStream 

func (s *TrafficGRPCServer) TrafficStream(stream pb.TrafficService_TrafficStreamServer) error

TrafficStream 双向流：实时流量上报

type UserGRPCServer 

type UserGRPCServer struct {
	pb.UnimplementedUserServiceServer
	// contains filtered or unexported fields
}

UserGRPCServer 用户 gRPC 服务实现

func NewUserGRPCServer 

func NewUserGRPCServer() *UserGRPCServer

NewUserGRPCServer 创建用户 gRPC 服务

func (*UserGRPCServer) GetUsers 

func (s *UserGRPCServer) GetUsers(ctx context.Context, req *pb.UserListRequest) (*pb.UserListResponse, error)

GetUsers 获取用户列表

func (*UserGRPCServer) UserChanges 

func (s *UserGRPCServer) UserChanges(stream pb.UserService_UserChangesServer) error

UserChanges 双向流：用户变更实时推送

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.