Merge pull request #101 from pyroscope-io/ebpf3

* adds eBPF spy
* adds a new new connect command

.gitignore

@@ -10,6 +10,7 @@ pkged.go
 /scripts/packages/
 .eslintcache
 *.pyc
+*.internal
 .idea
 .vscode
 vendor/

Dockerfile

@@ -40,9 +40,9 @@ RUN apk add --no-cache make
 
 WORKDIR /opt/pyroscope
 
-COPY package.json yarn.lock babel.config.js .eslintrc .eslintignore Makefile ./
 COPY scripts ./scripts
 COPY webapp ./webapp
+COPY package.json yarn.lock babel.config.js .eslintrc .eslintignore Makefile ./
 
 ARG EXTRA_METADATA=""
 RUN EXTRA_METADATA=$EXTRA_METADATA make assets-release
@@ -76,7 +76,7 @@ COPY scripts ./scripts
 COPY go.mod go.sum pyroscope.go ./
 COPY Makefile ./
 
-RUN EMBEDDED_ASSETS_DEPS="" EXTRA_LDFLAGS="-linkmode external -extldflags \"-static\"" make build-release
+RUN EMBEDDED_ASSETS_DEPS="" EXTRA_LDFLAGS="-linkmode external -extldflags '-static'" make build-release
 
 
 #   __ _             _   _

Makefile

@@ -1,11 +1,16 @@
 GOBUILD=go build -trimpath
 
-ENABLED_SPIES ?= "rbspy,pyspy"
 ifeq ("$(shell go env GOARCH || true)", "arm64")
 	# this makes it work better on M1 machines
 	GODEBUG=asyncpreemptoff=1
 endif
 
+ifeq ("$(shell go env GOOS || true)", "linux")
+	ENABLED_SPIES ?= "ebpfspy,rbspy,pyspy"
+else
+	ENABLED_SPIES ?= "rbspy,pyspy"
+endif
+
 EMBEDDED_ASSETS ?= ""
 EMBEDDED_ASSETS_DEPS ?= "assets-release"
 EXTRA_LDFLAGS ?= ""

pkg/agent/debugspy/debugspy.go

@@ -23,9 +23,9 @@ func (s *DebugSpy) Stop() error {
 }
 
 // Snapshot calls callback function with stack-trace or error.
-func (s *DebugSpy) Snapshot(cb func([]byte, error)) {
+func (s *DebugSpy) Snapshot(cb func([]byte, uint64, error)) {
 	stacktrace := fmt.Sprintf("debug_%d;debug", s.pid)
-	cb([]byte(stacktrace), nil)
+	cb([]byte(stacktrace), 1, nil)
 }
 
 func init() {

pkg/agent/ebpfspy/ebpfspy.go

@@ -0,0 +1,69 @@
+// +build ebpfspy
+
+// Package ebpfspy provides integration with Linux eBPF. It calls profile.py from BCC tools:
+//   https://github.com/iovisor/bcc/blob/master/tools/profile.py
+// TODO: At some point we might extract the part that starts another process because it has good potential to be reused by similar profiling tools.
+package ebpfspy
+
+import (
+	"sync"
+
+	"github.com/pyroscope-io/pyroscope/pkg/agent/spy"
+)
+
+type EbpfSpy struct {
+	resetMutex sync.Mutex
+	reset      bool
+	stop       bool
+
+	profilingSession *session
+
+	stopCh chan struct{}
+}
+
+func Start(pid int) (spy.Spy, error) {
+	s := newSession(pid)
+	err := s.Start()
+	if err != nil {
+		return nil, err
+	}
+	return &EbpfSpy{
+		profilingSession: s,
+		stopCh:           make(chan struct{}),
+	}, nil
+}
+
+func (s *EbpfSpy) Stop() error {
+	s.stop = true
+	<-s.stopCh
+	return nil
+}
+
+func (s *EbpfSpy) Snapshot(cb func([]byte, uint64, error)) {
+	s.resetMutex.Lock()
+	defer s.resetMutex.Unlock()
+
+	if !s.reset {
+		return
+	}
+
+	s.reset = false
+	s.profilingSession.Reset(func(name []byte, v uint64) {
+		cb(name, v, nil)
+	})
+	if s.stop {
+		s.stopCh <- struct{}{}
+		s.profilingSession.Stop()
+	}
+}
+
+func (s *EbpfSpy) Reset() {
+	s.resetMutex.Lock()
+	defer s.resetMutex.Unlock()
+
+	s.reset = true
+}
+
+func init() {
+	spy.RegisterSpy("ebpfspy", Start)
+}

pkg/agent/ebpfspy/placeholder.go

@@ -0,0 +1,3 @@
+// +build !ebpfspy
+
+package ebpfspy

pkg/agent/ebpfspy/session.go

@@ -0,0 +1,100 @@
+// +build ebpfspy
+
+// Package ebpfspy provides integration with Linux eBPF.
+package ebpfspy
+
+import (
+	"fmt"
+	"os/exec"
+	"strconv"
+	"sync"
+	"syscall"
+
+	"github.com/pyroscope-io/pyroscope/pkg/convert"
+	"github.com/pyroscope-io/pyroscope/pkg/util/file"
+)
+
+type line struct {
+	name []byte
+	val  int
+}
+
+type session struct {
+	pid int
+
+	cmd *exec.Cmd
+	ch  chan line
+
+	stopMutex sync.Mutex
+	stop      bool
+}
+
+const helpURL = "https://github.com/iovisor/bcc/blob/master/INSTALL.md"
+
+var command = "/usr/share/bcc/tools/profile"
+
+// TODO: make these configurable
+var commandArgs = []string{"-F", "100", "-f", "11"}
+
+func newSession(pid int) *session {
+	return &session{pid: pid}
+}
+
+func (s *session) Start() error {
+	if !file.Exists(command) {
+		return fmt.Errorf("Could not find profile.py at '%s'. Visit %s for instructions on how to install it", command, helpURL)
+	}
+
+	args := commandArgs
+	if s.pid != -1 {
+		args = append(commandArgs, "-p", strconv.Itoa(s.pid))
+	}
+
+	s.cmd = exec.Command(command, args...)
+	stdout, err := s.cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+
+	s.ch = make(chan line)
+
+	go func() {
+		convert.ParseGroups(stdout, func(name []byte, val int) {
+			s.ch <- line{
+				name: name,
+				val:  val,
+			}
+		})
+		stdout.Close()
+		close(s.ch)
+	}()
+
+	err = s.cmd.Start()
+	return err
+}
+
+func (s *session) Reset(cb func([]byte, uint64)) error {
+	s.cmd.Process.Signal(syscall.SIGINT)
+
+	for v := range s.ch {
+		cb(v.name, uint64(v.val))
+	}
+	s.cmd.Wait()
+
+	s.stopMutex.Lock()
+	defer s.stopMutex.Unlock()
+
+	if s.stop {
+		return nil
+	} else {
+		return s.Start()
+	}
+}
+
+func (s *session) Stop() error {
+	s.stopMutex.Lock()
+	defer s.stopMutex.Unlock()
+
+	s.stop = true
+	return nil
+}

pkg/agent/gospy/gospy.go

@@ -33,14 +33,15 @@ func (*GoSpy) Stop() error {
 }
 
 // Snapshot calls callback function with stack-trace or error.
-func (s *GoSpy) Snapshot(cb func([]byte, error)) {
+func (s *GoSpy) Snapshot(cb func([]byte, uint64, error)) {
 	if s.selfFrame == nil {
 		// Determine the runtime.Frame of this func so we can hide it from our
 		// profiling output.
 		rpc := make([]uintptr, 1)
 		n := runtime.Callers(1, rpc)
 		if n < 1 {
-			panic("could not determine selfFrame")
+			// TODO: log the error
+			return
 		}
 		selfFrame, _ := runtime.CallersFrames(rpc).Next()
 		s.selfFrame = &selfFrame
@@ -60,7 +61,7 @@ func (s *GoSpy) Snapshot(cb func([]byte, error)) {
 				}
 			}
 			if !shouldExclude {
-				cb([]byte(stackStr), nil)
+				cb([]byte(stackStr), 1, nil)
 			}
 		}
 	}

pkg/agent/pyspy/pyspy.go

@@ -64,12 +64,12 @@ func (s *PySpy) Stop() error {
 }
 
 // Snapshot calls callback function with stack-trace or error.
-func (s *PySpy) Snapshot(cb func([]byte, error)) {
+func (s *PySpy) Snapshot(cb func([]byte, uint64, error)) {
 	r := C.pyspy_snapshot(C.int(s.pid), s.dataPtr, C.int(bufferLength), s.errorPtr, C.int(bufferLength))
 	if r < 0 {
-		cb(nil, errors.New(string(s.errorBuf[:-r])))
+		cb(nil, 0, errors.New(string(s.errorBuf[:-r])))
 	} else {
-		cb(s.dataBuf[:r], nil)
+		cb(s.dataBuf[:r], 1, nil)
 	}
 }
 

pkg/agent/rbspy/rbspy.go

@@ -64,12 +64,12 @@ func (s *RbSpy) Stop() error {
 }
 
 // Snapshot calls callback function with stack-trace or error.
-func (s *RbSpy) Snapshot(cb func([]byte, error)) {
+func (s *RbSpy) Snapshot(cb func([]byte, uint64, error)) {
 	r := C.rbspy_snapshot(C.int(s.pid), s.dataPtr, C.int(bufferLength), s.errorPtr, C.int(bufferLength))
 	if r < 0 {
-		cb(nil, errors.New(string(s.errorBuf[:-r])))
+		cb(nil, 0, errors.New(string(s.errorBuf[:-r])))
 	} else {
-		cb(s.dataBuf[:r], nil)
+		cb(s.dataBuf[:r], 1, nil)
 	}
 }
 

pkg/agent/session.go

@@ -8,6 +8,7 @@ import (
 	//   That's why we do a blank import here and then packages themselves register with the rest of the code.
 
 	_ "github.com/pyroscope-io/pyroscope/pkg/agent/debugspy"
+	_ "github.com/pyroscope-io/pyroscope/pkg/agent/ebpfspy"
 	_ "github.com/pyroscope-io/pyroscope/pkg/agent/gospy"
 	_ "github.com/pyroscope-io/pyroscope/pkg/agent/pyspy"
 	_ "github.com/pyroscope-io/pyroscope/pkg/agent/rbspy"
@@ -55,19 +56,27 @@ func (ps *ProfileSession) takeSnapshots() {
 	for {
 		select {
 		case <-ticker.C:
-			if ps.isDueForReset() {
-				ps.reset()
+			isdueToReset := ps.isDueForReset()
+			if isdueToReset {
+				for _, s := range ps.spies {
+					if sr, ok := s.(spy.Resettable); ok {
+						sr.Reset()
+					}
+				}
 			}
-			for _, spy := range ps.spies {
-				spy.Snapshot(func(stack []byte, err error) {
+			for _, s := range ps.spies {
+				s.Snapshot(func(stack []byte, v uint64, err error) {
 					if stack != nil && len(stack) > 0 {
 						ps.trieMutex.Lock()
 						defer ps.trieMutex.Unlock()
 
-						ps.trie.Insert(stack, 1, true)
+						ps.trie.Insert(stack, v, true)
 					}
 				})
 			}
+			if isdueToReset {
+				ps.reset()
+			}
 		case <-ps.stopCh:
 			ticker.Stop()
 			for _, spy := range ps.spies {

pkg/agent/spy/spy.go

@@ -7,7 +7,10 @@ import (
 
 type Spy interface {
 	Stop() error
-	Snapshot(cb func([]byte, error))
+	Snapshot(cb func([]byte, uint64, error))
+}
+type Resettable interface {
+	Reset()
 }
 
 type spyIntitializer func(pid int) (Spy, error)