Add __needs_symbolization__ metadata label & made debuginfo store non-optional

Author: marcsanmi

pkg/experiment/block/metadata/metadata_labels.go

@@ -17,8 +17,9 @@ import (
 // TODO(kolesnikovae): LabelBuilder pool.
 
 const (
-	LabelNameTenantDataset     = "__tenant_dataset__"
-	LabelValueDatasetTSDBIndex = "dataset_tsdb_index"
+	LabelNameTenantDataset      = "__tenant_dataset__"
+	LabelValueDatasetTSDBIndex  = "dataset_tsdb_index"
+	LabelNameNeedsSymbolization = "__needs_symbolization__"
 )
 
 type LabelBuilder struct {

pkg/experiment/ingester/segment.go

@@ -110,7 +110,7 @@ func (sh *shard) flushSegment(ctx context.Context) {
 		if s.debuginfo.movedHeads > 0 {
 			_ = level.Debug(s.logger).Log("msg",
 				"writing segment block done",
-				"heads-count", len(s.heads),
+				"heads-count", len(s.datasets),
 				"heads-moved-count", s.debuginfo.movedHeads,
 				"inflight-duration", s.debuginfo.waitInflight,
 				"flush-heads-duration", s.debuginfo.flushHeadsDuration,
@@ -198,7 +198,7 @@ func (sw *segmentsWriter) newSegment(sh *shard, sk shardKey, sl log.Logger) *seg
 	s := &segment{
 		logger:   log.With(sl, "segment-id", id.String()),
 		ulid:     id,
-		heads:    make(map[datasetKey]dataset),
+		datasets: make(map[datasetKey]*dataset),
 		sw:       sw,
 		sh:       sh,
 		shard:    sk,
@@ -211,7 +211,7 @@ func (sw *segmentsWriter) newSegment(sh *shard, sk shardKey, sl log.Logger) *seg
 func (s *segment) flush(ctx context.Context) (err error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "segment.flush", opentracing.Tags{
 		"block_id": s.ulid.String(),
-		"datasets": len(s.heads),
+		"datasets": len(s.datasets),
 		"shard":    s.shard,
 	})
 	defer span.Finish()
@@ -273,7 +273,7 @@ func (s *segment) flushBlock(stream flushStream) ([]byte, *metastorev1.BlockMeta
 		f := stream.At()
 		svc, err := concatSegmentHead(f, w, stringTable)
 		if err != nil {
-			level.Error(s.logger).Log("msg", "failed to concat segment head", "err", err)
+			level.Error(s.logger).Log("msg", "failed to concat segment dataset", "err", err)
 			continue
 		}
 		meta.MinTime = min(meta.MinTime, svc.MinTime)
@@ -339,6 +339,10 @@ func concatSegmentHead(f *headFlush, w *writerOffset, s *metadata.StringTable) (
 		lb.WithLabelSet(model.LabelNameServiceName, f.head.key.service, model.LabelNameProfileType, profileType)
 	}
 
+	if f.head.needsSymbolization {
+		lb.WithLabelSet(metadata.LabelNameNeedsSymbolization, "true")
+	}
+
 	// Other optional labels:
 	// lb.WithLabelSet("label_name", "label_value", ...)
 	ds.Labels = lb.Build()
@@ -347,8 +351,8 @@ func concatSegmentHead(f *headFlush, w *writerOffset, s *metadata.StringTable) (
 }
 
 func (s *segment) flushHeads(ctx context.Context) flushStream {
-	heads := maps.Values(s.heads)
-	slices.SortFunc(heads, func(a, b dataset) int {
+	heads := maps.Values(s.datasets)
+	slices.SortFunc(heads, func(a, b *dataset) int {
 		return a.key.compare(b.key)
 	})
 
@@ -363,15 +367,15 @@ func (s *segment) flushHeads(ctx context.Context) flushStream {
 			defer close(f.done)
 			flushed, err := s.flushHead(ctx, f.head)
 			if err != nil {
-				level.Error(s.logger).Log("msg", "failed to flush head", "err", err)
+				level.Error(s.logger).Log("msg", "failed to flush dataset", "err", err)
 				return
 			}
 			if flushed == nil {
-				level.Debug(s.logger).Log("msg", "skipping nil head")
+				level.Debug(s.logger).Log("msg", "skipping nil dataset")
 				return
 			}
 			if flushed.Meta.NumSamples == 0 {
-				level.Debug(s.logger).Log("msg", "skipping empty head")
+				level.Debug(s.logger).Log("msg", "skipping empty dataset")
 				return
 			}
 			f.flushed = flushed
@@ -402,24 +406,24 @@ func (s *flushStream) Next() bool {
 	return false
 }
 
-func (s *segment) flushHead(ctx context.Context, e dataset) (*memdb.FlushedHead, error) {
+func (s *segment) flushHead(ctx context.Context, e *dataset) (*memdb.FlushedHead, error) {
 	th := time.Now()
 	flushed, err := e.head.Flush(ctx)
 	if err != nil {
 		s.sw.metrics.flushServiceHeadDuration.WithLabelValues(s.sshard, e.key.tenant).Observe(time.Since(th).Seconds())
 		s.sw.metrics.flushServiceHeadError.WithLabelValues(s.sshard, e.key.tenant).Inc()
-		return nil, fmt.Errorf("failed to flush head : %w", err)
+		return nil, fmt.Errorf("failed to flush dataset : %w", err)
 	}
 	s.sw.metrics.flushServiceHeadDuration.WithLabelValues(s.sshard, e.key.tenant).Observe(time.Since(th).Seconds())
 	level.Debug(s.logger).Log(
-		"msg", "flushed head",
+		"msg", "flushed dataset",
 		"tenant", e.key.tenant,
 		"service", e.key.service,
 		"profiles", flushed.Meta.NumProfiles,
 		"profiletypes", fmt.Sprintf("%v", flushed.Meta.ProfileTypeNames),
 		"mintime", flushed.Meta.MinTimeNanos,
 		"maxtime", flushed.Meta.MaxTimeNanos,
-		"head-flush-duration", time.Since(th).String(),
+		"dataset-flush-duration", time.Since(th).String(),
 	)
 	return flushed, nil
 }
@@ -437,12 +441,13 @@ func (k datasetKey) compare(x datasetKey) int {
 }
 
 type dataset struct {
-	key  datasetKey
-	head *memdb.Head
+	key                datasetKey
+	head               *memdb.Head
+	needsSymbolization bool
 }
 
 type headFlush struct {
-	head    dataset
+	head    *dataset
 	flushed *memdb.FlushedHead
 	// protects head
 	done chan struct{}
@@ -453,10 +458,12 @@ type segment struct {
 	shard            shardKey
 	sshard           string
 	inFlightProfiles sync.WaitGroup
-	heads            map[datasetKey]dataset
-	headsLock        sync.RWMutex
-	logger           log.Logger
-	sw               *segmentsWriter
+
+	mu       sync.RWMutex
+	datasets map[datasetKey]*dataset
+
+	logger log.Logger
+	sw     *segmentsWriter
 
 	// TODO(kolesnikovae): Revisit.
 	doneChan      chan struct{}
@@ -499,11 +506,13 @@ func (s *segment) ingest(tenantID string, p *profilev1.Profile, id uuid.UUID, la
 		tenant:  tenantID,
 		service: model.Labels(labels).Get(model.LabelNameServiceName),
 	}
+	ds := s.datasetForIngest(k)
+	ds.needsSymbolization = !hasSymbols(p)
 	size := p.SizeVT()
 	rules := s.sw.limits.IngestionRelabelingRules(tenantID)
 	usage := s.sw.limits.DistributorUsageGroups(tenantID).GetUsageGroups(tenantID, labels)
 	appender := &sampleAppender{
-		head:    s.headForIngest(k),
+		dataset: ds,
 		profile: p,
 		id:      id,
 	}
@@ -514,9 +523,79 @@ func (s *segment) ingest(tenantID string, p *profilev1.Profile, id uuid.UUID, la
 	// CountReceivedBytes is tracked in distributors.
 }
 
+func hasSymbols(p *profilev1.Profile) bool {
+	if p == nil {
+		return false
+	}
+
+	// If there are no functions or locations, the profile can't be symbolized
+	if len(p.Function) == 0 || len(p.Location) == 0 {
+		return false
+	}
+
+	if len(p.StringTable) <= 1 {
+		return false
+	}
+
+	for _, loc := range p.Location {
+		// If a location has no line information, it's missing symbols
+		if len(loc.Line) == 0 {
+			return false
+		}
+
+		// Check if all lines have valid function references
+		for _, line := range loc.Line {
+			if line.FunctionId == 0 {
+				return false
+			}
+
+			// Verify the function exists and has a name
+			fn, exists := getFunctionById(p, line.FunctionId)
+			if !exists || fn.Name == 0 || int(fn.Name) >= len(p.StringTable) {
+				return false
+			}
+
+			// Check if the function name is just a hex address (unsymbolized)
+			name := p.StringTable[fn.Name]
+			if isHexAddress(name) {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+// getFunctionById returns a function by its ID
+func getFunctionById(p *profilev1.Profile, id uint64) (*profilev1.Function, bool) {
+	for _, fn := range p.Function {
+		if fn.Id == id {
+			return fn, true
+		}
+	}
+	return nil, false
+}
+
+// isHexAddress checks if a string looks like a hexadecimal memory address
+func isHexAddress(s string) bool {
+	// Most hex addresses in profiles start with "0x"
+	if len(s) < 3 || !strings.HasPrefix(s, "0x") {
+		return false
+	}
+
+	// Check if the rest of the string is hexadecimal
+	for _, c := range s[2:] {
+		if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) {
+			return false
+		}
+	}
+
+	return true
+}
+
 type sampleAppender struct {
 	id       uuid.UUID
-	head     *memdb.Head
+	dataset  *dataset
 	profile  *profilev1.Profile
 	exporter *pprofmodel.SampleExporter
 
@@ -525,7 +604,7 @@ type sampleAppender struct {
 }
 
 func (v *sampleAppender) VisitProfile(labels []*typesv1.LabelPair) {
-	v.head.Ingest(v.profile, v.id, labels)
+	v.dataset.head.Ingest(v.profile, v.id, labels)
 }
 
 func (v *sampleAppender) VisitSampleSeries(labels []*typesv1.LabelPair, samples []*profilev1.Sample) {
@@ -534,37 +613,36 @@ func (v *sampleAppender) VisitSampleSeries(labels []*typesv1.LabelPair, samples
 	}
 	var n profilev1.Profile
 	v.exporter.ExportSamples(&n, samples)
-	v.head.Ingest(&n, v.id, labels)
+	v.dataset.head.Ingest(&n, v.id, labels)
 }
 
 func (v *sampleAppender) Discarded(profiles, bytes int) {
 	v.discardedProfiles += profiles
 	v.discardedBytes += bytes
 }
 
-func (s *segment) headForIngest(k datasetKey) *memdb.Head {
-	s.headsLock.RLock()
-	h, ok := s.heads[k]
-	s.headsLock.RUnlock()
+func (s *segment) datasetForIngest(k datasetKey) *dataset {
+	s.mu.RLock()
+	ds, ok := s.datasets[k]
+	s.mu.RUnlock()
 	if ok {
-		return h.head
+		return ds
 	}
 
-	s.headsLock.Lock()
-	defer s.headsLock.Unlock()
-	h, ok = s.heads[k]
-	if ok {
-		return h.head
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if ds, ok = s.datasets[k]; ok {
+		return ds
 	}
 
-	nh := memdb.NewHead(s.sw.headMetrics)
-
-	s.heads[k] = dataset{
+	h := memdb.NewHead(s.sw.headMetrics)
+	ds = &dataset{
 		key:  k,
-		head: nh,
+		head: h,
 	}
 
-	return nh
+	s.datasets[k] = ds
+	return ds
 }
 
 func (sw *segmentsWriter) uploadBlock(ctx context.Context, blockData []byte, meta *metastorev1.BlockMeta, s *segment) error {
@@ -652,6 +730,12 @@ func (sw *segmentsWriter) storeMetadata(ctx context.Context, meta *metastorev1.B
 		sw.metrics.storeMetadataDLQ.WithLabelValues(statusLabelValue(err)).Inc()
 	}()
 
+	// Log the metadata before storing it
+	for i, ds := range meta.Datasets {
+		fmt.Printf(">>> Storing dataset %d: Name=%s, Labels=%v\n",
+			i, meta.StringTable[ds.Name], ds.Labels)
+	}
+
 	if err = s.sw.storeMetadataDLQ(ctx, meta); err == nil {
 		return nil
 	}

pkg/experiment/ingester/segment_test.go

@@ -820,7 +820,7 @@ func gprofileFromProtoProfile(t *testing.T, profile *profilev1.Profile) *gprofil
 func staticTestData() []inputChunk {
 	return []inputChunk{
 		{
-			//todo check why it takes 10ms for each head
+			//todo check why it takes 10ms for each dataset
 			{shard: 1, tenant: "t1", profile: cpuProfile(42, 480, "svc1", "foo", "bar")},
 			{shard: 1, tenant: "t1", profile: cpuProfile(13, 233, "svc1", "qwe", "foo", "bar")},
 			{shard: 1, tenant: "t1", profile: cpuProfile(13, 472, "svc1", "qwe", "foo", "bar")},

pkg/experiment/symbolizer/debuginfo_store.go

@@ -12,7 +12,6 @@ import (
 
 // DebugInfoStoreConfig holds configuration for the debug info cache
 type DebugInfoStoreConfig struct {
-	Enabled bool                  `yaml:"enabled"`
 	MaxAge  time.Duration         `yaml:"max_age"`
 	Storage objstoreclient.Config `yaml:"storage"`
 }
@@ -87,12 +86,11 @@ func (c *ObjstoreDebugInfoStore) Put(ctx context.Context, buildID string, reader
 	return nil
 }
 
-// ObjstoreDebugInfoStore implementation
 func (o *ObjstoreDebugInfoStore) IsEnabled() bool {
 	return true
 }
 
-// NullCache implements DebugInfoStore but performs no caching
+// NullDebugInfoStore implements DebugInfoStore but performs no caching
 type NullDebugInfoStore struct{}
 
 // NewNullDebugInfoStore creates a new null debug info store