commit

kinglong/common/cfg/cloud-config-all.yaml

@@ -4,6 +4,7 @@ platform:
   url-device-auth: http://139.9.199.227:30991/device/auth
   url-task-poll: http://139.9.199.227:30991/device/task/poll
   url-task: http://139.9.199.227:30991/device/task
+full-collect: true
 bag-number: 120
 config-refresh-interval: 60
 disk-usage: 90

kinglong/common/cfg/cloud-config.yaml

@@ -3,6 +3,7 @@ platform:
   url-device-auth: http://139.9.199.227:30991/device/auth
   url-task-poll: http://139.9.199.227:30991/device/task/poll
   url-task: http://139.9.199.227:30991/device/task
+full-collect: false
 bag-number: 120
 config-refresh-interval: 60
 disk-usage: 90

kinglong/common/cfg/cloud_cfg.go

@@ -26,6 +26,7 @@ type Ros struct {
 }
 
 type cloudConfig struct {
+	FullCollect           bool      `yaml:"full-collect"`
 	ConfigRefreshInterval int       `yaml:"config-refresh-interval"` // 配置刷新时间间隔
 	BagNumber             int       `yaml:"bag-number"`
 	TimeWindowSendGap     int       `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
@@ -149,7 +150,3 @@ func checkConfig(check cloudConfig) bool {
 	}
 	return true
 }
-
-func handleErr() {
-
-}

kinglong/common/svc/rosbag_upload.go

@@ -31,25 +31,29 @@ outLoop:
 			}
 		default:
 		}
-
+		// 每一秒扫一次
+		time.Sleep(time.Duration(1) * time.Second)
 		waitLength := len(global.TimeWindowConsumerQueue)
-		if waitLength > 0 {
-			log.GlobalLogger.Infof("待处理窗口个数为：%v", len(global.TimeWindowConsumerQueue))
-			// 1 获取即将处理的窗口
-			currentTimeWindow := global.TimeWindowConsumerQueue[0]
-			util.RemoveHeaOfdTimeWindowConsumerQueue()
-			log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		if waitLength == 0 {
+			continue outLoop
+		}
+		log.GlobalLogger.Infof("待处理窗口个数为：%v", len(global.TimeWindowConsumerQueue))
+		// 1 获取即将处理的窗口
+		currentTimeWindow := global.TimeWindowConsumerQueue[0]
+		util.RemoveHeaOfdTimeWindowConsumerQueue()
+		log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
 
-			// 2 获取目录
-			dir := util.GetCopyDir(currentTimeWindow.FaultTime)
-			bags := util.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
-			bagNumber := len(bags)
-			if bagNumber > currentTimeWindow.Length {
-				bagNumber = currentTimeWindow.Length
-				bags = bags[0:currentTimeWindow.Length]
-			}
+		// 2 获取目录
+		dir := util.GetCopyDir(currentTimeWindow.FaultTime)
+		bags := util.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
+		bagNumber := len(bags)
+		if bagNumber > currentTimeWindow.Length {
+			bagNumber = currentTimeWindow.Length
+			bags = bags[0:currentTimeWindow.Length]
+		}
 
-			// 3 filter包，必须顺序执行
+		// 3 如果不是全量采集，则使用 filter 命令对 bag 包进行主题过滤。
+		if commonConfig.CloudConfig.FullCollect == false {
 			var filterTopics []string
 			if nodeName == commonConfig.CloudConfig.Hosts[0].Name {
 				filterTopics = currentTimeWindow.MasterTopics
@@ -60,17 +64,16 @@ outLoop:
 			for _, topic := range filterTopics {
 				topicsFilterSlice = append(topicsFilterSlice, "topic=='"+topic+"'")
 			}
-			for _, bag := range bags {
+			for i, bag := range bags {
 				oldName := bag
 				newName := bag + "_filter"
 				filterCommand := []string{"filter", oldName, newName, "\"" + strings.Join(topicsFilterSlice, " or ") + "\""}
-				log.GlobalLogger.Info("执行bag包过滤命令：", filterCommand)
 				_, output, err := util.Execute("rosbag", filterCommand...)
+				log.GlobalLogger.Info("正在过滤中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
 				if err != nil {
-					log.GlobalLogger.Errorf("filter命令执行出错【命令】=%v，【输出】=%v,【err】=%v", filterCommand, output, err)
+					log.GlobalLogger.Errorf("filter 命令执行出错【命令】=%v，【输出】=%v,【err】=%v", filterCommand, output, err)
 					continue
 				}
-				log.GlobalLogger.Info("filter bag包：", oldName)
 				// 删除旧文件
 				util.DeleteFile(oldName)
 				// 将新文件改回旧文件名
@@ -79,66 +82,66 @@ outLoop:
 					continue outLoop
 				}
 			}
+		}
 
-			// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
-			log.GlobalLogger.Info("压缩bag数据包，故障时间为：", currentTimeWindow.FaultTime)
-			for _, bag := range bags {
-				oldName := bag
-				compressCommand := []string{"compress", "--bz2", oldName}
-				if _, output, err := util.Execute("rosbag", compressCommand...); err != nil {
-					log.GlobalLogger.Errorf("compress命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
-					continue
-				}
-			}
-			// 5 upload，必须顺序执行
-			log.GlobalLogger.Info("发送bag数据包，故障时间为：", currentTimeWindow.FaultTime)
-			start := time.Now()
-			objectKey1 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data/" + nodeName + "_" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
-			objectKey2 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_merge/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + ".bag"
-			objectKey3 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_parse/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
-			for i, bag := range bags {
-				bagSlice := strings.Split(bag, "/")
-				log.GlobalLogger.Info("正在上传中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】->【", objectKey1+bagSlice[len(bagSlice)-1], "】")
-				err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
-				if err != nil {
-					log.GlobalLogger.Info("上传包 ", bag, " 时报错：", err)
-					continue
-				}
-			}
-			log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
-			// 在上传完成的包目录同级下添加一个目录同名的json
-			triggerIds := make([]string, 0)
-			for _, label := range currentTimeWindow.Labels {
-				triggerIdToAppend := masterConfig.LabelMapTriggerId[label]
-				log.GlobalLogger.Info("添加一个【triggerId】=", triggerIdToAppend)
-				triggerIds = append(triggerIds, triggerIdToAppend)
-			}
-			callBackMap := map[string]interface{}{
-				"dataName":    currentTimeWindow.FaultTime,
-				"dataSize":    "", // 由合并程序补充
-				"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
-				"secretKey":   commonConfig.LocalConfig.SecretKey,
-				"rosBagPath":  objectKey2,
-				"filePath":    objectKey3,
-				"taskId":      commonConfig.PlatformConfig.TaskConfigId,
-				"triggerId":   triggerIds,
-			}
-			callBackJson, err := util.MapToJsonString(callBackMap)
-			if err != nil {
-				log.GlobalLogger.Error("callBackMap", callBackMap, "转json失败：", err)
+		// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
+		log.GlobalLogger.Info("压缩bag数据包，故障时间为：", currentTimeWindow.FaultTime)
+		for i, bag := range bags {
+			oldName := bag
+			compressCommand := []string{"compress", "--bz2", oldName}
+			log.GlobalLogger.Info("正在压缩中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+			if _, output, err := util.Execute("rosbag", compressCommand...); err != nil {
+				log.GlobalLogger.Errorf("compress 命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
+				continue
 			}
-			err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
+		}
+		// 5 upload，必须顺序执行
+		log.GlobalLogger.Info("发送bag数据包，故障时间为：", currentTimeWindow.FaultTime)
+		start := time.Now()
+		objectKey1 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data/" + nodeName + "_" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		objectKey2 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_merge/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + ".bag"
+		objectKey3 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_parse/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		for i, bag := range bags {
+			bagSlice := strings.Split(bag, "/")
+			log.GlobalLogger.Info("正在上传中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】-------【", objectKey1+bagSlice[len(bagSlice)-1], "】")
+			err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
 			if err != nil {
-				log.GlobalLogger.Error("上传callback.json", callBackJson, "失败：", err)
+				log.GlobalLogger.Info("上传包 ", bag, " 时报错：", err)
+				continue
 			}
+		}
+		log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
+		// 在上传完成的包目录同级下添加一个目录同名的json
+		triggerIds := make([]string, 0)
+		for _, label := range currentTimeWindow.Labels {
+			triggerIdToAppend := masterConfig.LabelMapTriggerId[label]
+			log.GlobalLogger.Info("添加一个【triggerId】=", triggerIdToAppend)
+			triggerIds = append(triggerIds, triggerIdToAppend)
+		}
+		callBackMap := map[string]interface{}{
+			"dataName":    currentTimeWindow.FaultTime,
+			"dataSize":    "", // 由合并程序补充
+			"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
+			"secretKey":   commonConfig.LocalConfig.SecretKey,
+			"rosBagPath":  objectKey2,
+			"filePath":    objectKey3,
+			"taskId":      commonConfig.PlatformConfig.TaskConfigId,
+			"triggerId":   triggerIds,
+		}
+		callBackJson, err := util.MapToJsonString(callBackMap)
+		if err != nil {
+			log.GlobalLogger.Error("callBackMap", callBackMap, "转json失败：", err)
+		}
+		err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
+		if err != nil {
+			log.GlobalLogger.Error("上传 callback.json", callBackJson, "失败：", err)
+		}
 
-			// 删除本地所有已上传的bag文件
-			log.GlobalLogger.Infof("结束处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
-			if err = util.RemoveDir(dir); err != nil {
-				continue outLoop
-			}
+		// 删除本地所有已上传的bag文件
+		log.GlobalLogger.Infof("结束处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		if err = util.RemoveDir(dir); err != nil {
+			continue outLoop
 		}
-		// 每一秒扫一次
-		time.Sleep(time.Duration(1) * time.Second)
+
 	}
 }

kinglong/slave/pkg/svc/move_bag.go

@@ -33,10 +33,10 @@ func RunTimeWindowProducerQueue() {
 			// 将时间窗口移出准备队列
 			util.RemoveHeadOfdTimeWindowProducerQueue()
 			if currentTimeWindow.CanUpload == "yes" {
-				log.GlobalLogger.Info("接收到的可上传的timeWindow")
+				log.GlobalLogger.Info("从节点接收到可上传的timeWindow")
 			}
 			if currentTimeWindow.CanUpload == "no" {
-				log.GlobalLogger.Info("接收到的不可上传的timeWindow")
+				log.GlobalLogger.Info("从节点接收到不可上传的timeWindow")
 			}
 
 			// 2 timeWindow不可以上传，则将data目录下的数据move到copy目录

pji/common/cfg/cloud-config.yaml

@@ -3,6 +3,7 @@ platform:
   url-device-auth: http://139.9.199.227:30991/device/auth
   url-task-poll: http://139.9.199.227:30991/device/task/poll
   url-task: http://139.9.199.227:30991/device/task
+full-collect: false
 bag-number: 10
 config-refresh-interval: 60
 disk-usage: 90

pji/common/cfg/cloud_cfg.go

@@ -26,6 +26,7 @@ type Ros struct {
 }
 
 type cloudConfig struct {
+	FullCollect           bool      `yaml:"full-collect"`
 	ConfigRefreshInterval int       `yaml:"config-refresh-interval"` // 配置刷新时间间隔
 	BagNumber             int       `yaml:"bag-number"`
 	TimeWindowSendGap     int       `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔

pji/common/svc/rosbag_upload.go

@@ -16,6 +16,7 @@ func RunTimeWindowConsumerQueue(nodeName string) {
 	log.GlobalLogger.Info("处理消费者队列goroutine - 启动")
 outLoop:
 	for {
+
 		// 收到自杀信号
 		select {
 		case signal := <-ChannelKillConsume:
@@ -31,23 +32,29 @@ outLoop:
 			}
 		default:
 		}
+		// 每一秒扫一次
+		time.Sleep(time.Duration(1) * time.Second)
+
 		waitLength := len(global.TimeWindowConsumerQueue)
-		if waitLength > 0 {
-			log.GlobalLogger.Infof("待处理窗口个数为：%v", len(global.TimeWindowConsumerQueue))
-			// 1 获取即将处理的窗口
-			currentTimeWindow := global.TimeWindowConsumerQueue[0]
-			util.RemoveHeaOfdTimeWindowConsumerQueue()
-			log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
-			// 2 获取目录
-			dir := util.GetCopyDir(currentTimeWindow.FaultTime)
-			bags := util.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
-			bagNumber := len(bags)
-			if bagNumber > currentTimeWindow.Length {
-				bagNumber = currentTimeWindow.Length
-				bags = bags[0:currentTimeWindow.Length]
-			}
+		if waitLength == 0 {
+			continue outLoop
+		}
+		log.GlobalLogger.Infof("待处理窗口个数为：%v", len(global.TimeWindowConsumerQueue))
+		// 1 获取即将处理的窗口
+		currentTimeWindow := global.TimeWindowConsumerQueue[0]
+		util.RemoveHeaOfdTimeWindowConsumerQueue()
+		log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		// 2 获取目录
+		dir := util.GetCopyDir(currentTimeWindow.FaultTime)
+		bags := util.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
+		bagNumber := len(bags)
+		if bagNumber > currentTimeWindow.Length {
+			bagNumber = currentTimeWindow.Length
+			bags = bags[0:currentTimeWindow.Length]
+		}
 
-			// 3 filter包，必须顺序执行
+		// 3 如果不是全量采集，则使用 filter 命令对 bag 包进行主题过滤。
+		if commonConfig.CloudConfig.FullCollect == false {
 			var filterTopics []string
 			if nodeName == commonConfig.CloudConfig.Hosts[0].Name {
 				filterTopics = currentTimeWindow.MasterTopics
@@ -58,17 +65,16 @@ outLoop:
 			for _, topic := range filterTopics {
 				topicsFilterSlice = append(topicsFilterSlice, "topic=='"+topic+"'")
 			}
-			for _, bag := range bags {
+			for i, bag := range bags {
 				oldName := bag
 				newName := bag + "_filter"
 				filterCommand := []string{"filter", oldName, newName, "\"" + strings.Join(topicsFilterSlice, " or ") + "\""}
-				log.GlobalLogger.Info("执行bag包过滤命令：", filterCommand)
 				_, output, err := util.Execute("rosbag", filterCommand...)
+				log.GlobalLogger.Info("正在过滤中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
 				if err != nil {
 					log.GlobalLogger.Errorf("filter命令执行出错【命令】=%v，【输出】=%v,【err】=%v", filterCommand, output, err)
 					continue
 				}
-				log.GlobalLogger.Info("filter bag包：", oldName)
 				// 删除旧文件
 				util.DeleteFile(oldName)
 				// 将新文件改回旧文件名
@@ -77,65 +83,65 @@ outLoop:
 					continue outLoop
 				}
 			}
+		}
 
-			// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
-			log.GlobalLogger.Info("压缩bag数据包，故障时间为：", currentTimeWindow.FaultTime)
-			for _, bag := range bags {
-				oldName := bag
-				compressCommand := []string{"compress", "--bz2", oldName}
-				if _, output, err := util.Execute("rosbag", compressCommand...); err != nil {
-					log.GlobalLogger.Errorf("compress命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
-					continue
-				}
+		// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
+		log.GlobalLogger.Info("压缩 bag 数据包，故障时间为：", currentTimeWindow.FaultTime)
+		for i, bag := range bags {
+			oldName := bag
+			compressCommand := []string{"compress", "--bz2", oldName}
+			log.GlobalLogger.Info("正在压缩中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+			if _, output, err := util.Execute("rosbag", compressCommand...); err != nil {
+				log.GlobalLogger.Errorf("compress命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
+				continue
 			}
-			// 5 upload，必须顺序执行
-			log.GlobalLogger.Info("发送bag数据包，故障时间为：", currentTimeWindow.FaultTime)
-			start := time.Now()
-			objectKey1 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
-			objectKey2 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_merge/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + ".bag"
-			objectKey3 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_parse/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		}
+		// 5 upload，必须顺序执行
+		log.GlobalLogger.Info("发送bag数据包，故障时间为：", currentTimeWindow.FaultTime)
+		start := time.Now()
+		objectKey1 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		objectKey2 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_merge/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + ".bag"
+		objectKey3 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_parse/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
 
-			for i, bag := range bags {
-				bagSlice := strings.Split(bag, "/")
-				log.GlobalLogger.Info("正在上传中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】->【", objectKey1+bagSlice[len(bagSlice)-1], "】")
-				err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
-				if err != nil {
-					log.GlobalLogger.Info("上传包 ", bag, " 时报错：", err)
-					continue
-				}
-			}
-			log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
-			// 在上传完成的包目录同级下添加一个目录同名的json
-			triggerIds := make([]string, 0)
-			for _, label := range currentTimeWindow.Labels {
-				triggerIds = append(triggerIds, masterConfig.LabelMapTriggerId[label])
-			}
-			callBackMap := map[string]interface{}{
-				"dataName":    currentTimeWindow.FaultTime, // 云端callback程序会将该值加8小时，因为UTC和CSV时区相差8小时
-				"dataSize":    "",                          // 由合并程序补充
-				"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
-				"secretKey":   commonConfig.LocalConfig.SecretKey,
-				"rosBagPath":  objectKey2,
-				"filePath":    objectKey3,
-				"taskId":      commonConfig.PlatformConfig.TaskConfigId,
-				"triggerId":   triggerIds,
-			}
-			callBackJson, err := util.MapToJsonString(callBackMap)
+		for i, bag := range bags {
+			bagSlice := strings.Split(bag, "/")
+			log.GlobalLogger.Info("正在上传中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】-------【", objectKey1+bagSlice[len(bagSlice)-1], "】")
+			err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
 			if err != nil {
-				log.GlobalLogger.Error("callBackMap", callBackMap, "转json失败：", err)
-			}
-			err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
-			if err != nil {
-				log.GlobalLogger.Error("上传callback.json", callBackJson, "失败：", err)
+				log.GlobalLogger.Info("上传包 ", bag, " 时报错：", err)
+				continue
 			}
+		}
+		log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
+		// 在上传完成的包目录同级下添加一个目录同名的json
+		triggerIds := make([]string, 0)
+		for _, label := range currentTimeWindow.Labels {
+			triggerIds = append(triggerIds, masterConfig.LabelMapTriggerId[label])
+		}
+		callBackMap := map[string]interface{}{
+			"dataName":    currentTimeWindow.FaultTime, // 云端callback程序会将该值加8小时，因为UTC和CSV时区相差8小时
+			"dataSize":    "",                          // 由合并程序补充
+			"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
+			"secretKey":   commonConfig.LocalConfig.SecretKey,
+			"rosBagPath":  objectKey2,
+			"filePath":    objectKey3,
+			"taskId":      commonConfig.PlatformConfig.TaskConfigId,
+			"triggerId":   triggerIds,
+		}
+		callBackJson, err := util.MapToJsonString(callBackMap)
+		if err != nil {
+			log.GlobalLogger.Error("callBackMap", callBackMap, "转json失败：", err)
+		}
+		err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
+		if err != nil {
+			log.GlobalLogger.Error("上传 callback.json", callBackJson, "失败：", err)
+		}
 
-			// 删除本地所有已上传的bag文件
-			log.GlobalLogger.Infof("结束处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
-			if err = util.RemoveDir(dir); err != nil {
-				continue outLoop
-			}
+		// 删除本地所有已上传的bag文件
+		log.GlobalLogger.Infof("结束处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		if err = util.RemoveDir(dir); err != nil {
+			continue outLoop
 		}
-		// 每一秒扫一次
-		time.Sleep(time.Duration(1) * time.Second)
+
 	}
 }