commit

amd64/dispatch_server/README.md

@@ -1,2 +0,0 @@
-# 依赖
-1. Redis

amd64/dispatch_server/main.go

@@ -1,86 +0,0 @@
-package main
-
-/*
-1. 算法清理可以用仿真云平台的调度系统
-*/
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/global"
-	"cicv-data-closedloop/amd64/dispatch_server/package/handler"
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"cicv-data-closedloop/amd64/dispatch_server/package/service"
-	"cicv-data-closedloop/common/gin/middleware"
-	"cicv-data-closedloop/common/util"
-	_ "embed"
-	"encoding/json"
-	"github.com/gin-gonic/gin"
-	"os"
-)
-
-func init() {
-	// 1 解析YAML内容
-	infra.InitApplication()
-	// 2 初始化 日志
-	infra.InitLog(infra.ApplicationYaml.Log.Dir, infra.ApplicationYaml.Log.Prefix)
-	// 3 初始化 阿里云oss 客户端
-	infra.InitOss(
-		infra.ApplicationYaml.Oss.IsUseCname,
-		infra.ApplicationYaml.Oss.Endpoint,
-		infra.ApplicationYaml.Oss.AccessKeyId,
-		infra.ApplicationYaml.Oss.AccessKeySecret,
-		infra.ApplicationYaml.Oss.BucketName,
-	)
-	// 4 初始化 Redis 客户端
-	infra.InitRedisClient(
-		infra.ApplicationYaml.Redis.Addr,
-		infra.ApplicationYaml.Redis.Password,
-		infra.ApplicationYaml.Redis.Db,
-	)
-	// 5 将 gpu-node-list 写入redis
-	err := infra.GlobalRedisClient.Del(global.KeyGpuNodeList).Err()
-	if err != nil {
-		infra.GlobalLogger.Error("程序崩溃。gpu-node-list 初始化失败1：", err)
-		os.Exit(-1)
-	}
-	for _, gpuNode := range infra.ApplicationYaml.GpuNodeList {
-		gpuNodeJson, err := json.MarshalIndent(gpuNode, "", "    ")
-		if err != nil {
-			infra.GlobalLogger.Error("程序崩溃。gpu-node-list 初始化失败2：", err)
-			os.Exit(-1)
-		}
-		_, err = infra.GlobalRedisClient.RPush("gpu-node-list", gpuNodeJson).Result()
-		if err != nil {
-			infra.GlobalLogger.Error("程序崩溃。gpu-node-list 初始化失败3：", err)
-			os.Exit(-1)
-		}
-	}
-
-	// 6 初始化 kafka 客户端
-	infra.InitKafkaProducer(infra.ApplicationYaml.Kafka.Broker)
-	// 7 初始化 k8s 客户端
-	infra.InitKubernetesClient()
-
-}
-
-func main() {
-
-	// 启动任务处理进程
-	go service.RunWaitingUser()
-	go service.RunWaitingCluster()
-
-	// 启动 web 服务器
-	router := gin.Default()
-	//router.Use(commonHandler.ValidateHeaders()) // todo 暂时不需要校验
-	router.Use(middleware.Cors()) // 解决cors
-	api1 := router.Group(infra.ApplicationYaml.Web.RoutePrefix)
-	api1.POST("/start-project", handler.StartProject)
-	api2 := router.Group("/simulation/resource/scheduler/task")
-	api2.GET("/confirm", handler.Confirm)
-	api2.GET("/tick", handler.Tick)
-	api2.GET("/state", handler.State)
-	err := router.Run(":" + infra.ApplicationYaml.Web.Port)
-	if err != nil {
-		infra.GlobalLogger.Error("程序崩溃，监听端口 " + util.ToString(infra.ApplicationYaml.Web.Port) + " 失败。")
-		os.Exit(-1)
-	}
-}

amd64/dispatch_server/package/domain/comm_with_k8s.go

@@ -1 +0,0 @@
-package domain

amd64/dispatch_server/package/domain/comm_with_redis.go

@@ -1,161 +0,0 @@
-package domain
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/entity"
-	"cicv-data-closedloop/amd64/dispatch_server/package/global"
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"cicv-data-closedloop/amd64/dispatch_server/package/util"
-	"encoding/json"
-	"errors"
-	"fmt"
-)
-
-func CanRunUser(userId string, userParallelism int64) bool {
-	TaskQueueRunningUserNumber, _ := infra.GlobalRedisClient.LLen(global.KeyTaskQueueRunningUserPrefix + ":" + userId).Result()
-	if TaskQueueRunningUserNumber < userParallelism {
-		return true
-	}
-	return false
-}
-
-// 集群是否有剩余并行度可以运行任务
-//   - return 可运行任务的节点，取并行度剩余最大的节点
-func CanRunCluster() (bool, infra.GpuNode, error) {
-	// 1 获取集群并行度
-	gpuNodeJsons, err := infra.GlobalRedisClient.LRange(global.KeyGpuNodeList, 0, -1).Result()
-	if err != nil {
-		return false, infra.GpuNode{}, errors.New("获取集群并行度列表 " + global.KeyGpuNodeList + " 失败，错误信息为:" + util.ToString(err))
-	}
-	//infra.GlobalLogger.Infof("集群并行度列表为：%v", util.ToString(gpuNodeJsons))
-	var maxNode infra.GpuNode
-	var maxParallelism int64
-	var can bool
-	maxParallelism = 0
-	for _, gpuNodeJson := range gpuNodeJsons {
-		node, err := JsonToGpuNode(gpuNodeJson)
-		if err != nil {
-			return false, infra.GpuNode{}, err
-		}
-		if node.Parallelism != 0 && node.Parallelism > maxParallelism {
-			maxNode = node
-			maxParallelism = node.Parallelism
-			can = true
-		}
-	}
-	return can, maxNode, nil
-}
-
-func AddWaitingUser(userId string, userParallelism int64, algorithmObjectKey string, task entity.Task) error {
-	taskCacheTemp := entity.TaskCache{
-		UserId:             userId,
-		AlgorithmObjectKey: algorithmObjectKey,
-		UserParallelism:    userParallelism,
-		Task:               task,
-	}
-	// 转 json
-	taskCacheJson, err := TaskCacheToJson(taskCacheTemp)
-	if err != nil {
-		return errors.New("任务缓存对象" + util.ToString(taskCacheTemp) + " 转 json 失败，错误信息为: " + util.ToString(err))
-	}
-	// 添加到用户等待队列
-	key := global.KeyTaskQueueWaitingUser
-	err = infra.GlobalRedisClient.RPush(key, taskCacheJson).Err()
-	if err != nil {
-		return errors.New("任务缓存对象json " + taskCacheJson + "  " + key + " 失败，错误信息为: " + util.ToString(err))
-	}
-	return nil
-}
-
-func DeleteWaitingUser(taskCacheJson string) error {
-	// 定义要删除的列表和值
-	key := global.KeyTaskQueueWaitingUser
-	value := taskCacheJson
-	// 使用 LREM 命令删除列表中的特定值
-	_, err := infra.GlobalRedisClient.LRem(key, 0, value).Result()
-	if err != nil {
-		return errors.New("删除用户等待队列中的值 " + taskCacheJson + " 失败，错误信息为: " + util.ToString(err))
-	}
-	return nil
-}
-
-func AddWaitingCluster(userId string, userParallelism int64, algorithmObjectKey string, task entity.Task) error {
-	taskCacheTemp := entity.TaskCache{
-		UserId:             userId,
-		AlgorithmObjectKey: algorithmObjectKey,
-		UserParallelism:    userParallelism,
-		Task:               task,
-	}
-	// 转 json
-	taskCacheJson, err := TaskCacheToJson(taskCacheTemp)
-	if err != nil {
-		return errors.New("任务缓存对象" + util.ToString(taskCacheTemp) + " 转 json 失败，错误信息为: " + util.ToString(err))
-	}
-	// 添加到集群等待队列
-	err = infra.GlobalRedisClient.RPush(global.KeyTaskQueueWaitingCluster, taskCacheJson).Err()
-	if err != nil {
-		return errors.New("任务缓存对象json " + taskCacheJson + " 添加到集群等待队列失败，错误信息为: " + util.ToString(err))
-	}
-	return nil
-}
-
-func AddRunningCluster(taskCache entity.TaskCache, nodeName string) error {
-	taskCacheTemp := taskCache
-	// 转 json
-	taskCacheJson, err := TaskCacheToJson(taskCacheTemp)
-	if err != nil {
-		return errors.New("任务缓存对象" + util.ToString(taskCacheTemp) + " 转 json 失败，错误信息为: " + util.ToString(err))
-	}
-	// 添加到集群运行队列
-	err = infra.GlobalRedisClient.RPush(global.KeyTaskQueueRunningCluster, taskCacheJson).Err()
-	if err != nil {
-		return errors.New("任务缓存对象json " + taskCacheJson + " 添加到集群等待队列失败，错误信息为: " + util.ToString(err))
-	}
-
-	// 设置任务和节点映射
-	err = infra.GlobalRedisClient.Set(global.KeyTaskToNode+":"+taskCache.Task.Info.TaskId, nodeName, -1).Err()
-	if err != nil {
-		return errors.New("设置任务与节点关联键值对失败，错误信息为: " + util.ToString(err))
-	}
-	// 减少并行度
-	global.ParallelismMutex.Lock()
-	gpuNodeJsons, _ := infra.GlobalRedisClient.LRange(global.KeyGpuNodeList, 0, -1).Result()
-	for i, gpuNodeJson := range gpuNodeJsons {
-		node, _ := JsonToGpuNode(gpuNodeJson)
-		if node.Hostname == nodeName {
-			node.Parallelism--
-			nodeJson, _ := GpuNodeToJson(node)
-			_, _ = infra.GlobalRedisClient.LSet(global.KeyGpuNodeList, int64(i), nodeJson).Result()
-			infra.GlobalLogger.Infof("节点 %v 并行度减一后剩余并行度为 %v。", node.Hostname, node.Parallelism)
-		}
-	}
-	global.ParallelismMutex.Unlock()
-	return nil
-}
-
-func TaskCacheToJson(taskCache entity.TaskCache) (string, error) {
-	jsonData, err := json.MarshalIndent(taskCache, "", "    ")
-	if err != nil {
-		return "", err
-	}
-	return string(jsonData), nil
-}
-
-func JsonToGpuNode(jsonData string) (infra.GpuNode, error) {
-	// 创建一个 Person 类型的变量
-	var taskCache infra.GpuNode
-
-	// 使用 json.Unmarshal 解析 JSON 字符串到结构体
-	err := json.Unmarshal([]byte(jsonData), &taskCache)
-	if err != nil {
-		return infra.GpuNode{}, errors.New("对象json " + jsonData + " 转对象失败错误信息为: " + fmt.Sprintf("%v", err))
-	}
-	return taskCache, nil
-}
-
-func GpuNodeToJson(gpuNode infra.GpuNode) (string, error) {
-	jsonData, err := json.MarshalIndent(gpuNode, "", "    ")
-	if err != nil {
-		return "", err
-	}
-	return string(jsonData), nil
-}

amd64/dispatch_server/package/entity/camera_default.ini

@@ -1,20 +0,0 @@
-;默认的摄像头配置
-
-[Camera]
-; xyz unit: millimeter
-; hpr uint: degree
-; CameraFrameRate only affect video,
-CustomCamera = false
-CameraName = "custom_camera"
-CameraFovH = 45
-CameraFovV = 27
-CameraNear = 1
-CameraFar = 1500
-CameraResolution = "480*270"
-CameraFrameRate = 25
-CameraX = -15000
-CameraY = 0
-CameraZ = 5000
-CameraH = 0
-CameraP = 10
-CameraR = 0

amd64/dispatch_server/package/entity/project.go

@@ -1,9 +0,0 @@
-package entity
-
-type Project struct {
-	ProjectId          string `json:"projectId"`
-	AlgorithmObjectKey string `json:"algorithmObjectKey"`
-	UserId             string `json:"userId"`
-	Parallelism        int64  `json:"parallelism"`
-	Tasks              []Task `json:"tasks"`
-}

amd64/dispatch_server/package/entity/sensor.go

@@ -1,44 +0,0 @@
-package entity
-
-type Sensor struct {
-	SensorName string  `json:"sensor_name"` // 传感器名称
-	SensorNear float64 `json:"sensor_near"` // 盲区距离
-	SensorFar  float64 `json:"sensor_far"`  // 探测距离
-	SensorX    float64 `json:"sensor_x"`    // 传感器横向偏移量（x轴）
-	SensorY    float64 `json:"sensor_y"`    // 传感器纵向偏移量（y轴）
-	SensorZ    float64 `json:"sensor_z"`    // 传感器安装高度（z轴）
-	SensorH    float64 `json:"sensor_h"`    // 传感器横摆角
-	SensorP    float64 `json:"sensor_p"`    // 传感器俯仰角
-	SensorR    float64 `json:"sensor_r"`    // 传感器横滚角
-}
-
-// SensorParam 从数据闭环平台接收到的参数，通用
-type SensorParam struct {
-	Sensor
-	SensorForH       float64 `json:"sensor_fovH"`        // 水平视场角
-	SensorForV       float64 `json:"sensor_fovV"`        // 垂直视场角
-	SensorForHOffset float64 `json:"sensor_fovH_offset"` // 水平视场角偏移量
-	SensorForVOffset float64 `json:"sensor_fovV_offset"` // 垂直视场角偏移量
-}
-
-// SensorCamera 摄像头
-type SensorCamera struct {
-	Sensor
-	SensorForH       float64 `json:"sensor_fovH"`       // 水平现场角
-	SensorForV       float64 `json:"sensor_fovV"`       // 垂直现场角
-	SensorResolution string  `json:"sensor_resolution"` // 分辨率
-	SensorFrameRate  float64 `json:"sensor_frameRate"`  // 帧率
-}
-
-// SensorOgt 完美传感器
-type SensorOgt struct {
-	Sensor
-	SensorForHLeft   float64 `json:"sensor_fovHLeft"`   // 水平现场角左
-	SensorForHRight  float64 `json:"sensor_fovHRight"`  // 水平现场角右
-	SensorForHTop    float64 `json:"sensor_fovVTop"`    // 垂直现场角顶
-	SensorForHBottom float64 `json:"sensor_fovVBottom"` // 垂直现场角底
-	SensorFilter     string  `json:"sensor_filter"`     // 目标物筛选序列（0,1,2）
-	SensorDisplay    bool    `json:"sensor_display"`    // 显示目标物
-	SensorMaxObjects int     `json:"sensor_maxObjects"` // 最大目标物个数
-	SensorPort       int     `json:"sensor_port"`       // 端口
-}

amd64/dispatch_server/package/entity/sensor.sql

@@ -1,73 +0,0 @@
-
-create table if not exists simulation.relation_config_sensor
-(
-    id             varchar(32)    not null comment '主键id'
-    primary key,
-    config_id      varchar(32)    null comment '配置id',
-    sensor_id      varchar(32)    null comment '传感器id',
-    sensor_type    varchar(32)    null comment '传感器类型',
-    sensor_x       decimal(19, 4) null comment '传感器横向偏移量（x轴）',
-    sensor_y       decimal(19, 4) null comment '传感器纵向偏移量（y轴）',
-    sensor_z       decimal(19, 4) null comment '传感器安装高度（z轴）',
-    sensor_h       decimal(19, 4) null comment '传感器横摆角',
-    sensor_p       decimal(19, 4) null comment '传感器俯仰角',
-    sensor_r       decimal(19, 4) null comment '传感器横滚角',
-    sensor_port    int            null comment '传感器端口',
-    create_time    datetime       null,
-    modify_time    datetime       null,
-    create_user_id varchar(32)    null,
-    modify_user_id varchar(32)    null,
-    is_deleted     varchar(1)     null
-    )
-    comment '配置与传感器关联表';
-
-
-create table if not exists simulation.model_sensor_camera
-(
-    id             varchar(32)    not null comment '主键id'
-    primary key,
-    sensor_code    varchar(32)    null comment '传感器编码',
-    sensor_name    varchar(500)   null comment '传感器名称',
-    description    text           null comment '传感器描述',
-    near_distance  decimal(19, 4) null comment '盲区距离',
-    far_distance   decimal(19, 4) null comment '探测距离',
-    fov_h          decimal(19, 4) null comment '水平现场角',
-    fov_v          decimal(19, 4) null comment '垂直现场角',
-    resolution     varchar(32)    null comment '分辨率',
-    frame_rate     decimal(19, 4) null comment '帧率',
-    self_display   varchar(1)     null comment '是否显示本车（1显示0不显示）',
-    share          varchar(32)    null comment '是否分享',
-    create_time    datetime       null,
-    modify_time    datetime       null,
-    create_user_id varchar(32)    null,
-    modify_user_id varchar(32)    null,
-    is_deleted     varchar(1)     null
-    )
-    comment '摄像头表';
-
-create table if not exists simulation.model_sensor_ogt
-(
-    id                varchar(32)    not null comment '主键id'
-        primary key,
-    sensor_code       varchar(32)    null comment '传感器编码',
-    sensor_name       varchar(500)   null comment '传感器名称',
-    description       text           null comment '传感器描述',
-    near_distance     decimal(19, 4) null comment '盲区距离',
-    far_distance      decimal(19, 4) null comment '探测距离',
-    fov_h_left        decimal(19, 4) null comment '水平现场角左',
-    fov_h_right       decimal(19, 4) null comment '水平现场角右',
-    fov_v_top         decimal(19, 4) null comment '垂直现场角顶',
-    fov_v_bottom      decimal(19, 4) null comment '垂直现场角底',
-    range_display     varchar(1)     null comment '显示探测锥',
-    max_objects       int            null comment '最大目标物个数',
-    coordinate_system varchar(32)    null comment '坐标系',
-    target_filter     varchar(640)   null comment '目标物筛选',
-    share             varchar(32)    null comment '是否分享',
-    create_time       datetime       null,
-    modify_time       datetime       null,
-    create_user_id    varchar(32)    null,
-    modify_user_id    varchar(32)    null,
-    is_deleted        varchar(1)     null
-)
-    comment '完美传感器表';
-

amd64/dispatch_server/package/entity/sensor_filter.yaml

@@ -1,15 +0,0 @@
-sensor-filter:
-  - code: 0
-    name: 车辆
-  - code: 1
-    name: 行人
-  - code: 2
-    name: 交通信号灯
-  - code: 3
-    name: 交通标志
-  - code: 4
-    name: 障碍物
-  - code: 5
-    name: 车道信息
-  - code: 6
-    name: 路标

amd64/dispatch_server/package/entity/task.go

@@ -1,26 +0,0 @@
-package entity
-
-type Task struct {
-	Info     TaskInfo     `json:"info"`
-	Scenario TaskScenario `json:"scenario"`
-	Vehicle  TaskVehicle  `json:"vehicle"`
-}
-
-type TaskInfo struct {
-	ProjectId   string `json:"project_id"`
-	TaskId      string `json:"task_id"`
-	TaskPath    string `json:"task_path"`
-	DefaultTime int    `json:"default_time"`
-}
-
-type TaskScenario struct {
-	ScenarioOsc  string `json:"scenario_osc"`
-	ScenarioOdr  string `json:"scenario_odr"`
-	ScenarioOsgb string `json:"scenario_osgb"`
-}
-
-type TaskVehicle struct {
-	Model    VehicleModel    `json:"model"`
-	Dynamics VehicleDynamics `json:"dynamics"`
-	Sensors  VehicleSensors  `json:"sensors"`
-}

amd64/dispatch_server/package/entity/task_cache.go

@@ -1,8 +0,0 @@
-package entity
-
-type TaskCache struct {
-	UserId             string `json:"userId"`
-	UserParallelism    int64  `json:"userParallelism"`
-	AlgorithmObjectKey string `json:"algorithmObjectKey"`
-	Task               Task   `json:"task"`
-}

amd64/dispatch_server/package/entity/vehicle.go

@@ -1,31 +0,0 @@
-package entity
-
-type VehicleModel struct {
-	ModelLabel string `json:"model_label"`
-}
-
-// VehicleDynamics 动力学配置
-type VehicleDynamics struct {
-	DynamicsMaxspeed              float64 `json:"dynamics_maxspeed"`              // 最大速度（千米/小时）
-	DynamicsEnginepower           float64 `json:"dynamics_enginepower"`           // 发动机功率（千瓦）
-	DynamicsMaxdecel              float64 `json:"dynamics_maxdecel"`              // 最大减速度（米/秒2）
-	DynamicsMaxsteering           float64 `json:"dynamics_maxsteering"`           // 最大减速度（米/秒2）
-	DynamicsMass                  float64 `json:"dynamics_mass"`                  // 质量（千克）
-	DynamicsFrontsurfaceeffective float64 `json:"dynamics_frontsurfaceeffective"` // 前表面有效面积（平方米）
-	DynamicsAirdragcoefficient    float64 `json:"dynamics_airdragcoefficient"`    // 空气阻力系数
-	DynamicsRollingresistance     float64 `json:"dynamics_rollingresistance"`     // 滚动阻力系数
-	DynamicsWheeldiameter         float64 `json:"dynamics_wheeldiameter"`         // 车轮直径（米）
-	DynamicsWheeldrive            string  `json:"dynamics_wheeldrive"`            // 驱动方式
-	DynamicsOverallefficiency     float64 `json:"dynamics_overallefficiency"`     // 总效率
-	DynamicsDistfront             float64 `json:"dynamics_distfront"`             // 车前距（米）
-	DynamicsDistrear              float64 `json:"dynamics_distrear"`              // 车后距（米）
-	DynamicsDistleft              float64 `json:"dynamics_distleft"`              // 车左距（米）
-	DynamicsDistright             float64 `json:"dynamics_distright"`             // 车右距（米）
-	DynamicsDistheight            float64 `json:"dynamics_distheight"`            // 车高（米）
-	DynamicsWheelbase             float64 `json:"dynamics_wheelbase"`             // 轴距（米）
-}
-
-type VehicleSensors struct {
-	Camera []SensorCamera `json:"camera"`
-	OGT    []SensorOgt    `json:"OGT"`
-}

amd64/dispatch_server/package/entity/水平视场角和垂直视场角计算公式.txt

@@ -1,4 +0,0 @@
-sensor_fovHLeft=水平视场角/2+水平视场角偏移量
-sensor_fovHRight=水平视场角/2-水平视场角偏移量
-SensorFovVTop=垂直视场角/2-垂直视场角偏移量
-SensorFovVBottom=垂直视场角/2+垂直视场角偏移量

amd64/dispatch_server/package/global/redis_key.go

@@ -1,16 +0,0 @@
-package global
-
-import "sync"
-
-var (
-	KeyGpuNodeList   = "gpu-node-list" // 用户运行队列
-	GpuNodeListMutex sync.Mutex
-
-	KeyTaskQueueRunningUserPrefix = "task-queue-running-user"    // 用户运行队列
-	KeyTaskQueueWaitingUser       = "task-queue-waiting-user"    // 用户等待队列，等待队列不需要根据userId区分开，资源先到先得
-	KeyTaskQueueRunningCluster    = "task-queue-running-cluster" // 集群运行队列
-	KeyTaskQueueWaitingCluster    = "task-queue-waiting-cluster" // 集群等待队列
-	KeyTaskToNode                 = "task-to-node"               // 集群等待队列
-
-	ParallelismMutex sync.Mutex
-)

amd64/dispatch_server/package/global/run_sync.go

@@ -1,5 +0,0 @@
-package global
-
-import "sync"
-
-var RunTaskMutex sync.Mutex

amd64/dispatch_server/package/handler/old_interface_adapter.go

@@ -1,108 +0,0 @@
-package handler
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/domain"
-	"cicv-data-closedloop/amd64/dispatch_server/package/global"
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"cicv-data-closedloop/amd64/dispatch_server/package/service"
-	"cicv-data-closedloop/amd64/dispatch_server/package/util"
-	"cicv-data-closedloop/common/entity"
-	"github.com/gin-gonic/gin"
-	"net/http"
-)
-
-// 返回 true
-func Confirm(c *gin.Context) {
-	c.String(http.StatusOK, "true")
-}
-
-func Tick(c *gin.Context) {
-	c.String(http.StatusOK, "true")
-}
-
-func State(c *gin.Context) {
-	taskId := c.Query("taskId")
-	state := c.Query("state")
-	podName := c.Query("podName") // todo 根据pod名称杀死pod并清理redis缓存
-	if taskId == "" || state == "" || podName == "" {
-		infra.GlobalLogger.Error("任务状态接口请求参数错误，需要GET请求和【taskId】【state】【podName】三个参数。")
-		c.JSON(http.StatusBadRequest, entity.HttpResult{Status: false, Code: "1003", Message: "请求参数格式错误。"})
-		return
-	}
-	if state != "Running" {
-		err := util.DeletePod(infra.ClientSet, podName, infra.ApplicationYaml.K8s.NamespaceName)
-		if err != nil {
-			infra.GlobalLogger.Errorf("删除pod【%v】失败", podName)
-		}
-		// 根据taskId删除缓存中的pod元素
-		runningTaskCacheJsons, err := infra.GlobalRedisClient.LRange(global.KeyTaskQueueRunningCluster, 0, -1).Result()
-		if err != nil {
-			infra.GlobalLogger.Errorf("获取运行中的任务队列【%v】失败", global.KeyTaskQueueRunningCluster)
-		}
-		for _, json := range runningTaskCacheJsons {
-			cache, err := service.JsonToTaskCache(json)
-			if err != nil {
-				infra.GlobalLogger.Error(err)
-			}
-			if cache.Task.Info.TaskId == taskId {
-				// 1 获取任务id关联的节点
-				nodeName, _ := infra.GlobalRedisClient.Get(global.KeyTaskToNode + ":" + taskId).Result()
-				// 2 归还并行度
-				global.ParallelismMutex.Lock()
-				gpuNodeJsons, _ := infra.GlobalRedisClient.LRange(global.KeyGpuNodeList, 0, -1).Result()
-				for i, gpuNodeJson := range gpuNodeJsons {
-					node, _ := domain.JsonToGpuNode(gpuNodeJson)
-					if node.Hostname == nodeName {
-						node.Parallelism++
-						nodeJson, _ := domain.GpuNodeToJson(node)
-						_, _ = infra.GlobalRedisClient.LSet(global.KeyGpuNodeList, int64(i), nodeJson).Result()
-					}
-				}
-				global.ParallelismMutex.Unlock()
-				// 3 删除任务队列中的元素
-				_, _ = infra.GlobalRedisClient.LRem(global.KeyTaskQueueRunningCluster, 0, json).Result()
-			}
-		}
-
-	}
-
-	// 发送http请求到索为系统
-	/*
-		仿真测试任务回调接口：
-			1、接口url
-				http://1.202.169.139:8081/project/task/callback
-
-			2、请求方式
-				POST
-
-			3、请求体
-				{
-					"taskId":1764907332423110657,
-					"state":"RUNNING"
-				}
-
-			4、响应结果
-
-				{
-					"data":true/false,
-					"success": true,
-					"message": "ok",
-					"code": 1,
-					"nowTime": "2024-05-10 15:57:42"
-				}
-	*/
-	_, err := util.PostJsonResponseJson(
-		//"http://1.202.169.139:8081/project/task/callback",
-		//"http://10.14.86.127:9081/project/task/callback",
-		infra.ApplicationYaml.K8s.CallbackUri,
-		map[string]string{
-			"taskId": taskId,
-			"state":  state,
-		},
-	)
-	if err != nil {
-		infra.GlobalLogger.Error(err)
-		c.String(http.StatusOK, "false")
-	}
-	c.String(http.StatusOK, "true")
-}

amd64/dispatch_server/package/handler/receive_task_state.go

@@ -1 +0,0 @@
-package handler

amd64/dispatch_server/package/handler/start_project.go

@@ -1,157 +0,0 @@
-package handler
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/domain"
-	entity2 "cicv-data-closedloop/amd64/dispatch_server/package/entity"
-	"cicv-data-closedloop/amd64/dispatch_server/package/global"
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"cicv-data-closedloop/common/entity"
-	"github.com/gin-gonic/gin"
-	"net/http"
-)
-
-/*
-{
-  "projectId": "项目ID",
-  "algorithmObjectKey": "算法在阿里云的存储路径",
-  "userId": "用户ID",
-  "parallelism": 10,
-  "tasks": [
-    {
-      "info": {
-        "project_id": "项目ID",
-        "task_id": "任务ID",
-        "task_path": "任务结果路径",
-        "default_time": "最大仿真时间"
-      },
-      "scenario": {
-        "scenario_osc": "xosc或xml路径",
-        "scenario_odr": "xodr路径",
-        "scenario_osgb": "osgb路径"
-      },
-      "vehicle": {
-        "model": {
-          "model_label": "模型标签"
-        },
-        "dynamics": {
-            "dynamics_maxspeed": 0.0, //最大速度（千米/小时）
-            "dynamics_enginepower": 0.0, // 发动机功率（千瓦）
-            "dynamics_maxdecel": 0.0, // 最大减速度（米/秒2）
-            "dynamics_maxsteering": 0.0, // 最大减速度（米/秒2）
-            "dynamics_mass": 0.0, // 质量（千克）
-            "dynamics_frontsurfaceeffective": 0.0, // 前表面有效面积（平方米）
-            "dynamics_airdragcoefficient": 0.0, // 空气阻力系数
-            "dynamics_rollingresistance": 0.0, // 滚动阻力系数
-            "dynamics_wheeldiameter": 0.0, // 车轮直径（米）
-            "dynamics_wheeldrive": "wheel_drive_front", // 驱动方式
-            "dynamics_overallefficiency": 0.0, // 总效率
-            "dynamics_distfront": 0.0, // 车前距（米）
-            "dynamics_distrear": 0.0, // 车后距（米）
-            "dynamics_distleft": 0.0, // 车左距（米）
-            "dynamics_distright": 0.0, // 车右距（米）
-            "dynamics_distheight": 0.0, // 车高（米）
-            "dynamics_wheelbase": 0.0 // 轴距（米）
-        },
-        "sensors": {
-          "camera": [
-            {
-              "sensor_name": "",
-              "sensor_near": 0,
-              "sensor_far": 0,
-              "sensor_x": 0,
-              "sensor_y": 0,
-              "sensor_z": 0,
-              "sensor_h": 0,
-              "sensor_p": 0,
-              "sensor_r": 0,
-              "sensor_fovH": 0,	// 水平视场角
-              "sensor_fovV": 0,	// 垂直视场角
-              "sensor_resolution": 0, // 水平视场角偏移量
-              "sensor_frameRate": 0	// 垂直视场角偏移量
-            }
-          ],
-          "OGT": [
-            {
-                "sensor_name": "", // 传感器名称
-                "sensor_near": 0, // 盲区距离
-                "sensor_far": 0, // 探测距离
-                "sensor_x": 0, // 传感器横向偏移量（x轴）
-                "sensor_y": 0, // 传感器纵向偏移量（y轴）
-                "sensor_z": 0, // 传感器安装高度（z轴）
-                "sensor_h": 0, // 传感器横摆角
-                "sensor_p": 0, // 传感器俯仰角
-                "sensor_r": 0, // 传感器横滚角
-                "sensor_fovHLeft": 0, // 水平现场角左
-                "sensor_fovHRight": 0, // 水平现场角右
-                "sensor_fovVTop": 0, // 垂直现场角顶
-                "sensor_fovVBottom": 0, // 垂直现场角底
-                "sensor_filter": "0,5,6", // 目标物筛选序列（0,1,2）
-                "sensor_display": false, // 显示目标物
-                "sensor_maxObjects": 0, // 最大目标物个数
-                "sensor_port": 0 // 端口
-            },
-            {
-              "sensor_name": "",
-              "sensor_near": 0,
-              "sensor_far": 0,
-              "sensor_x": 0,
-              "sensor_y": 0,
-              "sensor_z": 0,
-              "sensor_h": 0,
-              "sensor_p": 0,
-              "sensor_r": 0,
-              "sensor_fovHLeft": 0,
-              "sensor_fovHRight": 0,
-              "sensor_fovVTop": 0,
-              "sensor_fovVBottom": 0,
-              "sensor_filter": "1,2,3,4",
-              "sensor_display": false,
-              "sensor_maxObjects": 0,
-              "sensor_port": 0
-            }
-          ]
-        }
-      }
-    }
-  ]
-}
-*/
-
-func StartProject(c *gin.Context) {
-	projectStartParam := new(entity2.Project)
-	if err := c.ShouldBindJSON(&projectStartParam); err != nil {
-		infra.GlobalLogger.Error("项目启动接收请求参数报错：", err)
-		c.JSON(http.StatusBadRequest, entity.HttpResult{Status: false, Code: "1003", Message: "请求参数格式错误。"})
-		return
-	}
-	// ------------ 维护一个运行任务队列，绑定用户id和节点名称，供下面两个判断同时使用（使用redis的队列）
-	userId := projectStartParam.UserId               // 用户ID
-	taskReceived := projectStartParam.Tasks          // 接收到的所有任务
-	userParallelism := projectStartParam.Parallelism // 用户的并行度上限
-	algorithmObjectKey := projectStartParam.AlgorithmObjectKey
-
-	// 1 判断用户并行度
-	for _, task := range taskReceived {
-		global.RunTaskMutex.Lock()
-		// 1 判断用户并行度是否有剩余，有剩余则加入集群等待队列，并从用户等待队列中拿出，没有剩余则不需要改动
-		if domain.CanRunUser(userId, userParallelism) { // 可以运行
-			err := domain.AddWaitingCluster(userId, userParallelism, algorithmObjectKey, task)
-			if err != nil {
-				infra.GlobalLogger.Errorf("将任务 %v 添加到【集群等待队列】失败，错误信息为：%v", task, err)
-				continue
-			}
-			infra.GlobalLogger.Infof("将任务 %v 添加到【集群等待队列】成功。", task.Info.TaskId)
-		} else { // 不能运行
-			err := domain.AddWaitingUser(userId, userParallelism, algorithmObjectKey, task)
-			if err != nil {
-				infra.GlobalLogger.Errorf("将任务 %v 添加到【用户等待队列】失败，错误信息为：%v", task, err)
-				continue
-			}
-			infra.GlobalLogger.Infof("将任务 %v 添加到【用户等待队列】成功。", task.Info.TaskId)
-		}
-		global.RunTaskMutex.Unlock()
-	}
-
-	// 4 返回
-	c.JSON(http.StatusOK, entity.HttpResult{Status: true, Code: "2000", Message: "项目启动请求已被成功接收，等待调度处理。"})
-}

amd64/dispatch_server/package/handler/stop_project.go

@@ -1 +0,0 @@
-package handler

amd64/dispatch_server/package/infra/application.yaml

@@ -1,51 +0,0 @@
-application-name: dispatch_server
-web:
-  ip-private: 10.14.85.241
-  port: 12342
-  route-prefix: /dispatch_server
-  token: U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM
-  white-list:
-    - "/simulation/resource/scheduler/confirm"
-    - "/simulation/resource/scheduler/tick"
-    - "/simulation/resource/scheduler/state"
-log:
-  dir: ./log/
-  prefix: dispatch_server
-
-redis:
-#  addr: 36.110.106.156:6379
-  addr: 10.14.85.240:6379
-  password: 1qaz2wsx!
-  db: 3
-
-kafka:
-  partition: 0
-  broker: 10.14.85.239:9092
-
-oss:
-  type: Aliyun # Minio 或 Aliyun
-#  is-user-cname: true
-#  endpoint: open-bucket.oss.icvdc.com
-  is-user-cname: false
-  endpoint: oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com
-  access-key-id: n8glvFGS25MrLY7j
-  access-key-secret: xZ2Fozoarpfw0z28FUhtg8cu0yDc5d
-  bucket-name: open-bucket
-
-gpu-node-list:
-  - hostname: gpu003
-    ip: 10.14.85.238
-    parallelism: 6
-
-k8s:
-  pod-yaml-dir: /mnt/disk001/cicv-data-closedloop/pod-yaml/
-  vtd-pod-template-yaml: /mnt/disk001/cicv-data-closedloop/amd64/dispatch_server/vtd-pod-template_20240522.yaml
-  algorithm-tar-temp-dir: /mnt/disk001/cicv-data-closedloop/temp/algorithm/
-  registry-uri: 10.14.85.237:5000
-  namespace-name: cicvdcl
-  vtd-image: 10.14.85.237:5000/vtd.run.perception.release:latest
-#  vtd-command: /Controller/config/docker_cloud_algContest.ini
-  vtd-command: /Controller/config/docker_cloud.ini
-  callback-uri: http://10.14.86.127:9081/project/task/callback
-  platform-type: 2
-

amd64/dispatch_server/package/infra/i_application.go

@@ -1,83 +0,0 @@
-package infra
-
-import (
-	"cicv-data-closedloop/common/util"
-	_ "embed"
-	"fmt"
-	"gopkg.in/yaml.v2"
-)
-
-type ApplicationYamlStruct struct {
-	ApplicationName string      `yaml:"application-name"`
-	Web             WebStruct   `yaml:"web"`
-	Log             LogStruct   `yaml:"log"`
-	Redis           RedisStruct `yaml:"redis"`
-	Kafka           KafkaStruct `yaml:"kafka"`
-	Oss             OssStruct   `yaml:"oss"`
-	GpuNodeList     []GpuNode   `yaml:"gpu-node-list"`
-	K8s             K8sStruct   `yaml:"k8s"`
-}
-
-type WebStruct struct {
-	IpPrivate   string   `yaml:"ip-private"`
-	Port        string   `yaml:"port"`
-	RoutePrefix string   `yaml:"route-prefix"`
-	Token       string   `yaml:"token"`
-	WhiteList   []string `yaml:"white-list"`
-}
-
-type LogStruct struct {
-	Dir    string `yaml:"dir"`
-	Prefix string `yaml:"prefix"`
-}
-
-type RedisStruct struct {
-	Addr     string `yaml:"addr"`
-	Password string `yaml:"password"`
-	Db       int    `yaml:"db"`
-}
-
-type KafkaStruct struct {
-	Partition int32  `yaml:"partition"`
-	Broker    string `yaml:"broker"`
-}
-
-type OssStruct struct {
-	Type            string `yaml:"type"`
-	IsUseCname      bool   `yaml:"is-use-cname"`
-	Endpoint        string `yaml:"endpoint"`
-	AccessKeyId     string `yaml:"access-key-id"`
-	AccessKeySecret string `yaml:"access-key-secret"`
-	BucketName      string `yaml:"bucket-name"`
-}
-
-type GpuNode struct {
-	Hostname    string `yaml:"hostname"`
-	Ip          string `yaml:"ip"`
-	Parallelism int64  `yaml:"parallelism"`
-}
-
-type K8sStruct struct {
-	PodYamlDir          string `yaml:"pod-yaml-dir"`
-	VtdPodTemplateYaml  string `yaml:"vtd-pod-template-yaml"`
-	AlgorithmTarTempDir string `yaml:"algorithm-tar-temp-dir"`
-	RegistryUri         string `yaml:"registry-uri"`
-	NamespaceName       string `yaml:"namespace-name"`
-	VtdImage            string `yaml:"vtd-image"`
-	VtdCommand          string `yaml:"vtd-command"`
-	CallbackUri         string `yaml:"callback-uri"`
-	PlatformType        string `yaml:"platform-type"`
-}
-
-var (
-	//go:embed application.yaml
-	applicationYamlBytes []byte
-	ApplicationYaml      ApplicationYamlStruct
-)
-
-func InitApplication() {
-	_ = yaml.Unmarshal(applicationYamlBytes, &ApplicationYaml)
-	fmt.Println("加载配置文件内容为：", ApplicationYaml)
-	// 创建镜像下载目录
-	util.CreateDir(ApplicationYaml.K8s.AlgorithmTarTempDir)
-}

amd64/dispatch_server/package/infra/i_kafka.go

@@ -1,25 +0,0 @@
-package infra
-
-import (
-	"github.com/confluentinc/confluent-kafka-go/kafka"
-)
-
-// 定义一个全局的Kafka生产者变量
-var GlobalKafkaProducer *kafka.Producer
-
-// 初始化Kafka生产者的函数
-func InitKafkaProducer(broker string) {
-	// 创建配置
-	conf := &kafka.ConfigMap{
-		"bootstrap.servers": broker,
-		"client.id":         "kafka-client-cicv-data-closedloop",
-	}
-
-	// 创建生产者
-	var err error
-	GlobalKafkaProducer, err = kafka.NewProducer(conf)
-	if err != nil {
-		GlobalLogger.Errorf("初始化kafka生产者失败，配置信息为：%v，错误信息为：%v", broker, err)
-	}
-	GlobalLogger.Errorf("初始化kafka生产者成功：%v", broker)
-}

amd64/dispatch_server/package/infra/i_kubernetes.go

@@ -1,26 +0,0 @@
-package infra
-
-import (
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/clientcmd"
-	"k8s.io/client-go/util/homedir"
-	"os"
-	"path/filepath"
-)
-
-var ClientSet *kubernetes.Clientset
-
-func InitKubernetesClient() {
-	kubeConfigFile := filepath.Join(homedir.HomeDir(), ".kube", "config")
-	config, err := clientcmd.BuildConfigFromFlags("", kubeConfigFile)
-	if err != nil {
-		GlobalLogger.Error("读取k8s配置文件失败", err.Error())
-		os.Exit(0)
-	}
-
-	ClientSet, err = kubernetes.NewForConfig(config)
-	if err != nil {
-		GlobalLogger.Error("读取k8s配置文件失败", err.Error())
-		os.Exit(0)
-	}
-}

amd64/dispatch_server/package/infra/i_log.go

@@ -1,39 +0,0 @@
-package infra
-
-import (
-	"cicv-data-closedloop/common/util"
-	"fmt"
-	"github.com/sirupsen/logrus"
-	"os"
-	"path/filepath"
-	"runtime"
-	"time"
-)
-
-var GlobalLogger *logrus.Logger
-
-// InitLog 初始化日志配置
-func InitLog(logDir string, prefix string) {
-	time.Sleep(time.Duration(1) * time.Second)
-	// 创建、追加、读写，777，所有权限
-	logPath := logDir + prefix + "-" + time.Now().Format("2006-01-02-15-04-05") + ".log"
-	err := util.CreateParentDir(logPath)
-	if err != nil {
-		os.Exit(-1)
-	}
-	f, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm)
-	if err != nil {
-		os.Exit(-1)
-	}
-	GlobalLogger = logrus.New()
-	GlobalLogger.SetOutput(f)
-	GlobalLogger.SetReportCaller(true) // 开启行号显示
-	GlobalLogger.SetFormatter(&logrus.JSONFormatter{
-		CallerPrettyfier: func(frame *runtime.Frame) (string, string) {
-			fileName := filepath.Base(frame.File)
-			return "", fmt.Sprintf("%s:%d", fileName, frame.Line)
-		},
-	})
-	GlobalLogger.Info("初始化GlobalLogger - 成功")
-
-}

amd64/dispatch_server/package/infra/i_oss.go

@@ -1,29 +0,0 @@
-package infra
-
-import (
-	"github.com/aliyun/aliyun-oss-go-sdk/oss"
-	"os"
-)
-
-var (
-	GlobalOssClient *oss.Client
-	GlobalOssBucket *oss.Bucket
-)
-
-func InitOss(isUseCname bool, endpoint string, accessKeyId string, accessKeySecret string, bucketName string) {
-	var err error
-	if isUseCname {
-		GlobalOssClient, err = oss.New(endpoint, accessKeyId, accessKeySecret, oss.UseCname(true)) // 公网
-	} else {
-		GlobalOssClient, err = oss.New(endpoint, accessKeyId, accessKeySecret, oss.UseCname(false)) // 内网
-	}
-	if err != nil {
-		GlobalLogger.Error("无法创建阿里云client:", err)
-		os.Exit(-1)
-	}
-	GlobalOssBucket, err = GlobalOssClient.Bucket(bucketName)
-	if err != nil {
-		GlobalLogger.Error("无法创建阿里云bucket:", err)
-		os.Exit(-1)
-	}
-}

amd64/dispatch_server/package/infra/i_redis.go

@@ -1,36 +0,0 @@
-package infra
-
-import (
-	"github.com/go-redis/redis"
-	"os"
-	"sync"
-)
-
-var (
-	once              sync.Once
-	GlobalRedisClient *redis.Client
-)
-
-// 获取全局的Redis客户端
-func InitRedisClient(addr string, password string, db int) {
-	once.Do(func() {
-		// 创建Redis连接
-		GlobalRedisClient = redis.NewClient(&redis.Options{
-			Addr:     addr,     // Redis地址
-			Password: password, // Redis密码，如果没有密码则留空
-			DB:       db,       // Redis数据库索引
-		})
-	})
-	// 检查连接是否成功
-	_, err := GlobalRedisClient.Ping().Result()
-	if err != nil {
-		GlobalLogger.Error("初始化 Redis 客户端报错:", err)
-	}
-	// 清空当前选择的数据库
-	err = GlobalRedisClient.FlushDB().Err()
-	if err != nil {
-		GlobalLogger.Errorf("清空 redis 【%v】号库的数据时报错，初始化 Redis 客户端失败，错误信息为：%v", db, err)
-		os.Exit(0)
-	}
-	GlobalLogger.Infof("已清空 redis 【%v】号库的数据，并初始化 Redis 全局客户端成功。", db)
-}

amd64/dispatch_server/package/kafka/consumer.go

@@ -1 +0,0 @@
-package kafka

amd64/dispatch_server/package/kafka/producer.go

@@ -1 +0,0 @@
-package kafka

amd64/dispatch_server/package/service/push_algorithm_image.go

@@ -1,6 +0,0 @@
-package service
-
-// PushAlgorithmImage 将镜像文件推送到镜像仓库，共集群内所有节点使用
-func PushAlgorithmImage() {
-
-}

amd64/dispatch_server/package/service/run_task.go

@@ -1,314 +0,0 @@
-package service
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/domain"
-	"cicv-data-closedloop/amd64/dispatch_server/package/entity"
-	"cicv-data-closedloop/amd64/dispatch_server/package/global"
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"cicv-data-closedloop/amd64/dispatch_server/package/util"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"github.com/confluentinc/confluent-kafka-go/kafka"
-	"path/filepath"
-	"strconv"
-	"strings"
-	"time"
-)
-
-/*
- 负责处理用户等待队列中的任务
- 负责运行集群等待队列中的任务
-*/
-
-// 判断用户等待队列中的任务是否可以加入到集群等待队列
-func RunWaitingUser() {
-	infra.GlobalLogger.Infof("启动【用户等待队列】监控进程。")
-	for {
-		time.Sleep(2 * time.Second)
-		global.RunTaskMutex.Lock()
-		// 获取Redis列表中的值
-		taskCacheJsons, err := infra.GlobalRedisClient.LRange(global.KeyTaskQueueWaitingUser, 0, -1).Result()
-		if err != nil {
-			infra.GlobalLogger.Errorf("遍历用户等待队列 %v 失败，错误信息为: %v", global.KeyTaskQueueWaitingUser, err)
-			continue
-		}
-
-		for _, taskCacheJson := range taskCacheJsons {
-			taskCache, err := JsonToTaskCache(taskCacheJson)
-			if err != nil {
-				infra.GlobalLogger.Error(err)
-				continue
-			}
-			userId := taskCache.UserId
-			userParallelism := taskCache.UserParallelism
-			algorithmObjectKey := taskCache.AlgorithmObjectKey
-			task := taskCache.Task
-			// 1 判断用户并行度是否有剩余，有剩余则加入集群等待队列，并从用户等待队列中拿出，没有剩余则不需要改动
-			if domain.CanRunUser(userId, userParallelism) { // 可以运行
-				err = domain.AddWaitingCluster(userId, userParallelism, algorithmObjectKey, task)
-				if err != nil {
-					infra.GlobalLogger.Error(err)
-					continue
-				}
-				err = domain.DeleteWaitingUser(task.Info.TaskId)
-				if err != nil {
-					infra.GlobalLogger.Error(err)
-					continue
-				}
-			}
-		}
-		global.RunTaskMutex.Unlock()
-	}
-}
-
-// 集群等待队列中的任务判断是否可以加入集群运行队列
-func RunWaitingCluster() {
-	infra.GlobalLogger.Infof("启动【集群等待队列】监控进程。")
-	for {
-		var algorithmTarName string
-		var algorithmTarPath string
-		var algorithmImageName string
-
-		time.Sleep(2 * time.Second)
-		global.GpuNodeListMutex.Lock()
-		// 1 判断用户并行度是否有剩余，有剩余则从集群等待队列取出第一个加入集群运行队列，并运行pod，没有剩余则不需要改动
-		can, gpuNode, err := domain.CanRunCluster()
-		if err != nil {
-			infra.GlobalLogger.Error(err)
-			global.GpuNodeListMutex.Unlock()
-			continue
-		}
-		var firstTaskCache entity.TaskCache
-		if can {
-			//infra.GlobalLogger.Infof("节点 %v 有剩余并行度。", gpuNode)
-			// 判断是否有待运行的任务
-			waitingClusterNumber, _ := infra.GlobalRedisClient.LLen(global.KeyTaskQueueWaitingCluster).Result()
-			if waitingClusterNumber == 0 {
-				//infra.GlobalLogger.Info("集群没有等待运行的任务。")
-				global.GpuNodeListMutex.Unlock()
-				continue
-			} else {
-				infra.GlobalLogger.Infof("集群存在 %v 个等待运行的任务。", waitingClusterNumber)
-			}
-
-			// 取出但不移除
-			{
-				firstTaskCacheJson, err := infra.GlobalRedisClient.LIndex(global.KeyTaskQueueWaitingCluster, 0).Result()
-				if err != nil {
-					infra.GlobalLogger.Error("取出集群等待队列中的头元素报错，错误信息为：", err)
-					global.GpuNodeListMutex.Unlock()
-					continue
-				}
-				firstTaskCache, err = JsonToTaskCache(firstTaskCacheJson)
-				if err != nil {
-					infra.GlobalLogger.Error(err)
-					global.GpuNodeListMutex.Unlock()
-					continue
-				}
-			}
-			// --------------- 下载算法 ---------------
-			{
-				infra.GlobalLogger.Infof("开始下载算法 %v。", firstTaskCache.AlgorithmObjectKey)
-				algorithmTarName = filepath.Base(firstTaskCache.AlgorithmObjectKey)
-				algorithmTarPath = infra.ApplicationYaml.K8s.AlgorithmTarTempDir + util.NewShortUUID() + "/" + algorithmTarName
-				_ = util.CreateParentDir(algorithmTarPath)
-				algorithmImageName = infra.ApplicationYaml.K8s.RegistryUri + "/cicvdcl_" + util.MD5HashShort(algorithmTarName)
-				_ = infra.GlobalOssBucket.GetObjectToFile(firstTaskCache.AlgorithmObjectKey, algorithmTarPath)
-				if err != nil {
-					infra.GlobalLogger.Error("下载oss上的算法镜像 "+firstTaskCache.AlgorithmObjectKey+" 失败，错误信息为：", err)
-					time.Sleep(time.Duration(2) * time.Second)
-					global.GpuNodeListMutex.Unlock()
-					continue
-				}
-				infra.GlobalLogger.Infof("下载算法 %v 成功。", firstTaskCache.AlgorithmObjectKey)
-			}
-		} else {
-			infra.GlobalLogger.Infof("集群没有剩余并行度。")
-			global.GpuNodeListMutex.Unlock()
-			continue
-		}
-		global.GpuNodeListMutex.Unlock()
-		// 获取项目ID
-		projectId := firstTaskCache.Task.Info.ProjectId
-		offsetKey := "offset:" + projectId
-		offset := 0
-
-		// 根据项目ID获取偏移量
-		val, err := infra.GlobalRedisClient.Get(offsetKey).Result()
-		if err != nil {
-			infra.GlobalLogger.Infof("偏移量键 %v 不存在，初始化设置为 0。", offsetKey)
-			err = infra.GlobalRedisClient.Set(offsetKey, 0, 0).Err()
-			if err != nil {
-				infra.GlobalLogger.Infof("偏移量键值对 %v 初始化失败，错误信息为： %v", offsetKey, err)
-				continue
-			}
-		} else {
-			offset, err = strconv.Atoi(val)
-			if err != nil {
-				infra.GlobalLogger.Infof("字符串 %v 转整数失败，错误信息为： %v", val, err)
-				continue
-			}
-			infra.GlobalLogger.Infof("当前任务使用偏移量【%v】", offset)
-		}
-		// 取出偏移量后将缓存中的加一，给下个任务使用。
-		_, err = infra.GlobalRedisClient.Incr(offsetKey).Result()
-		if err != nil {
-			infra.GlobalLogger.Infof("偏移量 %v 加一失败，错误信息为： %v", offsetKey, err)
-			continue
-		}
-		infra.GlobalLogger.Infof("偏移量【%v】加一给下个任务使用。", offsetKey)
-
-		// --------------- 发送 kafka 消息（获取偏移量和分区） ---------------
-		// 获取任务消息转json
-		// todo 将摄像头写死一个参数
-		firstTaskCache.Task.Vehicle.Sensors.Camera = []entity.SensorCamera{
-			{
-				Sensor: entity.Sensor{
-					SensorName: "custom_camera",
-					SensorNear: 1.0,
-					SensorFar:  1500.0,
-					SensorX:    -15000.0,
-					SensorY:    0.0,
-					SensorZ:    5000.0,
-					SensorH:    0.0,
-					SensorP:    10.0,
-					SensorR:    0.0,
-				},
-				SensorForH:       45,
-				SensorForV:       27,
-				SensorResolution: "480*270",
-				SensorFrameRate:  25,
-			},
-		}
-		taskJson, err := TaskToJson(firstTaskCache.Task)
-		if err != nil {
-			infra.GlobalLogger.Error(err)
-			continue
-		}
-		topic := projectId
-		// 创建一个Message，并指定分区为0
-		msg := &kafka.Message{
-			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: infra.ApplicationYaml.Kafka.Partition, Offset: kafka.Offset(offset)},
-			Value:          []byte(taskJson),
-		}
-		// 发送消息，并处理结果
-		err = infra.GlobalKafkaProducer.Produce(msg, nil)
-		if err != nil {
-			infra.GlobalLogger.Infof("发送任务消息 %v 失败，错误信息为： %v", msg, err)
-			continue
-		}
-		infra.GlobalLogger.Infof("发送任务消息成功，话题为【%v】，偏移量为【%v】。", topic, offset)
-
-		// 导入算法
-		_, s, err := util.Execute("docker", "import", algorithmTarPath, algorithmImageName)
-		_, s, err = util.Execute("docker", "push", algorithmImageName)
-		if err != nil {
-			infra.GlobalLogger.Errorf("导入算法镜像 %v 为 %v 失败，执行结果为：%v，错误信息为：%v", algorithmTarPath, algorithmImageName, s, err)
-			time.Sleep(time.Duration(2) * time.Second)
-			continue
-		}
-		infra.GlobalLogger.Infof("导入算法镜像 %v 为 %v 成功，执行结果为：%v", algorithmTarPath, algorithmImageName, s)
-		err = util.RemoveFile(algorithmTarPath)
-		if err != nil {
-			infra.GlobalLogger.Errorf("删除算法镜像文件 %v 失败，错误信息为：%v", algorithmTarPath, err)
-		}
-		// --------------- 启动 k8s pod ---------------
-		podName := "project-" + projectId + "-" + util.NewShortUUID()
-		namespaceName := infra.ApplicationYaml.K8s.NamespaceName
-		nodeName := gpuNode.Hostname
-		restParallelism := gpuNode.Parallelism
-		vtdContainer := "vtd-" + projectId
-		algorithmContainer := "algorithm-" + projectId
-		vtdImage := infra.ApplicationYaml.K8s.VtdImage
-		// 2 生成模板文件名称
-		podYaml := nodeName + "#" + podName + ".yaml"
-		// 3 模板yaml存储路径
-		yamlPath := infra.ApplicationYaml.K8s.PodYamlDir + podYaml
-		// 4 模板yaml备份路径
-		yamlPathBak := infra.ApplicationYaml.K8s.PodYamlDir + "bak/" + podYaml
-		fmt.Println(yamlPath, yamlPathBak)
-		// 5
-		podString, err := util.ReadFile(infra.ApplicationYaml.K8s.VtdPodTemplateYaml)
-		if err != nil {
-			infra.GlobalLogger.Error(err)
-			continue
-		}
-		podString = strings.Replace(podString, "pod-name", podName, -1)
-		podString = strings.Replace(podString, "namespace-name", namespaceName, -1)
-		podString = strings.Replace(podString, "node-name", nodeName, -1)
-		podString = strings.Replace(podString, "algorithm-image", algorithmImageName, -1)
-		podString = strings.Replace(podString, "vtd-container", vtdContainer, -1)
-		podString = strings.Replace(podString, "vtd-image", vtdImage, -1)
-		podString = strings.Replace(podString, "vtd-command", infra.ApplicationYaml.K8s.VtdCommand, -1)
-		podString = strings.Replace(podString, "platform-ip", infra.ApplicationYaml.Web.IpPrivate+":"+infra.ApplicationYaml.Web.Port, -1)
-		podString = strings.Replace(podString, "simulation-cloud-ip", infra.ApplicationYaml.Web.IpPrivate+":"+infra.ApplicationYaml.Web.Port, -1)
-		podString = strings.Replace(podString, "platform-type", "\""+infra.ApplicationYaml.K8s.PlatformType+"\"", -1)
-		podString = strings.Replace(podString, "oss-type", infra.ApplicationYaml.Oss.Type, -1)
-		podString = strings.Replace(podString, "oss-ip", infra.ApplicationYaml.Oss.Endpoint, -1) // 不带http://前缀
-		podString = strings.Replace(podString, "oss-access-key", infra.ApplicationYaml.Oss.AccessKeyId, -1)
-		podString = strings.Replace(podString, "oss-secret-key", infra.ApplicationYaml.Oss.AccessKeySecret, -1)
-		podString = strings.Replace(podString, "oss-bucket", infra.ApplicationYaml.Oss.BucketName, -1)
-		podString = strings.Replace(podString, "kafka-ip", infra.ApplicationYaml.Kafka.Broker, -1)
-		podString = strings.Replace(podString, "kafka-topic", projectId, -1)
-		podString = strings.Replace(podString, "kafka-partition", "\""+util.ToString(infra.ApplicationYaml.Kafka.Partition)+"\"", -1)
-		podString = strings.Replace(podString, "kafka-offset", "\""+util.ToString(offset)+"\"", -1)
-		podString = strings.Replace(podString, "cpu-order", "\""+util.ToString(restParallelism-1)+"\"", -1) // cpu编号是剩余并行度-1
-		podString = strings.Replace(podString, "algorithm-container", algorithmContainer, -1)
-
-		// --------------- 保存成文件
-		err = util.WriteFile(podString, yamlPath)
-		err = util.WriteFile(podString, yamlPathBak)
-		if err != nil {
-			infra.GlobalLogger.Error("保存yaml字符串失败，错误信息为", err)
-			continue
-		}
-		infra.GlobalLogger.Infof("保存yaml文件到执行路径【%v】和备份路径【%v】", yamlPath, yamlPathBak)
-		// --------------- 启动 pod
-		_, s2, err := util.Execute("kubectl", "apply", "-f", yamlPath)
-		if err != nil {
-			infra.GlobalLogger.Errorf("启动pod失败，执行结果为 %v，错误信息为 %v", s2, err)
-			continue
-		}
-		infra.GlobalLogger.Errorf("启动pod成功，执行结果为 %v。", s2)
-		// 收尾
-		{
-			// --------------- 添加到运行队列
-			err = domain.AddRunningCluster(firstTaskCache, gpuNode.Hostname)
-			if err != nil {
-				infra.GlobalLogger.Error(err)
-				global.GpuNodeListMutex.Unlock()
-				continue
-			}
-			// --------------- 从等待队列中移除
-			_, err = infra.GlobalRedisClient.LPop(global.KeyTaskQueueWaitingCluster).Result()
-			if err != nil {
-				infra.GlobalLogger.Error("取出集群等待队列中的头元素报错，错误信息为：", err)
-				continue
-			}
-			// --------------- 删除镜像文件
-			_ = util.RemoveFile(algorithmTarPath)
-		}
-
-	}
-}
-
-func JsonToTaskCache(jsonData string) (entity.TaskCache, error) {
-	// 创建一个 Person 类型的变量
-	var taskCache entity.TaskCache
-
-	// 使用 json.Unmarshal 解析 JSON 字符串到结构体
-	err := json.Unmarshal([]byte(jsonData), &taskCache)
-	if err != nil {
-		return entity.TaskCache{}, errors.New("对象json " + jsonData + " 转对象失败错误信息为: " + fmt.Sprintf("%v", err))
-	}
-	return taskCache, nil
-}
-
-func TaskToJson(task entity.Task) (string, error) {
-	jsonData, err := json.Marshal(task)
-	if err != nil {
-		return "", errors.New("转json失败，错误信息为：" + err.Error())
-	}
-	return string(jsonData), nil
-}

amd64/dispatch_server/package/util/u_crypto.go

@@ -1,37 +0,0 @@
-package util
-
-import (
-	"crypto/md5"
-	"encoding/hex"
-	"io"
-)
-
-// MD5Hash 函数接收一个字符串作为输入，返回该字符串的MD5哈希值（十六进制格式）
-func MD5Hash(text string) string {
-	// 创建一个新的hash.Hash接口来写入数据
-	hasher := md5.New()
-
-	// 写入需要哈希的数据
-	io.WriteString(hasher, text)
-
-	// 计算哈希值的Sum（字节切片）
-	sum := hasher.Sum(nil)
-
-	// 将字节切片转换为十六进制字符串
-	return hex.EncodeToString(sum)
-}
-
-// MD5Hash 函数接收一个字符串作为输入，返回该字符串的MD5哈希值（十六进制格式）
-func MD5HashShort(text string) string {
-	// 创建一个新的hash.Hash接口来写入数据
-	hasher := md5.New()
-
-	// 写入需要哈希的数据
-	io.WriteString(hasher, text)
-
-	// 计算哈希值的Sum（字节切片）
-	sum := hasher.Sum(nil)
-
-	// 将字节切片转换为十六进制字符串
-	return hex.EncodeToString(sum)[0:8]
-}

amd64/dispatch_server/package/util/u_exec.go

@@ -1,16 +0,0 @@
-package util
-
-import (
-	"cicv-data-closedloop/common/config/c_log"
-	"os/exec"
-)
-
-func Execute(name string, arg ...string) (*exec.Cmd, string, error) {
-	cmd := exec.Command(name, arg...)
-	combinedOutput, err := cmd.CombinedOutput()
-	if err != nil {
-		c_log.GlobalLogger.Info("命令", name, " ", arg, "执行出错:", err)
-		return nil, string(combinedOutput), err
-	}
-	return cmd, string(combinedOutput), nil
-}

amd64/dispatch_server/package/util/u_file.go

@@ -1,72 +0,0 @@
-package util
-
-import (
-	"io"
-	"os"
-	"path/filepath"
-)
-
-func ReadFile(filePath string) (string, error) {
-	// 1 打开文件
-	file, err := os.Open(filePath)
-	if err != nil {
-		return "", err
-	}
-	defer file.Close()
-
-	// 2 读取文件内容
-	content, err := io.ReadAll(file)
-	if err != nil {
-		return "", err
-	}
-	return string(content), err
-}
-
-func RemoveFile(path string) error {
-	// 检查文件是否存在
-	if _, err := os.Stat(path); err == nil {
-		// 文件存在，执行删除操作
-		err = os.Remove(path)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func WriteFile(sourceContent string, targetFilePath string) error {
-	if err := CreateFile(targetFilePath); err != nil {
-		return err
-	}
-	if err := os.WriteFile(targetFilePath, []byte(sourceContent), 0644); err != nil {
-		return err
-	}
-	return nil
-}
-
-// CreateFile 存在则覆盖，不存在则创建文件
-func CreateFile(filePath string) error {
-	if err := CreateParentDir(filePath); err != nil {
-		return err
-	}
-	// 创建文件，如果文件已存在则覆盖
-	file, err := os.Create(filePath)
-	if err != nil {
-		return err
-	}
-	defer func(file *os.File) {
-		err := file.Close()
-		if err != nil {
-
-		}
-	}(file)
-	return nil
-}
-
-// CreateParentDir 存在不创建，不存在则创建父目录
-func CreateParentDir(filePath string) error {
-	if err := os.MkdirAll(filepath.Dir(filePath), os.ModePerm); err != nil {
-		return err
-	}
-	return nil
-}

amd64/dispatch_server/package/util/u_http.go

@@ -1,28 +0,0 @@
-package util
-
-import (
-	"bytes"
-	"encoding/json"
-	"errors"
-	"io"
-	"net/http"
-)
-
-func PostJsonResponseJson(url string, params map[string]string) (string, error) {
-	jsonData, err := json.Marshal(params)
-	if err != nil {
-		return "", errors.New("参数转换成json失败，错误信息为：" + err.Error())
-	}
-	resp, err := http.Post(url, "application/json", bytes.NewBuffer(jsonData))
-	if err != nil {
-		return "", errors.New("发送请求失败，错误信息为：" + err.Error())
-	}
-	defer func(Body io.ReadCloser) {
-		_ = Body.Close()
-	}(resp.Body)
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return "", errors.New("读取响应结果失败，错误信息为：" + err.Error())
-	}
-	return string(body), nil
-}

amd64/dispatch_server/package/util/u_kubernetes.go

@@ -1,18 +0,0 @@
-package util
-
-import (
-	"cicv-data-closedloop/amd64/dispatch_server/package/infra"
-	"context"
-	"fmt"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-)
-
-func DeletePod(clientSet *kubernetes.Clientset, podName, namespace string) error {
-	err := clientSet.CoreV1().Pods(namespace).Delete(context.TODO(), podName, metav1.DeleteOptions{})
-	if err != nil {
-		return fmt.Errorf("删除pod报错: %s", err)
-	}
-	infra.GlobalLogger.Infof("Pod %s in namespace %s 删除成功\n", podName, namespace)
-	return nil
-}

amd64/dispatch_server/package/util/u_string.go

@@ -1,17 +0,0 @@
-package util
-
-import (
-	"fmt"
-	"github.com/google/uuid"
-)
-
-func ToString(value interface{}) string {
-	return fmt.Sprintf("%v", value)
-}
-
-func NewUUID() string {
-	return uuid.New().String()[:8]
-}
-func NewShortUUID() string {
-	return uuid.New().String()[:8]
-}

amd64/dispatch_server/test/test.go

@@ -1,29 +0,0 @@
-package main
-
-import (
-	"crypto/md5"
-	"encoding/hex"
-	"fmt"
-	"io"
-)
-
-// MD5Hash 函数接收一个字符串作为输入，返回该字符串的MD5哈希值（十六进制格式）
-func MD5Hash(text string) string {
-	// 创建一个新的hash.Hash接口来写入数据
-	hasher := md5.New()
-
-	// 写入需要哈希的数据
-	io.WriteString(hasher, text)
-
-	// 计算哈希值的Sum（字节切片）
-	sum := hasher.Sum(nil)
-
-	// 将字节切片转换为十六进制字符串
-	return hex.EncodeToString(sum)
-}
-
-func main() {
-	text := "算法比赛.tar"
-	hash := MD5Hash(text)
-	fmt.Printf("The MD5 hash of '%s' is: %s\n", text, hash[0:8])
-}

amd64/dispatch_server/vtd-pod-template_20240522.yaml

@@ -1,68 +0,0 @@
-apiVersion: v1
-kind: Pod
-metadata:
-  name: pod-name
-  namespace: namespace-name
-  labels:
-    user: CICV
-spec:
-  nodeName: node-name
-  dnsPolicy: None
-  dnsConfig:
-    nameservers:
-      - 10.16.11.1
-      - 10.16.11.2
-  hostAliases:
-    - ip: 10.14.85.239
-      hostnames:
-        - simulation004
-    - ip: 10.14.85.237
-      hostnames:
-        - gpu001
-  initContainers:
-    - name: init
-      image: algorithm-image
-      imagePullPolicy: Always
-      command: ['sh', '-c', 'echo algorithm image downloaded && sleep 2']
-  containers:
-    - name: vtd-container
-      image: vtd-image
-      imagePullPolicy: Always
-      command: [ "/Controller/VTDController", "vtd-command", "kafka-topic" ]
-      resources:
-        limits:
-          nvidia.com/gpu: 1
-      env:
-        - name: PodName
-          valueFrom:
-            fieldRef:
-              fieldPath: metadata.name
-        - name: LM_LICENSE_FILE
-          value: 27500@10.14.85.247
-        - name: SIMULATION_CLOUD_IP
-          value: simulation-cloud-ip
-        - name: PLATFORM_TYPE
-          value: platform-type
-        - name: KAFKA_IP
-          value: kafka-ip
-        - name: OSS_TYPE
-          value: oss-type
-        - name: OSS_IP
-          value: oss-ip
-        - name: OSS_ACCESS_KEY
-          value: oss-access-key
-        - name: OSS_SECRET_KEY
-          value: oss-secret-key
-        - name: OSS_BUCKET_NAME
-          value: oss-bucket
-        - name: KAFKA_PARTITION
-          value: kafka-partition
-        - name: KAFKA_OFFSET
-          value: kafka-offset
-        - name: CPU_ORDER
-          value: cpu-order
-    - name: algorithm-container
-      image: algorithm-image
-      imagePullPolicy: Never
-      command: [ "/bin/sh", "-c", "/run.sh; touch /tmp/hello.txt;while true;do /bin/echo $(date +%T) >> /tmp/hello.txt; sleep 600; done;" ]
-  restartPolicy: Never

build-dispatch_server.sh

@@ -1,4 +0,0 @@
-#!/bin/bash
-
-# 监控接口
-go build -o ./exe/dispatch_server.exe ./amd64/dispatch_server/main.go

kill-dispatch_server.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-
-pkill -9 -f dispatch_server.exe

run-dispatch_server.sh

@@ -1,6 +0,0 @@
-#!/bin/bash
-
-# 监控接口
-mkdir -p ./log
-rm -rf ./log/dispatch_server*
-nohup ./exe/dispatch_server.exe > ./log/dispatch_server.out 2>&1 &