fix: fix logstream state

pkg/collectconfig/executor/inputsmanager.go

@@ -42,30 +42,29 @@ func (im *inputsManager) checkInputsChange() {
 	newInputs := make(map[string]*inputWrapper, len(paths))
 
 	for _, fatPath := range paths {
-		path := fatPath.Path
+		key := logstream.BuildFileKey(fatPath.Path, fatPath.Attrs)
 		// deduplication
-		if _, ok := usedPaths[path]; ok {
+		if _, ok := usedPaths[key]; ok {
 			continue
 		}
-		usedPaths[path] = struct{}{}
+		usedPaths[key] = struct{}{}
 
-		if iw, ok := im.inputs[path]; ok {
-			newInputs[path] = iw
+		if iw, ok := im.inputs[key]; ok {
+			newInputs[key] = iw
 		} else {
 			// create if not exist
 			logger.Infoz("[pipeline] [log] [input] add", //
 				zap.String("key", im.key), //
-				zap.String("path", path))
+				zap.String("path", key))
 
 			var ls logstream.LogStream
 
 			if fatPath.IsSls {
 				ls = im.lsm.AcquireSls(fatPath.SlsConfig)
 			} else {
-				ls = im.lsm.AcquireFile(path, fatPath.Attrs)
+				ls = im.lsm.AcquireFile(fatPath.Path, fatPath.Attrs)
 			}
-
-			newInputs[path] = &inputWrapper{
+			newInputs[key] = &inputWrapper{
 				ls: ls,
 				inputStateObj: inputStateObj{
 					FatPath:   fatPath,
@@ -77,13 +76,15 @@ func (im *inputsManager) checkInputsChange() {
 		}
 	}
 
-	for path := range im.inputs {
-		if _, ok := newInputs[path]; !ok {
+	for key := range im.inputs {
+		if _, ok := newInputs[key]; !ok {
 			// 删除那些不再匹配的
-			im.releaseStream(im.inputs[path])
-			logger.Infoz("[pipeline] [log] [input] input", //
+			im.releaseStream(im.inputs[key])
+			logger.Infoz("[pipeline] [log] [input] remove input", //
 				zap.String("key", im.key), //
-				zap.String("path", path))
+				zap.String("path", key),
+				zap.Any("new", paths),
+			)
 		}
 	}
 

pkg/collectconfig/executor/logstream/logstream_g.go

@@ -87,6 +87,12 @@ func (f *GLogStream) getFromCache(cursor int64) *ReadResponse {
 }
 
 func (f *GLogStream) Read(reqCursor int64) (*ReadResponse, int64, error) {
+	//logger.Debugz("debug read",
+	//	zap.Uintptr("f", uintptr(unsafe.Pointer(f))),
+	//	zap.String("key", f.Key),
+	//	zap.Int64("cursor", reqCursor),
+	//	zap.Int("listeners", len(f.Listeners)))
+
 	// check cache
 	if c := f.getFromCache(reqCursor); c != nil {
 		return c, c.Cursor + 1, c.error
@@ -159,7 +165,7 @@ func (f *GLogStream) RemoveListener(listener Listener, cursor int64) {
 		}
 	}
 	if x := len(f.Listeners) - len(listeners); x > 1 {
-		logger.Errorz("remove multi listeners", zap.String("key", f.Key), zap.Int("old", len(f.Listeners)), zap.Int("new", len(listeners)))
+		logger.Errorz("remove multi listeners", zap.String("key", f.Key), zap.Int("old", len(f.Listeners)), zap.Int("new", len(listeners)), zap.Stack("stack"))
 	}
 	f.Listeners = listeners
 

pkg/collectconfig/executor/logstream/logstream_manager.go

@@ -218,6 +218,10 @@ func (m *Manager) AcquireFile(path string, attrs map[string]string) LogStream {
 	return m.acquire0(path, attrs, nil)
 }
 
+func BuildFileKey(path string, attrs map[string]string) string {
+	return buildFileKey(path, attrs)
+}
+
 func buildFileKey(path string, attrs map[string]string) string {
 	if len(attrs) == 0 {
 		return path

pkg/cri/impl/default_cri.go

@@ -606,7 +606,7 @@ func (e *defaultCri) buildCriContainer(criPod *cri.Pod, dc *cri.EngineDetailCont
 
 	criPod.All = append(criPod.All, criContainer)
 
-	if criContainer.IsRunning() && criContainer.Hacked == cri.HackInit && !criContainer.Sandbox {
+	if criContainer.IsRunning() && criContainer.Hacked == cri.HackInit && criContainer.MainBiz {
 		criContainer.Hacked = cri.HackIng
 
 		var err error
@@ -935,7 +935,7 @@ func (e *defaultCri) buildPod(pod *v1.Pod, oldState *internalState, newState *in
 			if cached != nil && !isContainerChanged(cached.engineContainer, container) {
 				cached.criContainer.Pod = criPod
 
-				if forceUpdateContainerInfo {
+				if forceUpdateContainerInfo && cached.criContainer.MainBiz {
 					go e.updateZombieCheck(cached.criContainer)
 				}