fix: Resources are released repeatedly when the pipeline scripts update

pipeline/manager/scriptstore.go

@@ -294,33 +294,27 @@ func (store *ScriptStore) UpdateScriptsWithNS(ns string,
 		stats.WriteEvent(&change, sTags)
 	}
 
-	needDelete := map[string]*PlScript{}
-
-	// 在 storage & index 执行删除以及更新操作
+	// 如果上一次的集合中的脚本不存在于当前结果，则删除
 	for name, curScript := range store.storage.scripts[ns] {
-		if newScript, ok := retScripts[name]; ok {
+		if newScript, ok := retScripts[name]; ok { // 有更新
 			store.storage.scripts[ns][name] = newScript
 			stats.WriteUpdateTime(newScript.tags)
 			store.indexUpdate(newScript)
+		} else { // 删除
+			stats.WriteUpdateTime(curScript.tags)
+			store.indexDeleteAndBack(name, ns, store.storage.scripts)
+			delete(store.storage.scripts[ns], name)
 		}
-		needDelete[name] = curScript
-	}
-
-	for name, scriptDel := range needDelete {
-		stats.WriteUpdateTime(scriptDel.tags)
-		store.indexDeleteAndBack(name, ns, store.storage.scripts)
 
-		if v, ok := store.storage.scripts[ns][name]; ok {
-			if v.plBuks != nil {
-				v.plBuks.StopAllBukScanner()
-			}
-			if v.cache != nil {
-				v.cache.Stop()
-			}
-			if v.ptWindow != nil {
-				v.ptWindow.Deprecated()
-			}
-			delete(store.storage.scripts[ns], name)
+		// 清理之前一个脚本的资源
+		if curScript.plBuks != nil {
+			curScript.plBuks.StopAllBukScanner()
+		}
+		if curScript.cache != nil {
+			curScript.cache.Stop()
+		}
+		if curScript.ptWindow != nil {
+			curScript.ptWindow.Deprecated()
 		}
 	}
 

point/ptpool.go

@@ -316,7 +316,10 @@
 		arr, err = NewAnyArray(x...)
 
 	default: // for nil or other types
-		return nil
+		return &Field{
+			Key: k,
+			Val: newVal(v),
+		}
 	}
 
 	// there are array types.
@@ -404,7 +407,7 @@
 	ch <- p8s.MustNewConstMetric(poolGetDesc, p8s.CounterValue, float64(cpp.poolGet()))
 	ch <- p8s.MustNewConstMetric(poolPutDesc, p8s.CounterValue, float64(cpp.poolPut()))
 
 	ch <- p8s.MustNewConstMetric(reservedCapacityDesc, p8s.CounterValue, float64(cpp.capacity))
 	ch <- p8s.MustNewConstMetric(poolMallocDesc, p8s.CounterValue, float64(cpp.malloc.Load()))
 	ch <- p8s.MustNewConstMetric(poolEscaped, p8s.CounterValue, float64(cpp.escaped.Load()))
 }