diff --git a/CHANGELOG.md b/CHANGELOG.md index 11956f155..2be2e3706 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,11 @@ to docs, or any other relevant information. workflow flow types to activities, adds flow/step/subflow/RPC metric tags, and rewrites Temporal SDK metric names into the `dex_*` namespace. +### Changed + +- Activities with a Dex `FlowTypeProvider` no longer store the parent workflow flow type in their + headers. The provider is authoritative, and an empty result produces the `none` metric label. + ### Breaking Changes - SDK metrics beginning with `temporal_` are now emitted under their corresponding `dex_*` names. diff --git a/internal/activity.go b/internal/activity.go index 44095c3cb..d95d71626 100644 --- a/internal/activity.go +++ b/internal/activity.go @@ -103,9 +103,8 @@ type ( // invoking the activity function. Empty values become "none". Provider // results are metric labels and should have bounded cardinality. // - // FlowTypeProvider extracts the flow type. A non-empty value overrides the - // flow type inherited from the parent workflow; an empty value falls back to - // the inherited value, or "none" when no inherited value exists. + // FlowTypeProvider extracts the flow type instead of inheriting it from the + // parent workflow. An empty value becomes "none". FlowTypeProvider func(input any) string // StepTypeProvider identifies the activity as a step and extracts its step // type. At most one of StepTypeProvider, SubFlowTypeProvider, and diff --git a/internal/dex_metrics.go b/internal/dex_metrics.go index bf9935b8b..9f33ab056 100644 --- a/internal/dex_metrics.go +++ b/internal/dex_metrics.go @@ -117,18 +117,16 @@ func (p dexActivityMetricProviders) values(typeName string, input any, inherited rpcName: metrics.NoneTagValue, kind: p.metricKind(), } - if inheritedFlowType != "" { + if p.flowTypeProvider == nil && inheritedFlowType != "" { values.flowType = inheritedFlowType } var err error if p.flowTypeProvider != nil { - flowType, providerErr := invokeDexMetricsProviderRaw("activity", typeName, "FlowTypeProvider", p.flowTypeProvider, input) + flowType, providerErr := invokeDexMetricsProvider("activity", typeName, "FlowTypeProvider", p.flowTypeProvider, input) if providerErr != nil { return values, providerErr } - if flowType != "" { - values.flowType = flowType - } + values.flowType = flowType } switch p.kind { case dexActivityMetricKindStep: @@ -155,7 +153,14 @@ type dexWorkflowMetricsEnvironment interface { dexWorkflowFlowType() (string, bool) } -func addDexFlowTypeHeader(header *commonpb.Header, env WorkflowEnvironment) { +func addDexFlowTypeHeader( + header *commonpb.Header, + env WorkflowEnvironment, + activityProviders dexActivityMetricProviders, +) { + if activityProviders.flowTypeProvider != nil { + return + } dexEnv, ok := env.(dexWorkflowMetricsEnvironment) if !ok { return diff --git a/internal/dex_metrics_test.go b/internal/dex_metrics_test.go index 257b1800e..66e7a73fb 100644 --- a/internal/dex_metrics_test.go +++ b/internal/dex_metrics_test.go @@ -46,13 +46,16 @@ func TestDexActivityMetricsProviders(t *testing.T) { values, err = providers.values("activity", &dexMetricsTestInput{}, "parent") require.NoError(t, err) - require.Equal(t, "parent", values.flowType) + require.Equal(t, "none", values.flowType) require.Equal(t, "none", values.stepType) - providers.flowTypeProvider = func(any) string { return "none" } - values, err = providers.values("activity", &dexMetricsTestInput{}, "parent") + inheritedProviders := dexActivityProviders(RegisterActivityOptions{ + StepTypeProvider: func(input any) string { return input.(*dexMetricsTestInput).StepType }, + }) + values, err = inheritedProviders.values("activity", &dexMetricsTestInput{StepType: "step"}, "parent") require.NoError(t, err) - require.Equal(t, "none", values.flowType) + require.Equal(t, "parent", values.flowType) + require.Equal(t, "step", values.stepType) } func TestDexMetricsProviderPanicReturnsError(t *testing.T) { @@ -86,11 +89,25 @@ func TestDexFlowTypeHeader(t *testing.T) { require.Empty(t, popDexFlowTypeHeader(nil)) require.Empty(t, popDexFlowTypeHeader(&commonpb.Header{})) - header := &commonpb.Header{Fields: map[string]*commonpb.Payload{}} env := &workflowEnvironmentImpl{dexFlowType: "flow", dexFlowTypeConfigured: true} - addDexFlowTypeHeader(header, env) + header := &commonpb.Header{Fields: map[string]*commonpb.Payload{}} + addDexFlowTypeHeader(header, env, dexActivityMetricProviders{}) require.Equal(t, "flow", popDexFlowTypeHeader(header)) require.NotContains(t, header.Fields, dexFlowTypeHeaderName) + + header = &commonpb.Header{Fields: map[string]*commonpb.Payload{}} + providers := dexActivityProviders(RegisterActivityOptions{ + FlowTypeProvider: func(input any) string { return input.(*dexMetricsTestInput).FlowType }, + }) + addDexFlowTypeHeader(header, env, providers) + require.NotContains(t, header.Fields, dexFlowTypeHeaderName) + + header = &commonpb.Header{Fields: map[string]*commonpb.Payload{}} + providers = dexActivityProviders(RegisterActivityOptions{ + StepTypeProvider: func(input any) string { return input.(*dexMetricsTestInput).StepType }, + }) + addDexFlowTypeHeader(header, env, providers) + require.Equal(t, "flow", popDexFlowTypeHeader(header)) } func TestDexActivityProviderUsesDecodedFirstArgumentOnce(t *testing.T) { @@ -177,7 +194,11 @@ func TestDexWorkflowAndActivityMetricsPropagation(t *testing.T) { }, }) - input := &dexMetricsTestInput{FlowType: "OrderFlow", StepType: "ChargeCard"} + input := &dexMetricsTestInput{ + FlowType: "OrderFlow", + ActivityFlowType: "OrderFlow", + StepType: "ChargeCard", + } env.ExecuteWorkflow("dexMetricsWorkflow", input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) diff --git a/internal/workflow.go b/internal/workflow.go index 805c72435..45266073e 100644 --- a/internal/workflow.go +++ b/internal/workflow.go @@ -1083,7 +1083,8 @@ func (wc *workflowEnvironmentInterceptor) ExecuteActivity(ctx Context, typeName settable.Set(nil, err) return future } - addDexFlowTypeHeader(header, getWorkflowEnvironment(ctx)) + activityProviders := registry.getDexActivityMetricsProviders(typeName) + addDexFlowTypeHeader(header, getWorkflowEnvironment(ctx), activityProviders) env := getWorkflowEnvironment(ctx) // Generate activity ID before serialization so it's available to context-aware data converters @@ -1214,6 +1215,7 @@ func ExecuteLocalActivity(ctx Context, activity interface{}, args ...interface{} func (wc *workflowEnvironmentInterceptor) ExecuteLocalActivity(ctx Context, typeName string, args ...interface{}) Future { future, settable := newDecodeFuture(ctx, typeName) + activityProviders := getRegistryFromWorkflowContext(ctx).getDexActivityMetricsProviders(typeName) envOptions := getWorkflowEnvOptions(ctx) header, err := workflowHeaderPropagated(ctx, envOptions.ContextPropagators) @@ -1221,7 +1223,7 @@ func (wc *workflowEnvironmentInterceptor) ExecuteLocalActivity(ctx Context, type settable.Set(nil, err) return future } - addDexFlowTypeHeader(header, getWorkflowEnvironment(ctx)) + addDexFlowTypeHeader(header, getWorkflowEnvironment(ctx), activityProviders) var activityFn interface{} localCtx := ctx.Value(localActivityFnContextKey).(*localActivityContext) @@ -1299,7 +1301,7 @@ func (wc *workflowEnvironmentInterceptor) ExecuteLocalActivity(ctx Context, type ScheduledTime: Now(ctx), // initial scheduled time Header: header, Attempt: 1, // Attempts always start at one - DexMetricsProviders: getRegistryFromWorkflowContext(ctx).getDexActivityMetricsProviders(typeName), + DexMetricsProviders: activityProviders, } Go(ctx, func(ctx Context) { diff --git a/test/dex_metrics_test.go b/test/dex_metrics_test.go index bef348499..cc32e0285 100644 --- a/test/dex_metrics_test.go +++ b/test/dex_metrics_test.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/activity" "go.temporal.io/sdk/client" "go.temporal.io/sdk/temporal" @@ -16,6 +17,7 @@ import ( ) const ( + dexFlowTypeHeaderName = "__temporal_sdk_dex_flow_type" dexMetricsWorkflowName = "dex-metrics-workflow" dexSystemWorkflowName = "dex-system-workflow" dexSyncStepActivityName = "dex-sync-step-activity" @@ -148,6 +150,53 @@ func (ts *IntegrationTestSuite) TestDexMetricsProvidersAndNames() { } } +func (ts *IntegrationTestSuite) TestDexFlowTypeHeaderInheritance() { + input := &dexMetricsIntegrationInput{ + FlowType: "OrderFlow", + StepType: "ChargeCard", + SubFlowType: "Fulfillment", + RPCName: "ReserveInventory", + } + run, err := ts.client.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ + ID: "dex-flow-type-header-" + uuid.NewString(), + TaskQueue: ts.taskQueueName, + }, dexMetricsWorkflowName, input) + ts.NoError(err) + ts.NoError(run.Get(context.Background(), nil)) + + expectedHeaderByActivityType := map[string]bool{ + dexSyncStepActivityName: false, + dexSubFlowActivityName: true, + dexRPCActivityName: false, + dexSystemActivityName: true, + } + observedHeaderByActivityType := make(map[string]bool, len(expectedHeaderByActivityType)) + history := ts.client.GetWorkflowHistory( + context.Background(), + run.GetID(), + run.GetRunID(), + false, + enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT, + ) + for history.HasNext() { + event, historyErr := history.Next() + ts.NoError(historyErr) + attributes := event.GetActivityTaskScheduledEventAttributes() + if attributes == nil { + continue + } + activityType := attributes.GetActivityType().GetName() + expectedHeader, tracked := expectedHeaderByActivityType[activityType] + if !tracked { + continue + } + _, hasHeader := attributes.GetHeader().GetFields()[dexFlowTypeHeaderName] + ts.Equal(expectedHeader, hasHeader, activityType) + observedHeaderByActivityType[activityType] = hasHeader + } + ts.Equal(expectedHeaderByActivityType, observedHeaderByActivityType) +} + func (ts *IntegrationTestSuite) TestDexMetricsProviderPanicsSkipBusinessFunctions() { run, err := ts.client.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ ID: "dex-workflow-provider-panic-" + uuid.NewString(),