WIP pyroscope java

component/discovery/process/discover.go

@@ -18,13 +18,15 @@ const (
 	labelProcessID          = "__process_pid__"
 	labelProcessExe         = "__meta_process_exe"
 	labelProcessCwd         = "__meta_process_cwd"
+	labelProcessCommandline = "__meta_process_commandline"
 	labelProcessContainerID = "__container_id__"
 )
 
 type process struct {
 	pid         string
 	exe         string
 	cwd         string
+	commandline string
 	containerID string
 }
 
@@ -76,6 +78,11 @@ func discover(l log.Logger) ([]process, error) {
 			loge(int(p.Pid), err)
 			continue
 		}
+		commandline, err := p.Cmdline()
+		if err != nil {
+			loge(int(p.Pid), err)
+			continue
+		}
 
 		containerID, err := getLinuxProcessContainerID(l, spid)
 		if err != nil {
@@ -86,9 +93,10 @@ func discover(l log.Logger) ([]process, error) {
 			pid:         spid,
 			exe:         exe,
 			cwd:         cwd,
+			commandline: commandline,
 			containerID: containerID,
 		})
-		_ = level.Debug(l).Log("msg", "found process", "pid", p.Pid, "exe", exe, "cwd", cwd, "container_id", containerID)
+		//_ = level.Debug(l).Log("msg", "found process", "pid", p.Pid, "exe", exe, "cwd", cwd, "container_id", containerID)
 	}
 
 	return res, nil

component/pyroscope/java/asprof/asprof.go

@@ -0,0 +1,128 @@
+package asprof
+
+import (
+	"bytes"
+	_ "embed"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"sync"
+
+	"k8s.io/utils/path"
+)
+
+// option1: embed the tar.gz file todo make this a module outside of agent repo
+// option2: distribute the tar.gz file with the agent docker image
+
+type Profiler struct {
+	tmpDir     string
+	unpackOnce sync.Once
+
+	mutex       sync.Mutex
+	unpackError error
+}
+
+func NewProfiler(tmpDir string) *Profiler {
+	return &Profiler{tmpDir: tmpDir}
+}
+
+type Distribution struct {
+	targz        []byte
+	fname        string
+	version      int
+	extractedDir string
+}
+
+const tmpDirMarker = "grafana-agent-asprof"
+
+func (d *Distribution) AsprofPath() string {
+	if d.version < 300 {
+		return filepath.Join(d.extractedDir, "bin/profiler.sh")
+	}
+	return filepath.Join(d.extractedDir, "bin/asprof")
+}
+
+func (p *Profiler) Extract() error {
+	p.unpackOnce.Do(func() {
+		for _, d := range AllDistributions() {
+			err := d.Extract(p.tmpDir, tmpDirMarker)
+			if err != nil {
+				p.unpackError = err
+				break
+			}
+		}
+	})
+	return p.unpackError
+}
+
+func (p *Profiler) Execute(dist *Distribution, argv []string) (string, string, error) {
+	stdout := bytes.NewBuffer(nil)
+	stderr := bytes.NewBuffer(nil)
+
+	exe := dist.AsprofPath()
+	cmd := exec.Command(exe, argv...)
+	cmd.Stdout = stdout
+	cmd.Stderr = stderr
+	err := cmd.Start()
+	if err != nil {
+		return stdout.String(), stderr.String(), fmt.Errorf("asprof failed to start %s: %w", exe, err)
+	}
+	err = cmd.Wait()
+	if err != nil {
+		return stdout.String(), stderr.String(), fmt.Errorf("asprof failed to run %s: %w", exe, err)
+	}
+	return stdout.String(), stderr.String(), nil
+}
+
+func (p *Profiler) CopyLib(dist *Distribution, pid int) error {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	src := dist.LibPath()
+
+	libBytes, err := os.ReadFile(src)
+	if err != nil {
+		return err
+	}
+
+	dst := ProcessPath(dist.LibPath(), pid)
+	targetExists, err := path.Exists(path.CheckSymlinkOnly, dst)
+	if err != nil {
+		return err
+	}
+	if targetExists {
+		targetLibBytes, err := os.ReadFile(dst)
+		if err != nil {
+			return err
+		}
+		if !bytes.Equal(libBytes, targetLibBytes) {
+			return fmt.Errorf("file %s already exists and is different", dst)
+		}
+		return nil
+	} else {
+		fd, err := os.OpenFile(dst, os.O_CREATE|os.O_EXCL, 0644)
+		if err != nil {
+			return fmt.Errorf("failed to create file %s: %w", dst, err)
+		}
+		defer fd.Close()
+		//todo check fd was not manipulated with symlinks
+		n, err := fd.Write(libBytes)
+		if err != nil {
+			return fmt.Errorf("failed to write to file %s: %w", dst, err)
+		}
+		if n != len(libBytes) {
+			return fmt.Errorf("failed to write to file %s %d", dst, n)
+		}
+		return nil
+	}
+}
+
+type ProcFile struct {
+	Path string
+	PID  int
+}
+
+func (f *ProcFile) ProcRootPath() string {
+	return filepath.Join("/proc", strconv.Itoa(f.PID), "root", f.Path)
+}

component/pyroscope/java/asprof/asprof_amd64.go

@@ -0,0 +1,18 @@
+//go:build linux && amd64
+
+package asprof
+
+import _ "embed"
+
+//go:embed async-profiler-3.0-ea-linux-x64.tar.gz
+var glibcDistribution []byte
+var glibcDistributionName = "async-profiler-3.0-ea-linux-x64"
+
+// todo
+var muslDistribution []byte
+var muslDistributionName = "TODO"
+
+func (p *Profiler) TargetLibPath(dist *Distribution, pid int) string {
+	f := ProcFile{Path: dist.LibPath(), PID: pid}
+	return f.ProcRootPath()
+}

component/pyroscope/java/asprof/asprof_arm64.go

@@ -0,0 +1,16 @@
+//go:build linux && arm64
+
+package asprof
+
+import _ "embed"
+
+var glibcDistribution []byte
+var glibcDistributionName = "TODO"
+
+var muslDistribution []byte
+var muslDistributionName = "TODO"
+
+func (p *Profiler) TargetLibPath(dist *Distribution, pid int) string {
+	f := ProcFile{Path: dist.LibPath(), PID: pid}
+	return f.ProcRootPath()
+}

component/pyroscope/java/asprof/asprof_darwin.go

@@ -0,0 +1,33 @@
+//go:build darwin
+
+package asprof
+
+import (
+	_ "embed"
+	"path/filepath"
+)
+
+//go:embed async-profiler-3.0-ea-macos-arm64.tar.gz
+var distribution []byte
+
+var macDist = &Distribution{
+	targz:   distribution,
+	fname:   "async-profiler-3.0-ea-macos-arm64.tar.gz",
+	version: 300,
+}
+
+func AllDistributions() []*Distribution {
+	return []*Distribution{macDist}
+}
+
+func DistributionForProcess(pid int) *Distribution {
+	return macDist
+}
+
+func (d *Distribution) LibPath() string {
+	return filepath.Join(d.extractedDir, "lib/libasyncProfiler.dylib")
+}
+
+func ProcessPath(path string, pid int) string {
+	return path
+}

component/pyroscope/java/asprof/asprof_test.go

@@ -0,0 +1,74 @@
+//go:build darwin || linux
+
+package asprof
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDistributionExtract(t *testing.T) {
+	d := *DistributionForProcess(os.Getpid())
+	tmpDir := t.TempDir()
+	t.Log(tmpDir)
+
+	err := d.Extract(tmpDir, tmpDirMarker)
+	assert.NoError(t, err)
+	assert.NotEmpty(t, d.extractedDir)
+	assert.FileExists(t, d.AsprofPath())
+	assert.FileExists(t, d.LibPath())
+	libStat1, err := os.Stat(d.LibPath())
+	assert.NoError(t, err)
+
+	d = *DistributionForProcess(os.Getpid()) // extracting second time should just verify
+	err = d.Extract(tmpDir, tmpDirMarker)
+	assert.NoError(t, err)
+	assert.NotEmpty(t, d.extractedDir)
+	assert.FileExists(t, filepath.Join(d.AsprofPath()))
+	assert.FileExists(t, filepath.Join(d.LibPath()))
+	libStat2, err := os.Stat(d.LibPath())
+	require.NoError(t, err)
+	require.Equal(t, libStat1.ModTime(), libStat2.ModTime())
+
+	file, err := os.OpenFile(d.AsprofPath(), os.O_RDWR, 0)
+	require.NoError(t, err)
+	defer file.Close()
+	file.Write([]byte("hello")) //modify binary
+	file.Close()
+
+	d = *DistributionForProcess(os.Getpid()) // extracting second time should just verify
+	err = d.Extract(tmpDir, tmpDirMarker)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "already exists and is different")
+	assert.Empty(t, d.extractedDir)
+
+}
+
+func TestDistributionExtractRace(t *testing.T) {
+	tmpDir := t.TempDir()
+	race = func(stage, extra string) {
+		switch stage {
+		case "mkdir dist":
+			err := os.MkdirAll(filepath.Join(tmpDir, extra), 0755)
+			assert.NoError(t, err)
+		default:
+			t.Fatal("unexpected stage")
+		}
+	}
+	defer func() {
+		race = func(stage, extra string) {
+
+		}
+	}()
+	d := *DistributionForProcess(os.Getpid())
+	t.Log(tmpDir)
+	err := d.Extract(tmpDir, tmpDirMarker)
+	assert.Error(t, err)
+	assert.Empty(t, d.extractedDir)
+	assert.NoFileExists(t, filepath.Join(d.AsprofPath()))
+	assert.NoFileExists(t, filepath.Join(d.LibPath()))
+}