installer

package
v1.0.0 Latest Latest
Warning

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

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

Documentation

Overview

Package installer provides SeaTunnel installation management APIs for Control Plane. installer 包提供 Control Plane 的 SeaTunnel 安装管理 API。

Index

Constants

View Source
const ApacheArchiveURL = seatunnel.ArchiveVersionsURL

ApacheArchiveURL is the URL to fetch version list from Apache Archive ApacheArchiveURL 是从 Apache Archive 获取版本列表的 URL

View Source
const PackageTransferChunkSize = 1024 * 1024

PackageTransferChunkSize is the size of each chunk for package transfer (1MB) PackageTransferChunkSize 是安装包传输每个块的大小(1MB)

View Source
const VersionCacheDuration = 1 * time.Hour

VersionCacheDuration is how long to cache the version list VersionCacheDuration 是版本列表的缓存时间

Variables

View Source
var (
	ErrPackageNotFound        = errors.New("package not found / 安装包未找到")
	ErrPackageAlreadyExists   = errors.New("package already exists / 安装包已存在")
	ErrInvalidPackageVersion  = errors.New("invalid package version / 安装包版本不合法")
	ErrInvalidPackageFile     = errors.New("invalid package file / 安装包文件不合法")
	ErrInvalidPackagePath     = errors.New("invalid package path / 安装包路径不合法")
	ErrPackageTooLarge        = errors.New("package too large / 安装包过大")
	ErrSourcePackageNotFound  = errors.New("source package not found / 源码包未找到")
	ErrInvalidSourcePackage   = errors.New("invalid source package / 源码包不合法")
	ErrInvalidUploadID        = errors.New("invalid upload id / 上传会话 ID 不合法")
	ErrInvalidChunkIndex      = errors.New("invalid chunk index / 分片索引不合法")
	ErrChunkOutOfOrder        = errors.New("chunk out of order / 分片顺序错误")
	ErrInstallationNotFound   = errors.New("installation not found / 安装任务未找到")
	ErrInstallationInProgress = errors.New("installation already in progress / 安装任务正在进行中")
	ErrHostNotConnected       = errors.New("host agent not connected / 主机 Agent 未连接")
	ErrAgentNotFound          = errors.New("agent not found / Agent 未找到")
)

Common errors / 常见错误

View Source
var DefaultPrecheckPorts = []int{5801, 5802, 8080, 18080}

DefaultPrecheckPorts is the default list of ports to check for SeaTunnel installation DefaultPrecheckPorts 是 SeaTunnel 安装时默认检查的端口列表

View Source
var ErrDownloadInProgress = errors.New("download already in progress / 下载已在进行中")

ErrDownloadInProgress indicates a download is already in progress for this version ErrDownloadInProgress 表示该版本的下载已在进行中

View Source
var ErrDownloadNotFound = errors.New("download task not found / 下载任务未找到")

ErrDownloadNotFound indicates the download task was not found ErrDownloadNotFound 表示下载任务未找到

View Source
var MirrorURLs = map[MirrorSource]string{
	MirrorAliyun:      "https://mirrors.aliyun.com/apache/seatunnel",
	MirrorApache:      "https://archive.apache.org/dist/seatunnel",
	MirrorHuaweiCloud: "https://mirrors.huaweicloud.com/apache/seatunnel",
}

MirrorURLs maps mirror sources to their base URLs MirrorURLs 将镜像源映射到其基础 URL

Functions

This section is empty.

Types

type AgentManager

type AgentManager interface {
	// GetAgentByHostID returns the agent connection for a host
	// GetAgentByHostID 返回主机的 Agent 连接
	GetAgentByHostID(hostID uint) (agentID string, connected bool)

	// SendInstallCommand sends an installation command to an agent
	// SendInstallCommand 向 Agent 发送安装命令
	SendInstallCommand(ctx context.Context, agentID string, params map[string]string) (commandID string, err error)

	// GetCommandStatus returns the status of a command
	// GetCommandStatus 返回命令的状态
	GetCommandStatus(commandID string) (status string, progress int, message string, err error)

	// 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) (success bool, output string, err error)

	// SendTransferPackageCommand sends a package transfer chunk to an agent
	// SendTransferPackageCommand 向 Agent 发送安装包传输块
	SendTransferPackageCommand(ctx context.Context, agentID string, version string, fileName string, chunk []byte, offset int64, totalSize int64, isLast bool, checksum string) (success bool, receivedBytes int64, localPath string, err error)
}

AgentManager is the interface for communicating with agents AgentManager 是与 Agent 通信的接口

type AvailableVersions

type AvailableVersions struct {
	Versions            []string                                 `json:"versions"`
	RecommendedVersion  string                                   `json:"recommended_version"`
	LocalPackages       []PackageInfo                            `json:"local_packages"`
	VersionCapabilities map[string]seatunnel.VersionCapabilities `json:"version_capabilities"`
}

AvailableVersions contains available SeaTunnel versions AvailableVersions 包含可用的 SeaTunnel 版本

type CheckStatus

type CheckStatus string

CheckStatus represents the status of a precheck item CheckStatus 表示预检查项的状态

const (
	CheckStatusPassed  CheckStatus = "passed"
	CheckStatusFailed  CheckStatus = "failed"
	CheckStatusWarning CheckStatus = "warning"
)

type CheckpointConfig

type CheckpointConfig struct {
	StorageType      CheckpointStorageType `json:"storage_type"`
	Namespace        string                `json:"namespace"`
	HDFSNameNodeHost string                `json:"hdfs_namenode_host,omitempty"`
	HDFSNameNodePort int                   `json:"hdfs_namenode_port,omitempty"`
	// HDFS Kerberos authentication / HDFS Kerberos 认证
	KerberosPrincipal      string `json:"kerberos_principal,omitempty"`
	KerberosKeytabFilePath string `json:"kerberos_keytab_file_path,omitempty"`
	// HDFS HA mode configuration / HDFS HA 模式配置
	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"`
	// HdfsSitePath 是可选的 hdfs-site.xml 路径,对应 SeaTunnel hdfs_site_path。
	// HdfsSitePath is the optional hdfs-site.xml path, mapped to SeaTunnel hdfs_site_path.
	HdfsSitePath string `json:"hdfs_site_path,omitempty"`
	// DisableCache 对应 plugin-config.disable.cache;nil 表示沿用默认(hdfs 类型默认关闭缓存)。
	// DisableCache maps to plugin-config.disable.cache; nil keeps the default (cache off for hdfs type).
	DisableCache *bool `json:"disable_cache,omitempty"`
	// OSS/S3 configuration / OSS/S3 配置
	StorageEndpoint  string `json:"storage_endpoint,omitempty"`
	StorageAccessKey string `json:"storage_access_key,omitempty"`
	StorageSecretKey string `json:"storage_secret_key,omitempty"`
	StorageBucket    string `json:"storage_bucket,omitempty"`
	// S3CredentialsProvider 对应 fs.s3a.aws.credentials.provider;空则使用 SimpleAWSCredentialsProvider。
	// S3CredentialsProvider maps to fs.s3a.aws.credentials.provider; empty uses SimpleAWSCredentialsProvider.
	S3CredentialsProvider string `json:"s3_credentials_provider,omitempty"`
}

CheckpointConfig contains checkpoint storage configuration CheckpointConfig 包含检查点存储配置

type CheckpointStorageType

type CheckpointStorageType string

CheckpointStorageType represents the checkpoint storage type CheckpointStorageType 表示检查点存储类型

const (
	CheckpointStorageLocalFile CheckpointStorageType = "LOCAL_FILE"
	CheckpointStorageHDFS      CheckpointStorageType = "HDFS"
	CheckpointStorageOSS       CheckpointStorageType = "OSS"
	CheckpointStorageS3        CheckpointStorageType = "S3"
)

type ConfigInitializer

type ConfigInitializer interface {
	// InitClusterConfigs initializes cluster configs by pulling from a host
	// InitClusterConfigs 通过从主机拉取来初始化集群配置
	InitClusterConfigs(ctx context.Context, clusterID uint, hostID uint, installDir string, userID uint) error
}

ConfigInitializer is the interface for initializing cluster configs after installation ConfigInitializer 是安装完成后初始化集群配置的接口

type ConnectorConfig

type ConnectorConfig struct {
	InstallConnectors bool         `json:"install_connectors"`
	Connectors        []string     `json:"connectors,omitempty"`
	PluginRepo        MirrorSource `json:"plugin_repo,omitempty"`
	// SelectedPlugins is the list of plugin names to install during SeaTunnel setup
	// SelectedPlugins 是 SeaTunnel 安装过程中要安装的插件名称列表
	SelectedPlugins []string `json:"selected_plugins,omitempty"`
	// SelectedPluginProfiles stores selected profile keys for profile-aware plugins
	// SelectedPluginProfiles 保存带画像插件选择的画像键
	SelectedPluginProfiles map[string][]string `json:"selected_plugin_profiles,omitempty"`
}

ConnectorConfig contains connector installation configuration ConnectorConfig 包含连接器安装配置

type DeletePackageResponse

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

DeletePackageResponse represents the response for deleting a package. DeletePackageResponse 表示删除安装包的响应。

type DeploymentMode

type DeploymentMode string

DeploymentMode represents the deployment mode DeploymentMode 表示部署模式

const (
	DeploymentModeHybrid    DeploymentMode = "hybrid"
	DeploymentModeSeparated DeploymentMode = "separated"
)

type DownloadListResponse

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

DownloadListResponse represents the response for listing downloads. DownloadListResponse 表示获取下载列表的响应。

type DownloadRequest

type DownloadRequest struct {
	Version    string       `json:"version" binding:"required"`
	Mirror     MirrorSource `json:"mirror"`
	WithSource *bool        `json:"with_source,omitempty"`
}

DownloadRequest represents a request to download a package DownloadRequest 表示下载安装包的请求

func (DownloadRequest) IncludeSource added in v1.0.0

func (r DownloadRequest) IncludeSource() bool

IncludeSource returns true when the caller omitted with_source or explicitly enabled it. IncludeSource 在调用方未传 with_source 或显式启用时返回 true。

type DownloadResponse

type DownloadResponse struct {
	ErrorMsg string        `json:"error_msg"`
	Data     *DownloadTask `json:"data"`
}

DownloadResponse represents the response for download operations. DownloadResponse 表示下载操作的响应。

type DownloadStatus

type DownloadStatus string

DownloadStatus represents the status of a download task DownloadStatus 表示下载任务的状态

const (
	DownloadStatusPending     DownloadStatus = "pending"
	DownloadStatusDownloading DownloadStatus = "downloading"
	DownloadStatusCancelling  DownloadStatus = "cancelling"
	DownloadStatusCompleted   DownloadStatus = "completed"
	DownloadStatusFailed      DownloadStatus = "failed"
	DownloadStatusCancelled   DownloadStatus = "cancelled"
)

type DownloadTask

type DownloadTask struct {
	ID              string         `json:"id"`
	Version         string         `json:"version"`
	Mirror          MirrorSource   `json:"mirror"`
	DownloadURL     string         `json:"download_url"`
	Status          DownloadStatus `json:"status"`
	Progress        int            `json:"progress"` // 0-100
	DownloadedBytes int64          `json:"downloaded_bytes"`
	TotalBytes      int64          `json:"total_bytes"`
	Speed           int64          `json:"speed"` // bytes per second
	Message         string         `json:"message,omitempty"`
	Error           string         `json:"error,omitempty"`
	StartTime       time.Time      `json:"start_time"`
	EndTime         *time.Time     `json:"end_time,omitempty"`
	ExecutionID     string         `json:"execution_id,omitempty"`
	SourceRequested bool           `json:"source_requested"`
	SourceStatus    DownloadStatus `json:"source_status,omitempty"`
	SourceURL       string         `json:"source_url,omitempty"`
	SourceProgress  int            `json:"source_progress,omitempty"`
	SourceBytes     int64          `json:"source_downloaded_bytes,omitempty"`
	SourceTotal     int64          `json:"source_total_bytes,omitempty"`
	SourceChecksum  string         `json:"source_checksum,omitempty"`
	SourceError     string         `json:"source_error,omitempty"`
	OwnerUserID     uint64         `json:"-"`
	// contains filtered or unexported fields
}

DownloadTask represents a package download task DownloadTask 表示安装包下载任务

type ExecutionRequest added in v1.0.0

type ExecutionRequest struct {
	RequestID      string
	IdempotencyKey string
	RequestHash    string
	Confirmed      bool
	ConfirmationID string
	ClientType     string
}

ExecutionRequest 保存安装包写操作的公共执行参数。 ExecutionRequest stores shared execution parameters for package write operations.

type GetPackageInfoResponse

type GetPackageInfoResponse struct {
	ErrorMsg string       `json:"error_msg"`
	Data     *PackageInfo `json:"data"`
}

GetPackageInfoResponse represents the response for getting package info. GetPackageInfoResponse 表示获取安装包信息的响应。

type Handler

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

Handler provides HTTP handlers for installation management. Handler 提供安装管理的 HTTP 处理器。

func NewHandler

func NewHandler(service *Service) *Handler

NewHandler creates a new Handler instance. NewHandler 创建一个新的 Handler 实例。

func (*Handler) CancelDownload

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

CancelDownload handles POST /api/v1/packages/download/:version/cancel - cancels a download. CancelDownload 处理 POST /api/v1/packages/download/:version/cancel - 取消下载。 @Tags packages @Produce json @Param version path string true "版本号" @Success 200 {object} DownloadResponse @Router /api/v1/packages/download/{version}/cancel [post]

func (*Handler) CancelInstallation

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

CancelInstallation handles POST /api/v1/hosts/:id/install/cancel - cancels installation. CancelInstallation 处理 POST /api/v1/hosts/:id/install/cancel - 取消安装。 @Tags installation @Produce json @Param id path int true "主机ID" @Success 200 {object} InstallResponse @Router /api/v1/hosts/{id}/install/cancel [post]

func (*Handler) DeletePackage

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

DeletePackage handles DELETE /api/v1/packages/:version - deletes a local package. DeletePackage 处理 DELETE /api/v1/packages/:version - 删除本地安装包。 @Tags packages @Produce json @Param version path string true "版本号" @Success 200 {object} DeletePackageResponse @Router /api/v1/packages/{version} [delete]

func (*Handler) DownloadSourcePackage added in v1.0.0

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

DownloadSourcePackage handles GET /api/v1/packages/:version/source/download. DownloadSourcePackage 处理 GET /api/v1/packages/:version/source/download。 @Tags packages @Produce application/gzip @Param version path string true "版本号" @Success 200 {file} binary @Router /api/v1/packages/{version}/source/download [get]

func (*Handler) FetchSourcePackage added in v1.0.0

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

FetchSourcePackage handles POST /api/v1/packages/:version/source/fetch. FetchSourcePackage 处理 POST /api/v1/packages/:version/source/fetch。 @Tags packages @Accept json @Produce json @Param version path string true "版本号" @Param request body SourceFetchRequest true "源码下载请求" @Success 200 {object} UploadPackageResponse @Router /api/v1/packages/{version}/source/fetch [post]

func (*Handler) GetDownloadStatus

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

GetDownloadStatus handles GET /api/v1/packages/download/:version - gets download status. GetDownloadStatus 处理 GET /api/v1/packages/download/:version - 获取下载状态。 @Tags packages @Produce json @Param version path string true "版本号" @Success 200 {object} DownloadResponse @Router /api/v1/packages/download/{version} [get]

func (*Handler) GetInstallationStatus

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

GetInstallationStatus handles GET /api/v1/hosts/:id/install/status - gets installation status. GetInstallationStatus 处理 GET /api/v1/hosts/:id/install/status - 获取安装状态。 @Tags installation @Produce json @Param id path int true "主机ID" @Success 200 {object} InstallResponse @Router /api/v1/hosts/{id}/install/status [get]

func (*Handler) GetPackageInfo

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

GetPackageInfo handles GET /api/v1/packages/:version - gets package info. GetPackageInfo 处理 GET /api/v1/packages/:version - 获取安装包信息。 @Tags packages @Produce json @Param version path string true "版本号" @Success 200 {object} GetPackageInfoResponse @Router /api/v1/packages/{version} [get]

func (*Handler) ListDownloads

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

ListDownloads handles GET /api/v1/packages/downloads - lists all download tasks. ListDownloads 处理 GET /api/v1/packages/downloads - 获取所有下载任务。 @Tags packages @Produce json @Success 200 {object} DownloadListResponse @Router /api/v1/packages/downloads [get]

func (*Handler) ListPackages

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

ListPackages handles GET /api/v1/packages - lists available packages. ListPackages 处理 GET /api/v1/packages - 获取可用安装包列表。 @Tags packages @Produce json @Success 200 {object} ListPackagesResponse @Router /api/v1/packages [get]

func (*Handler) RefreshVersions

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

RefreshVersions handles POST /api/v1/packages/versions/refresh - refreshes version list from Apache Archive. RefreshVersions 处理 POST /api/v1/packages/versions/refresh - 从 Apache Archive 刷新版本列表。 @Tags packages @Produce json @Success 200 {object} RefreshVersionsResponse @Router /api/v1/packages/versions/refresh [post]

func (*Handler) RetryStep

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

RetryStep handles POST /api/v1/hosts/:id/install/retry - retries a failed step. RetryStep 处理 POST /api/v1/hosts/:id/install/retry - 重试失败的步骤。 @Tags installation @Accept json @Produce json @Param id path int true "主机ID" @Param request body RetryStepRequest true "重试请求" @Success 200 {object} InstallResponse @Router /api/v1/hosts/{id}/install/retry [post]

func (*Handler) RunPrecheck

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

RunPrecheck handles POST /api/v1/hosts/:id/precheck - runs precheck on a host. RunPrecheck 处理 POST /api/v1/hosts/:id/precheck - 在主机上运行预检查。 @Tags installation @Accept json @Produce json @Param id path int true "主机ID" @Param request body PrecheckRequest false "预检查参数" @Success 200 {object} PrecheckResponse @Router /api/v1/hosts/{id}/precheck [post]

func (*Handler) StartDownload

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

StartDownload handles POST /api/v1/packages/download - starts downloading a package. StartDownload 处理 POST /api/v1/packages/download - 开始下载安装包。 @Tags packages @Accept json @Produce json @Param request body DownloadRequest true "下载请求" @Success 200 {object} DownloadResponse @Router /api/v1/packages/download [post]

func (*Handler) StartInstallation

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

StartInstallation handles POST /api/v1/hosts/:id/install - starts installation. StartInstallation 处理 POST /api/v1/hosts/:id/install - 开始安装。 @Tags installation @Accept json @Produce json @Param id path int true "主机ID" @Param request body InstallationRequest true "安装请求" @Success 200 {object} InstallResponse @Router /api/v1/hosts/{id}/install [post]

func (*Handler) UploadPackage

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

UploadPackage handles POST /api/v1/packages/upload - uploads a package. UploadPackage 处理 POST /api/v1/packages/upload - 上传安装包。 @Tags packages @Accept multipart/form-data @Produce json @Param file formData file true "安装包文件" @Param source_file formData file false "同版本源码包文件" @Param version formData string true "版本号" @Success 200 {object} UploadPackageResponse @Router /api/v1/packages/upload [post]

func (*Handler) UploadPackageChunk

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

UploadPackageChunk handles POST /api/v1/packages/upload/chunk - uploads one package chunk. UploadPackageChunk 处理 POST /api/v1/packages/upload/chunk - 上传安装包分片。 @Tags packages @Accept multipart/form-data @Produce json @Param file formData file true "分片文件" @Param version formData string true "版本号" @Param upload_id formData string true "上传会话 ID" @Param chunk_index formData int true "分片索引(从 0 开始)" @Param total_chunks formData int true "总分片数" @Param total_size formData int true "原始文件总大小(字节)" @Param file_name formData string false "原始文件名" @Success 200 {object} UploadChunkResponse @Router /api/v1/packages/upload/chunk [post]

func (*Handler) UploadSourcePackage added in v1.0.0

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

UploadSourcePackage handles POST /api/v1/packages/:version/source/upload. UploadSourcePackage 处理 POST /api/v1/packages/:version/source/upload。 @Tags packages @Accept multipart/form-data @Produce json @Param version path string true "版本号" @Param source_file formData file true "源码包文件" @Success 200 {object} UploadPackageResponse @Router /api/v1/packages/{version}/source/upload [post]

func (*Handler) ValidateRuntimeStorage

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

ValidateRuntimeStorage handles POST /api/v1/installer/runtime-storage/validate. ValidateRuntimeStorage 处理 POST /api/v1/installer/runtime-storage/validate。

type HostInfo

type HostInfo struct {
	ID          uint       `json:"id"`
	Name        string     `json:"name,omitempty"`
	AgentID     string     `json:"agent_id"`
	AgentStatus string     `json:"agent_status"`
	LastSeen    *time.Time `json:"last_seen"`
}

HostInfo contains host information for precheck HostInfo 包含预检查所需的主机信息

func (*HostInfo) IsOnline

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

IsOnline checks if the host agent is online within the timeout IsOnline 检查主机 Agent 是否在超时时间内在线

type HostProvider

type HostProvider interface {
	// GetHostByID returns host information by ID
	// GetHostByID 根据 ID 返回主机信息
	GetHostByID(ctx context.Context, hostID uint) (*HostInfo, error)
}

HostProvider is the interface for getting host information HostProvider 是获取主机信息的接口

type IMAPConfig

type IMAPConfig struct {
	StorageType IMAPStorageType `json:"storage_type"`
	Namespace   string          `json:"namespace"`
	// HDFS configuration / HDFS 配置
	HDFSNameNodeHost string `json:"hdfs_namenode_host,omitempty"`
	HDFSNameNodePort int    `json:"hdfs_namenode_port,omitempty"`
	// HDFS Kerberos authentication / HDFS Kerberos 认证
	KerberosPrincipal      string `json:"kerberos_principal,omitempty"`
	KerberosKeytabFilePath string `json:"kerberos_keytab_file_path,omitempty"`
	// HDFS HA mode configuration / HDFS HA 模式配置
	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"`
	// HdfsSitePath 是可选的 hdfs-site.xml 路径,对应 SeaTunnel hdfs_site_path。
	// HdfsSitePath is the optional hdfs-site.xml path, mapped to SeaTunnel hdfs_site_path.
	HdfsSitePath string `json:"hdfs_site_path,omitempty"`
	// DisableCache 对应 plugin-config.disable.cache;nil 表示沿用默认。
	// DisableCache maps to plugin-config.disable.cache; nil keeps the default.
	DisableCache *bool `json:"disable_cache,omitempty"`
	// OSS/S3 configuration / OSS/S3 配置
	StorageEndpoint  string `json:"storage_endpoint,omitempty"`
	StorageAccessKey string `json:"storage_access_key,omitempty"`
	StorageSecretKey string `json:"storage_secret_key,omitempty"`
	StorageBucket    string `json:"storage_bucket,omitempty"`
	// S3CredentialsProvider 对应 fs.s3a.aws.credentials.provider;空则使用 SimpleAWSCredentialsProvider。
	// S3CredentialsProvider maps to fs.s3a.aws.credentials.provider; empty uses SimpleAWSCredentialsProvider.
	S3CredentialsProvider string `json:"s3_credentials_provider,omitempty"`
}

IMAPConfig contains IMAP persistence configuration. IMAPConfig 包含 IMAP 持久化配置。

type IMAPStorageType

type IMAPStorageType string

IMAPStorageType represents the IMAP persistence storage type. IMAPStorageType 表示 IMAP 持久化存储类型。

const (
	IMAPStorageDisabled  IMAPStorageType = "DISABLED"
	IMAPStorageLocalFile IMAPStorageType = "LOCAL_FILE"
	IMAPStorageHDFS      IMAPStorageType = "HDFS"
	IMAPStorageOSS       IMAPStorageType = "OSS"
	IMAPStorageS3        IMAPStorageType = "S3"
)

type InstallMode

type InstallMode string

InstallMode represents the installation mode InstallMode 表示安装模式

const (
	InstallModeOnline  InstallMode = "online"
	InstallModeOffline InstallMode = "offline"
)

type InstallResponse

type InstallResponse struct {
	ErrorMsg string              `json:"error_msg"`
	Data     *InstallationStatus `json:"data"`
}

InstallResponse represents the response for installation. InstallResponse 表示安装响应。

type InstallStep

type InstallStep string

InstallStep represents an installation step InstallStep 表示安装步骤

const (
	InstallStepDownload            InstallStep = "download"
	InstallStepVerify              InstallStep = "verify"
	InstallStepExtract             InstallStep = "extract"
	InstallStepConfigureCluster    InstallStep = "configure_cluster"
	InstallStepConfigureCheckpoint InstallStep = "configure_checkpoint"
	InstallStepConfigureIMAP       InstallStep = "configure_imap"
	InstallStepConfigureJVM        InstallStep = "configure_jvm"
	InstallStepInstallPlugins      InstallStep = "install_plugins"
	InstallStepRegisterCluster     InstallStep = "register_cluster"
	InstallStepComplete            InstallStep = "complete"
)

type InstallationRequest

type InstallationRequest struct {
	HostID                  string                 `json:"host_id"`
	ClusterID               string                 `json:"cluster_id"`
	Version                 string                 `json:"version" binding:"required"`
	InstallDir              string                 `json:"install_dir"`
	InstallMode             InstallMode            `json:"install_mode"`
	Mirror                  MirrorSource           `json:"mirror,omitempty"`
	PackagePath             string                 `json:"package_path,omitempty"`
	DeploymentMode          DeploymentMode         `json:"deployment_mode"`
	NodeRole                NodeRole               `json:"node_role"`
	MasterAddresses         []string               `json:"master_addresses,omitempty"`
	WorkerAddresses         []string               `json:"worker_addresses,omitempty"` // Worker addresses for separated mode / 分离模式的 worker 地址
	ClusterPort             int                    `json:"cluster_port,omitempty"`     // Master hazelcast port / Master Hazelcast 端口
	WorkerPort              int                    `json:"worker_port,omitempty"`      // Worker hazelcast port / Worker Hazelcast 端口
	HTTPPort                int                    `json:"http_port,omitempty"`        // SeaTunnel HTTP API 端口
	JavaProxyPort           int                    `json:"java_proxy_port,omitempty"`  // Managed stx-java-proxy listen port / 托管 stx-java-proxy 监听端口
	EnableHTTP              *bool                  `json:"enable_http,omitempty"`      // 是否开启 SeaTunnel HTTP / Web UI
	DynamicSlot             *bool                  `json:"dynamic_slot,omitempty"`
	SlotNum                 *int                   `json:"slot_num,omitempty"`
	SlotAllocationStrategy  SlotAllocationStrategy `json:"slot_allocation_strategy,omitempty"`
	JobScheduleStrategy     JobScheduleStrategy    `json:"job_schedule_strategy,omitempty"`
	HistoryJobExpireMinutes *int                   `json:"history_job_expire_minutes,omitempty"`
	ScheduledDeletionEnable *bool                  `json:"scheduled_deletion_enable,omitempty"`
	JobLogMode              JobLogMode             `json:"job_log_mode,omitempty"`
	JVM                     *JVMConfig             `json:"jvm,omitempty"`
	Checkpoint              *CheckpointConfig      `json:"checkpoint,omitempty"`
	IMAP                    *IMAPConfig            `json:"imap,omitempty"`
	Connector               *ConnectorConfig       `json:"connector,omitempty"`
}

InstallationRequest is the request for one-click installation InstallationRequest 是一键安装的请求

type InstallationStatus

type InstallationStatus struct {
	ID          string      `json:"id"`
	HostID      string      `json:"host_id"`
	ClusterID   string      `json:"cluster_id,omitempty"`
	Status      StepStatus  `json:"status"`
	CurrentStep InstallStep `json:"current_step"`
	Steps       []StepInfo  `json:"steps"`
	Progress    int         `json:"progress"`
	Message     string      `json:"message,omitempty"`
	Error       string      `json:"error,omitempty"`
	Warnings    []string    `json:"warnings,omitempty"`
	StartTime   time.Time   `json:"start_time"`
	EndTime     *time.Time  `json:"end_time,omitempty"`
}

InstallationStatus represents the current installation status InstallationStatus 表示当前安装状态

type JVMConfig

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

JVMConfig contains JVM memory configuration JVMConfig 包含 JVM 内存配置

type JobLogMode

type JobLogMode string

JobLogMode represents the SeaTunnel log output mode. JobLogMode 表示 SeaTunnel 日志输出模式。

const (
	JobLogModeMixed  JobLogMode = "mixed"
	JobLogModePerJob JobLogMode = "per_job"
)

type JobScheduleStrategy

type JobScheduleStrategy string

JobScheduleStrategy represents the SeaTunnel job schedule strategy. JobScheduleStrategy 表示 SeaTunnel 作业调度策略。

const (
	JobScheduleStrategyWait   JobScheduleStrategy = "WAIT"
	JobScheduleStrategyReject JobScheduleStrategy = "REJECT"
)

type ListPackagesResponse

type ListPackagesResponse struct {
	ErrorMsg string             `json:"error_msg"`
	Data     *AvailableVersions `json:"data"`
}

ListPackagesResponse represents the response for listing packages. ListPackagesResponse 表示获取安装包列表的响应。

type MirrorSource

type MirrorSource string

MirrorSource represents the download mirror source MirrorSource 表示下载镜像源

const (
	MirrorAliyun      MirrorSource = "aliyun"
	MirrorApache      MirrorSource = "apache"
	MirrorHuaweiCloud MirrorSource = "huaweicloud"
)

type NodeJVMResolver

type NodeJVMResolver interface {
	// ResolveNodeJVMByClusterAndHostAndRole resolves effective JVM config for one logical node.
	// ResolveNodeJVMByClusterAndHostAndRole 解析一个逻辑节点的生效 JVM 配置。
	ResolveNodeJVMByClusterAndHostAndRole(ctx context.Context, clusterID uint, hostID uint, role string) (*JVMConfig, error)
}

NodeJVMResolver resolves cluster/node scoped JVM config for installation. NodeJVMResolver 解析安装时需要的集群/节点级 JVM 配置。

type NodeRole

type NodeRole string

NodeRole represents the node role NodeRole 表示节点角色

const (
	NodeRoleMaster       NodeRole = "master"
	NodeRoleWorker       NodeRole = "worker"
	NodeRoleMasterWorker NodeRole = "master/worker"
)

type NodeStarter

type NodeStarter interface {
	// EnsureNodeForInstallation ensures the installed node exists in cluster metadata before startup.
	// EnsureNodeForInstallation 在启动前确保已安装节点存在于集群元数据中。
	EnsureNodeForInstallation(ctx context.Context, clusterID uint, hostID uint, role string, installDir string, hazelcastPort int, apiPort int, workerPort int) error
	// StartNodeByClusterAndHost starts a node by cluster ID and host ID
	// StartNodeByClusterAndHost 根据集群 ID 和主机 ID 启动节点
	StartNodeByClusterAndHost(ctx context.Context, clusterID uint, hostID uint) (bool, string, error)
	// StartNodeByClusterAndHostAndRole starts a node by cluster ID, host ID and role (for separated mode: same host can have master + worker)
	// StartNodeByClusterAndHostAndRole 根据集群 ID、主机 ID 和角色启动节点(分离模式下同一主机可有 master+worker)
	StartNodeByClusterAndHostAndRole(ctx context.Context, clusterID uint, hostID uint, role string) (bool, string, error)
}

NodeStarter is the interface for starting cluster nodes NodeStarter 是启动集群节点的接口

type NodeStatusUpdater

type NodeStatusUpdater interface {
	// UpdateNodeStatusByClusterAndHost updates the node status by cluster ID and host ID
	// UpdateNodeStatusByClusterAndHost 根据集群 ID 和主机 ID 更新节点状态
	UpdateNodeStatusByClusterAndHost(ctx context.Context, clusterID uint, hostID uint, status string) error
}

NodeStatusUpdater is the interface for updating cluster node status NodeStatusUpdater 是更新集群节点状态的接口

type PackageChunkUploadRequest

type PackageChunkUploadRequest struct {
	Version     string
	UploadID    string
	ChunkIndex  int
	TotalChunks int
	TotalSize   int64
	FileName    string
}

PackageChunkUploadRequest is the request for uploading a package chunk. PackageChunkUploadRequest 表示安装包分片上传请求。

type PackageChunkUploadResult

type PackageChunkUploadResult struct {
	UploadID       string       `json:"upload_id"`
	Completed      bool         `json:"completed"`
	ReceivedChunks int          `json:"received_chunks"`
	TotalChunks    int          `json:"total_chunks"`
	Package        *PackageInfo `json:"package,omitempty"`
}

PackageChunkUploadResult is the result of package chunk upload. PackageChunkUploadResult 表示安装包分片上传结果。

type PackageInfo

type PackageInfo struct {
	Version            string                  `json:"version"`
	FileName           string                  `json:"file_name"`
	FileSize           int64                   `json:"file_size"`
	Checksum           string                  `json:"checksum,omitempty"`
	DownloadURLs       map[MirrorSource]string `json:"download_urls"`
	IsLocal            bool                    `json:"is_local"`
	LocalPath          string                  `json:"local_path,omitempty"`
	UploadedAt         *time.Time              `json:"uploaded_at,omitempty"`
	HasSource          bool                    `json:"has_source"`
	SourceStatus       DownloadStatus          `json:"source_status,omitempty"`
	SourceFileName     string                  `json:"source_file_name,omitempty"`
	SourceFileSize     int64                   `json:"source_file_size,omitempty"`
	SourceChecksum     string                  `json:"source_checksum,omitempty"`
	SourceUploadedAt   *time.Time              `json:"source_uploaded_at,omitempty"`
	SourceDownloadURLs map[MirrorSource]string `json:"source_download_urls"`
	SourceError        string                  `json:"source_error,omitempty"`
}

PackageInfo contains information about a SeaTunnel package PackageInfo 包含 SeaTunnel 安装包信息

type PluginInstallInfo

type PluginInstallInfo struct {
	// Name is the plugin name / Name 是插件名称
	Name string `json:"name"`
	// Category is the plugin category (source/sink/transform) / Category 是插件分类
	Category string `json:"category"`
	// Version is the plugin version / Version 是插件版本
	Version string `json:"version"`
	// Status is the installation status / Status 是安装状态
	Status PluginInstallStatus `json:"status"`
	// Progress is the installation progress (0-100) / Progress 是安装进度
	Progress int `json:"progress"`
	// Message is the status message / Message 是状态消息
	Message string `json:"message,omitempty"`
	// Error is the error message if failed / Error 是失败时的错误信息
	Error string `json:"error,omitempty"`
}

PluginInstallInfo contains information about a plugin installation during SeaTunnel setup PluginInstallInfo 包含 SeaTunnel 安装过程中插件安装的信息

type PluginInstallStatus

type PluginInstallStatus string

PluginInstallStatus represents the status of a plugin installation PluginInstallStatus 表示插件安装的状态

const (
	// PluginInstallStatusPending indicates the plugin is waiting to be installed
	// PluginInstallStatusPending 表示插件等待安装
	PluginInstallStatusPending PluginInstallStatus = "pending"
	// PluginInstallStatusDownloading indicates the plugin is being downloaded
	// PluginInstallStatusDownloading 表示插件正在下载
	PluginInstallStatusDownloading PluginInstallStatus = "downloading"
	// PluginInstallStatusInstalling indicates the plugin is being installed
	// PluginInstallStatusInstalling 表示插件正在安装
	PluginInstallStatusInstalling PluginInstallStatus = "installing"
	// PluginInstallStatusCompleted indicates the plugin installation is completed
	// PluginInstallStatusCompleted 表示插件安装完成
	PluginInstallStatusCompleted PluginInstallStatus = "completed"
	// PluginInstallStatusFailed indicates the plugin installation failed
	// PluginInstallStatusFailed 表示插件安装失败
	PluginInstallStatusFailed PluginInstallStatus = "failed"
)

type PluginTransferer

type PluginTransferer interface {
	// TransferPluginToAgent transfers a plugin to an agent
	// TransferPluginToAgent 将插件传输到 Agent
	TransferPluginToAgent(ctx context.Context, agentID, pluginName, version, installDir string, profileKeys []string) error

	// GetPluginArtifactID returns the Maven artifact ID for a plugin name
	// GetPluginArtifactID 返回插件名称对应的 Maven artifact ID
	GetPluginArtifactID(pluginName string) string

	// IsPluginDownloaded checks if a plugin is downloaded locally
	// IsPluginDownloaded 检查插件是否已下载到本地
	IsPluginDownloaded(name, version string) bool

	// DownloadPluginSync downloads a plugin synchronously (blocking)
	// DownloadPluginSync 同步下载插件(阻塞)
	DownloadPluginSync(ctx context.Context, pluginName, version, mirror string, profileKeys []string) error

	// GetPluginPreparationFingerprint returns a stable fingerprint for the plugin's effective dependency set.
	// GetPluginPreparationFingerprint 返回插件当前生效依赖集合的稳定指纹。
	GetPluginPreparationFingerprint(ctx context.Context, pluginName, version string, profileKeys []string) (string, error)

	// RecordInstalledPlugin records a plugin as installed for a cluster
	// RecordInstalledPlugin 记录插件已安装到集群
	RecordInstalledPlugin(ctx context.Context, clusterID uint, pluginName, version string) error
}

PluginTransferer is the interface for transferring plugins to agents PluginTransferer 是向 Agent 传输插件的接口

type PrecheckItem

type PrecheckItem struct {
	Name    string                 `json:"name"`
	Status  CheckStatus            `json:"status"`
	Message string                 `json:"message"`
	Details map[string]interface{} `json:"details,omitempty"`
}

PrecheckItem represents a single precheck result item PrecheckItem 表示单个预检查结果项

type PrecheckRequest

type PrecheckRequest struct {
	MinMemoryMB    int64  `json:"min_memory_mb"`
	MinCPUCores    int    `json:"min_cpu_cores"`
	MinDiskSpaceMB int64  `json:"min_disk_space_mb"`
	InstallDir     string `json:"install_dir"`
	Ports          []int  `json:"ports"`
}

PrecheckRequest represents the request for precheck. PrecheckRequest 表示预检查请求。

type PrecheckResponse

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

PrecheckResponse represents the response for precheck. PrecheckResponse 表示预检查响应。

type PrecheckResult

type PrecheckResult struct {
	Items         []PrecheckItem `json:"items"`
	OverallStatus CheckStatus    `json:"overall_status"`
	Summary       string         `json:"summary"`
}

PrecheckResult contains all precheck results PrecheckResult 包含所有预检查结果

type RefreshVersionsResponse

type RefreshVersionsResponse struct {
	ErrorMsg string   `json:"error_msg"`
	Data     []string `json:"data"`
}

RefreshVersionsResponse represents the response for refreshing versions. RefreshVersionsResponse 表示刷新版本列表的响应。

type RetryStepRequest

type RetryStepRequest struct {
	Step string `json:"step" binding:"required"`
}

RetryStepRequest represents the request for retrying a step. RetryStepRequest 表示重试步骤的请求。

type RuntimeStorageValidationHostResult

type RuntimeStorageValidationHostResult struct {
	HostID   uint              `json:"host_id"`
	HostName string            `json:"host_name,omitempty"`
	Success  bool              `json:"success"`
	Message  string            `json:"message"`
	Details  map[string]string `json:"details,omitempty"`
}

RuntimeStorageValidationHostResult describes one host validation result. RuntimeStorageValidationHostResult 描述单个主机的校验结果。

type RuntimeStorageValidationKind

type RuntimeStorageValidationKind string

RuntimeStorageValidationKind indicates which runtime storage config to validate. RuntimeStorageValidationKind 表示需要校验的运行时存储配置类型。

const (
	RuntimeStorageValidationCheckpoint RuntimeStorageValidationKind = "checkpoint"
	RuntimeStorageValidationIMAP       RuntimeStorageValidationKind = "imap"
)

type RuntimeStorageValidationRequest

type RuntimeStorageValidationRequest struct {
	HostIDs    []uint                       `json:"host_ids" binding:"required,min=1"`
	Kind       RuntimeStorageValidationKind `json:"kind" binding:"required"`
	Checkpoint *CheckpointConfig            `json:"checkpoint,omitempty"`
	IMAP       *IMAPConfig                  `json:"imap,omitempty"`
}

RuntimeStorageValidationRequest validates checkpoint or IMAP storage reachability from hosts. RuntimeStorageValidationRequest 用于校验主机到 checkpoint 或 IMAP 存储的可达性。

type RuntimeStorageValidationResponse

type RuntimeStorageValidationResponse struct {
	ErrorMsg string                          `json:"error_msg"`
	Data     *RuntimeStorageValidationResult `json:"data"`
}

RuntimeStorageValidationResponse represents runtime storage validation response. RuntimeStorageValidationResponse 表示运行时存储校验响应。

type RuntimeStorageValidationResult

type RuntimeStorageValidationResult struct {
	Success bool                                  `json:"success"`
	Kind    RuntimeStorageValidationKind          `json:"kind"`
	Warning string                                `json:"warning,omitempty"`
	Hosts   []*RuntimeStorageValidationHostResult `json:"hosts"`
}

RuntimeStorageValidationResult is the aggregated validation result. RuntimeStorageValidationResult 是聚合后的校验结果。

type Service

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

Service provides installation management functionality. Service 提供安装管理功能。

func NewService

func NewService(packageDir string, agentManager AgentManager) *Service

NewService creates a new Service instance. NewService 创建一个新的 Service 实例。 If packageDir is empty, it uses the configured packages directory. 如果 packageDir 为空,则使用配置的安装包目录。

func NewServiceWithDefaults

func NewServiceWithDefaults() *Service

NewServiceWithDefaults creates a new Service instance with default configuration. NewServiceWithDefaults 使用默认配置创建新的 Service 实例。

func (*Service) CancelDownload

func (s *Service) CancelDownload(ctx context.Context, version string) (*DownloadTask, error)

CancelDownload cancels an ongoing download. CancelDownload 取消正在进行的下载。

func (*Service) CancelDownloadWithExecution added in v1.0.0

func (s *Service) CancelDownloadWithExecution(ctx context.Context, actor executionapp.Actor, version string, request ExecutionRequest) (*DownloadTask, error)

CancelDownloadWithExecution 在公共确认和幂等规则下请求真实停止下载。 CancelDownloadWithExecution requests a real download stop under shared confirmation and idempotency rules.

func (*Service) CancelInstallation

func (s *Service) CancelInstallation(ctx context.Context, hostID uint) (*InstallationStatus, error)

CancelInstallation cancels an ongoing installation. CancelInstallation 取消正在进行的安装。

func (*Service) DeletePackage

func (s *Service) DeletePackage(ctx context.Context, version string) error

DeletePackage deletes a local package. DeletePackage 删除本地安装包。

func (*Service) DeletePackageWithExecution added in v1.0.0

func (s *Service) DeletePackageWithExecution(ctx context.Context, actor executionapp.Actor, version string, request ExecutionRequest) error

DeletePackageWithExecution 在公共确认和幂等规则下删除安装包。 DeletePackageWithExecution deletes a package under shared confirmation and idempotency rules.

func (*Service) FetchSourcePackage added in v1.0.0

func (s *Service) FetchSourcePackage(ctx context.Context, version string, mirror MirrorSource) (*PackageInfo, error)

FetchSourcePackage downloads and validates source for an existing runtime package. FetchSourcePackage 为已有运行包下载并校验同版本源码。

func (*Service) FetchSourcePackageWithExecution added in v1.0.0

func (s *Service) FetchSourcePackageWithExecution(ctx context.Context, actor executionapp.Actor, version string, mirror MirrorSource, request ExecutionRequest) (*PackageInfo, error)

FetchSourcePackageWithExecution 从镜像为已有运行包补充源码包。 FetchSourcePackageWithExecution fetches source from a mirror for an existing runtime package.

func (*Service) GetDownloadStatus

func (s *Service) GetDownloadStatus(ctx context.Context, version string) (*DownloadTask, error)

GetDownloadStatus returns the current download status for a version. GetDownloadStatus 返回某版本的当前下载状态。

func (*Service) GetDownloadStatusForActor added in v1.0.0

func (s *Service) GetDownloadStatusForActor(ctx context.Context, actor executionapp.Actor, version string) (*DownloadTask, error)

GetDownloadStatusForActor 仅向任务所有者或管理员返回下载任务。 GetDownloadStatusForActor returns a download only to its owner or an administrator.

func (*Service) GetInstallationStatus

func (s *Service) GetInstallationStatus(ctx context.Context, hostID uint) (*InstallationStatus, error)

GetInstallationStatus returns the current installation status. GetInstallationStatus 返回当前安装状态。

func (*Service) GetPackageInfo

func (s *Service) GetPackageInfo(ctx context.Context, version string) (*PackageInfo, error)

GetPackageInfo returns information about a specific package version. GetPackageInfo 返回特定版本安装包的信息。

func (*Service) ListAvailableVersions

func (s *Service) ListAvailableVersions(ctx context.Context) (*AvailableVersions, error)

ListAvailableVersions returns available SeaTunnel versions. ListAvailableVersions 返回可用的 SeaTunnel 版本。

func (*Service) ListDownloads

func (s *Service) ListDownloads(ctx context.Context) []*DownloadTask

ListDownloads returns all download tasks. ListDownloads 返回所有下载任务。

func (*Service) ListDownloadsForActor added in v1.0.0

func (s *Service) ListDownloadsForActor(ctx context.Context, actor executionapp.Actor) []*DownloadTask

ListDownloadsForActor 返回当前用户的下载任务,管理员可以查看全部任务。 ListDownloadsForActor returns owned downloads, while administrators can inspect all downloads.

func (*Service) RefreshVersions

func (s *Service) RefreshVersions(ctx context.Context) ([]string, error)

RefreshVersions forces a refresh of the version list from Apache Archive. RefreshVersions 强制从 Apache Archive 刷新版本列表。

func (*Service) RequestCancel added in v1.0.0

RequestCancel stops a package download through the real downloader signal. RequestCancel 通过真实下载停止信号取消安装包下载。

func (*Service) RetryStep

func (s *Service) RetryStep(ctx context.Context, hostID uint, step string) (*InstallationStatus, error)

RetryStep retries a failed installation step. RetryStep 重试失败的安装步骤。

func (*Service) RunPrecheck

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

RunPrecheck runs precheck on a host via Agent. RunPrecheck 通过 Agent 在主机上运行预检查。 This is for INSTALLATION precheck - ports should be AVAILABLE (not in use). 这是安装预检查 - 端口应该可用(未被占用)。 This is opposite to PrecheckNode which checks if SeaTunnel is running. 这与 PrecheckNode 相反,后者检查 SeaTunnel 是否正在运行。

func (*Service) SetAgentManager

func (s *Service) SetAgentManager(manager AgentManager)

SetAgentManager sets the agent manager for sending commands to agents. SetAgentManager 设置用于向 Agent 发送命令的 Agent 管理器。

func (*Service) SetConfigInitializer

func (s *Service) SetConfigInitializer(initializer ConfigInitializer)

SetConfigInitializer sets the config initializer for initializing cluster configs. SetConfigInitializer 设置用于初始化集群配置的配置初始化器。

func (*Service) SetExecutionService added in v1.0.0

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

SetExecutionService sets the shared execution service used by package operations. SetExecutionService 设置安装包操作使用的公共执行服务。

func (*Service) SetHostProvider

func (s *Service) SetHostProvider(provider HostProvider)

SetHostProvider sets the host provider for precheck operations. SetHostProvider 设置用于预检查操作的主机提供者。

func (*Service) SetNodeJVMResolver

func (s *Service) SetNodeJVMResolver(resolver NodeJVMResolver)

SetNodeJVMResolver sets the resolver for node-level JVM config. SetNodeJVMResolver 设置节点级 JVM 配置解析器。

func (*Service) SetNodeStarter

func (s *Service) SetNodeStarter(starter NodeStarter)

SetNodeStarter sets the node starter for starting cluster nodes. SetNodeStarter 设置用于启动集群节点的节点启动器。

func (*Service) SetNodeStatusUpdater

func (s *Service) SetNodeStatusUpdater(updater NodeStatusUpdater)

SetNodeStatusUpdater sets the node status updater for updating cluster node status. SetNodeStatusUpdater 设置用于更新集群节点状态的节点状态更新器。

func (*Service) SetPluginTransferer

func (s *Service) SetPluginTransferer(transferer PluginTransferer)

SetPluginTransferer sets the plugin transferer for transferring plugins to agents. SetPluginTransferer 设置用于向 Agent 传输插件的插件传输器。

func (*Service) SourcePackagePath added in v1.0.0

func (s *Service) SourcePackagePath(version string) (string, error)

SourcePackagePath returns a validated source archive path for authenticated download handlers. SourcePackagePath 返回已校验的源码包路径,供鉴权后的下载接口使用。

func (*Service) StartDownload

func (s *Service) StartDownload(ctx context.Context, req *DownloadRequest) (*DownloadTask, error)

StartDownload starts downloading a package from mirror to local storage. StartDownload 开始从镜像源下载安装包到本地存储。

func (*Service) StartDownloadWithExecution added in v1.0.0

func (s *Service) StartDownloadWithExecution(ctx context.Context, actor executionapp.Actor, download *DownloadRequest, request ExecutionRequest) (*DownloadTask, error)

StartDownloadWithExecution 在公共确认和幂等规则下启动安装包下载。 StartDownloadWithExecution starts a package download under shared confirmation and idempotency rules.

func (*Service) StartInstallation

func (s *Service) StartInstallation(ctx context.Context, req *InstallationRequest) (*InstallationStatus, error)

StartInstallation starts a new installation. StartInstallation 开始新的安装。

func (*Service) TransferPackageToAgent

func (s *Service) TransferPackageToAgent(ctx context.Context, agentID string, version string, status *InstallationStatus) (remotePath string, err error)

TransferPackageToAgent transfers a package to an Agent via gRPC TransferPackageToAgent 通过 gRPC 将安装包传输到 Agent

func (*Service) UploadPackage

func (s *Service) UploadPackage(ctx context.Context, version string, file *multipart.FileHeader) (*PackageInfo, error)

UploadPackage handles package file upload. UploadPackage 处理安装包文件上传。

func (*Service) UploadPackageBundle added in v1.0.0

func (s *Service) UploadPackageBundle(ctx context.Context, version string, file, sourceFile *multipart.FileHeader) (*PackageInfo, error)

UploadPackageBundle stores the required runtime package and an optional matching source archive. UploadPackageBundle 保存必需的运行包以及可选的同版本源码包。

func (*Service) UploadPackageBundleWithExecution added in v1.0.0

func (s *Service) UploadPackageBundleWithExecution(ctx context.Context, actor executionapp.Actor, version string, file, sourceFile *multipart.FileHeader, request ExecutionRequest) (*PackageInfo, error)

UploadPackageBundleWithExecution 上传运行包,并可在同一次请求中附带源码包。 UploadPackageBundleWithExecution uploads the runtime package with an optional source archive in the same request.

func (*Service) UploadPackageChunk

UploadPackageChunk handles package chunk upload. UploadPackageChunk 处理安装包分片上传。

func (*Service) UploadPackageChunkWithExecution added in v1.0.0

func (s *Service) UploadPackageChunkWithExecution(ctx context.Context, actor executionapp.Actor, upload *PackageChunkUploadRequest, file *multipart.FileHeader, request ExecutionRequest) (*PackageChunkUploadResult, error)

UploadPackageChunkWithExecution 在公共确认和幂等规则下上传一个分片。 UploadPackageChunkWithExecution uploads one chunk under shared confirmation and idempotency rules.

func (*Service) UploadPackageWithExecution added in v1.0.0

func (s *Service) UploadPackageWithExecution(ctx context.Context, actor executionapp.Actor, version string, file *multipart.FileHeader, request ExecutionRequest) (*PackageInfo, error)

UploadPackageWithExecution 在公共确认和幂等规则下上传安装包。 UploadPackageWithExecution uploads a package under shared confirmation and idempotency rules.

func (*Service) UploadSourcePackage added in v1.0.0

func (s *Service) UploadSourcePackage(ctx context.Context, version string, file *multipart.FileHeader) (*PackageInfo, error)

UploadSourcePackage validates and atomically stores a source archive for an existing runtime package. UploadSourcePackage 校验源码包,并为已有运行包原子保存源码包。

func (*Service) UploadSourcePackageWithExecution added in v1.0.0

func (s *Service) UploadSourcePackageWithExecution(ctx context.Context, actor executionapp.Actor, version string, file *multipart.FileHeader, request ExecutionRequest) (*PackageInfo, error)

UploadSourcePackageWithExecution 为已有运行包补充或替换源码包。 UploadSourcePackageWithExecution supplements or replaces source for an existing runtime package.

func (*Service) ValidateRuntimeStorage

ValidateRuntimeStorage validates checkpoint or IMAP storage reachability from selected hosts. ValidateRuntimeStorage 校验所选主机到 checkpoint 或 IMAP 存储的可达性。

type SlotAllocationStrategy

type SlotAllocationStrategy string

SlotAllocationStrategy represents the SeaTunnel slot allocation strategy. SlotAllocationStrategy 表示 SeaTunnel slot 分配策略。

const (
	SlotAllocationStrategyRandom     SlotAllocationStrategy = "RANDOM"
	SlotAllocationStrategySystemLoad SlotAllocationStrategy = "SYSTEM_LOAD"
	SlotAllocationStrategySlotRatio  SlotAllocationStrategy = "SLOT_RATIO"
)

type SourceFetchRequest added in v1.0.0

type SourceFetchRequest struct {
	Mirror MirrorSource `json:"mirror"`
}

SourceFetchRequest requests a source-only online supplement for an existing runtime package. SourceFetchRequest 请求为已有运行包在线补充源码包。

type StepInfo

type StepInfo struct {
	Step        InstallStep `json:"step"`
	Name        string      `json:"name"`
	Description string      `json:"description"`
	Status      StepStatus  `json:"status"`
	Progress    int         `json:"progress"`
	Message     string      `json:"message,omitempty"`
	Error       string      `json:"error,omitempty"`
	StartTime   *time.Time  `json:"start_time,omitempty"`
	EndTime     *time.Time  `json:"end_time,omitempty"`
	Retryable   bool        `json:"retryable"`
}

StepInfo contains information about an installation step StepInfo 包含安装步骤的信息

type StepStatus

type StepStatus string

StepStatus represents the status of an installation step StepStatus 表示安装步骤的状态

const (
	StepStatusPending StepStatus = "pending"
	StepStatusRunning StepStatus = "running"
	StepStatusSuccess StepStatus = "success"
	StepStatusFailed  StepStatus = "failed"
	StepStatusSkipped StepStatus = "skipped"
)

type UploadChunkResponse

type UploadChunkResponse struct {
	ErrorMsg string                    `json:"error_msg"`
	Data     *PackageChunkUploadResult `json:"data"`
}

UploadChunkResponse represents the response for uploading one package chunk. UploadChunkResponse 表示上传单个安装包分片的响应。

type UploadPackageResponse

type UploadPackageResponse struct {
	ErrorMsg string       `json:"error_msg"`
	Data     *PackageInfo `json:"data"`
}

UploadPackageResponse represents the response for uploading a package. UploadPackageResponse 表示上传安装包的响应。

Jump to

Keyboard shortcuts

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