cluster

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Oct 1, 2026 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Overview

Package cluster provides cluster management functionality for the STX Agent system. cluster 包提供 STX Agent 系统的集群管理功能。

Package cluster provides cluster management functionality for the STX Agent system.

Package cluster provides cluster management functionality for the STX Agent system. cluster 包提供 STX Agent 系统的集群管理功能。

Index

Constants

View Source 
const (
	ErrCodeClusterNotFound       = 3001
	ErrCodeClusterNameDuplicate  = 3002
	ErrCodeClusterHasRunningTask = 3003
	ErrCodeNodeAgentNotInstalled = 3004
	ErrCodeNodeNotFound          = 3005
	ErrCodeNodeAlreadyExists     = 3006
)

Error codes for cluster management operations.

View Source 
const (
	PrecheckStatusPassed  = "passed"
	PrecheckStatusFailed  = "failed"
	PrecheckStatusSkipped = "skipped"
)

PrecheckStatus constants 预检查状态常量

Variables

View Source 
var (
	// ErrClusterNotFound indicates the requested cluster does not exist.
	ErrClusterNotFound = errors.New("cluster: cluster not found")
	// ErrClusterNameDuplicate indicates a cluster with the same name already exists.
	ErrClusterNameDuplicate = errors.New("cluster: cluster name already exists")
	// ErrClusterNameEmpty indicates the cluster name is empty.
	ErrClusterNameEmpty = errors.New("cluster: cluster name cannot be empty")
	// ErrClusterHasRunningTask indicates the cluster has running tasks and cannot be deleted.
	ErrClusterHasRunningTask = errors.New("cluster: cluster has running tasks and cannot be deleted")
	// ErrNodeNotFound indicates the requested cluster node does not exist.
	ErrNodeNotFound = errors.New("cluster: node not found")
	// ErrNodeAlreadyExists indicates the host already has a node with the same role in the cluster.
	// ErrNodeAlreadyExists 表示主机在集群中已有相同角色的节点。
	ErrNodeAlreadyExists = errors.New("cluster: host already has a node with this role in the cluster / 主机在此集群中已有相同角色的节点")
	// ErrNodeAgentNotInstalled indicates the host's agent is not installed.
	ErrNodeAgentNotInstalled = errors.New("cluster: host agent is not installed")
	// ErrInvalidDeploymentMode indicates an invalid deployment mode was specified.
	ErrInvalidDeploymentMode = errors.New("cluster: invalid deployment mode")
	// ErrInvalidNodeRole indicates an invalid node role was specified.
	ErrInvalidNodeRole = errors.New("cluster: invalid node role")
	// ErrInvalidHazelcastPort indicates an invalid or missing hazelcast port.
	// ErrInvalidHazelcastPort 表示无效或缺失的 Hazelcast 端口。
	ErrInvalidHazelcastPort = errors.New("cluster: hazelcast port is required and must be valid")
	// ErrInvalidAPIPort indicates an invalid API port.
	// ErrInvalidAPIPort 表示无效的 API 端口。
	ErrInvalidAPIPort = errors.New("cluster: api port must be valid when provided")
	// ErrInvalidWorkerPort indicates an invalid worker port.
	// ErrInvalidWorkerPort 表示无效的 worker 端口。
	ErrInvalidWorkerPort = errors.New("cluster: worker port must be valid when provided")
	// ErrNodeBatchEntriesRequired indicates batch add request has no entries.
	// ErrNodeBatchEntriesRequired 表示批量加节点请求未提供任何条目。
	ErrNodeBatchEntriesRequired = errors.New("cluster: at least one node entry is required")
	// ErrInvalidNodeJVMOverride indicates node-level JVM override is invalid.
	// ErrInvalidNodeJVMOverride 表示节点级 JVM override 非法。
	ErrInvalidNodeJVMOverride = errors.New("cluster: jvm override must be greater than 0 when provided")
	// ErrPrecheckFailed indicates the node precheck failed.
	// ErrPrecheckFailed 表示节点预检查失败。
	ErrPrecheckFailed = errors.New("cluster: node precheck failed")
	// ErrSeparatedClusterRequiresMasterAndWorker indicates separated mode cluster must have at least one master and one worker node.
	// ErrSeparatedClusterRequiresMasterAndWorker 表示分离模式集群必须至少包含一个 master 节点和一个 worker 节点。
	ErrSeparatedClusterRequiresMasterAndWorker = errors.New("cluster: separated deployment mode requires at least one master and one worker node / 分离模式集群必须至少包含一个 master 节点和一个 worker 节点")
)

Error definitions for cluster management operations.

View Source 
var DefaultPorts = struct {
	MasterHazelcast int // Master hazelcast port / Master Hazelcast 端口
	MasterAPI       int // Master REST API port / Master REST API 端口
	WorkerHazelcast int // Worker hazelcast port / Worker Hazelcast 端口
}{
	MasterHazelcast: 5801,
	MasterAPI:       8080,
	WorkerHazelcast: 5802,
}

DefaultPorts defines default port values for different node roles DefaultPorts 定义不同节点角色的默认端口值

Functions

func GetDefaultPorts 

func GetDefaultPorts(role NodeRole, deploymentMode DeploymentMode) (hazelcastPort, apiPort, workerPort int)

GetDefaultPorts returns default ports based on node role and deployment mode GetDefaultPorts 根据节点角色和部署模式返回默认端口

Types

type AddNodeEntryRequest 

type AddNodeEntryRequest struct {
	Role          NodeRole       `json:"role" binding:"required"`
	InstallDir    string         `json:"install_dir"`
	HazelcastPort int            `json:"hazelcast_port"`
	APIPort       int            `json:"api_port"`
	WorkerPort    int            `json:"worker_port"`
	Overrides     *NodeOverrides `json:"overrides,omitempty"`
}

AddNodeEntryRequest represents one logical node entry under the same host. AddNodeEntryRequest 表示同一主机下的一个逻辑节点条目。

type AddNodeRequest 

type AddNodeRequest struct {
	HostID     uint     `json:"host_id" binding:"required"`
	Role       NodeRole `json:"role" binding:"required"`
	InstallDir string   `json:"install_dir"` // SeaTunnel installation directory / SeaTunnel 安装目录

	// Port configuration based on node role / 基于节点角色的端口配置
	// Master: hazelcast_port (default 5801) + api_port (default 8080, optional)
	// Worker: hazelcast_port (default 5802)
	// Hybrid: hazelcast_port (5801) + worker_port (5802)
	HazelcastPort int            `json:"hazelcast_port"`      // Hazelcast cluster port / Hazelcast 集群端口
	APIPort       int            `json:"api_port"`            // REST API port (Master only, optional) / REST API 端口（仅 Master，可选）
	WorkerPort    int            `json:"worker_port"`         // Worker hazelcast port (Hybrid only) / Worker Hazelcast 端口（仅混合模式）
	Overrides     *NodeOverrides `json:"overrides,omitempty"` // Node-level JSON overrides / 节点级 JSON 覆盖配置

	// Whether to skip precheck / 是否跳过预检查
	SkipPrecheck bool `json:"skip_precheck"`
}

AddNodeRequest represents a request to add a node to a cluster. 添加节点请求，包含安装目录和端口配置

type AddNodeResponse 

type AddNodeResponse struct {
	ErrorMsg string    `json:"error_msg"`
	Data     *NodeInfo `json:"data"`
}

AddNodeResponse represents the response for adding a node to a cluster. AddNodeResponse 表示向集群添加节点的响应。

type AddNodesRequest 

type AddNodesRequest struct {
	HostID       uint                  `json:"host_id" binding:"required"`
	InstallDir   string                `json:"install_dir"`
	Entries      []AddNodeEntryRequest `json:"entries" binding:"required"`
	SkipPrecheck bool                  `json:"skip_precheck"`
}

AddNodesRequest represents a same-host multi-entry add-node request. AddNodesRequest 表示同一主机多条目加节点请求。

type AddNodesResponse 

type AddNodesResponse struct {
	ErrorMsg string      `json:"error_msg"`
	Data     []*NodeInfo `json:"data"`
}

AddNodesResponse represents the response for batch-adding nodes to a cluster. AddNodesResponse 表示批量向集群添加节点的响应。

type AgentCommandSender 

type AgentCommandSender interface {
	// SendCommand sends a command to an agent and returns the result.
	// SendCommand 向 Agent 发送命令并返回结果。
	SendCommand(ctx context.Context, agentID string, commandType string, params map[string]string) (bool, string, error)
}

AgentCommandSender is an interface for sending commands to agents. AgentCommandSender 是向 Agent 发送命令的接口。 This interface will be implemented by the Agent Manager in Phase 4. 此接口将在第 4 阶段由 Agent Manager 实现。

type ApplyRuntimeStorageRequest added in v1.0.0 

type ApplyRuntimeStorageRequest struct {
	Enabled     bool   `json:"enabled"`
	StorageType string `json:"storage_type"`
	Namespace   string `json:"namespace"`
	Endpoint    string `json:"endpoint"`
	Bucket      string `json:"bucket"`
	AccessKey   string `json:"access_key"`
	SecretKey   string `json:"secret_key"`
	// HDFS 单 NameNode / HA / Kerberos / hdfs-site
	// HDFS single NameNode / HA / Kerberos / hdfs-site
	HDFSNameNodeHost          string `json:"hdfs_namenode_host"`
	HDFSNameNodePort          int    `json:"hdfs_namenode_port"`
	HDFSHAEnabled             bool   `json:"hdfs_ha_enabled"`
	HDFSNameServices          string `json:"hdfs_name_services"`
	HDFSHANamenodes           string `json:"hdfs_ha_namenodes"`
	HDFSNamenodeRPCAddress1   string `json:"hdfs_namenode_rpc_address_1"`
	HDFSNamenodeRPCAddress2   string `json:"hdfs_namenode_rpc_address_2"`
	HDFSFailoverProxyProvider string `json:"hdfs_failover_proxy_provider"`
	KerberosPrincipal         string `json:"kerberos_principal"`
	KerberosKeytabFilePath    string `json:"kerberos_keytab_file_path"`
	HdfsSitePath              string `json:"hdfs_site_path"`
	// DisableCache 对应 disable.cache；nil 不写入。
	// DisableCache maps to disable.cache; nil omits the key.
	DisableCache *bool `json:"disable_cache"`
	// S3 凭证提供方，对应 fs.s3a.aws.credentials.provider。
	// S3 credentials provider, mapped to fs.s3a.aws.credentials.provider.
	S3CredentialsProvider string `json:"s3_credentials_provider"`
}

ApplyRuntimeStorageRequest 是可视化存储表单提交的草稿。 ApplyRuntimeStorageRequest is the draft submitted by the visual storage form.

type ApplyRuntimeStorageResponse added in v1.0.0 

type ApplyRuntimeStorageResponse struct {
	ErrorMsg string                     `json:"error_msg"`
	Data     *ApplyRuntimeStorageResult `json:"data"`
}

ApplyRuntimeStorageResponse 表示保存运行时存储配置的响应。 ApplyRuntimeStorageResponse is the response for applying runtime storage settings.

type ApplyRuntimeStorageResult added in v1.0.0 

type ApplyRuntimeStorageResult struct {
	Saved           bool                                         `json:"saved"`
	RestartRequired bool                                         `json:"restart_required"`
	Message         string                                       `json:"message"`
	Versions        []ApplyRuntimeStorageVersion                 `json:"versions,omitempty"`
	Validation      *installerapp.RuntimeStorageValidationResult `json:"validation,omitempty"`
}

ApplyRuntimeStorageResult 是保存存储配置的结果。 ApplyRuntimeStorageResult is the result of saving runtime storage settings.

type ApplyRuntimeStorageVersion added in v1.0.0 

type ApplyRuntimeStorageVersion struct {
	ConfigType string `json:"config_type"`
	ConfigID   uint   `json:"config_id"`
	Version    int    `json:"version"`
}

ApplyRuntimeStorageVersion 记录本次写回的配置版本。 ApplyRuntimeStorageVersion records one config version written by this apply.

type CleanupRuntimeStorageResponse 

type CleanupRuntimeStorageResponse struct {
	ErrorMsg string                       `json:"error_msg"`
	Data     *RuntimeStorageCleanupResult `json:"data"`
}

CleanupRuntimeStorageResponse represents runtime storage cleanup response. CleanupRuntimeStorageResponse 表示运行时存储清理响应。

type Cluster 

type Cluster struct {
	ID             uint           `json:"id" gorm:"primaryKey;autoIncrement"`
	Name           string         `json:"name" gorm:"size:100;uniqueIndex;not null"`
	Description    string         `json:"description" gorm:"type:text"`
	DeploymentMode DeploymentMode `json:"deployment_mode" gorm:"size:20;not null"`
	Version        string         `json:"version" gorm:"size:20"`
	Status         ClusterStatus  `json:"status" gorm:"size:20;default:created;index"`
	InstallDir     string         `json:"install_dir" gorm:"size:255"`
	Config         ClusterConfig  `json:"config" gorm:"type:json"`
	CreatedAt      time.Time      `json:"created_at" gorm:"autoCreateTime"`
	UpdatedAt      time.Time      `json:"updated_at" gorm:"autoUpdateTime"`
	CreatedBy      uint           `json:"created_by"`
	Nodes          []ClusterNode  `json:"nodes" gorm:"foreignKey:ClusterID"`
}

Cluster represents a SeaTunnel cluster consisting of multiple nodes.

func (Cluster) TableName 

func (Cluster) TableName() string

TableName specifies the table name for the Cluster model.

func (*Cluster) ToClusterInfo 

func (c *Cluster) ToClusterInfo() *ClusterInfo

ToClusterInfo converts a Cluster to ClusterInfo (OnlineNodes and HealthStatus are set by caller).

type ClusterConfig 

type ClusterConfig map[string]interface{}

ClusterConfig represents the JSON configuration for a cluster.

func (ClusterConfig) GetJVMConfig 

func (c ClusterConfig) GetJVMConfig() *JVMConfig

GetJVMConfig returns cluster-level JVM defaults from cluster config. GetJVMConfig 返回 cluster config 中的 JVM 默认值。

func (ClusterConfig) GetPortConfig 

func (c ClusterConfig) GetPortConfig() *ClusterPortConfig

GetPortConfig 返回 cluster config 中的端口默认值。 GetPortConfig returns cluster-level port defaults from cluster config.

func (ClusterConfig) GetSTXJavaProxyConfig added in v1.0.0 

func (c ClusterConfig) GetSTXJavaProxyConfig() *STXJavaProxyConfig

GetSTXJavaProxyConfig returns cluster-level stx-java-proxy configuration. GetSTXJavaProxyConfig 返回集群级 stx-java-proxy 配置。

func (*ClusterConfig) Scan 

func (c *ClusterConfig) Scan(value interface{}) error

Scan implements the sql.Scanner interface for database retrieval.

func (ClusterConfig) Value 

func (c ClusterConfig) Value() (driver.Value, error)

Value implements the driver.Valuer interface for database storage.

type ClusterFilter 

type ClusterFilter struct {
	Name           string         `json:"name"`
	Status         ClusterStatus  `json:"status"`
	DeploymentMode DeploymentMode `json:"deployment_mode"`
	Page           int            `json:"page"`
	PageSize       int            `json:"page_size"`
}

ClusterFilter represents filter criteria for querying clusters.

type ClusterInfo 

type ClusterInfo struct {
	ID             uint           `json:"id"`
	Name           string         `json:"name"`
	Description    string         `json:"description"`
	DeploymentMode DeploymentMode `json:"deployment_mode"`
	Version        string         `json:"version"`
	Status         ClusterStatus  `json:"status"`
	InstallDir     string         `json:"install_dir"`
	Config         ClusterConfig  `json:"config"`
	NodeCount      int            `json:"node_count"`
	OnlineNodes    int            `json:"online_nodes"`  // number of nodes whose host is online / 主机在线的节点数
	HealthStatus   string         `json:"health_status"` // healthy, unhealthy, unknown / 健康状态
	CreatedAt      time.Time      `json:"created_at"`
	UpdatedAt      time.Time      `json:"updated_at"`
}

ClusterInfo represents cluster information for API responses.

type ClusterNode 

type ClusterNode struct {
	ID            uint          `json:"id" gorm:"primaryKey;autoIncrement"`
	ClusterID     uint          `json:"cluster_id" gorm:"index;not null"`
	HostID        uint          `json:"host_id" gorm:"index;not null"`
	Role          NodeRole      `json:"role" gorm:"size:20;not null"`
	InstallDir    string        `json:"install_dir" gorm:"size:255"`           // SeaTunnel installation directory on this node / 此节点上的 SeaTunnel 安装目录
	HazelcastPort int           `json:"hazelcast_port"`                        // Hazelcast cluster port / Hazelcast 集群端口
	APIPort       int           `json:"api_port"`                              // REST API port (Master only) / REST API 端口（仅 Master）
	WorkerPort    int           `json:"worker_port"`                           // Worker hazelcast port (Hybrid only) / Worker Hazelcast 端口（仅混合模式）
	Overrides     NodeOverrides `json:"overrides" gorm:"type:json"`            // Node-level JSON overrides / 节点级 JSON 覆盖配置
	Status        NodeStatus    `json:"status" gorm:"size:20;default:pending"` // Unified status field: pending, installing, running, stopped, error / 统一状态字段
	ProcessPID    int           `json:"process_pid" gorm:"column:process_pid"` // SeaTunnel process PID / SeaTunnel 进程 PID
	LastEventAt   *time.Time    `json:"last_event_at"`                         // 最后事件时间 / Last event time
	CreatedAt     time.Time     `json:"created_at" gorm:"autoCreateTime"`
	UpdatedAt     time.Time     `json:"updated_at" gorm:"autoUpdateTime"`
}

ClusterNode represents a node within a SeaTunnel cluster. 集群节点，每个节点可以有独立的安装目录和端口配置

func (*ClusterNode) ResolveJVM 

func (n *ClusterNode) ResolveJVM(clusterConfig ClusterConfig) *JVMConfig

ResolveJVM resolves node-level JVM overrides on top of cluster defaults. ResolveJVM 基于集群默认值解析节点级 JVM overrides。

func (ClusterNode) TableName 

func (ClusterNode) TableName() string

TableName specifies the table name for the ClusterNode model.

type ClusterOperationResponse 

type ClusterOperationResponse struct {
	ErrorMsg string           `json:"error_msg"`
	Data     *OperationResult `json:"data"`
}

ClusterOperationResponse represents the response for cluster operations (start/stop/restart). ClusterOperationResponse 表示集群操作（启动/停止/重启）的响应。

type ClusterPortConfig 

type ClusterPortConfig struct {
	MasterHazelcastPort int `json:"master_hazelcast_port,omitempty"`
	MasterAPIPort       int `json:"master_api_port,omitempty"`
	WorkerPort          int `json:"worker_port,omitempty"`
	// JavaProxyPort is the managed stx-java-proxy listen port (default 18080).
	// JavaProxyPort 是托管 stx-java-proxy 监听端口（默认 18080）。
	JavaProxyPort int `json:"java_proxy_port,omitempty"`
}

ClusterPortConfig 表示 cluster config 中的集群级端口默认值。 ClusterPortConfig represents cluster-level port defaults from cluster config.

func (*ClusterPortConfig) HasValues 

func (c *ClusterPortConfig) HasValues() bool

HasValues 返回端口配置是否包含任意显式值。 HasValues returns whether the port config contains any explicit value.

type ClusterStatus 

type ClusterStatus string

ClusterStatus represents the current status of a cluster. 

const (
	// ClusterStatusCreated indicates the cluster has been created but not deployed.
	ClusterStatusCreated ClusterStatus = "created"
	// ClusterStatusDeploying indicates the cluster is being deployed.
	ClusterStatusDeploying ClusterStatus = "deploying"
	// ClusterStatusRunning indicates the cluster is running normally.
	ClusterStatusRunning ClusterStatus = "running"
	// ClusterStatusStopped indicates the cluster has been stopped.
	ClusterStatusStopped ClusterStatus = "stopped"
	// ClusterStatusError indicates the cluster is in an error state.
	ClusterStatusError ClusterStatus = "error"
)

type ClusterStatusInfo 

type ClusterStatusInfo struct {
	ClusterID    uint              `json:"cluster_id"`
	ClusterName  string            `json:"cluster_name"`
	Status       ClusterStatus     `json:"status"`
	HealthStatus HealthStatus      `json:"health_status"`
	TotalNodes   int               `json:"total_nodes"`
	OnlineNodes  int               `json:"online_nodes"`
	OfflineNodes int               `json:"offline_nodes"`
	Nodes        []*NodeStatusInfo `json:"nodes"`
}

ClusterStatusInfo represents detailed cluster status information. ClusterStatusInfo 表示详细的集群状态信息。

type ConfigAgentClient 

type ConfigAgentClient interface {
	PullConfig(ctx context.Context, hostID uint, installDir string, configType appconfig.ConfigType) (string, error)
	PushConfig(ctx context.Context, hostID uint, installDir string, configType appconfig.ConfigType, content string) error
}

ConfigAgentClient reads/writes node config files via Agent. ConfigAgentClient 通过 Agent 读写节点配置文件。

type CreateClusterRequest 

type CreateClusterRequest struct {
	Name           string         `json:"name" binding:"required,max=100"`
	Description    string         `json:"description"`
	DeploymentMode DeploymentMode `json:"deployment_mode" binding:"required"`
	Version        string         `json:"version" binding:"required"`
	InstallDir     string         `json:"install_dir"`
	Config         ClusterConfig  `json:"config"`
	// Nodes to auto-create from discovery (optional)
	// 从发现自动创建的节点（可选）
	Nodes []CreateNodeFromDiscovery `json:"nodes,omitempty"`
}

CreateClusterRequest represents a request to create a new cluster. CreateClusterRequest 表示创建新集群的请求。

type CreateClusterResponse 

type CreateClusterResponse struct {
	ErrorMsg string       `json:"error_msg"`
	Data     *ClusterInfo `json:"data"`
}

CreateClusterResponse represents the response for creating a cluster. CreateClusterResponse 表示创建集群的响应。

type CreateNodeFromDiscovery 

type CreateNodeFromDiscovery struct {
	HostID        uint   `json:"host_id"`                  // Host ID / 主机 ID
	InstallDir    string `json:"install_dir"`              // SeaTunnel installation directory / SeaTunnel 安装目录
	Role          string `json:"role"`                     // Node role: master, worker, hybrid / 节点角色
	PID           int    `json:"pid,omitempty"`            // Process ID / 进程 ID
	HazelcastPort int    `json:"hazelcast_port,omitempty"` // Hazelcast cluster port (optional) / Hazelcast 集群端口（可选）
	APIPort       int    `json:"api_port,omitempty"`       // REST API port (optional) / REST API 端口（可选）
}

CreateNodeFromDiscovery represents a node to be created from process discovery. CreateNodeFromDiscovery 表示从进程发现创建的节点。

type DeleteClusterResponse 

type DeleteClusterResponse struct {
	ErrorMsg string `json:"error_msg"`
	Data     any    `json:"data"`
}

DeleteClusterResponse represents the response for deleting a cluster. DeleteClusterResponse 表示删除集群的响应。

type DeploymentMode 

type DeploymentMode string

DeploymentMode represents the deployment mode of a SeaTunnel cluster. 

const (
	// DeploymentModeHybrid indicates a hybrid deployment where master and worker run on the same nodes.
	DeploymentModeHybrid DeploymentMode = "hybrid"
	// DeploymentModeSeparated indicates a separated deployment where master and worker run on different nodes.
	DeploymentModeSeparated DeploymentMode = "separated"
)

type GetClusterResponse 

type GetClusterResponse struct {
	ErrorMsg string       `json:"error_msg"`
	Data     *ClusterInfo `json:"data"`
}

GetClusterResponse represents the response for getting a cluster. GetClusterResponse 表示获取集群详情的响应。

type GetClusterStatusResponse 

type GetClusterStatusResponse struct {
	ErrorMsg string             `json:"error_msg"`
	Data     *ClusterStatusInfo `json:"data"`
}

GetClusterStatusResponse represents the response for getting cluster status. GetClusterStatusResponse 表示获取集群状态的响应。

type GetNodeLogsRequest 

type GetNodeLogsRequest struct {
	Lines  int    `json:"lines" form:"lines"`   // Number of lines / 行数
	Mode   string `json:"mode" form:"mode"`     // "tail" (default), "head", "all" / 模式
	Filter string `json:"filter" form:"filter"` // Filter pattern / 过滤模式
	Date   string `json:"date" form:"date"`     // Date for rolling logs / 滚动日志日期
}

GetNodeLogsRequest represents the request for getting node logs. GetNodeLogsRequest 表示获取节点日志的请求。

type GetNodeLogsResponse 

type GetNodeLogsResponse struct {
	ErrorMsg string `json:"error_msg"`
	Data     *struct {
		Logs string `json:"logs"`
	} `json:"data"`
}

GetNodeLogsResponse represents the response for getting node logs. GetNodeLogsResponse 表示获取节点日志的响应。

type GetNodesResponse 

type GetNodesResponse struct {
	ErrorMsg string      `json:"error_msg"`
	Data     []*NodeInfo `json:"data"`
}

GetNodesResponse represents the response for getting cluster nodes. GetNodesResponse 表示获取集群节点列表的响应。

type GetRuntimeStorageResponse 

type GetRuntimeStorageResponse struct {
	ErrorMsg string                 `json:"error_msg"`
	Data     *RuntimeStorageDetails `json:"data"`
}

GetRuntimeStorageResponse 表示运行时存储详情响应。

type Handler 

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

Handler provides HTTP handlers for cluster management operations. Handler 提供集群管理操作的 HTTP 处理器。

func NewHandler 

func NewHandler(service *Service, auditRepo *audit.Repository) *Handler

NewHandler creates a new Handler instance. NewHandler 创建一个新的 Handler 实例。 auditRepo may be nil; audit logging is skipped when nil.

func (*Handler) AddNode 

func (h *Handler) AddNode(c *gin.Context)

AddNode handles POST /api/v1/clusters/:id/nodes - adds a node to a cluster. AddNode 处理 POST /api/v1/clusters/:id/nodes - 向集群添加节点。 @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param request body AddNodeRequest true "添加节点请求" @Success 200 {object} AddNodeResponse @Router /api/v1/clusters/{id}/nodes [post]

func (*Handler) AddNodes 

func (h *Handler) AddNodes(c *gin.Context)

AddNodes handles POST /api/v1/clusters/:id/nodes/batch - adds multiple logical nodes for one host atomically. AddNodes 处理 POST /api/v1/clusters/:id/nodes/batch - 为同一主机原子添加多个逻辑节点。

func (*Handler) ApplyRuntimeStorage added in v1.0.0 

func (h *Handler) ApplyRuntimeStorage(c *gin.Context)

ApplyRuntimeStorage 保存可视化存储设置：先真实读写测试，通过后生成配置版本。 ApplyRuntimeStorage saves visual storage settings: probe first, then create a config version.

func (*Handler) CleanupIMAPStorage 

func (h *Handler) CleanupIMAPStorage(c *gin.Context)

CleanupIMAPStorage handles POST /api/v1/clusters/:id/runtime-storage/imap/cleanup. CleanupIMAPStorage 处理 POST /api/v1/clusters/:id/runtime-storage/imap/cleanup。

func (*Handler) CreateCluster 

func (h *Handler) CreateCluster(c *gin.Context)

CreateCluster handles POST /api/v1/clusters - creates a new cluster. CreateCluster 处理 POST /api/v1/clusters - 创建新集群。 @Tags clusters @Accept json @Produce json @Param request body CreateClusterRequest true "创建集群请求" @Success 200 {object} CreateClusterResponse @Router /api/v1/clusters [post]

func (*Handler) DeleteCluster 

func (h *Handler) DeleteCluster(c *gin.Context)

DeleteCluster handles DELETE /api/v1/clusters/:id - deletes a cluster. DeleteCluster 处理 DELETE /api/v1/clusters/:id - 删除集群。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} DeleteClusterResponse @Router /api/v1/clusters/{id} [delete]

func (*Handler) GetCluster 

func (h *Handler) GetCluster(c *gin.Context)

GetCluster handles GET /api/v1/clusters/:id - gets a cluster by ID. GetCluster 处理 GET /api/v1/clusters/:id - 根据 ID 获取集群详情。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} GetClusterResponse @Router /api/v1/clusters/{id} [get]

func (*Handler) GetClusterStatus 

func (h *Handler) GetClusterStatus(c *gin.Context)

GetClusterStatus handles GET /api/v1/clusters/:id/status - gets the status of a cluster. GetClusterStatus 处理 GET /api/v1/clusters/:id/status - 获取集群状态。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} GetClusterStatusResponse @Router /api/v1/clusters/{id}/status [get]

func (*Handler) GetNodeLogs 

func (h *Handler) GetNodeLogs(c *gin.Context)

GetNodeLogs handles GET /api/v1/clusters/:id/nodes/:nodeId/logs - gets node logs. GetNodeLogs 处理 GET /api/v1/clusters/:id/nodes/:nodeId/logs - 获取节点日志。 Query parameters: - lines: number of lines (default: 100) / 行数（默认：100） - mode: "tail" (default), "head", "all" / 模式 - filter: grep pattern / 过滤模式 - date: date for rolling logs (e.g., "2025-11-12-1") / 滚动日志日期

func (*Handler) GetNodes 

func (h *Handler) GetNodes(c *gin.Context)

GetNodes handles GET /api/v1/clusters/:id/nodes - gets all nodes for a cluster. GetNodes 处理 GET /api/v1/clusters/:id/nodes - 获取集群的所有节点。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} GetNodesResponse @Router /api/v1/clusters/{id}/nodes [get]

func (*Handler) GetRuntimeStorage 

func (h *Handler) GetRuntimeStorage(c *gin.Context)

GetRuntimeStorage handles GET /api/v1/clusters/:id/runtime-storage. GetRuntimeStorage 处理 GET /api/v1/clusters/:id/runtime-storage。

func (*Handler) GetSTXJavaProxyStatus 

func (h *Handler) GetSTXJavaProxyStatus(c *gin.Context)

GetSTXJavaProxyStatus handles GET /api/v1/clusters/:id/stx-java-proxy/status.

func (*Handler) InspectCheckpointRuntimeStorage 

func (h *Handler) InspectCheckpointRuntimeStorage(c *gin.Context)

InspectCheckpointRuntimeStorage handles POST /api/v1/clusters/:id/runtime-storage/checkpoint/inspect.

func (*Handler) InspectIMAPRuntimeStorage 

func (h *Handler) InspectIMAPRuntimeStorage(c *gin.Context)

InspectIMAPRuntimeStorage handles POST /api/v1/clusters/:id/runtime-storage/imap/inspect.

func (*Handler) ListClusters 

func (h *Handler) ListClusters(c *gin.Context)

ListClusters handles GET /api/v1/clusters - lists clusters with filtering and pagination. ListClusters 处理 GET /api/v1/clusters - 获取集群列表（支持过滤和分页）。 @Tags clusters @Param request query ListClustersRequest true "查询参数" @Produce json @Success 200 {object} ListClustersResponse @Router /api/v1/clusters [get]

func (*Handler) ListRuntimeStorage 

func (h *Handler) ListRuntimeStorage(c *gin.Context)

ListRuntimeStorage handles POST /api/v1/clusters/:id/runtime-storage/:kind/list.

func (*Handler) PrecheckNode 

func (h *Handler) PrecheckNode(c *gin.Context)

PrecheckNode handles POST /api/v1/clusters/:id/nodes/precheck - prechecks a node before adding. PrecheckNode 处理 POST /api/v1/clusters/:id/nodes/precheck - 添加节点前的预检查。 @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param request body PrecheckRequest true "预检查请求" @Success 200 {object} PrecheckNodeResponse @Router /api/v1/clusters/{id}/nodes/precheck [post]

func (*Handler) PreviewRuntimeStorage 

func (h *Handler) PreviewRuntimeStorage(c *gin.Context)

PreviewRuntimeStorage handles POST /api/v1/clusters/:id/runtime-storage/:kind/preview.

func (*Handler) PreviewSTXJavaProxyServiceLog 

func (h *Handler) PreviewSTXJavaProxyServiceLog(c *gin.Context)

PreviewSTXJavaProxyServiceLog handles GET /api/v1/clusters/:id/stx-java-proxy/logs.

func (*Handler) ProxyWebUI 

func (h *Handler) ProxyWebUI(c *gin.Context)

ProxyWebUI handles proxying the SeaTunnel Web UI through the control plane. ProxyWebUI 处理通过控制平面反向代理 SeaTunnel Web UI。

func (*Handler) RemoveNode 

func (h *Handler) RemoveNode(c *gin.Context)

RemoveNode handles DELETE /api/v1/clusters/:id/nodes/:nodeId - removes a node from a cluster. RemoveNode 处理 DELETE /api/v1/clusters/:id/nodes/:nodeId - 从集群移除节点。 @Tags clusters @Produce json @Param id path int true "集群ID" @Param nodeId path int true "节点ID" @Success 200 {object} RemoveNodeResponse @Router /api/v1/clusters/{id}/nodes/{nodeId} [delete]

func (*Handler) RestartCluster 

func (h *Handler) RestartCluster(c *gin.Context)

RestartCluster handles POST /api/v1/clusters/:id/restart - restarts a cluster. RestartCluster 处理 POST /api/v1/clusters/:id/restart - 重启集群。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} ClusterOperationResponse @Router /api/v1/clusters/{id}/restart [post]

func (*Handler) RestartNode 

func (h *Handler) RestartNode(c *gin.Context)

RestartNode handles POST /api/v1/clusters/:id/nodes/:nodeId/restart - restarts a node. RestartNode 处理 POST /api/v1/clusters/:id/nodes/:nodeId/restart - 重启节点。

func (*Handler) RestartSTXJavaProxy 

func (h *Handler) RestartSTXJavaProxy(c *gin.Context)

RestartSTXJavaProxy handles POST /api/v1/clusters/:id/stx-java-proxy/restart.

func (*Handler) SetOnOperationExecuted 

func (h *Handler) SetOnOperationExecuted(fn func(context.Context, *OperationEvent) error)

SetOnOperationExecuted sets an optional hook invoked after a cluster operation HTTP request succeeds. SetOnOperationExecuted 设置集群操作 HTTP 请求成功后的可选回调。

func (*Handler) StartCluster 

func (h *Handler) StartCluster(c *gin.Context)

StartCluster handles POST /api/v1/clusters/:id/start - starts a cluster. StartCluster 处理 POST /api/v1/clusters/:id/start - 启动集群。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} ClusterOperationResponse @Router /api/v1/clusters/{id}/start [post]

func (*Handler) StartNode 

func (h *Handler) StartNode(c *gin.Context)

StartNode handles POST /api/v1/clusters/:id/nodes/:nodeId/start - starts a node. StartNode 处理 POST /api/v1/clusters/:id/nodes/:nodeId/start - 启动节点。

func (*Handler) StartSTXJavaProxy 

func (h *Handler) StartSTXJavaProxy(c *gin.Context)

StartSTXJavaProxy handles POST /api/v1/clusters/:id/stx-java-proxy/start.

func (*Handler) StopCluster 

func (h *Handler) StopCluster(c *gin.Context)

StopCluster handles POST /api/v1/clusters/:id/stop - stops a cluster. StopCluster 处理 POST /api/v1/clusters/:id/stop - 停止集群。 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} ClusterOperationResponse @Router /api/v1/clusters/{id}/stop [post]

func (*Handler) StopNode 

func (h *Handler) StopNode(c *gin.Context)

StopNode handles POST /api/v1/clusters/:id/nodes/:nodeId/stop - stops a node. StopNode 处理 POST /api/v1/clusters/:id/nodes/:nodeId/stop - 停止节点。

func (*Handler) StopSTXJavaProxy 

func (h *Handler) StopSTXJavaProxy(c *gin.Context)

StopSTXJavaProxy handles POST /api/v1/clusters/:id/stx-java-proxy/stop.

func (*Handler) SwitchJobLogMode added in v1.0.0 

func (h *Handler) SwitchJobLogMode(c *gin.Context)

SwitchJobLogMode 一键切换集群作业日志输出模式 (单 Job 独立日志模式 vs 共享混合日志模式)

func (*Handler) UpdateCluster 

func (h *Handler) UpdateCluster(c *gin.Context)

UpdateCluster handles PUT /api/v1/clusters/:id - updates an existing cluster. UpdateCluster 处理 PUT /api/v1/clusters/:id - 更新现有集群。 @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param request body UpdateClusterRequest true "更新集群请求" @Success 200 {object} UpdateClusterResponse @Router /api/v1/clusters/{id} [put]

func (*Handler) UpdateNode 

func (h *Handler) UpdateNode(c *gin.Context)

UpdateNode handles PUT /api/v1/clusters/:id/nodes/:nodeId - updates a node in a cluster. UpdateNode 处理 PUT /api/v1/clusters/:id/nodes/:nodeId - 更新集群中的节点。 @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param nodeId path int true "节点ID" @Param request body UpdateNodeRequest true "更新节点请求" @Success 200 {object} AddNodeResponse @Router /api/v1/clusters/{id}/nodes/{nodeId} [put]

func (*Handler) UpdateSTXJavaProxyConfig added in v1.0.0 

func (h *Handler) UpdateSTXJavaProxyConfig(c *gin.Context)

UpdateSTXJavaProxyConfig handles PUT /api/v1/clusters/:id/stx-java-proxy/config.

func (*Handler) ValidateRuntimeStorage 

func (h *Handler) ValidateRuntimeStorage(c *gin.Context)

ValidateRuntimeStorage handles POST /api/v1/clusters/:id/runtime-storage/:kind/validate. ValidateRuntimeStorage 处理 POST /api/v1/clusters/:id/runtime-storage/:kind/validate。

type HealthStatus 

type HealthStatus string

HealthStatus represents the health status of a cluster. HealthStatus 表示集群的健康状态。 

const (
	// HealthStatusHealthy indicates all nodes are online and running.
	// HealthStatusHealthy 表示所有节点都在线且运行正常。
	HealthStatusHealthy HealthStatus = "healthy"
	// HealthStatusUnhealthy indicates one or more nodes are offline or in error state.
	// HealthStatusUnhealthy 表示一个或多个节点离线或处于错误状态。
	HealthStatusUnhealthy HealthStatus = "unhealthy"
	// HealthStatusUnknown indicates the health status cannot be determined.
	// HealthStatusUnknown 表示无法确定健康状态。
	HealthStatusUnknown HealthStatus = "unknown"
)

type HostInfo 

type HostInfo struct {
	ID               uint
	Name             string
	HostType         string
	IPAddress        string
	AgentID          string
	AgentStatus      string
	LastHeartbeat    *time.Time
	ProcessStartedAt *time.Time // when set, online requires heartbeat after this (e.g. API process start)
}

HostInfo represents host information needed by cluster service. HostInfo 表示集群服务所需的主机信息。 This interface decouples cluster from host package to avoid import cycles. 此接口将集群与主机包解耦以避免导入循环。

func (*HostInfo) IsOnline 

func (h *HostInfo) IsOnline(timeout time.Duration) bool

IsOnline checks if the host is online based on heartbeat timeout. When ProcessStartedAt is set, also requires last_heartbeat to be after that time.

type HostProvider 

type HostProvider interface {
	// GetHostByID retrieves host information by ID.
	// GetHostByID 根据 ID 获取主机信息。
	GetHostByID(ctx context.Context, id uint) (*HostInfo, error)
}

HostProvider is an interface for retrieving host information. HostProvider 是获取主机信息的接口。 This interface decouples cluster service from host package. 此接口将集群服务与主机包解耦。

type InspectCheckpointRuntimeStorageResponse 

type InspectCheckpointRuntimeStorageResponse struct {
	ErrorMsg string                                 `json:"error_msg"`
	Data     *RuntimeStorageCheckpointInspectResult `json:"data"`
}

type InspectIMAPRuntimeStorageResponse 

type InspectIMAPRuntimeStorageResponse struct {
	ErrorMsg string                           `json:"error_msg"`
	Data     *RuntimeStorageIMAPInspectResult `json:"data"`
}

type JVMConfig 

type JVMConfig struct {
	HybridHeapSize int `json:"hybrid_heap_size,omitempty"`
	MasterHeapSize int `json:"master_heap_size,omitempty"`
	WorkerHeapSize int `json:"worker_heap_size,omitempty"`
}

JVMConfig represents cluster-level effective JVM configuration. JVMConfig 表示集群级生效后的 JVM 配置。

type ListClustersRequest 

type ListClustersRequest struct {
	Current        int            `json:"current" form:"current" binding:"min=1"`
	Size           int            `json:"size" form:"size" binding:"min=1,max=100"`
	Name           string         `json:"name" form:"name"`
	Status         ClusterStatus  `json:"status" form:"status"`
	DeploymentMode DeploymentMode `json:"deployment_mode" form:"deployment_mode"`
}

ListClustersRequest represents the request for listing clusters. ListClustersRequest 表示获取集群列表的请求。

type ListClustersResponse 

type ListClustersResponse struct {
	ErrorMsg string `json:"error_msg"`
	Data     *struct {
		Total    int64          `json:"total"`
		Clusters []*ClusterInfo `json:"clusters"`
	} `json:"data"`
}

ListClustersResponse represents the response for listing clusters. ListClustersResponse 表示获取集群列表的响应。

type ListRuntimeStorageResponse 

type ListRuntimeStorageResponse struct {
	ErrorMsg string                    `json:"error_msg"`
	Data     *RuntimeStorageListResult `json:"data"`
}

type NodeInfo 

type NodeInfo struct {
	ID            uint          `json:"id"`
	ClusterID     uint          `json:"cluster_id"`
	HostID        uint          `json:"host_id"`
	HostName      string        `json:"host_name"`
	HostIP        string        `json:"host_ip"`
	Role          NodeRole      `json:"role"`
	InstallDir    string        `json:"install_dir"`    // SeaTunnel installation directory / SeaTunnel 安装目录
	HazelcastPort int           `json:"hazelcast_port"` // Hazelcast cluster port / Hazelcast 集群端口
	APIPort       int           `json:"api_port"`       // REST API port (Master only) / REST API 端口（仅 Master）
	WorkerPort    int           `json:"worker_port"`    // Worker hazelcast port (Hybrid only) / Worker Hazelcast 端口（仅混合模式）
	Overrides     NodeOverrides `json:"overrides"`      // Node-level JSON overrides / 节点级 JSON 覆盖配置
	Status        NodeStatus    `json:"status"`         // Unified status: pending, installing, running, stopped, error, offline / 统一状态
	IsOnline      bool          `json:"is_online"`      // Whether host is online; when false, status may be shown as offline / 主机是否在线
	ProcessPID    int           `json:"process_pid"`    // SeaTunnel process PID / SeaTunnel 进程 PID
	CreatedAt     time.Time     `json:"created_at"`
	UpdatedAt     time.Time     `json:"updated_at"`
}

NodeInfo represents node information for API responses. 节点信息，用于 API 响应

type NodeJVMOverrides 

type NodeJVMOverrides struct {
	HybridHeapSize *int `json:"hybrid_heap_size,omitempty"`
	MasterHeapSize *int `json:"master_heap_size,omitempty"`
	WorkerHeapSize *int `json:"worker_heap_size,omitempty"`
}

NodeJVMOverrides represents node-level JVM override values. NodeJVMOverrides 表示节点级 JVM 覆盖值。

func (NodeJVMOverrides) HasValues 

func (o NodeJVMOverrides) HasValues() bool

HasValues returns whether the override contains any explicit field. HasValues 返回 override 是否包含任何显式字段。

type NodeOperationResult 

type NodeOperationResult struct {
	NodeID   uint   `json:"node_id"`
	HostID   uint   `json:"host_id"`
	HostName string `json:"host_name"`
	Success  bool   `json:"success"`
	Message  string `json:"message"`
}

NodeOperationResult represents the result of an operation on a single node. NodeOperationResult 表示单个节点操作的结果。

type NodeOverrides 

type NodeOverrides struct {
	JVM *NodeJVMOverrides `json:"jvm,omitempty"`
}

NodeOverrides represents extensible node-level override settings. NodeOverrides 表示可扩展的节点级覆盖配置。

func (NodeOverrides) HasValues 

func (o NodeOverrides) HasValues() bool

HasValues returns whether the override contains any effective values. HasValues 返回 override 是否包含任何有效值。

func (NodeOverrides) Normalize 

func (o NodeOverrides) Normalize() NodeOverrides

Normalize removes empty nested override objects. Normalize 会移除空的嵌套 override 对象。

func (*NodeOverrides) Scan 

func (o *NodeOverrides) Scan(value interface{}) error

Scan implements sql.Scanner for database retrieval. Scan 实现 sql.Scanner 以便从数据库读取。

func (NodeOverrides) Value 

func (o NodeOverrides) Value() (driver.Value, error)

Value implements driver.Valuer for database storage. Value 实现 driver.Valuer 以便写入数据库。

type NodeRole 

type NodeRole string

NodeRole represents the role of a node in a cluster. 

const (
	// NodeRoleMaster indicates the node is a master node.
	NodeRoleMaster NodeRole = "master"
	// NodeRoleWorker indicates the node is a worker node.
	NodeRoleWorker NodeRole = "worker"
	// NodeRoleMasterWorker indicates the node is both master and worker (hybrid mode).
	// NodeRoleMasterWorker 表示节点同时是 master 和 worker（混合模式）。
	NodeRoleMasterWorker NodeRole = "master/worker"
)

type NodeStatus 

type NodeStatus string

NodeStatus represents the current status of a cluster node. 

const (
	// NodeStatusPending indicates the node is pending deployment.
	NodeStatusPending NodeStatus = "pending"
	// NodeStatusInstalling indicates the node is being installed.
	NodeStatusInstalling NodeStatus = "installing"
	// NodeStatusRunning indicates the node is running normally.
	NodeStatusRunning NodeStatus = "running"
	// NodeStatusStopped indicates the node has been stopped.
	NodeStatusStopped NodeStatus = "stopped"
	// NodeStatusError indicates the node is in an error state.
	NodeStatusError NodeStatus = "error"
	// NodeStatusOffline indicates the host/agent is offline; display-only, not persisted.
	NodeStatusOffline NodeStatus = "offline"
)

type NodeStatusInfo 

type NodeStatusInfo struct {
	NodeID     uint       `json:"node_id"`
	HostID     uint       `json:"host_id"`
	HostName   string     `json:"host_name"`
	HostIP     string     `json:"host_ip"`
	Role       NodeRole   `json:"role"`
	Status     NodeStatus `json:"status"`      // Unified status: pending, installing, running, stopped, error / 统一状态
	IsOnline   bool       `json:"is_online"`   // Whether host is online / 主机是否在线
	ProcessPID int        `json:"process_pid"` // SeaTunnel process PID / SeaTunnel 进程 PID
}

NodeStatusInfo represents detailed node status information. NodeStatusInfo 表示详细的节点状态信息。

type OperationEvent 

type OperationEvent struct {
	ClusterID   uint
	ClusterName string
	NodeID      uint
	HostID      uint
	HostName    string
	HostIP      string
	Role        string
	Operation   OperationType
	Success     bool
	Message     string
	Operator    string
	Trigger     string
}

OperationEvent represents one cluster operation notification hook payload. OperationEvent 表示一条集群操作通知钩子载荷。

type OperationResult 

type OperationResult struct {
	ClusterID   uint                   `json:"cluster_id"`
	Operation   OperationType          `json:"operation"`
	Success     bool                   `json:"success"`
	Message     string                 `json:"message"`
	NodeResults []*NodeOperationResult `json:"node_results"`
}

OperationResult represents the result of a cluster operation. OperationResult 表示集群操作的结果。

type OperationType 

type OperationType string

OperationType represents the type of cluster operation. OperationType 表示集群操作类型。 

const (
	// OperationStart starts the cluster.
	// OperationStart 启动集群。
	OperationStart OperationType = "start"
	// OperationStop stops the cluster.
	// OperationStop 停止集群。
	OperationStop OperationType = "stop"
	// OperationRestart restarts the cluster.
	// OperationRestart 重启集群。
	OperationRestart OperationType = "restart"
)

type PrecheckCheckItem 

type PrecheckCheckItem struct {
	Name    string `json:"name"`    // Check name / 检查名称
	Status  string `json:"status"`  // passed, failed, skipped / 通过、失败、跳过
	Message string `json:"message"` // Detail message / 详细信息
}

PrecheckCheckItem represents a single check item in precheck. PrecheckCheckItem 表示预检查中的单个检查项。

type PrecheckNodeResponse 

type PrecheckNodeResponse struct {
	ErrorMsg string          `json:"error_msg"`
	Data     *PrecheckResult `json:"data"`
}

PrecheckNodeResponse represents the response for node precheck. PrecheckNodeResponse 表示节点预检查的响应。

type PrecheckRequest 

type PrecheckRequest struct {
	HostID        uint     `json:"host_id" binding:"required"`
	Role          NodeRole `json:"role" binding:"required"`
	InstallDir    string   `json:"install_dir"`
	HazelcastPort int      `json:"hazelcast_port" binding:"required"`
	APIPort       int      `json:"api_port"`
	WorkerPort    int      `json:"worker_port"`
}

PrecheckRequest represents a request to precheck a node before adding. PrecheckRequest 表示添加节点前的预检查请求。

type PrecheckResult 

type PrecheckResult struct {
	Success bool                 `json:"success"`
	Message string               `json:"message"`
	Checks  []*PrecheckCheckItem `json:"checks"`
}

PrecheckResult represents the result of a node precheck. PrecheckResult 表示节点预检查的结果。

type PreviewRuntimeStorageResponse 

type PreviewRuntimeStorageResponse struct {
	ErrorMsg string                       `json:"error_msg"`
	Data     *RuntimeStoragePreviewResult `json:"data"`
}

type RemoveNodeResponse 

type RemoveNodeResponse struct {
	ErrorMsg string `json:"error_msg"`
	Data     any    `json:"data"`
}

RemoveNodeResponse represents the response for removing a node from a cluster. RemoveNodeResponse 表示从集群移除节点的响应。

type Repository 

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

Repository provides data access operations for Cluster and ClusterNode entities.

func NewRepository 

func NewRepository(db *gorm.DB) *Repository

NewRepository creates a new Repository instance.

func (*Repository) AddNode 

func (r *Repository) AddNode(ctx context.Context, node *ClusterNode) error

AddNode adds a node to a cluster. Returns ErrClusterNotFound if the cluster does not exist. Returns ErrNodeAlreadyExists if the host already has a node with the same role in the cluster. 如果集群不存在返回 ErrClusterNotFound。 如果主机在集群中已有相同角色的节点返回 ErrNodeAlreadyExists。

func (*Repository) CountNodesByClusterID 

func (r *Repository) CountNodesByClusterID(ctx context.Context, clusterID uint) (int64, error)

CountNodesByClusterID returns the number of nodes in a cluster.

func (*Repository) Create 

func (r *Repository) Create(ctx context.Context, cluster *Cluster) error

Create creates a new cluster record in the database. Returns ErrClusterNameDuplicate if a cluster with the same name already exists. Returns ErrClusterNameEmpty if the cluster name is empty.

func (*Repository) Delete 

func (r *Repository) Delete(ctx context.Context, id uint) error

Delete removes a cluster record from the database. Returns ErrClusterNotFound if the cluster does not exist. Note: This also deletes all associated cluster nodes due to foreign key cascade.

func (*Repository) ExistsByName 

func (r *Repository) ExistsByName(ctx context.Context, name string) (bool, error)

ExistsByName checks if a cluster with the given name exists.

func (*Repository) GetByID 

func (r *Repository) GetByID(ctx context.Context, id uint, preloadNodes bool) (*Cluster, error)

GetByID retrieves a cluster by its ID with optional node preloading. Returns ErrClusterNotFound if the cluster does not exist.

func (*Repository) GetByName 

func (r *Repository) GetByName(ctx context.Context, name string) (*Cluster, error)

GetByName retrieves a cluster by its name. Returns ErrClusterNotFound if no cluster with the given name exists.

func (*Repository) GetClustersWithHostID 

func (r *Repository) GetClustersWithHostID(ctx context.Context, hostID uint) ([]*Cluster, error)

GetClustersWithHostID retrieves all clusters that have a specific host as a node. This is useful for checking cluster associations before deleting a host.

func (*Repository) GetNodeByClusterAndHost 

func (r *Repository) GetNodeByClusterAndHost(ctx context.Context, clusterID uint, hostID uint) (*ClusterNode, error)

GetNodeByClusterAndHost retrieves a cluster node by cluster ID and host ID. GetNodeByClusterAndHost 根据集群 ID 和主机 ID 获取集群节点。 When a host has multiple nodes (e.g. master + worker in separated mode), use GetNodeByClusterAndHostAndRole instead. 当同一主机有多个节点（如分离模式下 master+worker）时，请使用 GetNodeByClusterAndHostAndRole。

func (*Repository) GetNodeByClusterAndHostAndRole 

func (r *Repository) GetNodeByClusterAndHostAndRole(ctx context.Context, clusterID uint, hostID uint, role string) (*ClusterNode, error)

GetNodeByClusterAndHostAndRole retrieves a cluster node by cluster ID, host ID and role. GetNodeByClusterAndHostAndRole 根据集群 ID、主机 ID 和角色获取集群节点。 Used when one host has both master and worker nodes (separated mode). 用于同一主机同时有 master 和 worker 节点（分离模式）时。

func (*Repository) GetNodeByHostAndInstallDirAndRole 

func (r *Repository) GetNodeByHostAndInstallDirAndRole(ctx context.Context, hostID uint, installDir, role string) (uint, uint, bool, error)

GetNodeByHostAndInstallDirAndRole retrieves a cluster node by host ID, install directory and role. GetNodeByHostAndInstallDirAndRole 根据主机 ID、安装目录和角色获取集群节点。 Returns clusterID, nodeID, found, error

func (*Repository) GetNodeByHostID 

func (r *Repository) GetNodeByHostID(ctx context.Context, hostID uint) (*ClusterNode, error)

GetNodeByHostID retrieves the first cluster node for a specific host. GetNodeByHostID 获取特定主机的第一个集群节点。

func (*Repository) GetNodeByID 

func (r *Repository) GetNodeByID(ctx context.Context, id uint) (*ClusterNode, error)

GetNodeByID retrieves a cluster node by its ID. Returns ErrNodeNotFound if the node does not exist.

func (*Repository) GetNodesByClusterID 

func (r *Repository) GetNodesByClusterID(ctx context.Context, clusterID uint) ([]*ClusterNode, error)

GetNodesByClusterID retrieves all nodes for a cluster.

func (*Repository) GetNodesByClusterIDWithHost 

func (r *Repository) GetNodesByClusterIDWithHost(ctx context.Context, clusterID uint) ([]*ClusterNode, error)

GetNodesByClusterIDWithHost retrieves all nodes for a cluster with host information. GetNodesByClusterIDWithHost 获取集群的所有节点及其主机信息。

func (*Repository) GetNodesByHostID 

func (r *Repository) GetNodesByHostID(ctx context.Context, hostID uint) ([]*ClusterNode, error)

GetNodesByHostID retrieves all cluster nodes for a specific host. This is useful for checking if a host is associated with any clusters.

func (*Repository) List 

func (r *Repository) List(ctx context.Context, filter *ClusterFilter) ([]*Cluster, int64, error)

List retrieves clusters based on filter criteria with pagination. Returns the list of clusters and total count.

func (*Repository) RemoveNode 

func (r *Repository) RemoveNode(ctx context.Context, nodeID uint) error

RemoveNode removes a node from a cluster. Returns ErrNodeNotFound if the node does not exist.

func (*Repository) RemoveNodeByClusterAndHost 

func (r *Repository) RemoveNodeByClusterAndHost(ctx context.Context, clusterID, hostID uint) error

RemoveNodeByClusterAndHost removes a node by cluster ID and host ID. Returns ErrNodeNotFound if the node does not exist.

func (*Repository) Transaction 

func (r *Repository) Transaction(ctx context.Context, fn func(tx *Repository) error) error

Transaction executes fn inside one database transaction. Transaction 在一个数据库事务内执行 fn。

func (*Repository) Update 

func (r *Repository) Update(ctx context.Context, cluster *Cluster) error

Update updates an existing cluster record. Returns ErrClusterNotFound if the cluster does not exist. Returns ErrClusterNameDuplicate if updating to a name that already exists.

func (*Repository) UpdateNode 

func (r *Repository) UpdateNode(ctx context.Context, node *ClusterNode) error

UpdateNode updates a cluster node's configuration. UpdateNode 更新集群节点的配置。

func (*Repository) UpdateNodeProcess 

func (r *Repository) UpdateNodeProcess(ctx context.Context, nodeID uint, pid int, processStatus string) error

UpdateNodeProcess updates the process information for a cluster node.

func (*Repository) UpdateNodeProcessStatus 

func (r *Repository) UpdateNodeProcessStatus(ctx context.Context, nodeID uint, pid int, processStatus string) error

UpdateNodeProcessStatus updates the process PID, status, and last event time for a cluster node. UpdateNodeProcessStatus 更新集群节点的进程 PID、状态和最后事件时间。 Uses the unified 'status' field (not 'process_status') for node state. 使用统一的 'status' 字段（而非 'process_status'）表示节点状态。 Also updates 'last_event_at' to track when the last process event occurred. 同时更新 'last_event_at' 以跟踪最后一次进程事件发生的时间。

func (*Repository) UpdateNodeStatus 

func (r *Repository) UpdateNodeStatus(ctx context.Context, nodeID uint, status NodeStatus) error

UpdateNodeStatus updates the status of a cluster node.

func (*Repository) UpdateStatus 

func (r *Repository) UpdateStatus(ctx context.Context, id uint, status ClusterStatus) error

UpdateStatus updates the status of a cluster.

type RuntimeStorageCheckpointInspectJobConfig 

type RuntimeStorageCheckpointInspectJobConfig struct {
	Content       string                 `json:"content,omitempty"`
	ContentFormat string                 `json:"content_format,omitempty"`
	Variables     map[string]interface{} `json:"variables,omitempty"`
}

type RuntimeStorageCheckpointInspectResult 

type RuntimeStorageCheckpointInspectResult struct {
	ClusterID           uint                                              `json:"cluster_id"`
	Path                string                                            `json:"path,omitempty"`
	FileName            string                                            `json:"file_name,omitempty"`
	StorageType         string                                            `json:"storage_type,omitempty"`
	SizeBytes           int64                                             `json:"size_bytes,omitempty"`
	Truncated           bool                                              `json:"truncated,omitempty"`
	Binary              bool                                              `json:"binary,omitempty"`
	Encoding            string                                            `json:"encoding,omitempty"`
	TextPreview         string                                            `json:"text_preview,omitempty"`
	HexPreview          string                                            `json:"hex_preview,omitempty"`
	PipelineState       map[string]interface{}                            `json:"pipeline_state,omitempty"`
	CompletedCheckpoint map[string]interface{}                            `json:"completed_checkpoint,omitempty"`
	ActionStates        []map[string]interface{}                          `json:"action_states,omitempty"`
	TaskStatistics      []map[string]interface{}                          `json:"task_statistics,omitempty"`
	SourceStateInspect  *RuntimeStorageCheckpointSourceStateInspectResult `json:"source_state_inspect,omitempty"`
}

type RuntimeStorageCheckpointSourceStateInspectResult 

type RuntimeStorageCheckpointSourceStateInspectResult struct {
	PipelineState       map[string]interface{}   `json:"pipeline_state,omitempty"`
	CompletedCheckpoint map[string]interface{}   `json:"completed_checkpoint,omitempty"`
	Sources             []map[string]interface{} `json:"sources,omitempty"`
	Sinks               []map[string]interface{} `json:"sinks,omitempty"`
	UnsupportedSources  []map[string]interface{} `json:"unsupported_sources,omitempty"`
	Warnings            []string                 `json:"warnings,omitempty"`
	ErrorMessage        string                   `json:"error_message,omitempty"`
}

type RuntimeStorageCleanupResult 

type RuntimeStorageCleanupResult struct {
	ClusterID uint                         `json:"cluster_id"`
	Success   bool                         `json:"success"`
	Message   string                       `json:"message"`
	Nodes     []*RuntimeStorageNodeCleanup `json:"nodes,omitempty"`
}

RuntimeStorageCleanupResult describes a cleanup operation result. RuntimeStorageCleanupResult 描述清理操作结果。

type RuntimeStorageDetails 

type RuntimeStorageDetails struct {
	ClusterID            uint                `json:"cluster_id"`
	Checkpoint           *RuntimeStorageSpec `json:"checkpoint,omitempty"`
	IMAP                 *RuntimeStorageSpec `json:"imap,omitempty"`
	ConfiguredCheckpoint *RuntimeStorageSpec `json:"configured_checkpoint,omitempty"`
	ConfiguredIMAP       *RuntimeStorageSpec `json:"configured_imap,omitempty"`
	ConfigSource         string              `json:"config_source,omitempty"`
}

RuntimeStorageDetails summarizes checkpoint and IMAP runtime storage. RuntimeStorageDetails 汇总 checkpoint 与 IMAP 运行时存储信息。

type RuntimeStorageIMAPInspectResult 

type RuntimeStorageIMAPInspectResult struct {
	ClusterID   uint                     `json:"cluster_id"`
	Path        string                   `json:"path,omitempty"`
	FileName    string                   `json:"file_name,omitempty"`
	StorageType string                   `json:"storage_type,omitempty"`
	SizeBytes   int64                    `json:"size_bytes,omitempty"`
	Truncated   bool                     `json:"truncated,omitempty"`
	Binary      bool                     `json:"binary,omitempty"`
	Encoding    string                   `json:"encoding,omitempty"`
	TextPreview string                   `json:"text_preview,omitempty"`
	HexPreview  string                   `json:"hex_preview,omitempty"`
	EntryCount  int                      `json:"entry_count,omitempty"`
	Entries     []map[string]interface{} `json:"entries,omitempty"`
}

type RuntimeStorageListItem 

type RuntimeStorageListItem struct {
	Path       string `json:"path,omitempty"`
	Name       string `json:"name,omitempty"`
	Directory  bool   `json:"directory,omitempty"`
	SizeBytes  int64  `json:"size_bytes,omitempty"`
	ModifiedAt string `json:"modified_at,omitempty"`
}

func (*RuntimeStorageListItem) UnmarshalJSON 

func (item *RuntimeStorageListItem) UnmarshalJSON(data []byte) error

type RuntimeStorageListResult 

type RuntimeStorageListResult struct {
	ClusterID uint                     `json:"cluster_id"`
	Kind      string                   `json:"kind"`
	Path      string                   `json:"path,omitempty"`
	Items     []RuntimeStorageListItem `json:"items"`
}

type RuntimeStorageNodeCleanup 

type RuntimeStorageNodeCleanup struct {
	HostID   uint   `json:"host_id"`
	HostName string `json:"host_name"`
	Success  bool   `json:"success"`
	Message  string `json:"message"`
}

RuntimeStorageNodeCleanup describes one node's cleanup result. RuntimeStorageNodeCleanup 描述单个节点的清理结果。

type RuntimeStorageNodeStat 

type RuntimeStorageNodeStat struct {
	NodeID    uint   `json:"node_id"`
	HostID    uint   `json:"host_id"`
	HostName  string `json:"host_name"`
	Path      string `json:"path,omitempty"`
	Exists    bool   `json:"exists"`
	SizeBytes int64  `json:"size_bytes"`
	Message   string `json:"message,omitempty"`
}

RuntimeStorageNodeStat describes one node's runtime storage usage. RuntimeStorageNodeStat 描述单个节点的运行时存储使用情况。

type RuntimeStoragePreviewResult 

type RuntimeStoragePreviewResult struct {
	ClusterID   uint   `json:"cluster_id"`
	Kind        string `json:"kind"`
	StorageType string `json:"storage_type,omitempty"`
	Path        string `json:"path,omitempty"`
	FileName    string `json:"file_name,omitempty"`
	SizeBytes   int64  `json:"size_bytes,omitempty"`
	Truncated   bool   `json:"truncated,omitempty"`
	Binary      bool   `json:"binary,omitempty"`
	Encoding    string `json:"encoding,omitempty"`
	TextPreview string `json:"text_preview,omitempty"`
	HexPreview  string `json:"hex_preview,omitempty"`
}

type RuntimeStorageSpec 

type RuntimeStorageSpec struct {
	Kind        string `json:"kind"`
	Enabled     bool   `json:"enabled"`
	StorageType string `json:"storage_type"`
	Namespace   string `json:"namespace,omitempty"`
	Endpoint    string `json:"endpoint,omitempty"`
	Bucket      string `json:"bucket,omitempty"`
	External    bool   `json:"external"`
	// HDFS HA / Kerberos / hdfs-site，仅 HDFS 类型有值。
	// HDFS HA / Kerberos / hdfs-site; populated only for HDFS.
	HDFSHAEnabled             bool   `json:"hdfs_ha_enabled,omitempty"`
	HDFSNameServices          string `json:"hdfs_name_services,omitempty"`
	HDFSHANamenodes           string `json:"hdfs_ha_namenodes,omitempty"`
	HDFSNamenodeRPCAddress1   string `json:"hdfs_namenode_rpc_address_1,omitempty"`
	HDFSNamenodeRPCAddress2   string `json:"hdfs_namenode_rpc_address_2,omitempty"`
	HDFSFailoverProxyProvider string `json:"hdfs_failover_proxy_provider,omitempty"`
	HDFSNameNodeHost          string `json:"hdfs_namenode_host,omitempty"`
	HDFSNameNodePort          int    `json:"hdfs_namenode_port,omitempty"`
	KerberosPrincipal         string `json:"kerberos_principal,omitempty"`
	KerberosKeytabFilePath    string `json:"kerberos_keytab_file_path,omitempty"`
	HdfsSitePath              string `json:"hdfs_site_path,omitempty"`
	DisableCache              *bool  `json:"disable_cache,omitempty"`
	// S3CredentialsProvider 对应 fs.s3a.aws.credentials.provider。
	// S3CredentialsProvider maps to fs.s3a.aws.credentials.provider.
	S3CredentialsProvider string                    `json:"s3_credentials_provider,omitempty"`
	SizeAvailable         bool                      `json:"size_available"`
	TotalSizeBytes        int64                     `json:"total_size_bytes"`
	CleanupSupported      bool                      `json:"cleanup_supported"`
	Warning               string                    `json:"warning,omitempty"`
	Nodes                 []*RuntimeStorageNodeStat `json:"nodes,omitempty"`
}

RuntimeStorageSpec describes one runtime storage area. RuntimeStorageSpec 描述一类运行时存储。

type STXJavaProxyConfig added in v1.0.0 

type STXJavaProxyConfig struct {
	Port    int    `json:"port,omitempty"`
	JvmOpts string `json:"jvm_opts,omitempty"`
}

STXJavaProxyConfig represents cluster-level stx-java-proxy configuration. STXJavaProxyConfig 表示集群级 stx-java-proxy 配置。

type STXJavaProxyLogPreviewResponse 

type STXJavaProxyLogPreviewResponse struct {
	ErrorMsg string                        `json:"error_msg"`
	Data     *STXJavaProxyLogPreviewResult `json:"data"`
}

type STXJavaProxyLogPreviewResult 

type STXJavaProxyLogPreviewResult struct {
	ClusterID uint   `json:"cluster_id"`
	LogPath   string `json:"log_path,omitempty"`
	Lines     int    `json:"lines,omitempty"`
	Logs      string `json:"logs,omitempty"`
}

STXJavaProxyLogPreviewResult represents a service.log preview result. STXJavaProxyLogPreviewResult 表示 service.log 预览结果。

type STXJavaProxyResponse 

type STXJavaProxyResponse struct {
	ErrorMsg string              `json:"error_msg"`
	Data     *STXJavaProxyStatus `json:"data"`
}

STXJavaProxyResponse represents a managed stx-java-proxy response.

type STXJavaProxyStatus 

type STXJavaProxyStatus struct {
	ClusterID   uint                   `json:"cluster_id"`
	ClusterName string                 `json:"cluster_name,omitempty"`
	NodeID      uint                   `json:"node_id,omitempty"`
	HostID      uint                   `json:"host_id,omitempty"`
	HostName    string                 `json:"host_name,omitempty"`
	HostIP      string                 `json:"host_ip,omitempty"`
	Role        string                 `json:"role,omitempty"`
	InstallDir  string                 `json:"install_dir,omitempty"`
	Version     string                 `json:"version,omitempty"`
	Service     string                 `json:"service,omitempty"`
	Managed     bool                   `json:"managed"`
	Running     bool                   `json:"running"`
	Healthy     bool                   `json:"healthy"`
	Endpoint    string                 `json:"endpoint,omitempty"`
	Port        int                    `json:"port,omitempty"`
	PID         int                    `json:"pid,omitempty"`
	LogPath     string                 `json:"log_path,omitempty"`
	Message     string                 `json:"message,omitempty"`
	JvmOpts     string                 `json:"jvm_opts,omitempty"`
	JvmMemory   map[string]interface{} `json:"jvm_memory,omitempty"`
}

STXJavaProxyStatus represents the managed stx-java-proxy state for a cluster.

type Service 

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

Service provides business logic for cluster management operations. Service 提供集群管理操作的业务逻辑。

func NewService 

func NewService(repo *Repository, hostProvider HostProvider, cfg *ServiceConfig) *Service

NewService creates a new Service instance. NewService 创建一个新的 Service 实例。

func (*Service) AddNode 

func (s *Service) AddNode(ctx context.Context, clusterID uint, req *AddNodeRequest) (*ClusterNode, error)

AddNode adds a node to a cluster with validation. AddNode 向集群添加节点并进行验证。 Requirements: 7.2 - Validates host Agent status is "installed" before association.

func (*Service) AddNodes 

func (s *Service) AddNodes(ctx context.Context, clusterID uint, req *AddNodesRequest) ([]*ClusterNode, error)

AddNodes adds one or more logical nodes for the same host atomically. AddNodes 原子地为同一主机添加一个或多个逻辑节点。

func (*Service) ApplyRuntimeStorage added in v1.0.0 

func (s *Service) ApplyRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	kind installerapp.RuntimeStorageValidationKind,
	req *ApplyRuntimeStorageRequest,
	userID uint,
) (*ApplyRuntimeStorageResult, error)

ApplyRuntimeStorage 先做真实读写测试，通过后再写回配置模板并生成新版本。 ApplyRuntimeStorage runs a real read/write probe first, then writes config templates and creates versions.

func (*Service) CleanupIMAPStorage 

func (s *Service) CleanupIMAPStorage(ctx context.Context, clusterID uint) (*RuntimeStorageCleanupResult, error)

CleanupIMAPStorage cleans local IMAP directories on all online nodes when IMAP uses LOCAL_FILE. CleanupIMAPStorage 在 IMAP 使用 LOCAL_FILE 时清理所有在线节点上的本地 IMAP 目录。

func (*Service) Create 

func (s *Service) Create(ctx context.Context, req *CreateClusterRequest) (*Cluster, error)

Create creates a new cluster with validation. Create 创建一个新集群并进行验证。 Requirements: 7.1 - Validates cluster name uniqueness and stores basic info.

func (*Service) Delete 

func (s *Service) Delete(ctx context.Context, id uint, forceRemoveInstallDir bool) error

Delete removes a cluster after checking for running tasks. Before DB deletion it best-effort stops SeaTunnel processes and stx-java-proxy on node agents. If forceRemoveInstallDir is true, also sends REMOVE_INSTALL_DIR to each agent. Delete 在检查运行中的任务后删除集群；删除前尽力向各节点 Agent 停止 SeaTunnel 与 stx-java-proxy； 若 forceRemoveInstallDir 为 true 则再发送删除安装目录命令。 Requirements: 7.5 - Checks if cluster has running tasks before deletion.

func (*Service) EnsureNodeForInstallation added in v1.0.0 

func (s *Service) EnsureNodeForInstallation(ctx context.Context, clusterID uint, hostID uint, role string, installDir string, hazelcastPort int, apiPort int, workerPort int) error

EnsureNodeForInstallation 确保一键安装完成后的节点已登记，并刷新安装目录与端口。 EnsureNodeForInstallation ensures a one-click installation node is registered and refreshes its install directory and ports.

func (*Service) Get 

func (s *Service) Get(ctx context.Context, id uint) (*Cluster, error)

Get retrieves a cluster by ID with optional node preloading. Get 根据 ID 获取集群，可选择预加载节点。 Requirements: 7.3 - Returns cluster name, status, node list, version info, creation time.

func (*Service) GetByName 

func (s *Service) GetByName(ctx context.Context, name string) (*Cluster, error)

GetByName retrieves a cluster by name. GetByName 根据名称获取集群。

func (*Service) GetClusterNodeDisplayInfo 

func (s *Service) GetClusterNodeDisplayInfo(ctx context.Context, clusterID uint, nodeID uint) (clusterName, nodeDisplay string)

GetClusterNodeDisplayInfo returns cluster name and node display string "主机名 - 角色" for audit log. GetClusterNodeDisplayInfo 返回集群名及节点展示串「主机名 - 角色」，用于审计日志。

func (*Service) GetClusterNodesWithAgentInfo 

func (s *Service) GetClusterNodesWithAgentInfo(ctx context.Context, clusterID uint) ([]*NodeInfo, error)

GetClusterNodesWithAgentInfo retrieves all nodes for a cluster with agent information. GetClusterNodesWithAgentInfo 获取集群的所有节点及其 Agent 信息。 Returns nodes with their associated agent IDs for config push. 返回节点及其关联的 Agent ID 用于配置下发。

func (*Service) GetClusterVersion 

func (s *Service) GetClusterVersion(ctx context.Context, clusterID uint) (string, error)

GetClusterVersion retrieves the version of a cluster by ID. GetClusterVersion 根据 ID 获取集群的版本。 This method implements the ClusterGetter interface for plugin version validation. 此方法实现 ClusterGetter 接口用于插件版本校验。

func (*Service) GetClustersByHostID 

func (s *Service) GetClustersByHostID(ctx context.Context, hostID uint) ([]*Cluster, error)

GetClustersByHostID retrieves all clusters that have a specific host as a node. GetClustersByHostID 获取将特定主机作为节点的所有集群。

func (*Service) GetJobLogMode added in v1.0.0 

func (s *Service) GetJobLogMode(ctx context.Context, clusterID uint) (string, error)

GetJobLogMode returns the configured runtime job log mode for the cluster.

func (*Service) GetNode 

func (s *Service) GetNode(ctx context.Context, nodeID uint) (*ClusterNode, error)

GetNode retrieves a specific node by ID. GetNode 根据 ID 获取特定节点。

func (*Service) GetNodeByHostAndInstallDirAndRole 

func (s *Service) GetNodeByHostAndInstallDirAndRole(ctx context.Context, hostID uint, installDir, role string) (clusterID, nodeID uint, found bool, err error)

GetNodeByHostAndInstallDirAndRole retrieves a cluster node by host ID, install directory and role. GetNodeByHostAndInstallDirAndRole 根据主机 ID、安装目录和角色获取集群节点。 This implements the grpc.ClusterNodeProvider interface. 这实现了 grpc.ClusterNodeProvider 接口。

func (*Service) GetNodeInstallDir 

func (s *Service) GetNodeInstallDir(ctx context.Context, clusterID uint, hostID uint) (string, error)

GetNodeInstallDir retrieves the install directory for a node by cluster ID and host ID. GetNodeInstallDir 根据集群 ID 和主机 ID 获取节点的安装目录。 This implements the config.NodeInfoProvider interface. 这实现了 config.NodeInfoProvider 接口。

func (*Service) GetNodeLogs 

func (s *Service) GetNodeLogs(ctx context.Context, clusterID uint, nodeID uint, req *GetNodeLogsRequest) (string, error)

GetNodeLogs gets the logs of a node. GetNodeLogs 获取节点的日志。

func (*Service) GetNodes 

func (s *Service) GetNodes(ctx context.Context, clusterID uint) ([]*NodeInfo, error)

GetNodes retrieves all nodes for a cluster with host information. GetNodes 获取集群的所有节点及其主机信息。 Requirements: 7.4 - Returns each node's host info, role, SeaTunnel process status, resource usage.

func (*Service) GetNodesByHostID 

func (s *Service) GetNodesByHostID(ctx context.Context, hostID uint) ([]*ClusterNode, error)

GetNodesByHostID retrieves all cluster nodes for a specific host. GetNodesByHostID 获取特定主机上的所有集群节点。 This is used for pushing monitor config to agent after registration. 这用于在 Agent 注册后推送监控配置。

func (*Service) GetRuntimeStorageDetails 

func (s *Service) GetRuntimeStorageDetails(ctx context.Context, clusterID uint) (*RuntimeStorageDetails, error)

GetRuntimeStorageDetails returns runtime storage config and local usage stats. GetRuntimeStorageDetails 返回运行时存储配置及本地使用统计。

func (*Service) GetSTXJavaProxyServiceLog 

func (s *Service) GetSTXJavaProxyServiceLog(
	ctx context.Context,
	clusterID uint,
	lines int,
) (*STXJavaProxyLogPreviewResult, error)

func (*Service) GetSTXJavaProxyStatus 

func (s *Service) GetSTXJavaProxyStatus(ctx context.Context, clusterID uint) (*STXJavaProxyStatus, error)

func (*Service) GetStatus 

func (s *Service) GetStatus(ctx context.Context, clusterID uint) (*ClusterStatusInfo, error)

GetStatus retrieves the detailed status of a cluster including node health. GetStatus 获取集群的详细状态，包括节点健康状况。 Requirements: 7.6 - Returns cluster health status based on node states.

func (*Service) InspectCheckpointRuntimeStorage 

func (s *Service) InspectCheckpointRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	path string,
	jobConfig *RuntimeStorageCheckpointInspectJobConfig,
) (*RuntimeStorageCheckpointInspectResult, error)

func (*Service) InspectIMAPRuntimeStorage 

func (s *Service) InspectIMAPRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	path string,
) (*RuntimeStorageIMAPInspectResult, error)

func (*Service) IsClusterHealthy 

func (s *Service) IsClusterHealthy(ctx context.Context, clusterID uint) (bool, error)

IsClusterHealthy checks if all nodes in a cluster are online. IsClusterHealthy 检查集群中的所有节点是否都在线。 Requirements: 7.6 - Returns false if any node is offline.

func (*Service) List 

func (s *Service) List(ctx context.Context, filter *ClusterFilter) ([]*Cluster, int64, error)

List retrieves clusters based on filter criteria. List 根据过滤条件获取集群列表。 Requirements: 7.3 - Returns cluster list with node count.

func (*Service) ListRuntimeStorage 

func (s *Service) ListRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	kind installerapp.RuntimeStorageValidationKind,
	path string,
	recursive bool,
	limit int,
) (*RuntimeStorageListResult, error)

func (*Service) ListWithInfo 

func (s *Service) ListWithInfo(ctx context.Context, filter *ClusterFilter) ([]*ClusterInfo, int64, error)

ListWithInfo retrieves clusters and converts them to ClusterInfo with online_nodes and health_status. ListWithInfo 获取集群列表并转换为 ClusterInfo，包含 online_nodes 与 health_status。

func (*Service) PrecheckNode 

func (s *Service) PrecheckNode(ctx context.Context, clusterID uint, req *PrecheckRequest) (*PrecheckResult, error)

PrecheckNode performs precheck on a node before adding to cluster. PrecheckNode 在将节点添加到集群之前执行预检查。 Checks: 1. Port is listening (SeaTunnel service is running) / 端口正在监听（SeaTunnel 服务正在运行） 2. Directory exists and is writable / 目录存在且可写 3. SeaTunnel REST API connectivity / SeaTunnel REST API 连通性

func (*Service) PreviewRuntimeStorage 

func (s *Service) PreviewRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	kind installerapp.RuntimeStorageValidationKind,
	path string,
	maxBytes int,
) (*RuntimeStoragePreviewResult, error)

func (*Service) RefreshClusterStatusFromNodes 

func (s *Service) RefreshClusterStatusFromNodes(ctx context.Context, clusterID uint)

RefreshClusterStatusFromNodes recalculates and updates cluster status from its nodes' status. Called after heartbeat updates node process status so cluster status stays in sync. RefreshClusterStatusFromNodes 根据节点状态重新计算并更新集群状态；心跳更新节点后调用以保持集群状态一致。

func (*Service) RemoveNode 

func (s *Service) RemoveNode(ctx context.Context, clusterID uint, nodeID uint) error

RemoveNode removes a node from a cluster. RemoveNode 从集群中移除节点。 Requirements: 7.4 - Removes node from cluster.

func (*Service) RemoveNodeByHostID 

func (s *Service) RemoveNodeByHostID(ctx context.Context, clusterID uint, hostID uint) error

RemoveNodeByHostID removes a node from a cluster by host ID. RemoveNodeByHostID 根据主机 ID 从集群中移除节点。

func (*Service) ResolveNodeJVMByClusterAndHostAndRole 

func (s *Service) ResolveNodeJVMByClusterAndHostAndRole(ctx context.Context, clusterID uint, hostID uint, role string) (*installerapp.JVMConfig, error)

ResolveNodeJVMByClusterAndHostAndRole resolves effective JVM config for one logical node. ResolveNodeJVMByClusterAndHostAndRole 解析一个逻辑节点的生效 JVM 配置。

func (*Service) Restart 

func (s *Service) Restart(ctx context.Context, clusterID uint) (*OperationResult, error)

Restart restarts all nodes in a cluster. Restart 重启集群中的所有节点。 Requirements: 6.3 - Executes stop first, waits for complete exit, then executes start.

func (*Service) RestartNode 

func (s *Service) RestartNode(ctx context.Context, clusterID uint, nodeID uint) (*OperationResult, error)

RestartNode restarts a single node in a cluster. RestartNode 重启集群中的单个节点。

func (*Service) RestartSTXJavaProxy 

func (s *Service) RestartSTXJavaProxy(ctx context.Context, clusterID uint) (*STXJavaProxyStatus, error)

func (*Service) SetAgentCommandSender 

func (s *Service) SetAgentCommandSender(sender AgentCommandSender)

SetAgentCommandSender sets the agent command sender for cluster operations. SetAgentCommandSender 设置用于集群操作的 Agent 命令发送器。

func (*Service) SetConfigAgentClient 

func (s *Service) SetConfigAgentClient(client ConfigAgentClient)

SetConfigAgentClient sets the config agent client used for config file synchronization. SetConfigAgentClient 设置用于配置文件同步的 Agent 配置客户端。

func (*Service) SetOnBeforeClusterDelete 

func (s *Service) SetOnBeforeClusterDelete(fn func(context.Context, uint))

SetOnBeforeClusterDelete sets an optional hook called before cluster DB deletion (e.g. monitor config cleanup). SetOnBeforeClusterDelete 设置删除集群前可选钩子（如清理监控配置）。

func (*Service) SetOnClusterTopologyChanged 

func (s *Service) SetOnClusterTopologyChanged(fn func(context.Context, uint))

SetOnClusterTopologyChanged sets an optional hook called when cluster topology changes (create/update/add-node/remove-node/delete), typically used for observability target sync. SetOnClusterTopologyChanged 设置集群拓扑变更回调（创建/更新/加减节点/删除），通常用于可观测目标同步。

func (*Service) SetRuntimeConfigStore added in v1.0.0 

func (s *Service) SetRuntimeConfigStore(store runtimeConfigStore)

SetRuntimeConfigStore 注入配置服务，用于存储可视化与配置版本联动。 SetRuntimeConfigStore injects the config service used to link storage edits with config versions.

func (*Service) Start 

func (s *Service) Start(ctx context.Context, clusterID uint) (*OperationResult, error)

Start starts all nodes in a cluster. Start 启动集群中的所有节点。 Requirements: 6.1 - Executes SeaTunnel start script, waits for process startup, verifies process alive.

func (*Service) StartNode 

func (s *Service) StartNode(ctx context.Context, clusterID uint, nodeID uint) (*OperationResult, error)

StartNode starts a single node in a cluster. StartNode 启动集群中的单个节点。

func (*Service) StartNodeByClusterAndHost 

func (s *Service) StartNodeByClusterAndHost(ctx context.Context, clusterID uint, hostID uint) (bool, string, error)

StartNodeByClusterAndHost starts a node by cluster ID and host ID. StartNodeByClusterAndHost 根据集群 ID 和主机 ID 启动节点。 When a host has multiple nodes (master + worker), use StartNodeByClusterAndHostAndRole to start the specific role. 当同一主机有多个节点（master+worker）时，请使用 StartNodeByClusterAndHostAndRole 启动指定角色。 This implements the installer.NodeStarter interface. 这实现了 installer.NodeStarter 接口。

func (*Service) StartNodeByClusterAndHostAndRole 

func (s *Service) StartNodeByClusterAndHostAndRole(ctx context.Context, clusterID uint, hostID uint, role string) (bool, string, error)

StartNodeByClusterAndHostAndRole starts a node by cluster ID, host ID and role. StartNodeByClusterAndHostAndRole 根据集群 ID、主机 ID 和角色启动节点。 Used after install when one host has both master and worker (separated mode); starts the node that was just installed. 安装完成后、同一主机兼有 master 与 worker 时使用，用于启动刚安装完成的那一个节点。

func (*Service) StartSTXJavaProxy 

func (s *Service) StartSTXJavaProxy(ctx context.Context, clusterID uint) (*STXJavaProxyStatus, error)

func (*Service) Stop 

func (s *Service) Stop(ctx context.Context, clusterID uint) (*OperationResult, error)

Stop stops all nodes in a cluster. Stop 停止集群中的所有节点。 Requirements: 6.2 - Sends SIGTERM, waits for graceful shutdown (max 30s), sends SIGKILL if timeout.

func (*Service) StopNode 

func (s *Service) StopNode(ctx context.Context, clusterID uint, nodeID uint) (*OperationResult, error)

StopNode stops a single node in a cluster. StopNode 停止集群中的单个节点。

func (*Service) StopSTXJavaProxy 

func (s *Service) StopSTXJavaProxy(ctx context.Context, clusterID uint) (*STXJavaProxyStatus, error)

func (*Service) SwitchJobLogMode added in v1.0.0 

func (s *Service) SwitchJobLogMode(
	ctx context.Context,
	clusterID uint,
	req *SwitchJobLogModeRequest,
	userID uint,
) (*SwitchJobLogModeResult, error)

SwitchJobLogMode switches the cluster's log4j2.properties between per_job and mixed mode, creates a new version of the template configuration, and syncs it to all cluster nodes.

func (*Service) Update 

func (s *Service) Update(ctx context.Context, id uint, req *UpdateClusterRequest) (*Cluster, error)

Update updates an existing cluster with validation. Update 更新现有集群并进行验证。

func (*Service) UpdateNode 

func (s *Service) UpdateNode(ctx context.Context, clusterID uint, nodeID uint, req *UpdateNodeRequest) (*ClusterNode, error)

UpdateNode updates a node's configuration (install_dir, ports). UpdateNode 更新节点配置（安装目录、端口）。

func (*Service) UpdateNodeProcess 

func (s *Service) UpdateNodeProcess(ctx context.Context, nodeID uint, pid int, processStatus string) error

UpdateNodeProcess updates the process information for a cluster node. UpdateNodeProcess 更新集群节点的进程信息。

func (*Service) UpdateNodeProcessStatus 

func (s *Service) UpdateNodeProcessStatus(ctx context.Context, nodeID uint, pid int, status string) error

UpdateNodeProcessStatus updates the process PID and status for a node. UpdateNodeProcessStatus 更新节点的进程 PID 和状态。 This is called when agent reports process events (started, stopped, crashed, restarted). 当 Agent 上报进程事件（启动、停止、崩溃、重启）时调用。

func (*Service) UpdateNodeStatus 

func (s *Service) UpdateNodeStatus(ctx context.Context, nodeID uint, status NodeStatus) error

UpdateNodeStatus updates the status of a cluster node. UpdateNodeStatus 更新集群节点的状态。

func (*Service) UpdateNodeStatusByClusterAndHost 

func (s *Service) UpdateNodeStatusByClusterAndHost(ctx context.Context, clusterID uint, hostID uint, status string) error

UpdateNodeStatusByClusterAndHost updates the node status by cluster ID and host ID. UpdateNodeStatusByClusterAndHost 根据集群 ID 和主机 ID 更新节点状态。 This implements the installer.NodeStatusUpdater interface. 这实现了 installer.NodeStatusUpdater 接口。

func (*Service) UpdateSTXJavaProxyConfig added in v1.0.0 

func (s *Service) UpdateSTXJavaProxyConfig(
	ctx context.Context,
	clusterID uint,
	req *UpdateSTXJavaProxyConfigRequest,
) (*STXJavaProxyStatus, error)

func (*Service) UpdateStatus 

func (s *Service) UpdateStatus(ctx context.Context, id uint, status ClusterStatus) error

UpdateStatus updates the status of a cluster. UpdateStatus 更新集群的状态。

func (*Service) ValidateRuntimeStorage 

func (s *Service) ValidateRuntimeStorage(
	ctx context.Context,
	clusterID uint,
	kind installerapp.RuntimeStorageValidationKind,
) (*installerapp.RuntimeStorageValidationResult, error)

type ServiceConfig 

type ServiceConfig struct {
	HeartbeatTimeout time.Duration
}

ServiceConfig holds configuration for the Cluster Service. ServiceConfig 保存 Cluster Service 的配置。

type SwitchJobLogModeRequest added in v1.0.0 

type SwitchJobLogModeRequest struct {
	Mode string `json:"mode" binding:"required"` // "per_job" or "mixed"
}

SwitchJobLogModeRequest represents input to change job log mode.

type SwitchJobLogModeResponse added in v1.0.0 

type SwitchJobLogModeResponse struct {
	ErrorMsg string                  `json:"error_msg"`
	Data     *SwitchJobLogModeResult `json:"data"`
}

SwitchJobLogModeResponse represents job log mode switch response.

type SwitchJobLogModeResult added in v1.0.0 

type SwitchJobLogModeResult struct {
	Saved           bool   `json:"saved"`
	RestartRequired bool   `json:"restart_required"`
	Mode            string `json:"mode"`
	Message         string `json:"message"`
	ConfigVersion   int    `json:"config_version,omitempty"`
}

SwitchJobLogModeResult represents the result of changing job log mode.

type UpdateClusterRequest 

type UpdateClusterRequest struct {
	Name        *string        `json:"name"`
	Description *string        `json:"description"`
	Version     *string        `json:"version"`
	InstallDir  *string        `json:"install_dir"`
	Config      *ClusterConfig `json:"config"`
}

UpdateClusterRequest represents a request to update an existing cluster.

type UpdateClusterResponse 

type UpdateClusterResponse struct {
	ErrorMsg string       `json:"error_msg"`
	Data     *ClusterInfo `json:"data"`
}

UpdateClusterResponse represents the response for updating a cluster. UpdateClusterResponse 表示更新集群的响应。

type UpdateNodeRequest 

type UpdateNodeRequest struct {
	InstallDir    *string        `json:"install_dir"`         // SeaTunnel installation directory / SeaTunnel 安装目录
	HazelcastPort *int           `json:"hazelcast_port"`      // Hazelcast cluster port / Hazelcast 集群端口
	APIPort       *int           `json:"api_port"`            // REST API port (Master only, optional) / REST API 端口（仅 Master，可选）
	WorkerPort    *int           `json:"worker_port"`         // Worker hazelcast port (Hybrid only) / Worker Hazelcast 端口（仅混合模式）
	Overrides     *NodeOverrides `json:"overrides,omitempty"` // Node-level JSON overrides / 节点级 JSON 覆盖配置
}

UpdateNodeRequest represents a request to update a node in a cluster. 更新节点请求，包含安装目录和端口配置

type UpdateSTXJavaProxyConfigRequest added in v1.0.0 

type UpdateSTXJavaProxyConfigRequest struct {
	JvmOpts *string `json:"jvm_opts"`
	Port    *int    `json:"port,omitempty"`
	Restart *bool   `json:"restart,omitempty"`
}

UpdateSTXJavaProxyConfigRequest represents a request to update proxy parameters. UpdateSTXJavaProxyConfigRequest 表示更新 proxy 参数请求。

type ValidateRuntimeStorageResponse 

type ValidateRuntimeStorageResponse struct {
	ErrorMsg string                                       `json:"error_msg"`
	Data     *installerapp.RuntimeStorageValidationResult `json:"data"`
}

ValidateRuntimeStorageResponse represents runtime storage connectivity validation response. ValidateRuntimeStorageResponse 表示运行时存储连通性校验响应。

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.