dropwatch: consolidate BPF/Go changes

- Rename drop_source to drop_reason in BPF/Go binding
- Add device filter via load-time constant rewrite
- Wire --device/--device-exclude flags to BPF constants
- Add DROP_REASON_UNSUPPORT and rename filter variables
- Propagate writer IO errors and stop mutating events
- Set Source=client in handleEvent instead of socketWriter
- Unify stack formatting via symbol.FormatStack
- Capture subprocess output in exit error
- Clean up DropWatchTracing schema
- Adopt pcapfilter.Load (final API)

Change-Id: Iafbc745ca16c31fb1b52580816a39bdc41f259af
(cherry picked from commit d77559401b3a65ed76e0eea7adeb20075d4e70e5)

Author: huhong-web

bpf/dropwatch.c

@@ -7,6 +7,7 @@
 #include "bpf_net_namespace.h"
 #include "bpf_netdevice.h"
 #include "bpf_pcap_stub.h"
+#include "bpf_skb_filter.h"
 #include "vmlinux_net.h"
 
 #define TYPE_TCP_COMMON_DROP 1
@@ -19,8 +20,9 @@
 #define SK_FL_TYPE_SHIFT 16
 #define SK_FL_TYPE_MASK 0xffff0000
 
-/* Drop source (who triggered the kfree_skb) */
-#define DROP_SOURCE_UNKNOWN 0
+/* Drop reason (kernel skb_drop_reason enum; UNKNOWN until plumbed). */
+#define DROP_REASON_UNKNOWN   0
+#define DROP_REASON_UNSUPPORT 1
 #define PKT_RAW_LEN 120
 
 struct packet_meta {
@@ -32,7 +34,7 @@ struct packet_meta {
 	u32 ifindex;             /* 4  */
 	u32 dev_flags;           /* 4  */
 	u32 queue_mapping;       /* 4  */
-	u32 drop_source;         /* 4  */
+	u32 drop_reason;         /* 4  */
 	u32 type;                /* 4  */
 	u32 net_inum;             /* 4  */
 	u8  dev_name[IFNAMSIZ];  /* 16 */
@@ -167,6 +169,13 @@ int bpf_kfree_skb_prog(struct trace_event_raw_kfree_skb *ctx)
 	 * Read directly from skb to avoid the ambiguity. */
 	skb_protocol = bpf_ntohs(BPF_CORE_READ(skb, protocol));
 
+	/* device filter: rewritten at load time via filter_ifindex_included /
+	 * filter_ifindex_excluded. filter_ifindex_included == 0 (default) means all
+	 * devices pass; cheap check, runs before the pcap bytecode below.
+	 */
+	if (!skb_filter_pass_dev(skb))
+		return 0;
+
 	/* pcap filter via bpf_pcap_stub.h: pass-through stub patched at load
 	 * time by internal/pcapinject with the compiled tcpdump expression.
 	 */
@@ -183,7 +192,7 @@ int bpf_kfree_skb_prog(struct trace_event_raw_kfree_skb *ctx)
 	bpf_get_current_comm(&data->meta.comm, sizeof(data->meta.comm));
 	data->meta.kfree_skb_addr = (u64)(unsigned long)ctx->location;
 	data->meta.queue_mapping = BPF_CORE_READ(skb, queue_mapping);
-	data->meta.drop_source = DROP_SOURCE_UNKNOWN;
+	data->meta.drop_reason = DROP_REASON_UNKNOWN;
 	data->meta.type = 0;
 
 	data->pkt_hdr.pkt_len = BPF_CORE_READ(skb, len);

bpf/include/bpf_skb_filter.h

@@ -0,0 +1,42 @@
+/* SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) */
+#ifndef __BPF_SKB_FILTER_H__
+#define __BPF_SKB_FILTER_H__
+
+#include <bpf/bpf_core_read.h>
+
+#include "vmlinux_net.h"
+
+/*
+ * Device filter injected via RewriteConstants at load time.
+ *
+ * filter_ifindex_included == 0  : disabled — all devices pass (default)
+ * filter_ifindex_included != 0  : compare skb->dev->ifindex against this value
+ * filter_ifindex_excluded == 0  : whitelist — only matching ifindex passes
+ * filter_ifindex_excluded == 1  : blacklist — matching ifindex is dropped
+ *
+ * Each .c file that includes this header gets its own private copies of
+ * the variables (static linkage), rewritten independently per ELF load.
+ */
+static volatile const __u32 filter_ifindex_included  = 0;
+static volatile const __u32 filter_ifindex_excluded = 0;
+
+/*
+ * skb_filter_pass_dev - device filter check for a kfree_skb-style program.
+ *
+ * Returns true if the SKB should be processed, false if it should be skipped.
+ * When filter_ifindex_included == 0 the check is a no-op (returns true immediately).
+ */
+static __always_inline bool skb_filter_pass_dev(struct sk_buff *skb)
+{
+	// Early return when filter is disabled.
+	if (filter_ifindex_included == 0)
+		return true;
+
+	struct net_device *dev = BPF_CORE_READ(skb, dev);
+	__u32 idx = dev ? BPF_CORE_READ(dev, ifindex) : 0;
+	bool match = (idx == filter_ifindex_included);
+
+	return filter_ifindex_excluded ? !match : match;
+}
+
+#endif /* __BPF_SKB_FILTER_H__ */

cmd/dropwatch/format.go

@@ -19,9 +19,9 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"strings"
 
 	"huatuo-bamai/internal/packet"
+	"huatuo-bamai/internal/symbol"
 	"huatuo-bamai/internal/toolstream"
 	"huatuo-bamai/pkg/types"
 )
@@ -34,15 +34,15 @@ type writer interface {
 type textWriter struct{ w io.Writer }
 
 func (s *textWriter) Write(ev *types.DropWatchTracing) error {
-	fmt.Fprintf(s.w, "%s %s %s len=%d dev=%s pid=%d[%s] addr=%s\n",
+	if _, err := fmt.Fprintf(s.w, "%s %s %s len=%d dev=%s pid=%d[%s] addr=%s\n",
 		ev.ObservedTimestamp, ev.PacketType, formatDetail(ev.PacketInfo),
-		ev.PacketLen, ev.NetdevName, ev.Pid, ev.Comm, ev.PacketSkbAddr)
+		ev.PacketLen, ev.NetdevName, ev.Pid, ev.Comm, ev.PacketSkbAddr); err != nil {
+		return err
+	}
 
 	if ev.Stack != "" {
-		for i, frame := range strings.Split(strings.TrimRight(ev.Stack, "\n"), "\n") {
-			if frame != "" {
-				fmt.Fprintf(s.w, "\t#%-2d  %s\n", i, frame)
-			}
+		if err := symbol.FormatStackLines(s.w, ev.Stack); err != nil {
+			return err
 		}
 	}
 
@@ -52,15 +52,18 @@ func (s *textWriter) Write(ev *types.DropWatchTracing) error {
 type jsonWriter struct{ w io.Writer }
 
 func (s *jsonWriter) Write(ev *types.DropWatchTracing) error {
-	b, _ := json.Marshal(ev)
-	_, _ = fmt.Fprintf(s.w, "%s\n", b)
-	return nil
+	b, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	b = append(b, '\n')
+	_, err = s.w.Write(b)
+	return err
 }
 
 type socketWriter struct{ client *toolstream.Client }
 
 func (s *socketWriter) Write(ev *types.DropWatchTracing) error {
-	ev.Source = "client"
 	return s.client.Send(ev)
 }
 

cmd/dropwatch/format_test.go

@@ -0,0 +1,53 @@
+// Copyright 2026 The HuaTuo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on 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 main
+
+import (
+	"errors"
+	"testing"
+
+	"huatuo-bamai/internal/packet"
+	"huatuo-bamai/pkg/types"
+)
+
+// errWriter always fails Write with the configured error. Used to verify that
+// the dropwatch writers propagate IO errors instead of swallowing them.
+type errWriter struct{ err error }
+
+func (w errWriter) Write(_ []byte) (int, error) { return 0, w.err }
+
+func TestTextWriterPropagatesIOError(t *testing.T) {
+	boom := errors.New("boom")
+	w := &textWriter{w: errWriter{err: boom}}
+
+	err := w.Write(&types.DropWatchTracing{
+		ObservedTimestamp: "now",
+		PacketType:        packet.PacketTypeIPv4TCP,
+		NetdevName:        "eth0",
+	})
+	if !errors.Is(err, boom) {
+		t.Fatalf("got %v, want %v", err, boom)
+	}
+}
+
+func TestJSONWriterPropagatesIOError(t *testing.T) {
+	boom := errors.New("boom")
+	w := &jsonWriter{w: errWriter{err: boom}}
+
+	err := w.Write(&types.DropWatchTracing{ObservedTimestamp: "now"})
+	if !errors.Is(err, boom) {
+		t.Fatalf("got %v, want %v", err, boom)
+	}
+}

cmd/dropwatch/main.go

@@ -17,6 +17,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"net"
 	"os"
 	"os/signal"
 	"strings"
@@ -54,7 +55,7 @@ type packetMeta struct {
 	NetdevIfindex      uint32
 	NetdevFlags        uint32
 	NetdevQueueMapping uint32
-	DropSource         uint32
+	DropReason         uint32
 	Type               uint32
 	NetInode           uint32
 	NetdevName         [bpf.NetdevNameLen]byte
@@ -85,16 +86,38 @@ var (
 	_ = [1]struct{}{}[240-unsafe.Offsetof(dropPacketEvent{}.Stack)]
 )
 
-// loadDropwatchBPF reads the BPF object at bpfPath, injects filterExpr into the
-// pcap_stub_l2/l3 stubs, and loads it. Each instance uses a unique BPF name to
-// allow multiple instances to coexist.
-func loadDropwatchBPF(bpfPath, filterExpr string) (bpf.BPF, error) {
+// loadBPFWithFilter reads the BPF object at bpfPath, injects filterExpr into the
+// pcap_stub_l2/l3 stubs, applies RewriteConstants for any non-nil consts, and
+// loads it. Each instance uses a unique BPF name to allow multiple instances
+// to coexist.
+func loadBPFWithFilter(bpfPath, filterExpr string, consts map[string]any) (bpf.BPF, error) {
 	bpfBytes, err := os.ReadFile(bpfPath)
 	if err != nil {
 		return nil, fmt.Errorf("read bpf object: %w", err)
 	}
 	bpfName := fmt.Sprintf("dropwatch_%d.o", time.Now().UnixNano())
-	return pcapfilter.Load(bpfName, bpfBytes, filterExpr, nil)
+	return pcapfilter.Load(bpfName, bpfBytes, filterExpr, consts)
+}
+
+// deviceFilterConsts resolves devName to an ifindex and returns the consts map
+// expected by bpf_skb_filter.h. Returns nil consts when devName is empty
+// (filter disabled at the BPF level by leaving filter_ifindex_included == 0).
+func deviceFilterConsts(devName string, exclude bool) (map[string]any, error) {
+	if devName == "" {
+		return nil, nil
+	}
+	iface, err := net.InterfaceByName(devName)
+	if err != nil {
+		return nil, fmt.Errorf("--device %q: %w", devName, err)
+	}
+	var excl uint32
+	if exclude {
+		excl = 1
+	}
+	return map[string]any{
+		"filter_ifindex_included": uint32(iface.Index),
+		"filter_ifindex_excluded": excl,
+	}, nil
 }
 
 func formatEvent(ev *dropPacketEvent) *types.DropWatchTracing {
@@ -158,7 +181,12 @@ func mainAction(c *cli.Context) error {
 		defer sockClient.End()
 	}
 
-	bpfObj, err := loadDropwatchBPF(c.String("bpf-path"), c.String("filter"))
+	consts, err := deviceFilterConsts(c.String("device"), c.Bool("device-exclude"))
+	if err != nil {
+		return fmt.Errorf("dropwatch: %w", err)
+	}
+
+	bpfObj, err := loadBPFWithFilter(c.String("bpf-path"), c.String("filter"), consts)
 	if err != nil {
 		return fmt.Errorf("dropwatch: load bpf: %w", err)
 	}
@@ -234,6 +262,14 @@ func main() {
 				Name:  "filter",
 				Usage: `tcpdump expression, e.g. "tcp and port 80"`,
 			},
+			&cli.StringFlag{
+				Name:  "device",
+				Usage: "interface to filter on (e.g. eth0); empty = all devices",
+			},
+			&cli.BoolFlag{
+				Name:  "device-exclude",
+				Usage: "treat --device as a blacklist (drop matching) instead of whitelist (pass matching)",
+			},
 			&cli.IntFlag{
 				Name:  "duration",
 				Usage: "run for N seconds then exit (0=forever)",

core/events/config.go

@@ -35,9 +35,8 @@ type Config struct {
 	}
 
 	Dropwatch struct {
-		ExcludedNeighInvalidate bool   `default:"true"`
-		Filter                  string `default:"tcp"`
-		ExcludeContainers       []string
+		Filter            string `default:"tcp"`
+		ExcludeContainers []string
 	}
 
 	Netdev struct {

core/events/dropwatch.go

@@ -179,10 +179,8 @@ func (c *dropWatchTracing) ignore(data *types.DropWatchTracing) bool {
 	// 4. neigh_timer_handler/ffffffff96d3a870
 	// 5. ...
 	// neigh_invalidate: ARP/neighbor table cleanup, filtered by config.
-	if cfg != nil && cfg.Dropwatch.ExcludedNeighInvalidate {
-		if len(stack) >= 3 && strings.HasPrefix(stack[2], "neigh_invalidate/") {
-			return true
-		}
+	if len(stack) >= 3 && strings.HasPrefix(stack[2], "neigh_invalidate/") {
+		return true
 	}
 
 	// stack:

huatuo-bamai.conf

@@ -401,19 +401,21 @@ BlackList = ["netdev_hw", "metax_gpu"]
     #
     # monitor packets dropped events in the Linux kernel.
     #
-    # - ExcludedNeighInvalidate
-    # Skip neigh_invalidate drop events when true.
-    # Default: true
-    #
-    # - Filter
+
+    # NOT supported (go-pcap limitation):
     # icmp, icmp6, igmp, vrrp, ah, esp, pim and other raw IP protocol keywords
     # CIDR prefix length notation (e.g. /24) is not supported for subnet masks;
     # use dotted-decimal notation instead (e.g. mask 255.255.255.0)
-    # Default: tcp
-    #
     [EventTracing.Dropwatch]
-        # ExcludedNeighInvalidate = true
-        # Filter = "tcp"
+        Filter = "tcp"
+
+    # event socket
+    #
+    # Unified event socket for receiving events from tools.
+    # Default: "" (disabled)
+    #
+    [EventTracing.EventSocket]
+        SocketPath = "/var/run/huatuo-bamai-event.sock"
 
     # ras
     #

pkg/types/dropwatch.go

@@ -21,8 +21,7 @@ import "huatuo-bamai/internal/packet"
 type DropWatchTracing struct {
 	ObservedTimestamp  string            `json:"observed_timestamp"`
 	Type               string            `json:"type"`
-	Reason             string            `json:"reason,omitempty"`
-	DropSource         string            `json:"drop_source"`
+	DropReason         string            `json:"drop_reason"`
 	Comm               string            `json:"comm"`
 	Pid                uint64            `json:"pid"`
 	ContainerID        string            `json:"container_id,omitempty"`