longpeng/Beacon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构配置结构，新增 Temporal 配置支持，优化 OpenAI 配置命名，新增工作流相关路由和服务逻辑

Browse Source
This commit is contained in:
longpeng 2025-06-25 14:49:45 +08:00
parent 182bbbd215
commit ae42336160
10 changed files with 265 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
client/main.go
View File

@ -1,58 +0,0 @@
package client
// Go 服务端入口，触发 Workflow
import (
"beacon/pkg/pb"
"beacon/workflows"
"context"
"fmt"
"log"
"github.com/google/uuid"
"go.temporal.io/sdk/client"
)
func main() {
// 创建 Temporal 客户端
c, err := client.Dial(client.Options{
HostPort: "temporal.newai.day:17233", // 根据你的 Temporal Server 配置
Namespace: "default",
})
if err != nil {
log.Fatalf("Unable to create Temporal client: %v", err)
}
defer c.Close()
// 模拟一个触发测试的事件 (例如来自 Web UI 或 CI/CD)
runID := uuid.New().String()
testInput := &pb.DynamicTestRunInput{
RunId: runID,
CompositeCaseId: "11",
}
workflowOptions := client.StartWorkflowOptions{
ID: "test_workflow_" + runID,
TaskQueue: "test-task-queue", // 保持与 Python Worker 一致
}
fmt.Printf("Starting TestRunWorkflow for run ID: %s\n", runID)
we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, workflows.DynamicTestSuiteWorkflow, testInput)
if err != nil {
log.Fatalf("Unable to execute workflows: %v", err)
}
fmt.Printf("Workflow started. Workflow ID: %s, Run ID: %s\n", we.GetID(), we.GetRunID())
// 等待 Workflow 完成并获取结果
var result pb.TestRunOutput
err = we.Get(context.Background(), &result)
if err != nil {
log.Fatalf("Unable to get workflows result: %v", err)
}
fmt.Printf("Workflow finished. Overall Success: %t, Message: %s\n", result.OverallSuccess, result.CompletionMessage)
fmt.Printf("API Test Results: %+v\n", result.ApiResults)
fmt.Printf("UI Test Results: %+v\n", result.UiResults)
// 后续可以根据 result 生成报告、发送通知等
}

10
config/default.go
View File

@ -38,7 +38,8 @@ type SrvConfig struct {
*ApolloConfig `mapstructure:"apollo"`
*Registrar `mapstructure:"registrar"`
*RcloneConfig `mapstructure:"rclone"`
*OpenAI `mapstructure:"openai"`
*OpenAIConfig `mapstructure:"openai"`
*TemporalConfig `mapstructure:"temporal"` // Temporal配置
}
type LogConfig struct {
@ -113,13 +114,18 @@ type RcloneConfig struct {
CustomDomains string `mapstructure:"custom_domains"` // 自定义域
}
type OpenAI struct {
type OpenAIConfig struct {
BaseURL string `mapstructure:"base_url"`
ApiKey string `mapstructure:"api_key"`
Model string `mapstructure:"model"`
Prompt string `mapstructure:"prompt"`
}
type TemporalConfig struct {
Host string `mapstructure:"host"` // Temporal服务地址
Port int `mapstructure:"port"` // Temporal服务端口
}
// Init 整个服务配置文件初始化的方法
func Init(yamlContent string) (err error) {
// 方式1：直接指定配置文件路径（相对路径或者绝对路径）

10
main.go
View File

@ -6,6 +6,7 @@ import (
"beacon/pkg/logger"
"beacon/pkg/validator"
"beacon/routers"
workers "beacon/workers/go"
"flag"
"fmt"
"os"
@ -60,6 +61,15 @@ func Run() {
zap.L().Info(fmt.Sprintf("Serving Gin on %s:%d", config.Conf.IP, config.Conf.Port))
zap.L().Info("service start...")
go func() {
// 启动 Temporal Worker
err := workers.StartWorkflow(config.Conf.TemporalConfig)
if err != nil {
zap.L().Error("启动 Temporal Worker 失败", zap.Error(err))
os.Exit(1) // 如果启动 Temporal Worker 失败，退出程序
}
}()
// 接收操作系统发来的中断信号
QuitChan := make(chan os.Signal)
signal.Notify(QuitChan, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL) // kill -15 CTRL+C kill -9

8
pkg/openai/openai.go
View File

@ -32,8 +32,8 @@ func encodeImageToBase64(file *multipart.FileHeader) (string, error) {
func GetOllamaAICaptcha(file *multipart.FileHeader) (error, string) {
client := openai.NewClient(
option.WithAPIKey(config.Conf.OpenAI.ApiKey), // defaults to os.LookupEnv("OPENAI_API_KEY")
option.WithBaseURL(config.Conf.OpenAI.BaseURL),
option.WithAPIKey(config.Conf.OpenAIConfig.ApiKey), // defaults to os.LookupEnv("OPENAI_API_KEY")
option.WithBaseURL(config.Conf.OpenAIConfig.BaseURL),
)
base64Image, _ := encodeImageToBase64(file)
@ -49,7 +49,7 @@ func GetOllamaAICaptcha(file *multipart.FileHeader) (error, string) {
},
openai.ChatCompletionContentPartUnionParam{
OfText: &openai.ChatCompletionContentPartTextParam{
Text: config.Conf.OpenAI.Prompt,
Text: config.Conf.OpenAIConfig.Prompt,
},
},
)
@ -57,7 +57,7 @@ func GetOllamaAICaptcha(file *multipart.FileHeader) (error, string) {
Messages: []openai.ChatCompletionMessageParamUnion{
openai.UserMessage(message),
},
Model: config.Conf.OpenAI.Model,
Model: config.Conf.OpenAIConfig.Model,
})
if err != nil {
return err, ""

76
routers/handlers/workflow.go Normal file
View File

@ -0,0 +1,76 @@
package handlers
import (
"beacon/services"
"github.com/gin-gonic/gin"
)
type WorkflowHandler struct {
service *services.WorkflowService
}
func NewWorkflowHandler() *WorkflowHandler {
return &WorkflowHandler{
service: &services.WorkflowService{},
}
}
// StartWorkflow POST /api/workflow/start
func (h *WorkflowHandler) StartWorkflow(c *gin.Context) {
err := h.service.Start("11")
if err != nil {
c.JSON(500, gin.H{
"code": -1,
"message": "Failed to start workflow",
})
return
}
c.JSON(200, gin.H{
"message": "Workflow started successfully",
})
}
// StopWorkflow POST /api/workflow/stop
func (h *WorkflowHandler) StopWorkflow(c *gin.Context) {
// 停止工作流的逻辑
// 这里可以调用 Temporal 的 API 来停止指定的工作流
// 例如，使用 WorkflowID 或 RunID 来停止工作流
c.JSON(200, gin.H{
"message": "Workflow stopped successfully",
})
}
// GetWorkflowStatus GET /api/workflow/status/{workflowID}
func (h *WorkflowHandler) GetWorkflowStatus(c *gin.Context) {
workflowID := c.Param("workflowID")
if workflowID == "" {
c.JSON(400, gin.H{"error": "Workflow ID is required"})
return
}
status, err := h.service.GetWorkflowStatus(workflowID)
if err != nil {
c.JSON(500, gin.H{"error": "Failed to get workflow status", "details": err.Error()})
return
}
c.JSON(200, gin.H{"status": status})
}
// GetWorkflowResults GET /api/workflow/results/{workflowID}
func (h *WorkflowHandler) GetWorkflowResults(c *gin.Context) {
workflowID := c.Param("workflowID")
if workflowID == "" {
c.JSON(400, gin.H{"error": "Workflow ID is required"})
return
}
results, err := h.service.GetWorkflowResults(workflowID)
if err != nil {
c.JSON(500, gin.H{"error": "Failed to get workflow results", "details": err.Error()})
return
}
c.JSON(200, gin.H{"results": results})
}

14
routers/router.go
View File

@ -86,6 +86,7 @@ func Init() *gin.Engine {
// 路由分组
v1 := r.Group(config.Conf.Version)
SetupCompositeCaseRoutes(v1)
SetupWorkflowRoutes(v1)
return r
}
@ -104,3 +105,16 @@ func SetupCompositeCaseRoutes(group *gin.RouterGroup) {
api.DELETE("/composite-cases/:id", handler.DeleteCompositeCase)
}
}
func SetupWorkflowRoutes(group *gin.RouterGroup) {
// 初始化服务和处理器
handler := *handlers.NewWorkflowHandler()
api := group.Group("/api")
{
// 工作流相关路由
api.POST("/workflows/start", handler.StartWorkflow)
api.GET("/workflows/:id", handler.GetWorkflowStatus)
api.GET("/workflows/:id/results", handler.GetWorkflowResults)
}
}

79
services/workflow.go Normal file
View File

@ -0,0 +1,79 @@
package services
// Go 服务端入口，触发 Workflow
import (
"beacon/pkg/pb"
"beacon/workflows"
"context"
"github.com/google/uuid"
"go.temporal.io/sdk/client"
"go.uber.org/zap"
)
type WorkflowService struct {
}
func (s *WorkflowService) Start(compositeCaseId string) error {
// 创建 Temporal 客户端
c, err := client.Dial(client.Options{
HostPort: "temporal.newai.day:17233", // 根据你的 Temporal Server 配置
Namespace: "default",
})
if err != nil {
zap.L().Error("Unable to create Temporal client", zap.Error(err))
return err
}
defer c.Close()
// 模拟一个触发测试的事件 (例如来自 Web UI 或 CI/CD)
runID := uuid.New().String()
testInput := &pb.DynamicTestRunInput{
RunId: runID,
CompositeCaseId: compositeCaseId,
}
workflowOptions := client.StartWorkflowOptions{
ID: "test_workflow_" + runID,
TaskQueue: "test-task-queue", // 保持与 Python Worker 一致
}
zap.L().Info("Starting TestRunWorkflow", zap.String("runID", runID))
we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, workflows.DynamicTestSuiteWorkflow, testInput)
if err != nil {
zap.L().Error("Unable to execute workflows", zap.Error(err))
return err
}
zap.L().Info("Workflow started", zap.String("workflowID", we.GetID()), zap.String("runID", we.GetRunID()))
// 等待 Workflow 完成并获取结果
var result pb.TestRunOutput
err = we.Get(context.Background(), &result)
if err != nil {
zap.L().Error("Unable to get workflows result", zap.Error(err))
return err
}
zap.L().Info("Workflow finished", zap.Bool("OverallSuccess", result.OverallSuccess), zap.String("Message", result.CompletionMessage))
zap.L().Debug("API Test Results", zap.Any("ApiResults", result.ApiResults))
zap.L().Debug("UI Test Results", zap.Any("UiResults", result.UiResults))
// 后续可以根据 result 生成报告、发送通知等
return nil
}
func (s *WorkflowService) GetWorkflowResults(workflowID string) (status string, err error) {
// 这里可以实现获取工作流结果的逻辑
// 例如，查询 Temporal Server 获取指定工作流的结果
// 可以使用 WorkflowID 或 RunID 来查询
zap.L().Debug("GetWorkflowResults", zap.Any("workflowID", workflowID))
return
}
func (s *WorkflowService) GetWorkflowStatus(workflowID string) (status string, err error) {
// 这里可以实现获取工作流状态的逻辑
// 例如，查询 Temporal Server 获取指定工作流的状态
// 可以使用 WorkflowID 或 RunID 来查询
zap.L().Debug("GetWorkflowStatus", zap.Any("workflowID", workflowID))
return
}

37
workers/go/main.go
View File

@ -1,37 +0,0 @@
package main
import (
"beacon/activities"
"beacon/workflows"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"log"
)
func main() {
// 创建使用默认选项的 Temporal 客户端。
c, err := client.Dial(client.Options{HostPort: "temporal.newai.day:17233"})
if err != nil {
log.Fatalln("Unable to create Temporal client", err)
}
defer c.Close()
taskQueue := "test-task-queue"
// 创建一个监听指定任务队列的 Worker。
w := worker.New(c, taskQueue, worker.Options{})
// 将工作流和带有真实 Go 后缀的活动注册到 Worker。
w.RegisterWorkflow(workflows.TestRunWorkflow)
w.RegisterWorkflow(workflows.DynamicTestSuiteWorkflow)
w.RegisterActivity(activities.LoadCompositeCaseSteps)
//(for later) w.RegisterActivity(activities.AddSuffixActivity)
// 注意：Python 和 ts 活动将由 Python/ts 进程处理，此处未进行注册。
// 启动 Worker。此调用会阻塞，直到 Worker 被中断。
err = w.Run(worker.InterruptCh())
if err != nil {
log.Fatalln("Unable to start Worker", err)
}
}

70
workers/go/workers.go Normal file
View File

@ -0,0 +1,70 @@
package workers
import (
"beacon/activities"
"beacon/config"
"beacon/workflows"
"fmt"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"go.uber.org/zap"
"log"
)
func StartWorkflow(cnf *config.TemporalConfig) error {
// 创建使用默认选项的 Temporal 客户端。
c, err := client.Dial(client.Options{HostPort: fmt.Sprintf("%s:%d", cnf.Host, cnf.Port)})
if err != nil {
log.Fatalln("Unable to create Temporal client", err)
}
defer c.Close()
taskQueue := "test-task-queue"
// 创建一个监听指定任务队列的 Worker。
w := worker.New(c, taskQueue, worker.Options{})
// 将工作流和带有真实 Go 后缀的活动注册到 Worker。
w.RegisterWorkflow(workflows.TestRunWorkflow)
w.RegisterWorkflow(workflows.DynamicTestSuiteWorkflow)
w.RegisterActivity(activities.LoadCompositeCaseSteps)
//(for later) w.RegisterActivity(activities.AddSuffixActivity)
// 注意：Python 和 ts 活动将由 Python/ts 进程处理，此处未进行注册。
// 启动 Worker。此调用会阻塞，直到 Worker 被中断。
err = w.Run(worker.InterruptCh())
if err != nil {
zap.L().Error(fmt.Sprintf("Unable to start worker: %v", err))
}
return nil
}
/*func main() {
// 创建使用默认选项的 Temporal 客户端。
c, err := client.Dial(client.Options{HostPort: "temporal.newai.day:17233"})
if err != nil {
log.Fatalln("Unable to create Temporal client", err)
}
defer c.Close()
taskQueue := "test-task-queue"
// 创建一个监听指定任务队列的 Worker。
w := worker.New(c, taskQueue, worker.Options{})
// 将工作流和带有真实 Go 后缀的活动注册到 Worker。
w.RegisterWorkflow(workflows.TestRunWorkflow)
w.RegisterWorkflow(workflows.DynamicTestSuiteWorkflow)
w.RegisterActivity(activities.LoadCompositeCaseSteps)
//(for later) w.RegisterActivity(activities.AddSuffixActivity)
// 注意：Python 和 ts 活动将由 Python/ts 进程处理，此处未进行注册。
// 启动 Worker。此调用会阻塞，直到 Worker 被中断。
err = w.Run(worker.InterruptCh())
if err != nil {
log.Fatalln("Unable to start Worker", err)
}
}
*/

4
workflows/dynamic_workflow.go
View File

@ -2,7 +2,9 @@ package workflows
import (
"encoding/json"
"fmt"
"go.temporal.io/sdk/temporal"
"reflect"
"sort"
"strconv"
"time"
@ -98,6 +100,8 @@ func DynamicTestSuiteWorkflow(ctx workflow.Context, input *pb.DynamicTestRunInpu
switch step.ActivityName {
case "RunApiTest":
// API 测试：创建 API 测试请求结构并解析参数
logger.Info("Running api test activity", "ParametersJson", step.ParametersJson)
fmt.Println(reflect.TypeOf(step.ParametersJson))
apiReq := &pb.ApiTestRequest{}
if err := json.Unmarshal([]byte(step.ParametersJson), apiReq); err != nil {
logger.Error("Failed to unmarshal API test parameters", "error", err)