文件驱动型eKuiper自动化管理工具

一、镜像概述与主要用途

1.1 概述

本工具是一款基于文件驱动的自动化管理程序，用于监控指定文件夹下的命令文件。当检测到新增或更新的命令文件时，程序将解析文件内容并执行其中定义的HTTP命令（主要针对eKuiper流处理引擎），执行完成后将文件信息记录至历史文件（.history）中，避免重复处理。

1.2 主要用途

• 自动化管理eKuiper流（Stream）和规则（Rule）的全生命周期（创建、查询、更新、删除等）
• 通过文件触发机制实现eKuiper操作的批量执行与定时调度
• 记录操作历史，确保命令文件仅被处理一次

二、核心功能与特性

• 文件夹监控：定期扫描指定目录，检测新增或修改的命令文件
• 命令解析执行：解析JSON格式的命令文件，生成HTTP请求并发送至eKuiper
• 灵活配置：通过YAML配置文件自定义监控间隔、超时时间、日志路径等参数
• 历史记录：自动生成.history文件，记录已处理的命令文件名及时间戳
• 日志管理：支持自定义日志路径，记录程序运行状态与命令执行结果
• 批量命令支持：单个命令文件可包含多个HTTP请求，按顺序执行

三、使用场景与适用范围

3.1 适用场景

• eKuiper自动化运维：通过命令文件批量创建/删除流和规则，替代手动API调用
• 定时任务调度：结合监控间隔，定期执行状态检查或数据同步命令
• CI/CD集成：在部署流程中自动配置eKuiper流和规则，实现环境一致性
• 批量操作处理：一次性执行多个关联命令（如创建流→创建规则→启动规则）

3.2 适用范围

• 使用eKuiper进行流处理的用户
• 需要通过文件驱动方式管理eKuiper资源的场景
• 需记录操作历史并避免重复执行的自动化流程

四、配置说明

4.1 配置文件（config.yaml）

程序通过YAML格式的配置文件进行参数设置，示例及参数说明如下：

port: 9081          # eKuiper服务端口
timeout: 500        # 命令执行超时时间（单位：毫秒）
intervalTime: 60    # 文件夹检查间隔（单位：秒）
ip: "127.0.0.1"     # eKuiper服务IP地址
logPath: "./log/kubernetes.log"  # 日志文件保存路径
commandDir: "./sample/"          # 命令文件所在目录路径

4.2 命令文件（JSON）格式

命令文件为JSON格式，用于定义待执行的HTTP请求集合。文件需放置在commandDir配置指定的目录下，程序将自动监控并处理。

4.2.1 字段说明

4.2.2 命令文件示例

单个命令文件可包含一个或多个HTTP请求，示例如下：

{
    "commands": [
        {
            "url": "/streams",
            "description": "创建stream1",
            "method": "post",
            "data": {
                "sql": "create stream stream1 (id bigint, name string) WITH (datasource = \"topic/temp\", FORMAT = \"json\");"
            }
        },
        {
            "url": "/streams/stream1",
            "description": "查询stream1",
            "method": "get"
        }
    ]
}

五、使用方法

5.1 环境准备

• 依赖：Go 1.16+（编译环境）、eKuiper服务（运行环境）
• 确保eKuiper服务已启动，且IP和端口可被工具访问

5.2 程序编译

通过以下命令编译生成可执行文件：

go build -o tools/kubernetes/kuiper-kubernetes-tool tools/kubernetes/main.go

编译后生成的可执行文件路径：tools/kubernetes/kuiper-kubernetes-tool

5.3 基本使用步骤

1. 准备配置文件
   创建config.yaml，配置eKuiper连接信息、监控目录等参数（详见4.1节）。

2. 准备命令文件
   在commandDir目录下创建JSON格式的命令文件（详见4.2节）。

3. 启动程序
   执行编译后的可执行文件，程序将自动加载配置并开始监控命令目录：

   ./kuiper-kubernetes-tool  # 需与config.yaml在同一目录，或通过参数指定配置文件路径
   

4. 查看结果

   • 命令执行日志：查看logPath配置的日志文件
   • 历史记录：命令文件所在目录生成.history文件，记录处理记录

5.4 Docker部署

5.4.1 Dockerfile示例

# 编译阶段
FROM golang:1.18-alpine AS builder
WORKDIR /app
COPY . .
RUN go mod download
RUN go build -o kuiper-kubernetes-tool tools/kubernetes/main.go

# 运行阶段
FROM alpine:3.16
WORKDIR /app
COPY --from=builder /app/kuiper-kubernetes-tool .
COPY config.yaml .  # 复制配置文件（或通过挂载外部配置）

# 创建命令目录和日志目录
RUN mkdir -p /app/commands /app/log

# 运行程序
CMD ["./kuiper-kubernetes-tool"]

5.4.2 Docker Run命令

docker run -d \
  --name kuiper-tool \
  -v /host/config.yaml:/app/config.yaml \  # 挂载外部配置文件
  -v /host/commands:/app/commands \      # 挂载命令文件目录（需与config.yaml中commandDir一致）
  -v /host/log:/app/log \                # 挂载日志目录（需与config.yaml中logPath一致）
  kuiper-kubernetes-tool:latest

5.4.3 Docker Compose配置

version: "3"
services:
  kuiper-tool:
    build: .
    volumes:
      - ./config.yaml:/app/config.yaml
      - ./commands:/app/commands
      - ./log:/app/log
    restart: always
    environment:
      - EKUIPER_IP=192.168.1.100  # 可通过环境变量覆盖配置文件中的eKuiper IP（需程序支持环境变量注入）

5.5 命令文件操作示例

5.5.1 流（Stream）操作示例

创建流

{
    "commands": [
        {
            "url": "/streams",
            "description": "创建stream1",
            "method": "post",
            "data": {
                "sql": "create stream stream1 (id bigint, name string, score float) WITH (datasource = \"topic/temperature\", FORMAT = \"json\", KEY = \"id\");"
            }
        }
    ]
}

查询流列表

{
    "commands": [
        {
            "url": "/streams",
            "description": "查询所有流",
            "method": "get"
        }
    ]
}

查询指定流

{
    "commands": [
        {
            "url": "/streams/stream1",
            "description": "查询stream1详情",
            "method": "get"
        }
    ]
}

删除流

{
    "commands": [
        {
            "url": "/streams/stream1",
            "description": "删除stream1",
            "method": "delete"
        }
    ]
}

5.5.2 规则（Rule）操作示例

创建规则

{
    "commands": [
        {
            "url": "/rules",
            "description": "创建rule1",
            "method": "post",
            "data": {
                "id": "rule1",
                "sql": "SELECT * FROM stream1",
                "actions": [{"log": {}}]
            }
        }
    ]
}

启停规则

{
    "commands": [
        {
            "url": "/rules/rule1/stop",
            "description": "停止rule1",
            "method": "post"
        },
        {
            "url": "/rules/rule1/start",
            "description": "启动rule1",
            "method": "post"
        }
    ]
}

查询规则状态

{
    "commands": [
        {
            "url": "/rules/rule1/status",
            "description": "查询rule1状态",
            "method": "get"
        }
    ]
}

5.5.3 多命令集示例

单个命令文件可按顺序执行多个关联操作，示例如下：

{
    "commands": [
        {
            "url": "/streams",
            "description": "创建stream1",
            "method": "post",
            "data": {
                "sql": "create stream stream1 (id bigint) WITH (datasource = \"topic/temp\");"
            }
        },
        {
            "url": "/rules",
            "description": "创建rule1",
            "method": "post",
            "data": {
                "id": "rule1",
                "sql": "SELECT * FROM stream1",
                "actions": [{"log": {}}]
            }
        },
        {
            "url": "/rules/rule1/start",
            "description": "启动rule1",
            "method": "post"
        }
    ]
}

六、历史记录文件（.history）

程序处理命令文件后，会在命令文件所在目录（commandDir）生成.history文件，记录已处理的文件信息，格式如下：

[
  {
    "name": "sample.json",    // 已处理的命令文件名
    "loadTime": 1594362344    // 处理时间戳（Unix时间，秒级）
  }
]

说明：

• .history文件用于避免重复处理同一命令文件
• 若需重新处理某文件，可删除.history中对应记录或修改文件名

七、注意事项

• 命令文件必须为合法JSON格式，否则程序将记录错误日志并跳过该文件
• commandDir目录需提前创建，且程序需有读写权限
• eKuiper服务需正常运行且可被工具访问（IP:Port需正确配置）
• 配置文件中的路径（如logPath、commandDir）若为相对路径，需相对于程序运行目录
• 大批量命令执行时，建议拆分多个文件，避免单个文件处理超时