Merge pull request #13 from chentao-kernel/dev

ebpf: Add cachestat tool

Makefile

@@ -41,8 +41,8 @@ CLANG_COMPILE := $(CLANG) $(CFLAGS) $(VMLINUX) $(BPF_HEADER) -target bpf -D__TAR
 
 generate: export BPF_CLANG := $(CLANG)
 generate: export BPF_CFLAGS := $(CFLAGS)
-generate:
-	go generate $(EBPF_SRC)/uprobe/uprobe.go
+#generate:
+#	go generate $(EBPF_SRC)/uprobe/uprobe.go
 
 libbpf:
 	make -C $(LIBBPF_SRC) libbpf
@@ -53,6 +53,7 @@ ebpf.o: libbpf
 	$(CLANG_COMPILE) -c $(EBPF_SRC)/cpu/oncpu/oncpu.bpf.c -o $(EBPF_SRC)/cpu/oncpu/oncpu.bpf.o
 	$(CLANG_COMPILE) -c $(EBPF_SRC)/cpu/futexsnoop/futexsnoop.bpf.c -o $(EBPF_SRC)/cpu/futexsnoop/futexsnoop.bpf.o
 	$(CLANG_COMPILE) -c $(EBPF_SRC)/cpu/syscall/syscall.bpf.c -o $(EBPF_SRC)/cpu/syscall/syscall.bpf.o
+	$(CLANG_COMPILE) -c $(EBPF_SRC)/mem/cachestat/cachestat.bpf.c -o $(EBPF_SRC)/mem/cachestat/cachestat.bpf.o
 
 all: generate ebpf.o
 	@echo "go build $(TARGET)"

pkg/app/cmd.go

@@ -14,6 +14,7 @@ import (
 	"github.com/chentao-kernel/spycat/pkg/core"
 	"github.com/chentao-kernel/spycat/pkg/core/model"
 	"github.com/chentao-kernel/spycat/pkg/ebpf/cpu"
+	"github.com/chentao-kernel/spycat/pkg/ebpf/mem"
 	"github.com/chentao-kernel/spycat/pkg/log"
 	"github.com/fatih/color"
 	"github.com/pyroscope-io/pyroscope/pkg/cli"
@@ -63,6 +64,7 @@ func SubCmdInit(cmd *Cmd) {
 		newOnCpuSpyCmd(&cmd.cfg.ONCPU),
 		newFutexSnoopSpyCmd(&cmd.cfg.FUTEXSNOOP),
 		newSyscallSpyCmd(&cmd.cfg.SYSCALL),
+		newCacheStatSpyCmd(&cmd.cfg.CACHESTAT),
 		newVersionCmd(),
 	}
 
@@ -235,6 +237,33 @@ func newSyscallSpyCmd(cfg *config.SYSCALL) *cobra.Command {
 	return connectCmd
 }
 
+func newCacheStatSpyCmd(cfg *config.CACHESTAT) *cobra.Command {
+	vpr := newViper()
+
+	connectCmd := &cobra.Command{
+		Use:   "cachestat [flags]",
+		Short: "eBPF snoop cachestat",
+		Args:  cobra.NoArgs,
+
+		RunE: cli.CreateCmdRunFn(cfg, vpr, func(_ *cobra.Command, _ []string) error {
+			conf := &appspy.Config{
+				Exporter: cfg.Exporter,
+				Server:   cfg.Server,
+			}
+			return RunSpy(cfg, conf, func(cfg interface{}, buf chan *model.SpyEvent) core.BpfSpyer {
+				config, ok := cfg.(*config.CACHESTAT)
+				if ok {
+					return mem.NewCacheStatBpfSession(model.CacheStat, config, buf)
+				}
+				return nil
+			})
+		}),
+	}
+
+	cli.PopulateFlagSet(cfg, connectCmd.Flags(), vpr)
+	return connectCmd
+}
+
 func NewRootCmd(cfg *config.Config) *cobra.Command {
 	vpr := newViper()
 	rootCmd := &cobra.Command{

pkg/app/config/config.go

@@ -11,6 +11,7 @@ type Config struct {
 	OFFCPU     OFFCPU     `skip:"true" mapstructure:",squash"`
 	FUTEXSNOOP FUTEXSNOOP `skip:"true" mapstructure:",squash"`
 	SYSCALL    SYSCALL    `skip:"true" mapstructure:",squash"`
+	CACHESTAT  CACHESTAT  `skip:"true" mapstructure:",squash"`
 }
 
 type FUTEXSNOOP struct {
@@ -77,6 +78,17 @@ type ONCPU struct {
 	Exporter        string `def:"" desc:"data exporter: loki,pyroscoe,prometheus,disk,etc." mapstructure:"exporter"`
 }
 
+type CACHESTAT struct {
+	LogLevel   string        `def:"info" desc:"log level: debug|info|warn|error" mapstructure:"log-level"`
+	AppName    string        `def:"" desc:"application name used when uploading profiling data" mapstructure:"app-name"`
+	Server     string        `def:"http://localhost:4040" desc:"the server address" mapstructure:"server"`
+	UploadRate time.Duration `def:"30s" desc:"upload for the cachestat data. 30 means 30s upload the trace data from kernel" mapstructure:"upload-rate"`
+	Pid        int           `def:"-1" desc:"pid to trace, -1 to trace all pids" mapstructure:"pid"`
+	CacheType  int           `def:"0" desc:"cache type to trace, 0:read/write cache, 1:read cache, 2:write cache" mapstructure:"cache-type"`
+	BtfPath    string        `def:"" desc:"btf file path" mapstructure:"btf-path"`
+	Exporter   string        `def:"" desc:"data exporter: loki,pyroscoe,prometheus,disk,etc." mapstructure:"exporter"`
+}
+
 type NET struct {
 	LogLevel string `def:"info" desc:"log level: debug|info|warn|error" mapstructure:"log-level"`
 

pkg/component/detector/memdetector/memdetector.go

@@ -45,14 +45,63 @@ func (mem *MemDetector) Start() error {
 }
 
 func (mem *MemDetector) Stop() error {
+	close(mem.stopChan)
 	return nil
 }
 
 func (mem *MemDetector) Name() string {
 	return DetectorMemType
 }
 
+func (mem *MemDetector) formatCacheStat(e *model.SpyEvent) (*model.AttributeMap, error) {
+	labels := model.NewAttributeMap()
+	for i := 0; i < int(e.ParamsCnt); i++ {
+		userAttributes := e.UserAttributes[i]
+		switch {
+		case userAttributes.GetKey() == "read_size_m":
+			labels.AddIntValue(model.ReadSizeM, int64(userAttributes.GetUintValue()))
+		case userAttributes.GetKey() == "write_size_m":
+			labels.AddIntValue(model.WriteSizeM, int64(userAttributes.GetUintValue()))
+		case userAttributes.GetKey() == "pid":
+			labels.AddIntValue(model.Pid, int64(userAttributes.GetUintValue()))
+		case userAttributes.GetKey() == "comm":
+			labels.AddStringValue(model.Comm, string(userAttributes.GetValue()))
+		case userAttributes.GetKey() == "file":
+			labels.AddStringValue(model.File, string(userAttributes.GetValue()))
+		}
+	}
+	return labels, nil
+}
+
+func (mem *MemDetector) cachestatHandler(e *model.SpyEvent) (*model.DataBlock, error) {
+	labels, _ := mem.formatCacheStat(e)
+	val := e.GetUintUserAttribute("write_size")
+	metric := model.NewIntMetric(model.CacheStatMetricName, int64(val))
+	return model.NewDataBlock(model.CacheStat, labels, e.TimeStamp, metric), nil
+}
 func (mem *MemDetector) ProcessEvent(e *model.SpyEvent) error {
+	var dataBlock *model.DataBlock
+	var err error
+	switch e.Name {
+	case model.CacheStat:
+		dataBlock, err = mem.cachestatHandler(e)
+	default:
+		return nil
+	}
+	if err != nil {
+		return nil
+	}
+	if dataBlock == nil {
+		return nil
+	}
+	// next consumer is default processor
+	for _, con := range mem.consumers {
+		err := con.Consume(dataBlock)
+		if err != nil {
+			log.Loger.Error("consumer consume event failed:%v", err)
+			return err
+		}
+	}
 	return nil
 }
 
@@ -72,11 +121,12 @@ func (mem *MemDetector) ConsumeChanEvents() {
 }
 
 // 公共接口
-func (mem *MemDetector) ConsumeEvent(*model.SpyEvent) error {
+func (mem *MemDetector) ConsumeEvent(e *model.SpyEvent) error {
+	mem.eventChan <- e
 	return nil
 }
 
 // hard code
 func (mem *MemDetector) OwnedEvents() []string {
-	return []string{}
+	return []string{model.CacheStat}
 }

pkg/component/processor/processor.go

@@ -156,6 +156,8 @@ func (d *DefaultProcessor) Consume(data *model.DataBlock) error {
 	case model.Syscall:
 		fallthrough
 	case model.FutexSnoop:
+		fallthrough
+	case model.CacheStat:
 		err := d.consumer.Consume(data)
 		if err != nil {
 			log.Loger.Error("exporter consume data failed:%v", err)

pkg/core/model/const.go

@@ -15,13 +15,15 @@ const (
 	OnCpu      = "oncpu"
 	FutexSnoop = "futexsnoop"
 	Syscall    = "syscall"
+	CacheStat  = "cachestat"
 	OtherEvent = "other_event"
 )
 
 // for metric
 const (
 	OffCpuMetricName     = "offcpu_dur_ms"
 	FutexMaxUerCountName = "max_futex_user_cnt"
+	CacheStatMetricName  = "cachestat_read_size"
 )
 
 // for labels
@@ -73,6 +75,11 @@ const (
 	// for syscall
 	DurMs = "dur_ms"
 	DurUs = "dur_us"
+
+	// for cachestat
+	ReadSizeM  = "read_size_m"
+	WriteSizeM = "write_size_m"
+	File       = "file"
 )
 
 const (