Compare `822434c ... +0 ... 5a85fc6`

56.99% 45.65% +0.03%

Showing 4 of 4 files from the diff.

53.88% 50.00% ø

self/logrunner.go changed.

69.72% 0.00% ø

reader/dirx/dirx.go changed.

61.38% 48.78% -1.08%

reader/tailx/tailx.go changed.

37.85% 0.00% -0.43%

reader/meta.go changed.

@@ -53,8 +53,8 @@

		"head_pattern":       `^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} \[(WARN)|(INFO)|(ERROR)]|(DEBUG)\])`,
	}
	parserConfig = conf.MapConf{
		"type":                   "raw",
		"name":                   "parser",
		"type": "raw",
		"name": "parser",
		"disable_record_errdata": "true",
	}
	senderConfig = conf.MapConf{

@@ -238,9 +238,9 @@

	defer timer.Stop()
	select {
	case <-lr.exitChan:
		log.Debugf("runner %s has been stopped", lr.Name())
		log.Debugf("Runner[%s] has been stopped", lr.Name())
	case <-timer.C:
		log.Debugf("runner %s exited timeout, start to force stop", lr.Name())
		log.Debugf("Runner[%s] exited timeout, start to force stop", lr.Name())
		atomic.StoreInt32(&lr.stopped, 1)
	}


@@ -323,7 +323,7 @@


		if r.hasStopped() || r.isStopping() {
			log.Warnf("Runner[%v] created new reader for log path %q but daemon reader has stopped/is stopping, will not run at this time", r.meta.RunnerName, logPath)
			continue
			break
		}

		go dr.Run()

@@ -385,6 +385,10 @@

		log.Debugf("Runner[%s] ActiveReader %s was closed", ar.runnerName, ar.originpath)
	}()
	ar.SyncMeta()
	return ar.CloseNoSync()
}

func (ar *ActiveReader) CloseNoSync() error {
	brCloseErr := ar.br.Close()
	if err := ar.Stop(); err != nil {
		return brCloseErr

@@ -721,6 +725,16 @@

			continue
		}

		//达到最大打开文件数，不再追踪
		if len(r.fileReaders) >= r.maxOpenFiles {
			if !IsSelfRunner(r.meta.RunnerName) {
				log.Warnf("Runner[%s] %s meet maxOpenFiles limit %d, ignore Stat new log...", r.meta.RunnerName, r.Name(), r.maxOpenFiles)
			} else {
				log.Debugf("Runner[%s] %s meet maxOpenFiles limit %d, ignore Stat new log...", r.meta.RunnerName, r.Name(), r.maxOpenFiles)
			}
			break
		}

		ar, err := NewActiveReader(mc, rp, r.whence, inodeStr, r)
		if err != nil {
			err = fmt.Errorf("Runner[%s] NewActiveReader for matches %s error %v ", r.meta.RunnerName, rp, err)

@@ -776,6 +790,8 @@

			} else {
				log.Debugf("Runner[%v] %v NewActiveReader but reader was stopped, ignore this...", r.meta.RunnerName, mc)
			}
			r.armapmux.Unlock()
			break
		}
		r.armapmux.Unlock()
		if !r.hasStopped() && !r.isStopping() {

@@ -786,6 +802,7 @@

			} else {
				log.Debugf("Runner[%s] %s NewActiveReader but reader was stopped, will not running...", r.meta.RunnerName, mc)
			}
			break
		}
	}


@@ -1001,14 +1018,23 @@


	// 停10ms为了管道中的数据传递完毕，确认reader run函数已经结束不会再读取，保证syncMeta的正确性
	time.Sleep(10 * time.Millisecond)
	r.SyncMeta()
	r.SyncMetaClose()

	// 在所有 active readers 关闭完成后再关闭管道
	close(r.msgChan)
	close(r.errChan)
	return nil
}

func (r *Reader) SyncMetaClose() {
	ars := r.getActiveReaders()
	var wg sync.WaitGroup
	for _, ar := range ars {
		readcache := ar.SyncMeta()
		wg.Add(1)
		go func(mar *ActiveReader) {
			defer wg.Done()
			xerr := mar.Close()
			xerr := mar.CloseNoSync()
			if xerr != nil {
				if !IsSelfRunner(r.meta.RunnerName) {
					log.Errorf("Runner[%s] Close ActiveReader %s error %v", r.meta.RunnerName, mar.originpath, xerr)

@@ -1017,13 +1043,38 @@

				}
			}
		}(ar)
		if readcache == "" {
			continue
		}
		r.armapmux.Lock()
		r.cacheMap[ar.realpath] = readcache
		r.armapmux.Unlock()
	}
	wg.Wait()
	r.armapmux.Lock()
	buf, err := jsoniter.Marshal(r.cacheMap)
	r.armapmux.Unlock()
	if err != nil {
		if !IsSelfRunner(r.meta.RunnerName) {
			log.Errorf("%s sync meta error %v, cacheMap %v", r.Name(), err, r.cacheMap)
		} else {
			log.Debugf("Runner[%s] %s sync meta error %v, cacheMap %v", r.meta.RunnerName, r.Name(), err, r.cacheMap)
		}
		return
	}
	err = r.meta.WriteBuf(buf, 0, 0, len(buf))
	if err != nil {
		if !IsSelfRunner(r.meta.RunnerName) {
			log.Errorf("%v sync meta WriteBuf error %v, buf %v", r.Name(), err, string(buf))
		} else {
			log.Debugf("Runner[%s] %s sync meta WriteBuf error %v, buf %v", r.meta.RunnerName, r.Name(), err, string(buf))
		}
		return
	}

	// 在所有 active readers 关闭完成后再关闭管道
	close(r.msgChan)
	close(r.errChan)
	return nil
	if IsSubMetaExpire(r.submetaExpire, r.expire) {
		r.meta.CleanExpiredSubMetas(r.submetaExpire)
	}
}

func (r *Reader) Reset() error {

@@ -283,8 +283,9 @@

		// 二次确认 submeta 目录在删除前的一刻仍旧是过期状态才执行删除操作
		if hasSubMetaExpired(path, expire) {
			numCleaned++
			err := os.RemoveAll(path)
			log.Infof("Expired submeta %q has been removed with error %v", path, err)
			if err := os.RemoveAll(path); err != nil {
				log.Errorf("Expired submeta %q has been removed with error %v", path, err)
			}
		}
		delete(m.subMetaExpired, path)
	}

Learn more Showing 4 files with coverage changes found.

Changes in mgr/mgr.go

-3

→

Loading file...

Changes in logkit.go

-2

→

Loading file...

Changes in reader/dirx/dir_reader.go

-3

→

Loading file...

Changes in sender/file/writer.go

-2

→

Loading file...

56.99% 45.65% +0.03%

tailx stop fix

redHJ

5a85fc6

CI Passed

+36

+28

-1

56.96% ø -0.03%

Merge pull request #1057 from redHJ/httpConfigFix

redHJ

822434c

CI Passed

Files		•	•	•	Coverage
conf	ø	ø	ø	ø	97.67%
mgr	ø	+3	ø	-3	^0.09% 66.15%
parser	ø	ø	ø	ø	87.12%
queue	ø	ø	ø	ø	74.13%
rateio	ø	ø	ø	ø	95.74%
reader	+36	+19	+1	+16	^0.02% 47.84%
router	ø	ø	ø	ø	54.13%
samples	ø	ø	ø	ø	100.00%
sender	ø	+4	-2	-2	^0.17% 45.15%
transforms	ø	ø	ø	ø	55.67%
utils	ø	ø	ø	ø	58.88%
audit/audit.go	ø	ø	ø	ø	73.63%
cleaner/cleaner.go	ø	ø	ø	ø	58.82%
cli/upgrade.go	ø	ø	ø	ø	46.75%
logkit.go	ø	+2	ø	-2	^+1.12% 36.31%
self/logrunner.go	ø	ø	ø	ø	53.88%
times/times.go	ø	ø	ø	ø	91.30%
Project Totals (139 files)	25,715	14,655	1,660	9,400	56.99%

238	238		defer timer.Stop()
239	239		select {
240	240		case <-lr.exitChan:
241		-	log.Debugf("runner %s has been stopped", lr.Name())
	241	+	log.Debugf("Runner[%s] has been stopped", lr.Name())
242	242		case <-timer.C:
243		-	log.Debugf("runner %s exited timeout, start to force stop", lr.Name())
	243	+	log.Debugf("Runner[%s] exited timeout, start to force stop", lr.Name())
244	244		atomic.StoreInt32(&lr.stopped, 1)
245	245		}
246	246

323	323
324	324		if r.hasStopped() \|\| r.isStopping() {
325	325		log.Warnf("Runner[%v] created new reader for log path %q but daemon reader has stopped/is stopping, will not run at this time", r.meta.RunnerName, logPath)
326		-	continue
	326	+	break
327	327		}
328	328
329	329		go dr.Run()

385	385		log.Debugf("Runner[%s] ActiveReader %s was closed", ar.runnerName, ar.originpath)
386	386		}()
387	387		ar.SyncMeta()
	388	+	return ar.CloseNoSync()
	389	+	}
	390	+
	391	+	func (ar *ActiveReader) CloseNoSync() error {
388	392		brCloseErr := ar.br.Close()
389	393		if err := ar.Stop(); err != nil {
390	394		return brCloseErr

721	725		continue
722	726		}
723	727
	728	+	//达到最大打开文件数，不再追踪
	729	+	if len(r.fileReaders) >= r.maxOpenFiles {
	730	+	if !IsSelfRunner(r.meta.RunnerName) {
	731	+	log.Warnf("Runner[%s] %s meet maxOpenFiles limit %d, ignore Stat new log...", r.meta.RunnerName, r.Name(), r.maxOpenFiles)
	732	+	} else {
	733	+	log.Debugf("Runner[%s] %s meet maxOpenFiles limit %d, ignore Stat new log...", r.meta.RunnerName, r.Name(), r.maxOpenFiles)
	734	+	}
	735	+	break
	736	+	}
	737	+
724	738		ar, err := NewActiveReader(mc, rp, r.whence, inodeStr, r)
725	739		if err != nil {
726	740		err = fmt.Errorf("Runner[%s] NewActiveReader for matches %s error %v ", r.meta.RunnerName, rp, err)

776	790		} else {
777	791		log.Debugf("Runner[%v] %v NewActiveReader but reader was stopped, ignore this...", r.meta.RunnerName, mc)
778	792		}
	793	+	r.armapmux.Unlock()
	794	+	break
779	795		}
780	796		r.armapmux.Unlock()
781	797		if !r.hasStopped() && !r.isStopping() {

Compare `822434c ... +0 ... 5a85fc6`

No flags found

2 Commits

53	53		"head_pattern": `^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} \[(WARN)\|(INFO)\|(ERROR)]\|(DEBUG)\])`,
54	54		}
55	55		parserConfig = conf.MapConf{
56		-	"type": "raw",
57		-	"name": "parser",
	56	+	"type": "raw",
	57	+	"name": "parser",
58	58		"disable_record_errdata": "true",
59	59		}
60	60		senderConfig = conf.MapConf{

786	802		} else {
787	803		log.Debugf("Runner[%s] %s NewActiveReader but reader was stopped, will not running...", r.meta.RunnerName, mc)
788	804		}
	805	+	break
789	806		}
790	807		}
791	808

1001	1018
1002	1019		// 停10ms为了管道中的数据传递完毕，确认reader run函数已经结束不会再读取，保证syncMeta的正确性
1003	1020		time.Sleep(10 * time.Millisecond)
1004		-	r.SyncMeta()
	1021	+	r.SyncMetaClose()
	1022	+
	1023	+	// 在所有 active readers 关闭完成后再关闭管道
	1024	+	close(r.msgChan)
	1025	+	close(r.errChan)
	1026	+	return nil
	1027	+	}
	1028	+
	1029	+	func (r *Reader) SyncMetaClose() {
1005	1030		ars := r.getActiveReaders()
1006	1031		var wg sync.WaitGroup
1007	1032		for _, ar := range ars {
	1033	+	readcache := ar.SyncMeta()
1008	1034		wg.Add(1)
1009	1035		go func(mar *ActiveReader) {
1010	1036		defer wg.Done()
1011		-	xerr := mar.Close()
	1037	+	xerr := mar.CloseNoSync()
1012	1038		if xerr != nil {
1013	1039		if !IsSelfRunner(r.meta.RunnerName) {
1014	1040		log.Errorf("Runner[%s] Close ActiveReader %s error %v", r.meta.RunnerName, mar.originpath, xerr)

1017	1043		}
1018	1044		}
1019	1045		}(ar)
	1046	+	if readcache == "" {
	1047	+	continue
	1048	+	}
	1049	+	r.armapmux.Lock()
	1050	+	r.cacheMap[ar.realpath] = readcache
	1051	+	r.armapmux.Unlock()
1020	1052		}
1021	1053		wg.Wait()
	1054	+	r.armapmux.Lock()
	1055	+	buf, err := jsoniter.Marshal(r.cacheMap)
	1056	+	r.armapmux.Unlock()
	1057	+	if err != nil {
	1058	+	if !IsSelfRunner(r.meta.RunnerName) {
	1059	+	log.Errorf("%s sync meta error %v, cacheMap %v", r.Name(), err, r.cacheMap)
	1060	+	} else {
	1061	+	log.Debugf("Runner[%s] %s sync meta error %v, cacheMap %v", r.meta.RunnerName, r.Name(), err, r.cacheMap)
	1062	+	}
	1063	+	return
	1064	+	}
	1065	+	err = r.meta.WriteBuf(buf, 0, 0, len(buf))
	1066	+	if err != nil {
	1067	+	if !IsSelfRunner(r.meta.RunnerName) {
	1068	+	log.Errorf("%v sync meta WriteBuf error %v, buf %v", r.Name(), err, string(buf))
	1069	+	} else {
	1070	+	log.Debugf("Runner[%s] %s sync meta WriteBuf error %v, buf %v", r.meta.RunnerName, r.Name(), err, string(buf))
	1071	+	}
	1072	+	return
	1073	+	}
1022	1074
1023		-	// 在所有 active readers 关闭完成后再关闭管道
1024		-	close(r.msgChan)
1025		-	close(r.errChan)
1026		-	return nil
	1075	+	if IsSubMetaExpire(r.submetaExpire, r.expire) {
	1076	+	r.meta.CleanExpiredSubMetas(r.submetaExpire)
	1077	+	}
1027	1078		}
1028	1079
1029	1080		func (r *Reader) Reset() error {

283	283		// 二次确认 submeta 目录在删除前的一刻仍旧是过期状态才执行删除操作
284	284		if hasSubMetaExpired(path, expire) {
285	285		numCleaned++
286		-	err := os.RemoveAll(path)
287		-	log.Infof("Expired submeta %q has been removed with error %v", path, err)
	286	+	if err := os.RemoveAll(path); err != nil {
	287	+	log.Errorf("Expired submeta %q has been removed with error %v", path, err)
	288	+	}
288	289		}
289	290		delete(m.subMetaExpired, path)
290	291		}

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

Compare 822434c ... +0 ... 5a85fc6

No flags found

2 Commits

Compare `822434c ... +0 ... 5a85fc6`