Documentation
¶
Index ¶
- func ArgOptionalPartition(args map[string]any) (*int32, error)
- func ClosersAsOne(closers []io.Closer) io.Closer
- func CopyBetweenAssets(ctx context.Context, srcAssetID int64, srcPath string, dstAssetID int64, ...) error
- func DialAssetSSH(ctx context.Context, assetID int64) (*ssh.Client, func(), error)
- func DialSSHClient(ctx context.Context, assetID int64) (*ssh.Client, func(), error)
- func ExecWithStdio(ctx context.Context, assetID int64, command string, stdin io.Reader, ...) error
- func ExecuteMongoDB(ctx context.Context, client *mongo.Client, ...) (string, error)
- func ExecuteRedis(ctx context.Context, client *redis.Client, command string) (string, error)
- func ExecuteRedisRaw(ctx context.Context, client *redis.Client, args []string) (string, error)
- func ExecuteSQL(ctx context.Context, db *sql.DB, sqlText string) (string, error)
- func ExecuteSQLPaged(ctx context.Context, db *sql.DB, sqlText string, page, pageSize int, ...) (string, error)
- func ExecuteSSHCommand(ctx context.Context, assetID int64, command string) (string, error)
- func ExecuteWithSFTP(ctx context.Context, assetID int64, fn func(*sftp.Client) error) error
- func FormatEtcdCommand(req *etcd_svc.ExecRequest) string
- func HandleExecEtcd(ctx context.Context, args map[string]any) (string, error)
- func HandleExecMongo(ctx context.Context, args map[string]any) (string, error)
- func HandleExecRedis(ctx context.Context, args map[string]any) (string, error)
- func HandleExecSQL(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaACL(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaCluster(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaConnect(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaConsumerGroup(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaMessage(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaSchema(ctx context.Context, args map[string]any) (string, error)
- func HandleKafkaTopic(ctx context.Context, args map[string]any) (string, error)
- func HandleRunSerialCommand(ctx context.Context, args map[string]any) (string, error)
- func IsExpectedCloseErr(err error) bool
- func KafkaACLCommand(operation string) (string, error)
- func KafkaAlterTopicConfigRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.AlterTopicConfigRequest, error)
- func KafkaCheckSchemaCompatibilityRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.CheckSchemaCompatibilityRequest, error)
- func KafkaClusterCommand(operation string) (string, error)
- func KafkaConnectCommand(operation, connector string) (string, error)
- func KafkaConnectorConfigRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.ConnectorConfigRequest, error)
- func KafkaConsumerGroupCommand(operation, group string) (string, error)
- func KafkaCreateACLRequestFromArgs(assetID int64, args map[string]any) kafka_svc.CreateACLRequest
- func KafkaCreateTopicRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.CreateTopicRequest, error)
- func KafkaDeleteACLRequestFromArgs(assetID int64, args map[string]any) kafka_svc.DeleteACLRequest
- func KafkaDeleteRecordsRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.DeleteRecordsRequest, error)
- func KafkaDeleteSchemaRequestFromArgs(assetID int64, args map[string]any) kafka_svc.DeleteSchemaRequest
- func KafkaInt32SliceFromJSON(raw string) ([]int32, error)
- func KafkaListACLsRequestFromArgs(assetID int64, args map[string]any) kafka_svc.ListACLsRequest
- func KafkaMessageCommand(operation, topic string) (string, error)
- func KafkaProduceHeadersFromArgs(args map[string]any) ([]kafka_svc.ProduceMessageHeader, error)
- func KafkaRegisterSchemaRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.RegisterSchemaRequest, error)
- func KafkaResetConsumerGroupOffsetRequestFromArgs(assetID int64, args map[string]any) (kafka_svc.ResetConsumerGroupOffsetRequest, error)
- func KafkaRestartConnectorRequestFromArgs(assetID int64, args map[string]any) kafka_svc.RestartConnectorRequest
- func KafkaSchemaCommand(operation, subject string) (string, error)
- func KafkaSchemaReferencesFromJSON(raw string) ([]kafka_svc.SchemaReference, error)
- func KafkaStringMapFromJSON(raw string) (map[string]string, error)
- func KafkaTopicCommand(operation, topic string) (string, error)
- func ListMongoCollections(ctx context.Context, client *mongo.Client, database string) ([]string, error)
- func ListMongoDatabases(ctx context.Context, client *mongo.Client) ([]string, error)
- func NormalizeKafkaOperation(operation, fallback string) string
- func RunSSHCommand(ctx context.Context, client *ssh.Client, command string) (string, error)
- func WithDatabaseCache(ctx context.Context, cache *DatabaseClientCache) context.Context
- func WithKafkaService(ctx context.Context, svc *kafka_svc.Service) context.Context
- func WithMongoDBCache(ctx context.Context, cache *MongoDBClientCache) context.Context
- func WithRedisCache(ctx context.Context, cache *RedisClientCache) context.Context
- func WithSSHPool(ctx context.Context, pool *sshpool.Pool) context.Context
- func WithSerialManager(ctx context.Context, mgr serial_svc.CommandManager) context.Context
- type AIPoolDialer
- type ConnCache
- type DatabaseClientCache
- type MongoDBClientCache
- type RedisClientCache
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ArgOptionalPartition ¶
ArgOptionalPartition parses the optional kafka_message partition field.
func ClosersAsOne ¶
ClosersAsOne 将多个 closer 打包成单个 io.Closer（用于只接受单 closer 的 API，如 ConnCache）。
func CopyBetweenAssets ¶
func CopyBetweenAssets(ctx context.Context, srcAssetID int64, srcPath string, dstAssetID int64, dstPath string) error
CopyBetweenAssets 在两个资产间直接传输文件（SFTP 流式，不经本地磁盘）
func DialAssetSSH ¶
DialAssetSSH 建立到资产的 SSH 连接，返回 client 与一次性清理函数。 底层委托给 credential_resolver.DialAssetSSH，统一支持 SOCKS5/HTTP 代理与跳板机链。 调用方必须在使用结束后调用返回的 cleanup，一并关闭 client 与跳板机链上的中间连接。
func DialSSHClient ¶
DialSSHClient 创建 SSH 客户端连接，自动解析凭据、代理、跳板机链。 调用者必须调用返回的 cleanup 关闭 client 与链路资源。
func ExecWithStdio ¶
func ExecWithStdio(ctx context.Context, assetID int64, command string, stdin io.Reader, stdout, stderr io.Writer) error
ExecWithStdio 在远程服务器执行命令，直接连接 stdio（支持管道）
func ExecuteMongoDB ¶
func ExecuteMongoDB(ctx context.Context, client *mongo.Client, database, collection, operation, query string) (string, error)
ExecuteMongoDB 执行 MongoDB 操作并返回 JSON 结果
func ExecuteRedis ¶
ExecuteRedis 执行 Redis 命令并返回 JSON 结果
func ExecuteRedisRaw ¶
ExecuteRedisRaw 使用预拆分的参数执行 Redis 命令（支持含空格的值）
func ExecuteSQL ¶
ExecuteSQL 执行 SQL 并返回 JSON 结果
func ExecuteSQLPaged ¶
func ExecuteSQLPaged(ctx context.Context, db *sql.DB, sqlText string, page, pageSize int, driver asset_entity.DatabaseDriver) (string, error)
ExecuteSQLPaged 对 SELECT/WITH 语句进行子查询包装分页，其他语句走 ExecuteSQL
func ExecuteSSHCommand ¶
ExecuteSSHCommand 执行一次性 SSH 命令并返回输出（每次新建连接）
func ExecuteWithSFTP ¶
ExecuteWithSFTP 创建临时 SSH+SFTP 连接并执行操作。 ctx 取消时主动关闭底层连接以打断 fn 内部可能的 io.Copy 阻塞， 从而让 AI 停止会话能立即生效（否则大文件传输会挂住 runner.Stop）。
func FormatEtcdCommand ¶
func FormatEtcdCommand(req *etcd_svc.ExecRequest) string
FormatEtcdCommand 把结构化 ExecRequest 还原为策略匹配 / 审计可读的命令字符串。
规则：
- 复合 op（"member_list" / "endpoint_status" 等）还原为 "member list" / "endpoint status"；
- 依次追加 key、value 与 --prefix 标志；
- 与 audit 提取器、SaveGrantPattern 共用同一份格式，避免策略匹配与审计文本漂移。
func HandleExecEtcd ¶
HandleExecEtcd 是 exec_etcd AI 工具的入口。
策略 / grant / 审批由 permission.CheckForAsset 在 svc 调用前完成； svc.Exec 内部已有三态日志与 dispatch，这里不重复记录。
func HandleExecMongo ¶
func HandleExecRedis ¶
func HandleKafkaCluster ¶
func HandleKafkaConnect ¶
func HandleKafkaMessage ¶
func HandleKafkaSchema ¶
func HandleKafkaTopic ¶
func HandleRunSerialCommand ¶
func IsExpectedCloseErr ¶
IsExpectedCloseErr 判断 SSH/网络连接关闭时的预期错误。 取消路径会主动 Close session/client 打断阻塞，随后的 defer 关闭就会返回这些错误； 归类为预期错误后，上层可以跳过 warn 日志，避免噪音。
func KafkaACLCommand ¶
func KafkaClusterCommand ¶
KafkaClusterCommand 将 kafka_cluster 工具的 operation 映射为策略命令字符串。
func KafkaConnectCommand ¶
func KafkaCreateACLRequestFromArgs ¶
func KafkaCreateACLRequestFromArgs(assetID int64, args map[string]any) kafka_svc.CreateACLRequest
func KafkaDeleteACLRequestFromArgs ¶
func KafkaDeleteACLRequestFromArgs(assetID int64, args map[string]any) kafka_svc.DeleteACLRequest
func KafkaDeleteSchemaRequestFromArgs ¶
func KafkaDeleteSchemaRequestFromArgs(assetID int64, args map[string]any) kafka_svc.DeleteSchemaRequest
func KafkaInt32SliceFromJSON ¶
func KafkaListACLsRequestFromArgs ¶
func KafkaListACLsRequestFromArgs(assetID int64, args map[string]any) kafka_svc.ListACLsRequest
func KafkaMessageCommand ¶
func KafkaProduceHeadersFromArgs ¶
func KafkaProduceHeadersFromArgs(args map[string]any) ([]kafka_svc.ProduceMessageHeader, error)
func KafkaRestartConnectorRequestFromArgs ¶
func KafkaRestartConnectorRequestFromArgs(assetID int64, args map[string]any) kafka_svc.RestartConnectorRequest
func KafkaSchemaCommand ¶
func KafkaSchemaReferencesFromJSON ¶
func KafkaSchemaReferencesFromJSON(raw string) ([]kafka_svc.SchemaReference, error)
func KafkaTopicCommand ¶
func ListMongoCollections ¶
func ListMongoCollections(ctx context.Context, client *mongo.Client, database string) ([]string, error)
ListMongoCollections 列出指定数据库的所有集合名称
func ListMongoDatabases ¶
ListMongoDatabases 列出所有数据库名称
func NormalizeKafkaOperation ¶
func RunSSHCommand ¶
RunSSHCommand 在已有的 SSH 客户端上执行命令
func WithDatabaseCache ¶
func WithDatabaseCache(ctx context.Context, cache *DatabaseClientCache) context.Context
WithDatabaseCache 将数据库缓存注入 context
func WithKafkaService ¶
WithKafkaService 将 Kafka 服务注入 context，让 AI handler 在同一次 AI Send 内复用 franz-go client（避免每次工具调用都重新 dial+ping）。
func WithMongoDBCache ¶
func WithMongoDBCache(ctx context.Context, cache *MongoDBClientCache) context.Context
WithMongoDBCache 将 MongoDB 缓存注入 context
func WithRedisCache ¶
func WithRedisCache(ctx context.Context, cache *RedisClientCache) context.Context
WithRedisCache 将 Redis 缓存注入 context
func WithSSHPool ¶
WithSSHPool 将 SSH 连接池注入 context（供 connpool 隧道使用）
func WithSerialManager ¶
func WithSerialManager(ctx context.Context, mgr serial_svc.CommandManager) context.Context
WithSerialManager 将串口管理器注入 context
Types ¶
type AIPoolDialer ¶
type AIPoolDialer struct{}
AIPoolDialer 实现 sshpool.PoolDialer，委托给 credential_resolver 统一 dial
type ConnCache ¶
ConnCache 泛型连接缓存，在同一次 AI Send 中复用连接。 并发模型：mu 保护 maps；sf 保证同一 assetID 并发首拨只发生一次（second waiter 拿到 first dialer 的结果），batch_command 多条命令并发命中同资产时只握手一次。
func (*ConnCache[C]) Forget ¶
Forget 仅把 assetID 对应的连接从缓存中摘除，但不主动调用 Close。 用于 client 已被上游提前关闭（例如 RunSSHCommand 在 ctx 取消时已 Close）的场景， 避免 Remove 触发二次 Close 及伴随的预期关闭错误日志。
func (*ConnCache[C]) GetOrDial ¶
func (c *ConnCache[C]) GetOrDial(assetID int64, dial func() (C, io.Closer, error)) (C, io.Closer, error)
GetOrDial 从缓存获取连接，不存在则通过 dial 创建并缓存。 返回的 closer 始终为 nil —— 缓存命中或新拨号都由 cache 持有 closer， 调用方不需要也不应该 Close 这个连接（应通过 Remove/Forget/Close 释放）。
并发安全：同一 assetID 的并发首拨经 singleflight 合流为一次 dial，second waiter 直接拿到 first dialer 的 client；不同 assetID 完全并行。
type DatabaseClientCache ¶
DatabaseClientCache 在同一次 AI Send 中复用数据库连接
func NewDatabaseClientCache ¶
func NewDatabaseClientCache() *DatabaseClientCache
NewDatabaseClientCache 创建数据库连接缓存
type MongoDBClientCache ¶
type MongoDBClientCache = ConnCache[*connpool.MongoClientCloser]
MongoDBClientCache 在同一次 AI Send 中复用 MongoDB 连接
func NewMongoDBClientCache ¶
func NewMongoDBClientCache() *MongoDBClientCache
NewMongoDBClientCache 创建 MongoDB 连接缓存
type RedisClientCache ¶
RedisClientCache 在同一次 AI Send 中复用 Redis 连接
func NewRedisClientCache ¶
func NewRedisClientCache() *RedisClientCache
NewRedisClientCache 创建 Redis 连接缓存
Source Files