loginfo

cgimport.conf

@@ -13,7 +13,7 @@
 ; count=20           ; 保留数量，默认 100
 ; scroll=1天         ; 时间滚动 scroll，覆盖 dialy 设置，默认 1 天，支持时间单位 d 天 h 小时 m 分钟 s 秒 ms 毫秒 us 微秒 ns 纳秒，默认毫秒
 ; expire=14d         ; 保留时间，默认 14 天
-level=debug
+level=info
 
 [cgi]
 # 文件解析并发数

importer/importer.go

@@ -38,6 +38,7 @@ type Importer struct {
 	odbcimporter     *ODBCImporter
 	starttime        time.Time
 	currentstarttime time.Time
+	lastlogtime      cmap.ConcurrentMap[string, time.Time]
 }
 
 func ImportDir(datapath string, parallel int, rebuild, reload bool) (totalfilescount, totallinecount, totalrecordscount, totalretrycount int64, totalusetime time.Duration, filescount, linescount, recordscount, retrycount int64, usetime time.Duration, err error) {
@@ -51,6 +52,7 @@ func ImportDir(datapath string, parallel int, rebuild, reload bool) (totalfilesc
 		fileimportrc: rc.NewRoutinesController("", parallel),
 		odbcqueryrc:  rc.NewRoutinesControllerLimit("", concurlimt, concurlimt*5),
 		odbcimporter: NewODBCImporter(),
+		lastlogtime:  cmap.NewSingle[string, time.Time](),
 	}
 	return importer.Import()
 }
@@ -275,13 +277,15 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 	if e != nil {
 		return linecount, blockcount, retrycount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}})
 	}
-	lastlogtime := time.Now()
 	skiplines := int(linefrom)
 	blockcount = blockfrom
 	doinglines := []int64{}
 	donelines := linefrom
 	doneblocks := blockfrom
+	savedlines := linefrom
+	savedblocks := blockfrom
 	retrycount = totalretrycount
+	linecount = linefrom
 	// maxresponsetime := time.Duration(0)
 	var wg sync.WaitGroup
 	defer importer.done()
@@ -290,6 +294,7 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 		if err != nil {
 			return
 		}
+		lastlinecount := linecount
 		block, line, linenumber, e := br.ReadBlock(skiplines)
 		linecount = int64(linenumber)
 		if e != nil {
@@ -329,6 +334,7 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 				err = merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(doingline)}, {"line": line}})
 				return
 			}
+			atomic.AddInt64(&donelines, doingline-lastlinecount)
 			atomic.AddInt64(&doneblocks, 1)
 			if logstatus {
 				readinglines := doinglines[len(doinglines)-1]
@@ -339,18 +345,11 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 						RecordsCount: doingblock,
 						RetryCount:   retrycount,
 					}
-					donelines = doingline
 					importer.importstatus.TotalUseTime = time.Since(importer.starttime)
 					importer.importstatus.Save()
+					savedlines = doingline
+					savedblocks = doingblock
 					doinglines = doinglines[1:]
-					if time.Since(lastlogtime) > 5*time.Second {
-						if retrycount > 0 {
-							logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records", retrycount, "retry times")
-						} else {
-							logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records")
-						}
-						lastlogtime = time.Now()
-					}
 					importer.importstatus.mutex.Unlock()
 				} else {
 					for i, l := range doinglines {
@@ -359,15 +358,8 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 							break
 						}
 					}
-					if time.Since(lastlogtime) > 5*time.Second {
-						if retrycount > 0 {
-							logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records", retrycount, "retry times")
-						} else {
-							logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records")
-						}
-						lastlogtime = time.Now()
-					}
 				}
+				importer.logInfo(filename, readinglines, doinglines, donelines, doneblocks, savedlines, savedblocks, retrycount)
 			}
 		})
 		if e != nil {
@@ -376,6 +368,25 @@ func (importer *Importer) importReader(filename string, buf io.Reader, linefrom,
 	}
 }
 
+func (importer *Importer) logInfo(filename string, readinglines int64, doinglines []int64, donelines, doneblocks, savedlines, savedblocks, retrycount int64) {
+	if time.Since(importer.lastlogtime.GetIFPresent(filename)) > 5*time.Second {
+		if odbc.LogDebug {
+			if retrycount > 0 {
+				logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records,", "saved", savedlines, "lines", savedblocks, "records", retrycount, "retry times")
+			} else {
+				logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records,", "saved", savedlines, "lines", savedblocks, "records")
+			}
+		} else {
+			if retrycount > 0 {
+				logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records", retrycount, "retry times")
+			} else {
+				logger.Info("file", filename, "read", readinglines, "lines,", "importing", len(doinglines), "chunks,", "imported", donelines, "lines", doneblocks, "records")
+			}
+		}
+		importer.lastlogtime.Set(filename, time.Now())
+	}
+}
+
 func (importer *Importer) importRecord(record map[string]any, line string, filename string, classaliasname string, linecount int) (retrycount int, err error) {
 	if odbc.LogDebug {
 		bs, e := json.MarshalIndent(record, "", "  ")