cluster

package
v1.4.0 Latest Latest
Warning

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

Go to latest
Published: Nov 8, 2026 License: Apache-2.0 Imports: 41 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 (
	ErrorCodeClusterScaleInProgress         = "cluster_scale_in_progress"
	ErrorCodeScaleRollingRestartUnavailable = "scale_rolling_restart_unavailable"
	ErrorCodeScaleOperationNotFound         = "scale_operation_not_found"
	ErrorCodeScaleNotRetryable              = "scale_operation_not_retryable"
	ErrorCodeScaleInvalidRequest            = "scale_invalid_request"
	ErrorCodeScalePlanBlocked               = "scale_plan_blocked"
	ErrorCodeScalePlanChanged               = "scale_plan_changed"
	ErrorCodeScaleRestartAckRequired        = "scale_restart_ack_required"
	ErrorCodeScaleJobsAckRequired           = "scale_jobs_risk_ack_required"
)

伸缩操作对外错误编码,前端据此本地化。 Public error codes of scaling operations, localized by the frontend.

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 节点")

	// ErrClusterScaleInProgress 集群已有进行中的伸缩操作(409 cluster_scale_in_progress)。
	// ErrClusterScaleInProgress means the cluster already has a running scaling operation (409 cluster_scale_in_progress).
	ErrClusterScaleInProgress = errors.New("cluster: another scaling operation is running on this cluster")
	// ErrScaleRollingRestartUnavailable 选择了滚动重启但计划不允许(无可重启节点或有运行中作业)。
	// ErrScaleRollingRestartUnavailable means a rolling restart was requested but the plan does not allow it (no node
	// to restart, or running jobs).
	ErrScaleRollingRestartUnavailable = errors.New("cluster: rolling restart is not available for this plan")
	// ErrScaleOperationNotFound 伸缩操作不存在或无权查看。
	// ErrScaleOperationNotFound means the scaling operation does not exist or is not visible to the caller.
	ErrScaleOperationNotFound = errors.New("cluster: scaling operation not found")
	// ErrScaleOperationCorrupted 伸缩操作的计划快照无法解析。
	// ErrScaleOperationCorrupted means the scaling operation's plan snapshot cannot be decoded.
	ErrScaleOperationCorrupted = errors.New("cluster: scaling operation plan snapshot is invalid")
	// ErrScaleOperationNotRetryable 只有失败的伸缩操作可以重试。
	// ErrScaleOperationNotRetryable means only failed scaling operations can be retried.
	ErrScaleOperationNotRetryable = errors.New("cluster: only failed scaling operations can be retried")
	// ErrScaleInvalidRequest 伸缩请求参数非法(如模式、角色)。
	// ErrScaleInvalidRequest means the scaling request is invalid (e.g. mode or role).
	ErrScaleInvalidRequest = errors.New("cluster: invalid scaling request")
	// ErrScalePlanBlocked 计划存在阻断项,不能提交。
	// ErrScalePlanBlocked means the plan has blocking items and cannot be submitted.
	ErrScalePlanBlocked = errors.New("cluster: scaling plan has blocking items")
	// ErrScalePlanChanged 提交的计划指纹与当前重算结果不一致,需要重新预检。
	// ErrScalePlanChanged means the submitted plan fingerprint differs from the recomputed plan; precheck again.
	ErrScalePlanChanged = errors.New("cluster: scaling plan changed since precheck, please precheck again")
	// ErrScaleRestartAckRequired 存在回环成员但未确认重启现有节点。
	// ErrScaleRestartAckRequired means loopback members exist but the restart of existing nodes was not acknowledged.
	ErrScaleRestartAckRequired = errors.New("cluster: restarting existing nodes with loopback member addresses must be acknowledged")
	// ErrScaleJobsRiskAckRequired 存在运行中作业(或无法确认)但未确认风险。
	// ErrScaleJobsRiskAckRequired means running jobs exist (or are unknown) but the risk was not acknowledged.
	ErrScaleJobsRiskAckRequired = errors.New("cluster: running jobs risk must be acknowledged before removing the node")
	// ErrScaleNotConfigured 伸缩编排缺少必要依赖。
	// ErrScaleNotConfigured means the scaling orchestrator lacks a required dependency.
	ErrScaleNotConfigured = errors.New("cluster: scaling orchestrator is not fully configured")
)

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 根据节点角色和部署模式返回默认端口

func IsValidClusterCode added in v1.3.0

func IsValidClusterCode(code string) bool

IsValidClusterCode 校验集群编码是否符合规范(2-32位纯小写字母与数字)。 IsValidClusterCode validates if cluster code matches format (2-32 chars of lowercase letters and digits).

func SanitizeClusterCode added in v1.3.0

func SanitizeClusterCode(raw string, fallbackID uint) string

SanitizeClusterCode 清洗并生成合法的集群编码。 规则:保留英文字母与数字,转小写;若清洗后有效字符不足2位,则使用 "stxcluster" 结合 fallbackID 兜底;最大截断至 32 位。 SanitizeClusterCode sanitizes and generates a valid cluster code. Rule: preserves letters & digits in lowercase; falls back to "stxcluster" + fallbackID if < 2 chars; max 32 chars.

func TryExtractClusterNameFromLocalInstallDir added in v1.3.0

func TryExtractClusterNameFromLocalInstallDir(installDir string) string

TryExtractClusterNameFromLocalInstallDir 尝试从本地安装目录下的 Hazelcast 配置文件解析 cluster-name。 TryExtractClusterNameFromLocalInstallDir tries to extract cluster-name from Hazelcast config under local installDir.

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 覆盖配置
	// JavaHome 是该节点指定的 JDK 目录,为空时自动探测。/ JavaHome is the node's JDK directory; empty means auto detection.
	JavaHome string `json:"java_home"`

	// 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"`
	// JavaHome 是该主机上这些节点共用的 JDK 目录(同一安装目录只能有一个 JDK),为空时自动探测。
	// JavaHome is the JDK directory shared by these nodes on the host (one install dir has one JDK); empty means
	// auto detection.
	JavaHome string `json:"java_home"`
}

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"`
	// ClusterCode 是底层的不可变集群标识,仅包含字符与数字 [a-z0-9]
	// ClusterCode is the immutable engine cluster identifier containing only characters and digits [a-z0-9]
	ClusterCode    string         `json:"cluster_code" gorm:"column:cluster_code;size:64;index"`
	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) EffectiveClusterCode added in v1.3.0

func (c *Cluster) EffectiveClusterCode() string

EffectiveClusterCode 返回当前集群的有效底层标识;若为空则返回兜底默认值 "seatunnel"。 EffectiveClusterCode returns the effective engine cluster code; falls back to "seatunnel" if empty.

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"`
	ClusterCode    string         `json:"cluster_code"`
	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 ClusterMonitorPusher added in v1.2.0

type ClusterMonitorPusher interface {
	PushClusterMonitorConfigToHost(ctx context.Context, clusterID uint, hostID uint) error
}

ClusterMonitorPusher 按集群向主机 Agent 下发监控跟踪列表(由 gRPC Server 实现)。 ClusterMonitorPusher pushes a cluster-scoped monitor tracking list to a host Agent (implemented by the gRPC server).

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
	// JavaHome 是该节点生效 JDK 的唯一存储位置:为空表示沿用自动探测;SeaTunnel、stx-java-proxy、预检与诊断工具都读这里。
	// JavaHome is the single storage place of this node's effective JDK: empty keeps auto detection; SeaTunnel,
	// stx-java-proxy, precheck and diagnostic tools all read it.
	JavaHome    string     `json:"java_home" gorm:"size:512"`
	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 ClusterScaleOperation added in v1.2.0

type ClusterScaleOperation struct {
	ID          uint                 `json:"id" gorm:"primaryKey;autoIncrement"`
	ExecutionID string               `json:"execution_id" gorm:"size:64;index"`
	ClusterID   uint                 `json:"cluster_id" gorm:"index;not null"`
	HostID      uint                 `json:"host_id" gorm:"index"`
	NodeIDs     string               `json:"-" gorm:"type:text"` // JSON []uint
	Kind        ScaleKind            `json:"kind" gorm:"size:20;not null"`
	Mode        ScaleMode            `json:"mode" gorm:"size:20"`
	Status      ScaleOperationStatus `json:"status" gorm:"size:20;index"`
	CurrentStep ScaleStepName        `json:"current_step" gorm:"size:40"`
	// LoopbackRestartAck / JobsRiskAck 记录提交时用户的显式确认,执行与重试时据此复核计划护栏。
	// LoopbackRestartAck / JobsRiskAck record the user's explicit acknowledgements at submission, re-checked by
	// the plan guardrails during execution and retry.
	LoopbackRestartAck bool `json:"loopback_restart_ack" gorm:"not null"`
	JobsRiskAck        bool `json:"jobs_risk_ack" gorm:"not null"`
	// RollingRestart 记录用户是否选择了扩容后的滚动重启。/ RollingRestart records whether the user chose the rolling restart.
	RollingRestart bool `json:"rolling_restart" gorm:"not null;default:false"`
	// Lang 是提交时的界面语言(zh/en),执行器据此输出单语言步骤说明。
	// Lang is the UI language at submission (zh/en); the executor writes single-language step messages with it.
	Lang      string        `json:"lang" gorm:"size:8"`
	PlanJSON  db.ScriptText `json:"-"` // 提交时的 Plan 快照(不含敏感内容) / plan snapshot at submission (no secrets)
	StepsJSON db.ScriptText `json:"-"` // []ScaleStepState
	Error     string        `json:"error" gorm:"type:text"`
	CreatedBy uint          `json:"created_by" gorm:"index"`
	// CreatedByName 是发起人登录名快照,用于并发冲突提示(用户改名不回溯)。
	// CreatedByName is a snapshot of the initiator's login name for conflict messages (not updated on rename).
	CreatedByName string     `json:"created_by_name" gorm:"size:100"`
	CreatedAt     time.Time  `json:"created_at" gorm:"autoCreateTime"`
	UpdatedAt     time.Time  `json:"updated_at" gorm:"autoUpdateTime"`
	FinishedAt    *time.Time `json:"finished_at"`
}

ClusterScaleOperation 伸缩操作(扩容/缩容)及其分步状态。 ClusterScaleOperation is a scale-out/scale-in operation with per-step status.

func (ClusterScaleOperation) TableName added in v1.2.0

func (ClusterScaleOperation) TableName() string

TableName 指定伸缩操作表名。 TableName returns the scaling operation table name.

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"`
	// RunningNodes 是主机在线且 SeaTunnel 进程运行中的节点数。
	// RunningNodes counts nodes whose host is online and whose SeaTunnel process is running.
	RunningNodes int `json:"running_nodes"`
	// HealthReasons 解释 health_status 不健康的原因(离线、进程未运行、成员分裂)。
	// HealthReasons explains why health_status is not healthy (offline, process down, split members).
	HealthReasons []string          `json:"health_reasons,omitempty"`
	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"`
	ClusterCode    string         `json:"cluster_code,omitempty"`
	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 端口(可选)
	ClusterName   string `json:"cluster_name,omitempty"`   // Hazelcast cluster-name (optional) / Hazelcast 集群标识(可选)
	// JavaHome 是纳管时给定的节点 JDK;空表示自动探测。优先填进程实际 JDK(AC1e / AC1f)。
	// JavaHome is the node JDK given at registration; empty means auto detection. Prefer the process's actual JDK (AC1e / AC1f).
	JavaHome string `json:"java_home,omitempty"`
}

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 GetNodeJavaRuntimeResponse added in v1.3.0

type GetNodeJavaRuntimeResponse struct {
	ErrorMsg string           `json:"error_msg"`
	Data     *NodeJavaRuntime `json:"data"`
}

GetNodeJavaRuntimeResponse 是节点 JDK 查询的响应。 GetNodeJavaRuntimeResponse is the response of the node JDK query.

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) GetNodeJavaRuntime added in v1.3.0

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

GetNodeJavaRuntime 返回节点实际生效的 JDK 路径与版本(AC7)。 GetNodeJavaRuntime returns the node's effective JDK path and version (AC7). @Summary 查询节点生效的 JDK / Get the node's effective JDK @Tags clusters @Produce json @Param id path int true "集群 ID / Cluster ID" @Param nodeId path int true "节点 ID / Node ID" @Success 200 {object} GetNodeJavaRuntimeResponse @Router /api/v1/clusters/{id}/nodes/{nodeId}/java-runtime [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) GetWebUIEndpoints added in v1.3.0

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

GetWebUIEndpoints 处理 GET /api/v1/clusters/:id/webui-endpoints:返回代理地址与原生地址(与代理选择同一个节点,优先 active master)。 GetWebUIEndpoints handles GET /api/v1/clusters/:id/webui-endpoints: it returns the proxy and native addresses (the same node the proxy uses, preferring the active master). @Summary 获取 SeaTunnel 集群观测地址 @Tags clusters @Produce json @Param id path int true "集群ID" @Success 200 {object} WebUIEndpointsResponse @Failure 400 {object} WebUIEndpointsResponse @Failure 404 {object} WebUIEndpointsResponse @Router /api/v1/clusters/{id}/webui-endpoints [get]

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)
	// DiskUsage 是 Agent 心跳上报的磁盘使用率(百分比,0 表示未知),扩容预检据此给出磁盘空间警告。
	// DiskUsage is the disk usage percentage reported by Agent heartbeats (0 means unknown); scale-out prechecks use
	// it for the disk space warning.
	DiskUsage float64
}

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
	JavaHome      string        `json:"java_home"`      // 节点指定的 JDK,为空表示自动探测 / Node JDK; empty means auto detection
	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 NodeJavaRuntime added in v1.3.0

type NodeJavaRuntime struct {
	NodeID     uint   `json:"node_id"`
	HostID     uint   `json:"host_id"`
	Configured string `json:"configured_java_home"`
	javaruntime.Result
}

NodeJavaRuntime 是节点实际生效的 JDK:配置值来自 cluster_nodes.java_home,路径与版本由 Agent 现场检查得出(AC7)。 NodeJavaRuntime is the node's effective JDK: the configured value comes from cluster_nodes.java_home, and the path and version are checked live by the Agent (AC7).

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
	// MemberCount 是该节点看到的 Hazelcast 成员数;未知(Agent 不支持或 REST 不可达)时为空。
	// MemberCount is the number of Hazelcast members seen by this node; nil when unknown.
	MemberCount *int `json:"member_count,omitempty"`
}

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 RecordRemovalCleanup added in v1.2.0

type RecordRemovalCleanup struct {
	// MembersPatched 表示剩余节点的 member-list 已修补;MembersSkippedReason 说明未修补的原因。
	// MembersPatched means the remaining nodes' member lists were patched; MembersSkippedReason explains why not.
	MembersPatched       bool     `json:"members_patched"`
	MembersSkippedReason string   `json:"members_skipped_reason,omitempty"`
	DeletedConfigs       int      `json:"deleted_configs"`
	MonitorRefreshed     bool     `json:"monitor_refreshed"`
	Warnings             []string `json:"warnings"`
}

RecordRemovalCleanup 是「仅移除记录」后清理的结果。/ RecordRemovalCleanup is the result of cleaning up after "remove record only".

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) GetNodeByHostAndInstallDir added in v1.3.0

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

GetNodeByHostAndInstallDir 按主机与安装目录查找任一节点(发现匹配用,不区分角色)。 GetNodeByHostAndInstallDir finds any node by host and install directory (for discovery matching; role-agnostic). Returns clusterID, nodeID, found, error

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 RunningJobsProvider added in v1.2.0

type RunningJobsProvider interface {
	ListClusterRunningJobs(ctx context.Context, clusterID uint) ([]ScaleRunningJob, error)
}

RunningJobsProvider 按集群维度读取运行中作业;由 router 适配 sync 服务注入,避免 cluster 直接依赖 sync。 RunningJobsProvider reads running jobs per cluster; router injects a sync adapter so cluster never imports sync.

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 ScaleCheck added in v1.2.0

type ScaleCheck struct {
	Name     string           `json:"name"`
	Status   ScaleCheckStatus `json:"status"`
	Blocking bool             `json:"blocking"`
	Message  string           `json:"message"`
	Target   string           `json:"target,omitempty"`
}

ScaleCheck 是一个预检项;Name 为稳定编码,Blocking 表示该项失败时阻断提交。 ScaleCheck is one precheck item; Name is a stable code and Blocking means a failure blocks submission.

type ScaleCheckStatus added in v1.2.0

type ScaleCheckStatus string

ScaleCheckStatus 是预检项结果。 ScaleCheckStatus is the result of one precheck item.

const (
	ScaleCheckPassed  ScaleCheckStatus = "passed"
	ScaleCheckFailed  ScaleCheckStatus = "failed"
	ScaleCheckWarning ScaleCheckStatus = "warning"
	ScaleCheckSkipped ScaleCheckStatus = "skipped"
)

type ScaleCheckpointStorage added in v1.2.0

type ScaleCheckpointStorage struct {
	Type     string `json:"type"`
	Location string `json:"location"`
	Local    bool   `json:"local"`
}

ScaleCheckpointStorage 描述集群 checkpoint 存储;Local 表示本地文件存储,作业 failover 后无法从 checkpoint 恢复。 ScaleCheckpointStorage describes the cluster checkpoint storage; Local means local file storage, so failed-over jobs cannot restore from checkpoints.

type ScaleConfigManager added in v1.2.0

type ScaleConfigManager interface {
	GetByCluster(ctx context.Context, clusterID uint) ([]*appconfig.ConfigInfo, error)
	PatchMemberList(ctx context.Context, clusterID uint, deploymentMode string, members []string, opRef string, userID uint) (*appconfig.MemberListPatchResult, error)
	ApplyTemplatesToHost(ctx context.Context, clusterID uint, hostID uint, deploymentMode string, members []string, opRef string, userID uint) (*appconfig.ConfigChangeResult, error)
	RegisterHostConfigsFromNode(ctx context.Context, clusterID uint, hostID uint, installDir string, userID uint) (*appconfig.ConfigChangeResult, error)
	DeleteHostConfigs(ctx context.Context, clusterID uint, hostID uint) (int, error)
}

ScaleConfigManager 是伸缩编排使用的配置服务能力(由 config.Service 实现)。 ScaleConfigManager is the config service capability used by the scaling orchestrator (implemented by config.Service).

type ScaleDiscoveredProcess added in v1.2.0

type ScaleDiscoveredProcess struct {
	PID           int
	Role          string
	InstallDir    string
	Version       string
	HazelcastPort int
	APIPort       int
	ClusterName   string
	// JavaHome 是进程实际使用的 JDK;纳管时优先预选(AC1f)。
	// JavaHome is the JDK the process actually uses; register mode prefers it as the preselection (AC1f).
	JavaHome string
}

ScaleDiscoveredProcess 是主机上发现的一个 SeaTunnel 进程。 ScaleDiscoveredProcess is one SeaTunnel process discovered on a host.

type ScaleExecutionRequest added in v1.2.0

type ScaleExecutionRequest struct {
	RequestID      string
	IdempotencyKey string
	RequestHash    string
	Confirmed      bool
	ConfirmationID string
	ClientType     string
	Lang           string
	// Username 是发起人登录名,随操作记录保存,用于并发冲突时告知其他用户是谁在操作。
	// Username is the initiator's login name, stored with the operation so a conflict can tell other users who
	// is running it.
	Username string
}

ScaleExecutionRequest 保存伸缩写请求的公共执行元数据(不含请求正文)。 ScaleExecutionRequest stores shared-execution metadata of a scaling write request (no request body).

type ScaleHandler added in v1.2.0

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

ScaleHandler 提供集群扩缩容的 HTTP 接口。 ScaleHandler serves the cluster scaling HTTP endpoints.

func NewScaleHandler added in v1.2.0

func NewScaleHandler(service *ScaleService, auditRepo *audit.Repository) *ScaleHandler

NewScaleHandler 创建扩缩容处理器;auditRepo 为 nil 时跳过业务审计(公共执行审计仍会记录)。 NewScaleHandler creates the scaling handler; with a nil auditRepo the business audit is skipped (shared execution audits are still recorded).

func (*ScaleHandler) GetScaleOperation added in v1.2.0

func (h *ScaleHandler) GetScaleOperation(c *gin.Context)

GetScaleOperation 处理 GET /api/v1/clusters/:id/scale-operations/:opId,返回操作详情与步骤。 GetScaleOperation handles GET /api/v1/clusters/:id/scale-operations/:opId and returns details and steps. @Summary 查询伸缩操作 / Get a scaling operation @Tags clusters @Produce json @Param id path int true "集群ID" @Param opId path int true "伸缩操作ID" @Success 200 {object} ScaleOperationResponse @Router /api/v1/clusters/{id}/scale-operations/{opId} [get]

func (*ScaleHandler) ListScaleOperations added in v1.2.0

func (h *ScaleHandler) ListScaleOperations(c *gin.Context)

ListScaleOperations 处理 GET /api/v1/clusters/:id/scale-operations,可按 status 过滤(页面刷新后恢复进行中操作)。 ListScaleOperations handles GET /api/v1/clusters/:id/scale-operations, optionally filtered by status (used to resume running operations after a page refresh). @Summary 列出伸缩操作 / List scaling operations @Tags clusters @Produce json @Param id path int true "集群ID" @Param status query string false "running / succeeded / failed" @Param limit query int false "最多返回条数(默认 20,最大 100)" @Success 200 {object} ScaleOperationListResponse @Router /api/v1/clusters/{id}/scale-operations [get]

func (*ScaleHandler) PlanScaleIn added in v1.2.0

func (h *ScaleHandler) PlanScaleIn(c *gin.Context)

PlanScaleIn 处理 POST /api/v1/clusters/:id/nodes/:nodeId/scale-in/plan,返回缩容计划(R0,只读)。 PlanScaleIn handles POST /api/v1/clusters/:id/nodes/:nodeId/scale-in/plan and returns a scale-in plan (R0). @Summary 计算缩容计划 / Compute a scale-in plan @Tags clusters @Produce json @Param id path int true "集群ID" @Param nodeId path int true "节点ID" @Param lang query string false "zh / en" @Success 200 {object} ScalePlanResponse @Router /api/v1/clusters/{id}/nodes/{nodeId}/scale-in/plan [post]

func (*ScaleHandler) PlanScaleOut added in v1.2.0

func (h *ScaleHandler) PlanScaleOut(c *gin.Context)

PlanScaleOut 处理 POST /api/v1/clusters/:id/scale-out/plan,返回扩容计划(R0,只读)。 PlanScaleOut handles POST /api/v1/clusters/:id/scale-out/plan and returns a scale-out plan (R0, read-only). @Summary 计算扩容计划 / Compute a scale-out plan @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param lang query string false "zh / en" @Param request body ScaleOutRequest true "扩容计划请求" @Success 200 {object} ScalePlanResponse @Router /api/v1/clusters/{id}/scale-out/plan [post]

func (*ScaleHandler) RetryScaleOperation added in v1.2.0

func (h *ScaleHandler) RetryScaleOperation(c *gin.Context)

RetryScaleOperation 处理 POST /api/v1/clusters/:id/scale-operations/:opId/retry,从失败步骤重试(R1)。 RetryScaleOperation handles POST /api/v1/clusters/:id/scale-operations/:opId/retry and retries from the failed step (R1). @Summary 重试伸缩操作 / Retry a scaling operation @Tags clusters @Produce json @Param id path int true "集群ID" @Param opId path int true "伸缩操作ID" @Param lang query string false "zh / en" @Param Idempotency-Key header string true "幂等键" @Param X-STX-Confirm header string true "显式确认(true)" @Success 200 {object} ScaleSubmitResponse @Router /api/v1/clusters/{id}/scale-operations/{opId}/retry [post]

func (*ScaleHandler) ScaleIn added in v1.2.0

func (h *ScaleHandler) ScaleIn(c *gin.Context)

ScaleIn 处理 POST /api/v1/clusters/:id/nodes/:nodeId/scale-in,提交缩容(R2,两步确认,异步执行)。 ScaleIn handles POST /api/v1/clusters/:id/nodes/:nodeId/scale-in and submits a scale-in (R2, two-step, async). @Summary 提交缩容 / Submit a scale-in @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param nodeId path int true "节点ID" @Param lang query string false "zh / en" @Param Idempotency-Key header string true "幂等键" @Param X-STX-Confirmation-ID header string false "一次性确认编号" @Param request body ScaleInSubmitRequest true "缩容提交请求" @Success 200 {object} ScaleSubmitResponse @Router /api/v1/clusters/{id}/nodes/{nodeId}/scale-in [post]

func (*ScaleHandler) ScaleOut added in v1.2.0

func (h *ScaleHandler) ScaleOut(c *gin.Context)

ScaleOut 处理 POST /api/v1/clusters/:id/scale-out,提交扩容(R2,两步确认,异步执行)。 ScaleOut handles POST /api/v1/clusters/:id/scale-out and submits a scale-out (R2, two-step confirmation, async). @Summary 提交扩容 / Submit a scale-out @Tags clusters @Accept json @Produce json @Param id path int true "集群ID" @Param lang query string false "zh / en" @Param Idempotency-Key header string true "幂等键" @Param X-STX-Confirmation-ID header string false "一次性确认编号" @Param request body ScaleOutSubmitRequest true "扩容提交请求" @Success 200 {object} ScaleSubmitResponse @Router /api/v1/clusters/{id}/scale-out [post]

type ScaleInProgressError added in v1.2.0

type ScaleInProgressError struct {
	OperationID   uint      `json:"operation_id"`
	Kind          ScaleKind `json:"kind"`
	CreatedBy     uint      `json:"created_by"`
	CreatedByName string    `json:"created_by_name"`
}

ScaleInProgressError 描述集群上正在进行的伸缩操作,用于 409 响应告知调用方是谁在操作。 它包装 ErrClusterScaleInProgress,errors.Is 判断保持不变。 ScaleInProgressError describes the scaling operation already running on a cluster, so a 409 response can tell the caller who is running it. It wraps ErrClusterScaleInProgress, so errors.Is checks keep working.

func (*ScaleInProgressError) Error added in v1.2.0

func (e *ScaleInProgressError) Error() string

Error 返回中英双语说明;拿不到用户名时显示用户 ID。 Error returns a bilingual message; the user ID is shown when the login name is unknown.

func (*ScaleInProgressError) Unwrap added in v1.2.0

func (e *ScaleInProgressError) Unwrap() error

Unwrap 让 errors.Is(err, ErrClusterScaleInProgress) 继续成立。 Unwrap keeps errors.Is(err, ErrClusterScaleInProgress) true.

type ScaleInSubmitRequest added in v1.2.0

type ScaleInSubmitRequest struct {
	PlanFingerprint string `json:"plan_fingerprint" binding:"required"`
	JobsRiskAck     bool   `json:"jobs_risk_ack"`
}

ScaleInSubmitRequest 是缩容提交请求:计划指纹 + 运行中作业风险确认。 ScaleInSubmitRequest is a scale-in submission: plan fingerprint + running jobs risk ack.

type ScaleInstaller added in v1.2.0

type ScaleInstaller interface {
	StartInstallation(ctx context.Context, req *installerapp.InstallationRequest) (*installerapp.InstallationStatus, error)
	GetInstallationStatusSnapshot(ctx context.Context, hostID uint) (*installerapp.InstallationStatus, error)
}

ScaleInstaller 是扩容安装模式使用的安装器能力(由 installer.Service 实现)。 ScaleInstaller is the installer capability used by install-mode scale-out (implemented by installer.Service).

type ScaleJavaProxyFollow added in v1.2.0

type ScaleJavaProxyFollow struct {
	FromNodeID     uint   `json:"from_node_id"`
	FromHostID     uint   `json:"from_host_id"`
	FromHostName   string `json:"from_host_name"`
	FromInstallDir string `json:"from_install_dir"`
	ToNodeID       uint   `json:"to_node_id"`
	ToHostID       uint   `json:"to_host_id"`
	ToHostName     string `json:"to_host_name"`
}

ScaleJavaProxyFollow 描述缩容后 java-proxy 承载节点的迁移。 ScaleJavaProxyFollow describes moving the java-proxy host after scale-in.

type ScaleKind added in v1.2.0

type ScaleKind string

ScaleKind 是伸缩操作类型:扩容或缩容。 ScaleKind is the scaling operation type: scale-out or scale-in.

const (
	// ScaleKindOut 扩容(添加节点)。
	// ScaleKindOut is a scale-out (add a node).
	ScaleKindOut ScaleKind = "scale_out"
	// ScaleKindIn 缩容(移除节点)。
	// ScaleKindIn is a scale-in (remove a node).
	ScaleKindIn ScaleKind = "scale_in"
)

type ScaleLoopbackNode added in v1.2.0

type ScaleLoopbackNode struct {
	NodeID   uint   `json:"node_id"`
	HostID   uint   `json:"host_id"`
	HostName string `json:"host_name"`
	Address  string `json:"address"`
	// Source 为 config(节点配置)或 runtime(Hazelcast 运行时成员)。
	// Source is config (node config) or runtime (Hazelcast runtime members).
	Source string `json:"source"`
}

ScaleLoopbackNode 是 member-list 或运行时成员中含回环地址的现有节点。 ScaleLoopbackNode is an existing node whose member-list or runtime members contain a loopback address.

type ScaleMode added in v1.2.0

type ScaleMode string

ScaleMode 是扩容模式:在主机上安装 SeaTunnel 后入群,或纳管主机上已运行的 SeaTunnel。 ScaleMode is the scale-out mode: install SeaTunnel on the host and join, or register an already running one.

const (
	// ScaleModeInstall 安装并加入集群。
	// ScaleModeInstall installs SeaTunnel and joins the cluster.
	ScaleModeInstall ScaleMode = "install"
	// ScaleModeRegister 纳管已运行的节点(以节点为准,只登记不覆盖)。
	// ScaleModeRegister registers an already running node (the node is the truth; register only, never overwrite).
	ScaleModeRegister ScaleMode = "register"
)

type ScaleNodePorts added in v1.2.0

type ScaleNodePorts struct {
	HazelcastPort int `json:"hazelcast_port"`
	APIPort       int `json:"api_port"`
	WorkerPort    int `json:"worker_port"`
}

ScaleNodePorts 是新节点的端口;为 0 时依次取发现结果、集群端口配置、同角色节点与角色默认值。 ScaleNodePorts are the new node's ports; zero values fall back to discovery, cluster port config, a same-role node and the role default, in that order.

type ScaleOperationFilter added in v1.2.0

type ScaleOperationFilter struct {
	ClusterID   uint
	Status      ScaleOperationStatus
	OwnerUserID uint
	IncludeAll  bool
	Limit       int
}

ScaleOperationFilter 是伸缩操作列表的过滤条件;OwnerUserID 与 IncludeAll 在仓库查询时限定归属。 ScaleOperationFilter filters scaling operation lists; OwnerUserID and IncludeAll scope ownership in the query.

type ScaleOperationInfo added in v1.2.0

type ScaleOperationInfo struct {
	ID                 uint                 `json:"id"`
	ExecutionID        string               `json:"execution_id"`
	ClusterID          uint                 `json:"cluster_id"`
	HostID             uint                 `json:"host_id"`
	NodeIDs            []uint               `json:"node_ids"`
	Kind               ScaleKind            `json:"kind"`
	Mode               ScaleMode            `json:"mode,omitempty"`
	Status             ScaleOperationStatus `json:"status"`
	CurrentStep        ScaleStepName        `json:"current_step"`
	LoopbackRestartAck bool                 `json:"loopback_restart_ack"`
	JobsRiskAck        bool                 `json:"jobs_risk_ack"`
	RollingRestart     bool                 `json:"rolling_restart"`
	Error              string               `json:"error,omitempty"`
	// Warnings 汇总 partial 步骤的说明,操作仍按成功结束。
	// Warnings summarizes partial steps; the operation still finishes as succeeded.
	Warnings  []string         `json:"warnings"`
	Retryable bool             `json:"retryable"`
	Plan      *ScalePlan       `json:"plan"`
	Steps     []ScaleStepState `json:"steps"`
	CreatedBy uint             `json:"created_by"`
	// CreatedByName 发起人登录名快照 / initiator login name snapshot
	CreatedByName string     `json:"created_by_name"`
	CreatedAt     time.Time  `json:"created_at"`
	UpdatedAt     time.Time  `json:"updated_at"`
	FinishedAt    *time.Time `json:"finished_at,omitempty"`
}

ScaleOperationInfo 是伸缩操作的 API 视图:计划与步骤已解析,数组字段永不为 null。 ScaleOperationInfo is the API view of a scaling operation with the plan and steps decoded; array fields are never null.

type ScaleOperationListData added in v1.2.0

type ScaleOperationListData struct {
	Items []*ScaleOperationInfo `json:"items"`
}

ScaleOperationListData 是伸缩操作列表数据,Items 永不为 null。 ScaleOperationListData is the scaling operation list payload; Items is never null.

type ScaleOperationListResponse added in v1.2.0

type ScaleOperationListResponse struct {
	ErrorCode string                  `json:"error_code,omitempty"`
	ErrorMsg  string                  `json:"error_msg"`
	Data      *ScaleOperationListData `json:"data"`
}

ScaleOperationListResponse 是伸缩操作列表响应。 ScaleOperationListResponse is the scaling operation list response.

type ScaleOperationResponse added in v1.2.0

type ScaleOperationResponse struct {
	ErrorCode string              `json:"error_code,omitempty"`
	ErrorMsg  string              `json:"error_msg"`
	Data      *ScaleOperationInfo `json:"data"`
}

ScaleOperationResponse 是伸缩操作详情响应。 ScaleOperationResponse is the scaling operation detail response.

type ScaleOperationStatus added in v1.2.0

type ScaleOperationStatus string

ScaleOperationStatus 是伸缩操作整体状态。 ScaleOperationStatus is the overall status of a scaling operation.

const (
	ScaleOperationRunning   ScaleOperationStatus = "running"
	ScaleOperationSucceeded ScaleOperationStatus = "succeeded"
	ScaleOperationFailed    ScaleOperationStatus = "failed"
)

type ScaleOutRequest added in v1.2.0

type ScaleOutRequest struct {
	HostID uint `json:"host_id" binding:"required"`
	// Mode 为 install / register;为空时使用自动识别结果。
	// Mode is install / register; empty uses the detected mode.
	Mode ScaleMode `json:"mode"`
	// Role 仅分离模式需要(master / worker);混合模式固定为 master/worker。
	// Role is only required in separated mode (master / worker); hybrid always uses master/worker.
	Role       NodeRole       `json:"role"`
	InstallDir string         `json:"install_dir"`
	Ports      ScaleNodePorts `json:"ports"`
	// JavaHome 是新节点指定的 JDK 目录;为空时自动探测,不继承已有节点的路径(AC1d)。
	// JavaHome is the new node's JDK directory; empty means auto detection and never inherits an existing node's
	// path (AC1d).
	JavaHome string `json:"java_home"`
}

ScaleOutRequest 是扩容计划请求。 ScaleOutRequest is a scale-out plan request.

type ScaleOutSubmitRequest added in v1.2.0

type ScaleOutSubmitRequest struct {
	ScaleOutRequest
	PlanFingerprint    string `json:"plan_fingerprint" binding:"required"`
	LoopbackRestartAck bool   `json:"loopback_restart_ack"`
	// RollingRestart 入群后滚动重启现有节点(可选,见 ScalePlan.RollingRestart)。
	// RollingRestart restarts existing nodes one by one after the join (optional, see ScalePlan.RollingRestart).
	RollingRestart bool `json:"rolling_restart"`
}

ScaleOutSubmitRequest 是扩容提交请求:计划参数 + 计划指纹 + 回环重启确认。 ScaleOutSubmitRequest is a scale-out submission: plan parameters + plan fingerprint + loopback restart ack.

type ScalePlan added in v1.2.0

type ScalePlan struct {
	Kind               ScaleKind               `json:"kind"`
	Mode               ScaleMode               `json:"mode,omitempty"`
	DetectedMode       ScaleMode               `json:"detected_mode,omitempty"`
	ClusterID          uint                    `json:"cluster_id"`
	DeploymentMode     DeploymentMode          `json:"deployment_mode"`
	ClusterVersion     string                  `json:"cluster_version"`
	Target             ScaleTarget             `json:"target"`
	Fingerprint        string                  `json:"fingerprint"`
	MembersBefore      []string                `json:"members_before"`
	MembersAfter       []string                `json:"members_after"`
	Checks             []ScaleCheck            `json:"checks"`
	LoopbackNodes      []ScaleLoopbackNode     `json:"loopback_nodes"`
	RequiresRestartAck bool                    `json:"requires_restart_ack"`
	PluginsToReplay    []ScalePluginItem       `json:"plugins_to_replay"`
	RunningJobs        []ScaleRunningJob       `json:"running_jobs"`
	RunningJobsKnown   bool                    `json:"running_jobs_known"`
	RequiresJobsAck    bool                    `json:"requires_jobs_ack"`
	CheckpointStorage  *ScaleCheckpointStorage `json:"checkpoint_storage"`
	JavaProxyFollow    *ScaleJavaProxyFollow   `json:"java_proxy_follow"`
	// RollingRestart 只用于扩容:入群后可选的滚动重启。/ Scale-out only: the optional rolling restart after the join.
	RollingRestart  *ScaleRollingRestart `json:"rolling_restart"`
	Steps           []ScaleStepName      `json:"steps"`
	CanSubmit       bool                 `json:"can_submit"`
	BlockingReasons []string             `json:"blocking_reasons"`
	Warnings        []string             `json:"warnings"`
	// Lang 是生成本计划文案所用的语言,不参与序列化。
	// Lang is the language used for this plan's texts and is not serialized.
	Lang string `json:"-"`
}

ScalePlan 是伸缩计划:预检结果、成员变化、护栏与执行步骤。所有数组字段输出 [],不输出 null。 ScalePlan is a scaling plan: precheck results, member changes, guardrails and execution steps. Every array field is rendered as [] instead of null.

type ScalePlanResponse added in v1.2.0

type ScalePlanResponse struct {
	ErrorCode string     `json:"error_code,omitempty"`
	ErrorMsg  string     `json:"error_msg"`
	Data      *ScalePlan `json:"data"`
}

ScalePlanResponse 是伸缩计划响应。 ScalePlanResponse is the scaling plan response.

type ScalePluginItem added in v1.2.0

type ScalePluginItem struct {
	Name           string   `json:"name"`
	ArtifactID     string   `json:"artifact_id"`
	Version        string   `json:"version"`
	Status         string   `json:"status"`
	ProfileKeys    []string `json:"profile_keys"`
	ProfileUnknown bool     `json:"profile_unknown"`
	// ProfileInferred 画像由控制面缓存中的依赖反推 / profile inferred from dependencies cached on the Control Plane
	ProfileInferred bool `json:"profile_inferred"`
}

ScalePluginItem 是新节点需要补齐的集群插件。 ScalePluginItem is a cluster plugin the joining node must receive.

type ScalePluginReplayer added in v1.2.0

type ScalePluginReplayer interface {
	ListReplayCandidates(ctx context.Context, clusterID uint) ([]pluginapp.PluginReplayCandidate, error)
	ReplayClusterPluginsToHost(ctx context.Context, clusterID uint, hostID uint, installDir string) (*pluginapp.PluginReplayReport, error)
}

ScalePluginReplayer 是新节点补齐集群插件的能力(由 plugin.Service 实现)。 ScalePluginReplayer replays cluster plugins to a joining node (implemented by plugin.Service).

type ScaleProcessDiscoverer added in v1.2.0

type ScaleProcessDiscoverer interface {
	DiscoverSeaTunnelProcesses(ctx context.Context, hostID uint) ([]ScaleDiscoveredProcess, error)
}

ScaleProcessDiscoverer 读取主机上运行中的 SeaTunnel 进程(由 router 适配 discovery 服务注入)。 ScaleProcessDiscoverer lists running SeaTunnel processes on a host (router injects a discovery adapter).

type ScaleRepository added in v1.2.0

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

ScaleRepository 是伸缩操作的数据访问层。 ScaleRepository is the data access layer of scaling operations.

func NewScaleRepository added in v1.2.0

func NewScaleRepository(database *gorm.DB) *ScaleRepository

NewScaleRepository 创建伸缩操作仓库。 NewScaleRepository creates a scaling operation repository.

func (*ScaleRepository) Create added in v1.2.0

Create 新建伸缩操作。 Create inserts a scaling operation.

func (*ScaleRepository) FindRunningByCluster added in v1.2.0

func (r *ScaleRepository) FindRunningByCluster(ctx context.Context, clusterID uint) (*ClusterScaleOperation, error)

FindRunningByCluster 返回集群上进行中的伸缩操作(不区分归属,用于互斥),没有时返回 nil。 FindRunningByCluster returns the running scaling operation of a cluster regardless of owner (for mutual exclusion), or nil when none.

func (*ScaleRepository) GetByID added in v1.2.0

GetByID 按 ID 读取伸缩操作,不存在时返回 ErrScaleOperationNotFound。 GetByID loads a scaling operation by ID and returns ErrScaleOperationNotFound when missing.

func (*ScaleRepository) GetForOwner added in v1.2.0

func (r *ScaleRepository) GetForOwner(ctx context.Context, clusterID, id, ownerUserID uint, includeAll bool) (*ClusterScaleOperation, error)

GetForOwner 在集群与归属条件下读取伸缩操作;非管理员只能读取自己创建的操作。 GetForOwner loads a scaling operation scoped by cluster and ownership; non-admins only see their own.

func (*ScaleRepository) List added in v1.2.0

List 按过滤条件列出伸缩操作(最新在前),归属条件在查询中加入。 List lists scaling operations (newest first) with ownership applied in the query.

func (*ScaleRepository) ListRunning added in v1.2.0

func (r *ScaleRepository) ListRunning(ctx context.Context) ([]*ClusterScaleOperation, error)

ListRunning 返回全部进行中的伸缩操作(控制面启动时用于标记被打断的操作)。 ListRunning returns every running scaling operation (used at Control Plane startup to mark interruptions).

func (*ScaleRepository) Save added in v1.2.0

Save 保存伸缩操作的全部字段。 Save persists every field of a scaling operation.

type ScaleRestartNode added in v1.2.0

type ScaleRestartNode struct {
	NodeID   uint   `json:"node_id"`
	HostID   uint   `json:"host_id"`
	HostName string `json:"host_name"`
	Address  string `json:"address"`
}

ScaleRestartNode 是滚动重启的候选节点。/ ScaleRestartNode is a rolling-restart candidate node.

type ScaleRollingRestart added in v1.2.0

type ScaleRollingRestart struct {
	Nodes       []ScaleRestartNode `json:"nodes"`
	Available   bool               `json:"available"`
	Recommended bool               `json:"recommended"`
	// BlockedReason 不可选的原因;Note 是可选但需留意的说明(如作业状态未知)。
	// BlockedReason explains why it cannot be chosen; Note is a caveat when it can (e.g. jobs unknown).
	BlockedReason string `json:"blocked_reason,omitempty"`
	Note          string `json:"note,omitempty"`
}

ScaleRollingRestart 描述扩容后可选的滚动重启。Hazelcast 只在启动时读取 tcp-ip member-list,且分裂后的合并只探测 启动时的列表;扩容只改写现有节点的文件、不重启,因此这些节点直到下次重启前都无法在分裂后主动找回新节点。 回环修正已重启的节点会直接写入扩容后的完整列表,不在候选之内。 ScaleRollingRestart describes the optional rolling restart after scaling out. Hazelcast reads the tcp-ip member list only at startup, and split-brain merging only probes that list; scaling out rewrites existing nodes' files without restarting them, so until their next restart they cannot find the new node again after a split. Nodes already restarted by the loopback fix get the full post-scale list and are not candidates.

type ScaleRunningJob added in v1.2.0

type ScaleRunningJob struct {
	JobID   string `json:"job_id"`
	JobName string `json:"job_name"`
}

ScaleRunningJob 是集群上一个运行中的作业。 ScaleRunningJob is one job running on the cluster.

type ScaleService added in v1.2.0

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

ScaleService 编排集群扩缩容:只读计划(Plan)+ 异步分步执行(Execute),步骤状态持久化并可从失败步骤重试。 ScaleService orchestrates cluster scaling: a read-only Plan plus asynchronous step execution whose state is persisted and can be retried from the failed step.

func NewScaleService added in v1.2.0

func NewScaleService(svc *Service, repo *ScaleRepository) *ScaleService

NewScaleService 创建伸缩编排服务;其余依赖通过 Set* 注入。 NewScaleService creates the scaling orchestrator; other dependencies are injected through Set*.

func (*ScaleService) CleanupAfterRecordRemoval added in v1.2.0

func (s *ScaleService) CleanupAfterRecordRemoval(ctx context.Context, clusterID uint, removed *ClusterNode) *RecordRemovalCleanup

CleanupAfterRecordRemoval 在「仅移除记录」(不停进程,主机离线或进程被其他集群共用时的降级入口)删除节点记录后补齐清理:

  • 修补剩余节点与模板的 member-list(只写文件、不重启),避免其他节点继续把它当成员; 若被移除节点或任一剩余节点的进程同时登记在其他集群,节点上的 hazelcast 文件为多个集群共用,修补会改到其他集群,跳过;
  • 该主机在本集群已无节点时,删除其在本集群的配置记录,避免 sync-all 永远报待推送;
  • 主机在线时按集群下发监控配置以移除本集群的跟踪项(共享进程时其他集群的跟踪项保留)。

清理尽力而为:节点记录已删除,失败只记入 Warnings 与日志。 CleanupAfterRecordRemoval completes the cleanup after "remove record only" (no process stop; the fallback when the host is offline or the process is shared with another cluster) deleted a node record:

  • patch the member list of the template and remaining nodes (files only, no restart) so they stop treating it as a member; skipped when the removed node or any remaining node shares its process with another cluster, because the hazelcast file on that host is then shared and patching it would change the other cluster;
  • when the host has no node of this cluster left, delete its config records of this cluster so sync-all no longer reports them as pending forever;
  • when the host is online, push the cluster-scoped monitor config to drop this cluster's tracked entry (another cluster's entry for a shared process is kept).

Best effort: the node record is already gone, so failures only go to Warnings and the log.

func (*ScaleService) GetOperation added in v1.2.0

func (s *ScaleService) GetOperation(ctx context.Context, actor executionapp.Actor, clusterID uint, operationID uint) (*ScaleOperationInfo, error)

GetOperation 在归属条件下读取伸缩操作详情。 GetOperation loads a scaling operation detail under ownership scoping.

func (*ScaleService) ListOperations added in v1.2.0

func (s *ScaleService) ListOperations(ctx context.Context, actor executionapp.Actor, clusterID uint, status ScaleOperationStatus, limit int) ([]*ScaleOperationInfo, error)

ListOperations 在归属条件下列出集群伸缩操作(最新在前),结果永不为 nil。 ListOperations lists a cluster's scaling operations under ownership scoping (newest first); never nil.

func (*ScaleService) PlanScaleIn added in v1.2.0

func (s *ScaleService) PlanScaleIn(ctx context.Context, clusterID uint, nodeID uint, lang string) (*ScalePlan, error)

PlanScaleIn 计算缩容计划(只读):最后节点 / 最后在线 master 阻断,运行中作业需强确认,java-proxy 承载节点被移除时规划跟随。 PlanScaleIn computes a scale-in plan (read-only): removing the last node or the last online master is blocked, running jobs require a strong acknowledgement and the java-proxy follows when its host node is removed.

func (*ScaleService) PlanScaleOut added in v1.2.0

func (s *ScaleService) PlanScaleOut(ctx context.Context, clusterID uint, req *ScaleOutRequest, lang string) (*ScalePlan, error)

PlanScaleOut 计算扩容计划(只读,不修改任何状态)。 PlanScaleOut computes a scale-out plan (read-only; nothing is modified).

func (*ScaleService) RecoverInterruptedOperations added in v1.2.0

func (s *ScaleService) RecoverInterruptedOperations(ctx context.Context) (int, error)

RecoverInterruptedOperations 在控制面启动时把仍为 running 的伸缩操作标记为失败(interrupted_by_restart), 运行中的步骤记为失败,公共执行记录同步失败;用户可从失败步骤重试。 RecoverInterruptedOperations marks scaling operations still running at Control Plane startup as failed (interrupted_by_restart); the running step is marked failed and the shared execution fails too, so the user can retry from the failed step.

func (*ScaleService) RetryOperation added in v1.2.0

func (s *ScaleService) RetryOperation(ctx context.Context, actor executionapp.Actor, clusterID uint, operationID uint, meta ScaleExecutionRequest) (*ScaleSubmitResult, error)

RetryOperation 从失败步骤重试伸缩操作(R1):创建新的公共执行记录并沿用原计划快照,已完成的步骤不会重跑。 RetryOperation retries a scaling operation from the failed step (R1): a new shared execution is created and the original plan snapshot is reused; completed steps are not re-run.

func (*ScaleService) SetConfigManager added in v1.2.0

func (s *ScaleService) SetConfigManager(manager ScaleConfigManager)

SetConfigManager 注入配置服务。 SetConfigManager injects the config service.

func (*ScaleService) SetExecutionService added in v1.2.0

func (s *ScaleService) SetExecutionService(service *executionapp.Service)

SetExecutionService 注入公共执行服务(R2 提交、R1 重试与审计)。 SetExecutionService injects the shared execution service (R2 submission, R1 retry and auditing).

func (*ScaleService) SetInstaller added in v1.2.0

func (s *ScaleService) SetInstaller(installer ScaleInstaller)

SetInstaller 注入安装器。 SetInstaller injects the installer.

func (*ScaleService) SetMonitorPusher added in v1.2.0

func (s *ScaleService) SetMonitorPusher(pusher ClusterMonitorPusher)

SetMonitorPusher 注入按集群下发监控配置的能力。 SetMonitorPusher injects the cluster-scoped monitor config pusher.

func (*ScaleService) SetPluginReplayer added in v1.2.0

func (s *ScaleService) SetPluginReplayer(replayer ScalePluginReplayer)

SetPluginReplayer 注入插件补齐能力。 SetPluginReplayer injects the plugin replayer.

func (*ScaleService) SetProcessDiscoverer added in v1.2.0

func (s *ScaleService) SetProcessDiscoverer(discoverer ScaleProcessDiscoverer)

SetProcessDiscoverer 注入主机进程发现能力。 SetProcessDiscoverer injects the host process discoverer.

func (*ScaleService) SetRunningJobsProvider added in v1.2.0

func (s *ScaleService) SetRunningJobsProvider(provider RunningJobsProvider)

SetRunningJobsProvider 注入运行中作业读取能力。 SetRunningJobsProvider injects the running jobs provider.

func (*ScaleService) SubmitScaleIn added in v1.2.0

func (s *ScaleService) SubmitScaleIn(ctx context.Context, actor executionapp.Actor, clusterID uint, nodeID uint, req *ScaleInSubmitRequest, meta ScaleExecutionRequest) (*ScaleSubmitResult, error)

SubmitScaleIn 校验 R2 确认与幂等后重算缩容计划,满足护栏时创建缩容操作并异步执行。 SubmitScaleIn validates R2 confirmation and idempotency, recomputes the scale-in plan and, when the guardrails pass, creates the scale-in operation and runs it asynchronously.

func (*ScaleService) SubmitScaleOut added in v1.2.0

func (s *ScaleService) SubmitScaleOut(ctx context.Context, actor executionapp.Actor, clusterID uint, req *ScaleOutSubmitRequest, meta ScaleExecutionRequest) (*ScaleSubmitResult, error)

SubmitScaleOut 校验 R2 确认与幂等后重算计划,计划指纹一致、无阻断项且确认齐全时创建扩容操作并异步执行。 SubmitScaleOut validates R2 confirmation and idempotency, recomputes the plan and, when the fingerprint matches, nothing blocks and acknowledgements are complete, creates the scale-out operation and runs it asynchronously.

type ScaleStepName added in v1.2.0

type ScaleStepName string

ScaleStepName 是伸缩操作中的步骤编码(稳定常量,前端据此本地化展示)。 ScaleStepName is a step code of a scaling operation (stable constant localized by the frontend).

const (
	ScaleStepValidatePlan     ScaleStepName = "validate_plan"
	ScaleStepFixLoopback      ScaleStepName = "fix_loopback"
	ScaleStepInstallFiles     ScaleStepName = "install_files"
	ScaleStepRegisterNode     ScaleStepName = "register_node"
	ScaleStepApplyConfigs     ScaleStepName = "apply_configs"
	ScaleStepReplayPlugins    ScaleStepName = "replay_plugins"
	ScaleStepStartNode        ScaleStepName = "start_node"
	ScaleStepPatchMembers     ScaleStepName = "patch_members"
	ScaleStepVerifyMembership ScaleStepName = "verify_membership"
	ScaleStepStopNode         ScaleStepName = "stop_node"
	ScaleStepCleanupRecords   ScaleStepName = "cleanup_records"
	ScaleStepRefreshMonitor   ScaleStepName = "refresh_monitor"
	ScaleStepFollowJavaProxy  ScaleStepName = "follow_java_proxy"
	ScaleStepVerify           ScaleStepName = "verify"
	// ScaleStepRollingRestart 扩容入群后可选的滚动重启:让现有节点加载含新节点的 member-list(Hazelcast 只在启动时读取,
	// 分裂后的合并只探测启动时的列表)。
	// ScaleStepRollingRestart is the optional rolling restart after the join: existing nodes load the member list that
	// includes the new node (Hazelcast reads it only at startup, and split-brain merging only probes that list).
	ScaleStepRollingRestart ScaleStepName = "rolling_restart"
)

type ScaleStepState added in v1.2.0

type ScaleStepState struct {
	Name       ScaleStepName   `json:"name"`
	Status     ScaleStepStatus `json:"status"`
	Attempts   int             `json:"attempts"`
	Message    string          `json:"message,omitempty"`
	Error      string          `json:"error,omitempty"`
	StartedAt  *time.Time      `json:"started_at,omitempty"`
	FinishedAt *time.Time      `json:"finished_at,omitempty"`
	Result     json.RawMessage `json:"result,omitempty"`
}

ScaleStepState 是持久化的单步状态,Result 保存该步骤的结构化结果(不含敏感内容)。 ScaleStepState is the persisted state of one step; Result keeps the step's structured result (no secrets).

type ScaleStepStatus added in v1.2.0

type ScaleStepStatus string

ScaleStepStatus 是单个步骤的状态;partial 表示完成但有可重试的局部失败(不阻断后续步骤)。 ScaleStepStatus is the status of one step; partial means done with retryable partial failures (does not block later steps).

const (
	ScaleStepPending   ScaleStepStatus = "pending"
	ScaleStepRunning   ScaleStepStatus = "running"
	ScaleStepSucceeded ScaleStepStatus = "succeeded"
	ScaleStepPartial   ScaleStepStatus = "partial"
	ScaleStepFailed    ScaleStepStatus = "failed"
	ScaleStepSkipped   ScaleStepStatus = "skipped"
)

type ScaleSubmitResponse added in v1.2.0

type ScaleSubmitResponse struct {
	ErrorCode string             `json:"error_code,omitempty"`
	ErrorMsg  string             `json:"error_msg"`
	Data      *ScaleSubmitResult `json:"data"`
}

ScaleSubmitResponse 是提交或重试伸缩操作的响应。 ScaleSubmitResponse is the response of submitting or retrying a scaling operation.

type ScaleSubmitResult added in v1.2.0

type ScaleSubmitResult struct {
	ExecutionID string              `json:"execution_id"`
	OperationID uint                `json:"operation_id"`
	Operation   *ScaleOperationInfo `json:"operation"`
}

ScaleSubmitResult 是提交或重试伸缩操作的响应。 ScaleSubmitResult is the response of submitting or retrying a scaling operation.

type ScaleTarget added in v1.2.0

type ScaleTarget struct {
	NodeID        uint     `json:"node_id,omitempty"`
	HostID        uint     `json:"host_id"`
	HostName      string   `json:"host_name"`
	HostIP        string   `json:"host_ip"`
	Role          NodeRole `json:"role"`
	InstallDir    string   `json:"install_dir"`
	HazelcastPort int      `json:"hazelcast_port"`
	APIPort       int      `json:"api_port"`
	WorkerPort    int      `json:"worker_port"`
	Address       string   `json:"address"`
	JavaHome      string   `json:"java_home,omitempty"`
}

ScaleTarget 是本次伸缩的目标节点(扩容为待加入节点,缩容为待移除节点)。 ScaleTarget is the node being scaled (the joining node for scale-out, the removed node for scale-in).

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) CreateClusterFromDiscovery added in v1.3.0

func (s *Service) CreateClusterFromDiscovery(ctx context.Context, hostID uint, dc *discovery.DiscoveredCluster) (uint, error)

CreateClusterFromDiscovery 从「发现集群」结果创建集群并写入节点 java_home(AC1e)。 CreateClusterFromDiscovery creates a cluster from a discovery result and persists node java_home (AC1e).

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, javaHome string) error

EnsureNodeForInstallation 确保一键安装完成后的节点已登记,并刷新安装目录与端口。 EnsureNodeForInstallation ensures a one-click installation node is registered and refreshes its install directory and ports. javaHome 为该节点指定的 JDK(为空表示自动探测),每次安装都覆盖写入,不沿用其他节点或旧值(AC1d)。 javaHome is the node's JDK (empty means auto detection); every installation overwrites it and never inherits from other nodes or old values (AC1d).

func (*Service) EnsureSTXJavaProxyOnHost added in v1.3.0

func (s *Service) EnsureSTXJavaProxyOnHost(ctx context.Context, hostID uint)

EnsureSTXJavaProxyOnHost 在 Agent(重新)就绪后,对 stx-java-proxy 选点落在该主机上的集群补一次启动。 以前只在注册集群时自动拉起一次,主机重启或进程退出后编辑器校验、DAG、表结构等依赖它的功能会一直不可用。 Agent 侧启动是幂等的,已在运行时只做健康检查。 EnsureSTXJavaProxyOnHost starts stx-java-proxy for clusters whose proxy node is on this host once its Agent is (re)ready. It used to be auto-started only at cluster registration, so after a host reboot or a process exit the features depending on it (editor validation, DAG, schemas) stayed unavailable. Starting is idempotent on the Agent.

func (*Service) FindClusterByInstallDir added in v1.3.0

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

FindClusterByInstallDir 按主机安装目录匹配已纳管集群(discovery.ClusterMatcher)。 FindClusterByInstallDir matches an already-managed cluster by host install directory (discovery.ClusterMatcher).

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) GetNodeJavaRuntime added in v1.3.0

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

GetNodeJavaRuntime 读取节点生效的 JDK:按节点记录的 java_home 让 Agent 检查 <java_home>/bin/java, 未指定时返回 Agent 自动探测到的 JDK。Agent 不在线时返回错误,不猜测结果。 GetNodeJavaRuntime reads the node's effective JDK: the Agent checks <java_home>/bin/java from the node record, or reports its auto-detected JDK when none is set. An offline Agent returns an error instead of a guess.

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) GetWithInfo added in v1.2.0

func (s *Service) GetWithInfo(ctx context.Context, id uint) (*ClusterInfo, error)

GetWithInfo 获取单个集群并填充 online_nodes 与 health_status,与列表接口口径一致。 GetWithInfo returns one cluster with online_nodes and health_status populated, consistent with the list API.

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) SetOnNodeRecordRemoved added in v1.2.0

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

SetOnNodeRecordRemoved 设置「仅移除记录」后的清理钩子。/ SetOnNodeRecordRemoved sets the cleanup hook run after "remove record only".

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) UpdateNodeFromDiscovery added in v1.3.0

func (s *Service) UpdateNodeFromDiscovery(ctx context.Context, nodeID uint, node *discovery.DiscoveredNode) error

UpdateNodeFromDiscovery 用发现到的进程信息更新已有节点状态(PID / 运行中)。 UpdateNodeFromDiscovery updates an existing node from a discovered process (PID / running).

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 更新集群的状态。

type ServiceConfig

type ServiceConfig struct {
	HeartbeatTimeout time.Duration
	// AssetBaseURL is the Control Plane base for Agent asset downloads
	// (e.g. http://cp:17800/api/v1/agent/assets). Empty disables push-on-start hints.
	// AssetBaseURL 是 Agent 拉取辅助资产的控制面基址
	// (如 http://cp:17800/api/v1/agent/assets)。为空则不下发启动拉取提示。
	AssetBaseURL string
}

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 表示运行时存储连通性校验响应。

type WebUIEndpoints added in v1.3.0

type WebUIEndpoints struct {
	Available     bool   `json:"available"`
	Reason        string `json:"reason,omitempty"`
	ProxyBaseURL  string `json:"proxy_base_url"`
	NativeBaseURL string `json:"native_base_url"`
	NodeID        uint   `json:"node_id,omitempty"`
	HostName      string `json:"host_name,omitempty"`
	// JobPathTemplate 是作业详情页的 hash 路由,前端把 {job_id} 替换为引擎作业 ID 后拼在基础地址后面。
	// JobPathTemplate is the job detail hash route; the frontend substitutes {job_id} and appends it to a base URL.
	JobPathTemplate string `json:"job_path_template"`
}

WebUIEndpoints 是 SeaTunnel 集群观测(引擎自带 Web UI)的访问地址:经 STX 代理,或浏览器直连原生地址。 WebUIEndpoints lists how to open the SeaTunnel cluster observation UI (the engine's own Web UI): through the STX proxy or directly at the native address.

type WebUIEndpointsResponse added in v1.3.0

type WebUIEndpointsResponse struct {
	ErrorMsg string          `json:"error_msg"`
	Data     *WebUIEndpoints `json:"data"`
}

WebUIEndpointsResponse 包装 SeaTunnel 集群观测地址。 WebUIEndpointsResponse wraps the SeaTunnel cluster observation addresses.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL