[fix][backend] add MaxBytes limit to ListSpansRepeat to prevent… (#568)

[fix][observability] add MaxBytes limit to ListSpansRepeat to prevent OOM in trajectory backfill

When a traceID contains a large number of spans, ListSpansRepeat would
fetch all pages into memory without any bound, potentially causing OOM
in TCE instances during backfill tasks.

This adds a configurable MaxBytes parameter to ListSpansRepeat. When the
accumulated span size exceeds the limit, it returns ErrMaxBytesExceeded.
GetTrajectories catches this error, logs a warning, and returns an empty
trajectory map to skip oversized traces gracefully.

The limit is configured per workspace via BackfillConfig.TrajectoryMaxBytes.
When unset (0), the original unlimited behavior is preserved.

Author: SharkeyChen

backend/modules/observability/domain/component/config/config.go

@@ -149,6 +149,8 @@ type BackfillConfig struct {
 	CkQueryLimit SpaceAwareParam[int] `mapstructure:"ck_query_limit" json:"ck_query_limit"`
 	// BatchDispatchGray 批量分发灰度配置
 	BatchDispatchGray *BatchGrayConfig `mapstructure:"batch_dispatch_gray" json:"batch_dispatch_gray"`
+	// TrajectoryMaxBytes 单次 GetTrajectories 中 ListSpansRepeat 的内存上限（字节），0 表示不限制
+	TrajectoryMaxBytes SpaceAwareParam[int64] `mapstructure:"trajectory_max_bytes" json:"trajectory_max_bytes"`
 }
 
 // GetDispatchBatchSize 获取指定 workspace 的分发批大小
@@ -166,6 +168,11 @@ func (c *BackfillConfig) GetCkQueryLimit(workspaceID int64) int {
 	return c.CkQueryLimit.Get(workspaceID)
 }
 
+// GetTrajectoryMaxBytes 获取指定 workspace 的 trajectory 内存上限
+func (c *BackfillConfig) GetTrajectoryMaxBytes(workspaceID int64) int64 {
+	return c.TrajectoryMaxBytes.Get(workspaceID)
+}
+
 // BatchGrayConfig 批量处理灰度开关配置
 type BatchGrayConfig struct {
 	// EnableAll 全开开关，为 true 时所有 workspace 走批量逻辑

backend/modules/observability/domain/trace/repo/trace.go

@@ -5,11 +5,14 @@ package repo
 
 import (
 	"context"
+	"errors"
 
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/entity"
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/entity/loop_span"
 )
 
+var ErrMaxBytesExceeded = errors.New("ListSpansRepeat: max bytes exceeded")
+
 type GetTraceParam struct {
 	WorkSpaceID        string
 	Tenants            []string
@@ -46,6 +49,7 @@ type ListSpansParam struct {
 	NotQueryAnnotation bool
 	OmitColumns        []string // omit specific columns
 	SelectColumns      []string // select specific columns, default select all columns
+	MaxBytes           int64    // max total bytes for ListSpansRepeat, 0 means no limit
 }
 
 type ListSpansResult struct {

backend/modules/observability/domain/trace/service/trace_service.go

@@ -5,6 +5,7 @@ package service
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sort"
@@ -2489,6 +2490,11 @@ func (r *TraceServiceImpl) GetTrajectories(ctx context.Context, workspaceID int6
 		metaRules = metaCfg.Spaces[workspaceID]
 	}
 
+	var maxBytes int64
+	if backfillCfg := r.traceConfig.GetBackfillConfig(ctx); backfillCfg != nil {
+		maxBytes = backfillCfg.GetTrajectoryMaxBytes(workspaceID)
+	}
+
 	allSpans, err := r.traceRepo.ListSpansRepeat(ctx, &repo.ListSpansParam{
 		Tenants: tenant,
 		Filters: &loop_span.FilterFields{
@@ -2506,8 +2512,13 @@ func (r *TraceServiceImpl) GetTrajectories(ctx context.Context, workspaceID int6
 		Limit:              1000,
 		NotQueryAnnotation: true,
 		SelectColumns:      []string{loop_span.SpanFieldTraceId, loop_span.SpanFieldSpanId, loop_span.SpanFieldParentID, loop_span.SpanFieldSpaceId, loop_span.SpanFieldSpanName},
+		MaxBytes:           maxBytes,
 	})
 	if err != nil {
+		if errors.Is(err, repo.ErrMaxBytesExceeded) {
+			logs.CtxWarn(ctx, "GetTrajectories skipped: allSpans exceeded max bytes, traceIDs:%v, maxBytes:%d", traceIDs, maxBytes)
+			return map[string]*loop_span.Trajectory{}, nil
+		}
 		logs.CtxError(ctx, "Failed to list all spans, err:%+v", err)
 		return nil, err
 	}
@@ -2520,8 +2531,13 @@ func (r *TraceServiceImpl) GetTrajectories(ctx context.Context, workspaceID int6
 		EndAt:              endTime,
 		Limit:              100,
 		NotQueryAnnotation: true,
+		MaxBytes:           maxBytes,
 	})
 	if err != nil {
+		if errors.Is(err, repo.ErrMaxBytesExceeded) {
+			logs.CtxWarn(ctx, "GetTrajectories skipped: selectedSpans exceeded max bytes, traceIDs:%v, maxBytes:%d", traceIDs, maxBytes)
+			return map[string]*loop_span.Trajectory{}, nil
+		}
 		logs.CtxError(ctx, "Failed to list selected spans, err:%+v", err)
 		return nil, err
 	}

backend/modules/observability/domain/trace/service/trace_trajectory_service_test.go

@@ -129,6 +129,7 @@ func TestTraceServiceImpl_GetTrajectories_and_ListTrajectory(t *testing.T) {
 	repoMock.EXPECT().GetTrajectoryConfig(gomock.Any(), repo.GetTrajectoryConfigParam{WorkspaceId: 1}).Return(nil, nil).AnyTimes()
 	traceConfigMock := configmocks.NewMockITraceConfig(ctrl)
 	traceConfigMock.EXPECT().GetTrajectoryMetadataConfig(gomock.Any()).Return(nil).AnyTimes()
+	traceConfigMock.EXPECT().GetBackfillConfig(gomock.Any()).Return(nil).AnyTimes()
 
 	svc := &TraceServiceImpl{traceRepo: repoMock, buildHelper: builder, tenantProvider: tenantProviderMock, traceConfig: traceConfigMock}
 	// mock list all spans

backend/modules/observability/infra/repo/trace.go

@@ -296,12 +296,22 @@ func (t *TraceRepoImpl) ListSpansRepeat(ctx context.Context, req *repo.ListSpans
 
 	clonedReq := *req
 	totalSpans := loop_span.SpanList{}
+	var totalBytes int64
 
 	for {
 		resp, err := t.ListSpans(ctx, &clonedReq)
 		if err != nil {
 			return nil, err
 		}
+
+		if req.MaxBytes > 0 {
+			pageBytes := int64(loop_span.SizeofSpans(resp.Spans))
+			if totalBytes+pageBytes > req.MaxBytes {
+				return nil, repo.ErrMaxBytesExceeded
+			}
+			totalBytes += pageBytes
+		}
+
 		totalSpans = append(totalSpans, resp.Spans...)
 		if !resp.HasMore || resp.PageToken == "" {
 			break