[fix][backend]: Task filter tenant

backend/modules/observability/domain/task/entity/event.go

@@ -145,6 +145,11 @@ func (s *RawSpan) RawSpanConvertToLoopSpan() *loop_span.Span {
 	callType := tagsString["call_type"]
 	spanType := tagsString["span_type"]
 
+	// RawSpan 中的 tenant 一般在 Tags 字段里
+	// 而 LoopSpan 在 systemTagsString 中
+	// 这里是为了方便转化为 loopSpan 对象后统一使用 getTenant 方法
+	systemTagsString["tenant"] = tagsString["tenant"]
+
 	result := &loop_span.Span{
 		StartTime:        s.StartTimeInUs,
 		SpanID:           s.SpanID,

backend/modules/observability/domain/task/service/taskexe/processor/auto_evaluate.go

@@ -120,7 +120,7 @@ func (p *AutoEvaluateProcessor) Invoke(ctx context.Context, trigger *taskexe.Tri
 	_ = p.taskRepo.IncrTaskRunCount(ctx, trigger.Task.ID, taskRun.ID, taskTTL)
 	taskCount, _ := p.taskRepo.GetTaskCount(ctx, trigger.Task.ID)
 	taskRunCount, _ := p.taskRepo.GetTaskRunCount(ctx, trigger.Task.ID, taskRun.ID)
-	if (trigger.Task.Sampler.CycleCount != 0 && taskRunCount > trigger.Task.Sampler.CycleCount) ||
+	if (trigger.Task.Sampler.IsCycle && trigger.Task.Sampler.CycleCount != 0 && taskRunCount > trigger.Task.Sampler.CycleCount) ||
 		(taskCount > trigger.Task.Sampler.SampleSize) {
 		logs.CtxInfo(ctx, "[task-debug] AutoEvaluateProcessor Invoke, subCount:%v,taskCount:%v", taskRunCount, taskCount)
 		_ = p.taskRepo.DecrTaskCount(ctx, trigger.Task.ID, taskTTL)

backend/modules/observability/domain/task/service/taskexe/processor/utils.go

@@ -120,12 +120,14 @@ func convertDatasetSchemaDTO2DO(schema *dataset0.DatasetSchema) entity.DatasetSc
 			name := fs.GetName()
 			description := fs.GetDescription()
 			textSchema := fs.GetTextSchema()
+			schemaKey := fs.GetSchemaKey()
 			result.FieldSchemas[i] = entity.FieldSchema{
 				Key:         &key,
 				Name:        name,
 				Description: description,
 				ContentType: convertContentTypeDTO2DO(fs.GetContentType()),
 				TextSchema:  textSchema,
+				SchemaKey:   entity.SchemaKey(schemaKey),
 			}
 		}
 	}

backend/modules/observability/domain/task/service/taskexe/tracehub/span_trigger.go

@@ -66,7 +66,6 @@ func (h *TraceHubServiceImpl) buildSubscriberOfSpan(ctx context.Context, span *l
 		logs.CtxError(ctx, "Failed to get consumer listening config, err: %v", err)
 		return nil, err
 	}
-
 	var subscribers []*spanSubscriber
 	taskDOs, err := h.listNonFinalTaskByRedis(ctx, span.WorkspaceID)
 	if err != nil {
@@ -86,13 +85,19 @@ func (h *TraceHubServiceImpl) buildSubscriberOfSpan(ctx context.Context, span *l
 		}
 
 		proc := h.taskProcessor.GetTaskProcessor(taskDO.TaskType)
+		tenants, err := h.getTenants(ctx, taskDO.GetPlatformType())
+		if err != nil {
+			logs.CtxError(ctx, "Failed to get tenants, err: %v", err)
+			return nil, err
+		}
 		subscribers = append(subscribers, &spanSubscriber{
 			taskID:      taskDO.ID,
 			t:           taskDO,
 			processor:   proc,
 			taskRepo:    h.taskRepo,
 			runType:     entity.TaskRunTypeNewData,
 			buildHelper: h.buildHelper,
+			tenants:     tenants,
 		})
 	}
 

backend/modules/observability/domain/task/service/taskexe/tracehub/span_trigger_test.go

@@ -16,6 +16,7 @@ import (
 	taskconvertor "github.com/coze-dev/coze-loop/backend/modules/observability/application/convertor/task"
 	componentconfig "github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/config"
 	config_mocks "github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/config/mocks"
+	tenant_mocks "github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/tenant/mocks"
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/entity"
 	repo_mocks "github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/repo/mocks"
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/service/taskexe/processor"
@@ -60,6 +61,7 @@ func TestTraceHubServiceImpl_SpanTriggerDispatchError(t *testing.T) {
 	mockBuilder := trace_service_mocks.NewMockTraceFilterProcessorBuilder(ctrl)
 	mockFilter := span_filter_mocks.NewMockFilter(ctrl)
 	configLoader := config_mocks.NewMockITraceConfig(ctrl)
+	tenantProvider := tenant_mocks.NewMockITenantProvider(ctrl)
 
 	now := time.Now()
 	workspaceID := int64(1)
@@ -112,6 +114,7 @@ func TestTraceHubServiceImpl_SpanTriggerDispatchError(t *testing.T) {
 	mockFilter.EXPECT().BuildBasicSpanFilter(gomock.Any(), gomock.Any()).Return(nil, false, nil).AnyTimes()
 	mockFilter.EXPECT().BuildALLSpanFilter(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 	mockBuilder.EXPECT().BuildPlatformRelatedFilter(gomock.Any(), gomock.Any()).Return(mockFilter, nil).AnyTimes()
+	tenantProvider.EXPECT().GetTenantsByPlatformType(gomock.Any(), loop_span.PlatformDefault, gomock.Any()).Return([]string{"tenant"}, nil).AnyTimes()
 
 	spanRun := &entity.TaskRun{
 		ID:          201,
@@ -132,11 +135,12 @@ func TestTraceHubServiceImpl_SpanTriggerDispatchError(t *testing.T) {
 	taskProcessor.Register(entity.TaskTypeAutoEval, procMock)
 
 	impl := &TraceHubServiceImpl{
-		taskRepo:      mockRepo,
-		buildHelper:   mockBuilder,
-		taskProcessor: taskProcessor,
-		localCache:    NewLocalCache(),
-		config:        configLoader,
+		taskRepo:       mockRepo,
+		buildHelper:    mockBuilder,
+		taskProcessor:  taskProcessor,
+		localCache:     NewLocalCache(),
+		config:         configLoader,
+		tenantProvider: tenantProvider,
 	}
 	impl.localCache.taskCache.Store("ObjListWithTask", TaskCacheInfo{WorkspaceIDs: []string{"space-1"}, Tasks: []*entity.ObservabilityTask{taskDO}})
 
@@ -146,13 +150,12 @@ func TestTraceHubServiceImpl_SpanTriggerDispatchError(t *testing.T) {
 		LogID:         "log",
 		StartTimeInUs: now.UnixMicro(),
 		Tags: map[string]any{
-			"fornax_space_id": "space-1",
-			"call_type":       "",
-			"bot_id":          "bot-1",
-		},
-		SystemTags: map[string]any{
+			"fornax_space_id":         "space-1",
+			"call_type":               "",
+			"bot_id":                  "bot-1",
 			loop_span.SpanFieldTenant: "tenant",
 		},
+		SystemTags:    map[string]any{},
 		SensitiveTags: &entity.SensitiveTags{},
 		ServerEnv:     &entity.ServerInRawSpan{},
 	}

backend/modules/observability/domain/task/service/taskexe/tracehub/subscriber.go

@@ -21,11 +21,11 @@ import (
 )
 
 type spanSubscriber struct {
-	taskID    int64
-	t         *entity.ObservabilityTask
-	tr        *entity.TaskRun
-	processor taskexe.Processor
-
+	taskID      int64
+	t           *entity.ObservabilityTask
+	tr          *entity.TaskRun
+	processor   taskexe.Processor
+	tenants     []string
 	taskRepo    repo.ITaskRepo
 	runType     entity.TaskRunType
 	buildHelper service.TraceFilterProcessorBuilder
@@ -66,15 +66,15 @@ func (s *spanSubscriber) Match(ctx context.Context, span *loop_span.Span) (bool,
 		return false, nil
 	}
 
-	filters := s.buildSpanFilters(ctx, task)
+	filters := s.buildSpanFilters(ctx, task, span)
 	if !filters.Satisfied(span) {
 		return false, nil
 	}
 
 	return true, nil
 }
 
-func (s *spanSubscriber) buildSpanFilters(ctx context.Context, taskDO *entity.ObservabilityTask) *loop_span.FilterFields {
+func (s *spanSubscriber) buildSpanFilters(ctx context.Context, taskDO *entity.ObservabilityTask, span *loop_span.Span) *loop_span.FilterFields {
 	// Additional filters can be constructed based on task configuration if needed.
 	// Simplified handling here: returning nil means no extra filters are applied.
 	filters := &loop_span.FilterFields{}
@@ -93,11 +93,30 @@ func (s *spanSubscriber) buildSpanFilters(ctx context.Context, taskDO *entity.Ob
 		logs.CtxError(ctx, "traverse filter fields failed, task_id=%d, err=%v", taskDO.ID, err)
 		return filters
 	}
-	filters = combineFilters(builtinFilter, &taskDO.SpanFilter.Filters)
+	var tenantFilter *loop_span.FilterFields = nil
+	if len(span.GetTenant()) > 0 {
+		tenantFilter = buildTenantFilter(s.tenants)
+	}
+	filters = combineFilters(builtinFilter, &taskDO.SpanFilter.Filters, tenantFilter)
 
 	return filters
 }
 
+func buildTenantFilter(tenants []string) *loop_span.FilterFields {
+	return &loop_span.FilterFields{
+		FilterFields: []*loop_span.FilterField{
+			{
+				FieldName: loop_span.SpanFieldTenant,
+				FieldType: loop_span.FieldTypeString,
+				Values:    tenants,
+				QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
+				IsSystem:  true,
+			},
+		},
+		QueryAndOr: ptr.Of(loop_span.QueryAndOrEnumAnd),
+	}
+}
+
 func buildBuiltinFilters(ctx context.Context, f span_filter.Filter, req *ListSpansReq) (*loop_span.FilterFields, error) {
 	filters := make([]*loop_span.FilterField, 0)
 	env := &span_filter.SpanEnv{

backend/modules/observability/domain/task/service/taskexe/tracehub/subscriber_test.go

@@ -15,6 +15,9 @@ import (
 	repo_mocks "github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/repo/mocks"
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/service/taskexe"
 	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/entity/loop_span"
+	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/service/trace/span_filter"
+	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/service/trace/span_processor"
+	"github.com/coze-dev/coze-loop/backend/pkg/lang/ptr"
 )
 
 type noopProcessor struct{ invoked bool }
@@ -45,6 +48,60 @@ func (n *noopProcessor) OnTaskCreated(ctx context.Context, currentTask *entity.O
 	return nil
 }
 
+type fakeSpanFilter struct {
+	basic []*loop_span.FilterField
+	root  []*loop_span.FilterField
+	llm   []*loop_span.FilterField
+	all   []*loop_span.FilterField
+	force bool
+}
+
+func (f *fakeSpanFilter) BuildBasicSpanFilter(ctx context.Context, env *span_filter.SpanEnv) ([]*loop_span.FilterField, bool, error) {
+	return f.basic, f.force, nil
+}
+
+func (f *fakeSpanFilter) BuildRootSpanFilter(ctx context.Context, env *span_filter.SpanEnv) ([]*loop_span.FilterField, error) {
+	return f.root, nil
+}
+
+func (f *fakeSpanFilter) BuildLLMSpanFilter(ctx context.Context, env *span_filter.SpanEnv) ([]*loop_span.FilterField, error) {
+	return f.llm, nil
+}
+
+func (f *fakeSpanFilter) BuildALLSpanFilter(ctx context.Context, env *span_filter.SpanEnv) ([]*loop_span.FilterField, error) {
+	return f.all, nil
+}
+
+type fakeBuilder struct{ f span_filter.Filter }
+
+func (b *fakeBuilder) BuildPlatformRelatedFilter(ctx context.Context, pt loop_span.PlatformType) (span_filter.Filter, error) {
+	return b.f, nil
+}
+
+func (b *fakeBuilder) BuildGetTraceProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
+func (b *fakeBuilder) BuildListSpansProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
+func (b *fakeBuilder) BuildAdvanceInfoProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
+func (b *fakeBuilder) BuildIngestTraceProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
+func (b *fakeBuilder) BuildSearchTraceOApiProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
+func (b *fakeBuilder) BuildListSpansOApiProcessors(ctx context.Context, set span_processor.Settings) ([]span_processor.Processor, error) {
+	return nil, nil
+}
+
 func TestSpanSubscriber_AddSpan_SkipNonRunning(t *testing.T) {
 	t.Parallel()
 	ctrl := gomock.NewController(t)
@@ -78,3 +135,132 @@ func TestSpanSubscriber_AddSpan_SkipNonRunning(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, proc.invoked, "Invoke should not be called for non-running TaskRun")
 }
+
+func TestSpanSubscriber_Match_PlatformAndTenant_Positive(t *testing.T) {
+	t.Parallel()
+	basic := []*loop_span.FilterField{
+		{
+			FieldName: loop_span.SpanFieldPSM,
+			FieldType: loop_span.FieldTypeString,
+			Values:    []string{"coze-loop"},
+			QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
+		},
+	}
+	f := &fakeSpanFilter{basic: basic, root: nil, force: true}
+	task := &entity.ObservabilityTask{
+		ID:          1,
+		WorkspaceID: 7,
+		TaskStatus:  entity.TaskStatusRunning,
+		SpanFilter: &entity.SpanFilterFields{
+			PlatformType: loop_span.PlatformCozeLoop,
+			SpanListType: loop_span.SpanListTypeRootSpan,
+		},
+	}
+	sub := &spanSubscriber{
+		taskID:      task.ID,
+		t:           task,
+		processor:   &noopProcessor{},
+		taskRepo:    nil,
+		runType:     entity.TaskRunTypeNewData,
+		buildHelper: &fakeBuilder{f: f},
+		tenants:     []string{"tenant1", "tenant2"},
+	}
+	span := &loop_span.Span{
+		SpanID:  "s1",
+		TraceID: "t1",
+		PSM:     "coze-loop",
+		SystemTagsString: map[string]string{
+			loop_span.SpanFieldTenant: "tenant1",
+		},
+		StartTime: time.Now().UnixMilli(),
+	}
+	matched, err := sub.Match(context.Background(), span)
+	assert.NoError(t, err)
+	assert.True(t, matched)
+}
+
+func TestSpanSubscriber_Match_NegativePSM(t *testing.T) {
+	t.Parallel()
+	basic := []*loop_span.FilterField{
+		{
+			FieldName: loop_span.SpanFieldPSM,
+			FieldType: loop_span.FieldTypeString,
+			Values:    []string{"coze-loop"},
+			QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
+		},
+	}
+	f := &fakeSpanFilter{basic: basic, root: nil, force: true}
+	task := &entity.ObservabilityTask{
+		ID:          2,
+		WorkspaceID: 7,
+		TaskStatus:  entity.TaskStatusRunning,
+		SpanFilter: &entity.SpanFilterFields{
+			PlatformType: loop_span.PlatformCozeLoop,
+			SpanListType: loop_span.SpanListTypeRootSpan,
+		},
+	}
+	sub := &spanSubscriber{
+		taskID:      task.ID,
+		t:           task,
+		processor:   &noopProcessor{},
+		taskRepo:    nil,
+		runType:     entity.TaskRunTypeNewData,
+		buildHelper: &fakeBuilder{f: f},
+		tenants:     []string{"tenant1"},
+	}
+	span := &loop_span.Span{
+		SpanID:  "s2",
+		TraceID: "t2",
+		PSM:     "other",
+		SystemTagsString: map[string]string{
+			loop_span.SpanFieldTenant: "tenant1",
+		},
+		StartTime: time.Now().UnixMilli(),
+	}
+	matched, err := sub.Match(context.Background(), span)
+	assert.NoError(t, err)
+	assert.False(t, matched)
+}
+
+func TestSpanSubscriber_Match_NegativeTenant(t *testing.T) {
+	t.Parallel()
+	basic := []*loop_span.FilterField{
+		{
+			FieldName: loop_span.SpanFieldPSM,
+			FieldType: loop_span.FieldTypeString,
+			Values:    []string{"coze-loop"},
+			QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
+		},
+	}
+	f := &fakeSpanFilter{basic: basic, root: nil, force: true}
+	task := &entity.ObservabilityTask{
+		ID:          3,
+		WorkspaceID: 7,
+		TaskStatus:  entity.TaskStatusRunning,
+		SpanFilter: &entity.SpanFilterFields{
+			PlatformType: loop_span.PlatformCozeLoop,
+			SpanListType: loop_span.SpanListTypeRootSpan,
+		},
+	}
+	sub := &spanSubscriber{
+		taskID:      task.ID,
+		t:           task,
+		processor:   &noopProcessor{},
+		taskRepo:    nil,
+		runType:     entity.TaskRunTypeNewData,
+		buildHelper: &fakeBuilder{f: f},
+		tenants:     []string{"tenantX"},
+	}
+	span := &loop_span.Span{
+		SpanID:  "s3",
+		TraceID: "t3",
+		PSM:     "coze-loop",
+		SystemTagsString: map[string]string{
+			loop_span.SpanFieldTenant: "tenant1",
+		},
+		StartTime: time.Now().UnixMilli(),
+	}
+	matched, err := sub.Match(context.Background(), span)
+	assert.NoError(t, err)
+	assert.False(t, matched)
+}