Parse JFR directly into profile model

Author: r1viollet

analysis/analysis.go

@@ -661,13 +661,8 @@ func AnalyzePprofFile(r Reporter, pprofFile string, typedStacks TypedStacks, tes
 }
 
 // AnalyzeResults loads the expected_profile.json at jsonFilePath and asserts
-// every pprof file under pprofFolder matches it. Failures are reported via r.
-// Any *.jfr files in pprofFolder are converted to per-metric pprof files
-// before the analysis loop runs.
+// every profile file under pprofFolder matches it. Failures are reported via r.
 func AnalyzeResults(r Reporter, jsonFilePath string, pprofFolder string) {
-	// Convert any JFR files to pprof before the analysis loop.
-	convertJFRFiles(r, pprofFolder)
-
 	stackTestData, err := ReadJSONFile(jsonFilePath)
 	if err != nil {
 		r.Fatalf("Error opening file %s: %v", jsonFilePath, err)

analysis/jfr.go

@@ -1,110 +1,285 @@
 // Package analysis — JFR support.
 //
-// This file adds the ability to read JFR (Java Flight Recorder) files and
-// convert them to pprof profiles that the rest of the analysis pipeline can
-// consume.
-//
-// Flow:
-//
-//  1. At the start of AnalyzeResults, convertJFRFiles walks the output folder
-//     for any *.jfr files.
-//  2. Each JFR file is parsed with github.com/grafana/jfr-parser/pprof.
-//  3. Per-metric pprof profiles are written as <stem>_<metric>.pprof alongside
-//     the source JFR file (e.g. profile.jfr → profile_cpu.pprof).
-//  4. The normal pprof analysis loop then picks those up via the usual
-//     filename regex.
+// This file adds the ability to read JFR (Java Flight Recorder) files and map
+// JFR events directly into the analyzer's neutral ProfileSet.  The mapping from
+// event names to profile semantics intentionally lives here: prof-correctness is
+// the consumer that knows which JDK / Datadog profiler events should satisfy a
+// given expected_profile.json assertion.
 package analysis
 
 import (
-	"bytes"
 	"fmt"
-	"os"
-	"path/filepath"
+	"io"
 	"strings"
-	"time"
 
-	"github.com/google/pprof/profile"
-	jfrpprof "github.com/grafana/jfr-parser/pprof"
+	"github.com/grafana/jfr-parser/parser"
+	"github.com/grafana/jfr-parser/parser/types"
+	"github.com/grafana/jfr-parser/parser/types/def"
 )
 
-// parseJFR converts raw JFR bytes into a map of profile-name → pprof profile.
-func parseJFR(data []byte) (map[string]*profile.Profile, error) {
-	profiles, err := jfrpprof.ParseJFR(data, &jfrpprof.ParseInput{
-		StartTime: time.Unix(0, 0),
-		EndTime:   time.Unix(0, 0),
-		// Keep prof-correctness values sample-like (1 per CPU/wall event) while
-		// reusing jfr-parser's pprof conversion, which otherwise scales CPU/wall
-		// samples by 1e9/SampleRate.
-		SampleRate: 1_000_000_000,
-	}, nil)
-	if err != nil {
-		return nil, fmt.Errorf("jfr ParseJFR: %w", err)
-	}
-
-	result := make(map[string]*profile.Profile, len(profiles.Profiles))
-	for _, parsed := range profiles.Profiles {
-		data, err := parsed.Profile.MarshalVT()
+type jfrValue struct {
+	u64 uint64
+	str string
+	set bool
+}
+
+// FromJFR builds a ProfileSet directly from Java Flight Recorder events.
+func FromJFR(data []byte) (*ProfileSet, error) {
+	p := parser.NewParser(data, parser.Options{
+		SymbolProcessor: parser.ProcessSymbols,
+	})
+	ps := newProfileSet()
+
+	for {
+		event, err := p.ParseRawEvent()
+		if err == io.EOF {
+			break
+		}
 		if err != nil {
-			return nil, fmt.Errorf("jfr marshal profile %s: %w", parsed.Metric, err)
+			return nil, fmt.Errorf("jfr ParseRawEvent: %w", err)
+		}
+		if event.Type == nil {
+			continue
 		}
-		prof, err := profile.ParseData(data)
+
+		fields, err := decodeJFREventFields(event.Type, event.Data, &p.TypeMap)
 		if err != nil {
-			return nil, fmt.Errorf("jfr parse pprof profile %s: %w", parsed.Metric, err)
+			return nil, fmt.Errorf("jfr decode %s: %w", event.Type.Name, err)
 		}
-		for _, fn := range prof.Function {
-			fn.Name = strings.ReplaceAll(fn.Name, "/", ".")
+
+		switch event.Type.Name {
+		case "jdk.ExecutionSample", "datadog.ExecutionSample":
+			addJFRCPU(ps, p, fields)
 		}
-		result[jfrProfileName(parsed.Metric, prof)] = prof
 	}
-	return result, nil
+
+	return ps.finalize(), nil
 }
 
-// convertJFRFiles walks dir for *.jfr files and converts each to a set of
-// per-metric pprof files written into the same directory.
-// Output files are named <stem>_<metric>.pprof (e.g. profile_cpu.pprof).
-// Errors are non-fatal: they are logged through r and the function continues.
-func convertJFRFiles(r Reporter, dir string) {
-	entries, err := os.ReadDir(dir)
-	if err != nil {
-		r.Logf("convertJFRFiles: reading dir %s: %v", dir, err)
+func addJFRCPU(ps *ProfileSet, p *parser.Parser, fields map[string]jfrValue) {
+	stack := types.StackTraceRef(fields["stackTrace"].u64)
+	state := types.ThreadStateRef(fields["state"].u64)
+
+	// Match jfr-parser/pprof's CPU semantics: execution samples from sleeping
+	// threads do not count as CPU samples. If the state is absent, keep the
+	// sample rather than silently dropping producer-specific events.
+	if fields["state"].set {
+		if ts := p.GetThreadState(state); ts != nil && ts.Name == "STATE_SLEEPING" {
+			return
+		}
+	}
+	folded := foldJFRStack(p, stack)
+	if folded == "" {
 		return
 	}
-	for _, entry := range entries {
-		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".jfr") {
-			continue
+
+	val := int64(1)
+	if weight := fields["weight"]; weight.set && weight.u64 > 0 {
+		val = int64(weight.u64)
+	}
+
+	ps.add("cpu", StackSample{Stack: folded, Val: val, Labels: executionSampleLabels(fields)})
+}
+
+func executionSampleLabels(fields map[string]jfrValue) map[string][]string {
+	labels := map[string][]string{}
+	if spanID := fields["spanId"]; spanID.set && spanID.u64 != 0 {
+		labels[LabelSpanID] = []string{fmt.Sprintf("%d", spanID.u64)}
+	}
+	traceHi := fields["traceIdHi"]
+	traceLo := fields["traceIdLo"]
+	if (traceHi.set || traceLo.set) && (traceHi.u64 != 0 || traceLo.u64 != 0) {
+		labels[LabelTraceID] = []string{fmt.Sprintf("%016x%016x", traceHi.u64, traceLo.u64)}
+	}
+	return labels
+}
+
+func foldJFRStack(p *parser.Parser, stackRef types.StackTraceRef) string {
+	st := p.GetStacktrace(stackRef)
+	if st == nil || len(st.Frames) == 0 {
+		return ""
+	}
+	// JFR frames are leaf-first.  The analyzer uses root-first folded stacks,
+	// matching FromPprof and the historical expected_profile.json captures.
+	frames := make([]string, 0, len(st.Frames))
+	for i := len(st.Frames) - 1; i >= 0; i-- {
+		name := jfrFrameName(p, st.Frames[i].Method)
+		if name != "" {
+			frames = append(frames, name)
 		}
-		jfrPath := filepath.Join(dir, entry.Name())
-		stem := strings.TrimSuffix(entry.Name(), ".jfr")
+	}
+	return strings.Join(frames, ";")
+}
+
+func jfrFrameName(p *parser.Parser, methodRef types.MethodRef) string {
+	m := p.GetMethod(methodRef)
+	if m == nil {
+		return ""
+	}
+	methodName := p.GetSymbolString(m.Name)
+	cls := p.GetClass(m.Type)
+	if cls == nil {
+		return methodName
+	}
+	clsName := strings.ReplaceAll(p.GetSymbolString(cls.Name), "/", ".")
+	return clsName + "." + methodName
+}
 
-		data, err := os.ReadFile(jfrPath)
+func decodeJFREventFields(class *def.Class, data []byte, typeMap *def.TypeMap) (map[string]jfrValue, error) {
+	pos := 0
+	out := map[string]jfrValue{}
+	for _, field := range class.Fields {
+		vals, next, err := decodeJFRField(data, pos, field, typeMap)
 		if err != nil {
-			r.Logf("convertJFRFiles: reading %s: %v", jfrPath, err)
-			continue
+			return nil, fmt.Errorf("field %s: %w", field.Name, err)
+		}
+		pos = next
+		if len(vals) > 0 {
+			out[field.Name] = vals[0]
+		}
+	}
+	return out, nil
+}
+
+func decodeJFRField(data []byte, pos int, field def.Field, typeMap *def.TypeMap) ([]jfrValue, int, error) {
+	count := 1
+	var err error
+	if field.Array {
+		var n uint64
+		n, pos, err = readJFRVar(data, pos, 32)
+		if err != nil {
+			return nil, pos, err
 		}
+		count = int(n)
+	}
 
-		profiles, err := parseJFR(data)
+	vals := make([]jfrValue, 0, count)
+	for i := 0; i < count; i++ {
+		var v jfrValue
+		v, pos, err = decodeJFRScalar(data, pos, field.Type, field.ConstantPool, typeMap)
 		if err != nil {
-			r.Logf("convertJFRFiles: parsing %s: %v", jfrPath, err)
-			continue
+			return nil, pos, err
 		}
-		if len(profiles) == 0 {
-			r.Logf("convertJFRFiles: no profiles found in %s", jfrPath)
-			continue
+		vals = append(vals, v)
+	}
+	return vals, pos, nil
+}
+
+func decodeJFRScalar(data []byte, pos int, typeID def.TypeID, constantPool bool, typeMap *def.TypeMap) (jfrValue, int, error) {
+	if constantPool {
+		v, next, err := readJFRVar(data, pos, 64)
+		return jfrValue{u64: v, set: true}, next, err
+	}
+
+	switch typeID {
+	case typeMap.T_STRING:
+		s, next, err := readJFRString(data, pos, typeMap)
+		return jfrValue{str: s, set: true}, next, err
+	case typeMap.T_INT:
+		v, next, err := readJFRVar(data, pos, 32)
+		return jfrValue{u64: v, set: true}, next, err
+	case typeMap.T_LONG:
+		v, next, err := readJFRVar(data, pos, 64)
+		return jfrValue{u64: v, set: true}, next, err
+	case typeMap.T_SHORT:
+		v, next, err := readJFRVar(data, pos, 16)
+		return jfrValue{u64: v, set: true}, next, err
+	case typeMap.T_BOOLEAN:
+		if pos >= len(data) {
+			return jfrValue{}, pos, io.ErrUnexpectedEOF
+		}
+		return jfrValue{u64: uint64(data[pos]), set: true}, pos + 1, nil
+	case typeMap.T_FLOAT:
+		v, next, err := readJFRVar(data, pos, 32)
+		return jfrValue{u64: v, set: true}, next, err
+	}
+
+	cls := typeMap.IDMap[typeID]
+	if cls == nil || len(cls.Fields) == 0 {
+		return jfrValue{}, pos, fmt.Errorf("unknown type %d", typeID)
+	}
+	for _, nested := range cls.Fields {
+		_, next, err := decodeJFRField(data, pos, nested, typeMap)
+		if err != nil {
+			return jfrValue{}, pos, err
 		}
+		pos = next
+	}
+	return jfrValue{set: true}, pos, nil
+}
 
-		for metric, prof := range profiles {
-			outPath := filepath.Join(dir, stem+"_"+metric+".pprof")
-			var buf bytes.Buffer
-			if err := prof.Write(&buf); err != nil {
-				r.Logf("convertJFRFiles: serialising %s metric %s: %v", entry.Name(), metric, err)
-				continue
+func readJFRVar(data []byte, pos int, bits uint) (uint64, int, error) {
+	v := uint64(0)
+	maxShift := bits
+	if bits == 64 {
+		maxShift = 56
+	}
+	for shift := uint(0); ; shift += 7 {
+		if (bits < 64 && shift >= maxShift) || (bits == 64 && shift > maxShift) {
+			return 0, pos, def.ErrIntOverflow
+		}
+		if pos >= len(data) {
+			return 0, pos, io.ErrUnexpectedEOF
+		}
+		b := data[pos]
+		pos++
+		if bits == 64 && shift == 56 {
+			v |= uint64(b&0xff) << shift
+			break
+		}
+		v |= uint64(b&0x7f) << shift
+		if b < 0x80 {
+			break
+		}
+	}
+	return v, pos, nil
+}
+
+func readJFRString(data []byte, pos int, typeMap *def.TypeMap) (string, int, error) {
+	if pos >= len(data) {
+		return "", pos, io.ErrUnexpectedEOF
+	}
+	kind := data[pos]
+	pos++
+	switch kind {
+	case 0, 1:
+		return "", pos, nil
+	case 3, 5:
+		n, next, err := readJFRVar(data, pos, 32)
+		if err != nil {
+			return "", next, err
+		}
+		pos = next
+		end := pos + int(n)
+		if end < pos || end > len(data) {
+			return "", pos, io.ErrUnexpectedEOF
+		}
+		bs := data[pos:end]
+		if kind == 5 {
+			decoded, err := typeMap.ISO8859_1Decoder.Bytes(bs)
+			if err != nil {
+				return "", pos, err
 			}
-			if err := os.WriteFile(outPath, buf.Bytes(), 0644); err != nil {
-				r.Logf("convertJFRFiles: writing %s: %v", outPath, err)
-				continue
+			bs = decoded
+		}
+		return string(bs), end, nil
+	case 4:
+		n, next, err := readJFRVar(data, pos, 32)
+		if err != nil {
+			return "", next, err
+		}
+		pos = next
+		runes := make([]rune, int(n))
+		for i := range runes {
+			c, next, err := readJFRVar(data, pos, 32)
+			if err != nil {
+				return "", next, err
 			}
-			r.Logf("Converted JFR %s → %s (%d samples)",
-				entry.Name(), filepath.Base(outPath), len(prof.Sample))
+			pos = next
+			runes[i] = rune(c)
 		}
+		return string(runes), pos, nil
+	default:
+		return "", pos, fmt.Errorf("unknown string type %d", kind)
 	}
 }