add some log

iqshell/common/flow/flow.go

@@ -104,26 +104,32 @@ func (f *Flow) Start() {
 		workList := make([]*WorkInfo, 0, f.doWorkInfoListCount)
 		for {
 			hasMore, workInfo, err := f.WorkProvider.Provide()
+			log.DebugF("work producer get work, hasMore:%v, workInfo: %+v, err: %+v", hasMore, workInfo.Data, err)
 			if err != nil {
 				if err.Code == data.ErrorCodeParamMissing ||
 					err.Code == data.ErrorCodeLineHeader {
+					log.DebugF("work producer get work, skip:%s because:%s", workInfo.Data, err)
 					f.notifyWorkSkip(workInfo, nil, err)
 				} else {
+					log.DebugF("work producer get work fail, error:%s info:%s", err, workInfo.Data)
 					f.notifyWorkFail(workInfo, err)
 				}
 				continue
 			}
 
 			if workInfo == nil || workInfo.Work == nil {
 				if !hasMore {
+					log.Info("work producer get work completed")
 					break
 				} else {
+					log.Info("work producer get work fail: work in empty")
 					continue
 				}
 			}
 
 			// 检测 work 是否需要过
 			if skip, cause := f.shouldWorkSkip(workInfo); skip {
+				log.DebugF("work producer get work, skip:%s cause:%s", workInfo.Data, cause)
 				f.notifyWorkSkip(workInfo, nil, cause)
 				continue
 			}
@@ -192,11 +198,15 @@ func (f *Flow) Start() {
 
 				workCount := len(workList)
 
+				log.InfoF("work consumer get works, count:%s", workCount)
+
 				_ = f.limitAcquire(workCount)
 				// workRecordList 有数据则长度和 workList 长度相同
 				workRecordList, workErr := worker.DoWork(workList)
 				f.limitRelease(workCount)
 
+				log.InfoF("work consumer handle works, count:%s error:%+v", workCount, workErr)
+
 				if len(workRecordList) == 0 && workErr != nil {
 					log.ErrorF("Do Worker Error:%+v", workErr)
 					for _, workInfo := range workList {

iqshell/storage/object/download/operations/work_provider.go

@@ -103,7 +103,9 @@ func (w *workProvider) getWorkInfoFromFile() {
 		var keys []string
 		for {
 			if len(keys) == 300 {
-				w.getWorkInfoOfKeys(keys)
+				if !w.getWorkInfoOfKeys(keys) {
+					break
+				}
 				keys = nil
 			}
 
@@ -140,9 +142,9 @@ func (w *workProvider) getWorkInfoFromFile() {
 	}()
 }
 
-func (w *workProvider) getWorkInfoOfKeys(keys []string) {
+func (w *workProvider) getWorkInfoOfKeys(keys []string) bool {
 	if len(keys) == 0 {
-		return
+		return true
 	}
 
 	operations := make([]batch.Operation, 0, len(keys))
@@ -155,6 +157,11 @@ func (w *workProvider) getWorkInfoOfKeys(keys []string) {
 		}
 	}
 
+	if len(operations) == 0 {
+		log.Error("get file info error: key invalid")
+		return false
+	}
+
 	results, err := batch.Some(operations)
 	if len(results) == len(operations) {
 		for i, result := range results {
@@ -203,6 +210,8 @@ func (w *workProvider) getWorkInfoOfKeys(keys []string) {
 		}
 		time.Sleep(10 * time.Second)
 	}
+
+	return true
 }
 
 func (w *workProvider) getWorkInfoFromBucket() {