feat: 审计采集数据分批发送 --story=118658293 (#449)

pkg/bkmonitorbeat/VERSION

@@ -1 +1 @@
-v3.41.x
+v3.42.x

pkg/bkmonitorbeat/beater/taskfactory/socketsnapshot.go

@@ -15,10 +15,10 @@ package taskfactory
 import (
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/bkmonitorbeat/configs"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/bkmonitorbeat/define"
-	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/bkmonitorbeat/tasks/socketsanpshot"
+	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/bkmonitorbeat/tasks/socketsnapshot"
 )
 
 func init() {
 	SetTaskConfigByName(define.ModuleSocketSnapshot, func() define.TaskMetaConfig { return new(configs.SocketSnapshotConfig) })
-	Register(define.ModuleSocketSnapshot, socketsanpshot.New)
+	Register(define.ModuleSocketSnapshot, socketsnapshot.New)
 }

pkg/bkmonitorbeat/tasks/exceptionbeat/collector/outofmem/oomparser_unix.go

@@ -167,7 +167,7 @@ func (p *OomParser) StreamOoms(ctx context.Context, outStream chan<- *OomInstanc
 				outStream <- oomCurrentInstance
 			}
 		case <-ctx.Done():
-			logger.Errorf("exiting analyzeLines. OOM events will not be reported.")
+			logger.Info("exiting analyzeLines. OOM events will not be reported.")
 			return
 		}
 	}

pkg/bkmonitorbeat/tasks/procbin/gather.go

@@ -53,14 +53,16 @@ func (g *Gather) Run(ctx context.Context, e chan<- define.Event) {
 		return
 	}
 
+	total := len(procs)
+	if total <= 0 {
+		return
+	}
+
+	const size = 5000
 	var procbins []ProcBin
 	pcs := make(map[pidCreated]struct{})
 	for i := 0; i < len(procs); i++ {
 		proc := procs[i]
-		if proc.Cmd == "" {
-			continue
-		}
-
 		pc := pidCreated{pid: proc.Pid, created: proc.Created}
 		si := readStatInfo(pc, proc.Exe, g.config.MaxBytes)
 		pcs[pc] = struct{}{}
@@ -79,8 +81,16 @@ func (g *Gather) Run(ctx context.Context, e chan<- define.Event) {
 			Change:      si.Change.Unix(),
 			Access:      si.Access.Unix(),
 		})
+		if len(procbins) >= size {
+			e <- &Event{dataid: g.config.DataID, data: procbins, utcTime: now}
+			procbins = make([]ProcBin, 0) // 重置状态
+		}
+	}
+
+	// 确保数据完整发送
+	if len(procbins) > 0 {
+		e <- &Event{dataid: g.config.DataID, data: procbins, utcTime: now}
 	}
-	e <- &Event{dataid: g.config.DataID, data: procbins, utcTime: now}
 	cleanupCached(pcs)
 }
 

pkg/bkmonitorbeat/tasks/processbeat/process/netlink_linux.go

@@ -429,7 +429,6 @@ func getConcernPidInodes(pids []int32) map[int32][]uint64 {
 
 		inodes, err := getProcInodes("/proc", pid)
 		if err != nil {
-			logger.Errorf("failed to get /proc info: %v", err)
 			continue
 		}
 		ret[pid] = inodes

pkg/bkmonitorbeat/tasks/procsnapshot/gather.go

@@ -11,6 +11,7 @@ package procsnapshot
 
 import (
 	"context"
+	"math"
 	"sync/atomic"
 	"time"
 
@@ -52,13 +53,22 @@ func (g *Gather) Run(ctx context.Context, e chan<- define.Event) {
 		return
 	}
 
-	ret := make([]ProcMeta, 0)
-	for i := 0; i < len(procs); i++ {
-		p := procs[i]
-		if p.Cmd == "" {
-			continue
+	total := len(procs)
+	if total <= 0 {
+		return
+	}
+
+	const size = 5000
+	batch := int(math.Ceil(float64(total) / float64(size))) // 向上取整
+	var start, end int
+	for i := 0; i < batch; i++ {
+		end = start + size
+		if end >= total {
+			end = total
 		}
-		ret = append(ret, p)
+
+		data := procs[start:end]
+		e <- &Event{dataid: g.config.DataID, data: data, utcTime: now}
+		start = end
 	}
-	e <- &Event{dataid: g.config.DataID, data: ret, utcTime: now}
 }

pkg/bkmonitorbeat/tasks/procsnapshot/procs.go

@@ -72,12 +72,16 @@ func AllProcsMeta() ([]ProcMeta, error) {
 			time.Sleep(time.Millisecond * socketPerformanceSleep)
 		}
 
-		stat, err := getProcMeta(pid)
+		meta, err := getProcMeta(pid)
 		if err != nil {
 			logger.Warnf("get process meta data failed, pid: %d, err: %v", pid, err)
 			continue
 		}
-		ret = append(ret, stat)
+		// 采集不到 cmd 的进程忽略
+		if meta.Cmd == "" {
+			continue
+		}
+		ret = append(ret, meta)
 	}
 
 	return ret, nil

pkg/bkmonitorbeat/tasks/socketsanpshot/event.go → pkg/bkmonitorbeat/tasks/socketsnapshot/event.go

@@ -7,7 +7,7 @@
 // an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-package socketsanpshot
+package socketsnapshot
 
 import (
 	"time"

pkg/bkmonitorbeat/tasks/socketsanpshot/gather.go → pkg/bkmonitorbeat/tasks/socketsnapshot/gather.go

@@ -7,10 +7,11 @@
 // an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-package socketsanpshot
+package socketsnapshot
 
 import (
 	"context"
+	"math"
 	"sync/atomic"
 	"time"
 
@@ -62,5 +63,23 @@ func (g *Gather) Run(ctx context.Context, e chan<- define.Event) {
 		logger.Errorf("faile to get procs sockets: %v", err)
 		return
 	}
-	e <- &Event{dataid: g.config.DataID, data: sockets, utcTime: now}
+
+	total := len(sockets)
+	if total <= 0 {
+		return
+	}
+
+	const size = 10000
+	batch := int(math.Ceil(float64(total) / float64(size))) // 向上取整
+	var start, end int
+	for i := 0; i < batch; i++ {
+		end = start + size
+		if end >= total {
+			end = total
+		}
+
+		data := sockets[start:end]
+		e <- &Event{dataid: g.config.DataID, data: data, utcTime: now}
+		start = end
+	}
 }

pkg/bkmonitorbeat/tasks/socketsanpshot/socket.go → pkg/bkmonitorbeat/tasks/socketsnapshot/socket.go

@@ -7,7 +7,7 @@
 // an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-package socketsanpshot
+package socketsnapshot
 
 import (
 	"fmt"