CHASM telemetry

Author: stephanos

chasm/registrable_task.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 )
 
 type (
 	RegistrableTask struct {
 		taskType                string
+		telemetryTypeName       string
 		goType                  reflect.Type
 		componentGoType         reflect.Type // It is not clear how this one is used.
 		validateFn              validateFn
@@ -155,6 +157,7 @@ func (rt *RegistrableTask) registerToLibrary(
 
 	fqn := rt.fqType()
 	rt.taskTypeID = GenerateTypeID(fqn)
+	rt.telemetryTypeName = deriveTelemetryTypeName(rt.goType, fqn)
 	return fqn, rt.taskTypeID, nil
 }
 
@@ -173,3 +176,23 @@ func (rt *RegistrableTask) fqType() string {
 	}
 	return FullyQualifiedName(rt.library.Name(), rt.taskType)
 }
+
+func (rt *RegistrableTask) telemetryType() string {
+	if rt.telemetryTypeName != "" {
+		return rt.telemetryTypeName
+	}
+	return rt.fqType()
+}
+
+func deriveTelemetryTypeName(taskType reflect.Type, fallback string) string {
+	for taskType.Kind() == reflect.Pointer {
+		taskType = taskType.Elem()
+	}
+	// Remove common prefix/suffix from type name
+	if name := taskType.Name(); name != "" {
+		name = strings.TrimPrefix(name, "Chasm")
+		name = strings.TrimSuffix(name, "Data")
+		return name
+	}
+	return fallback
+}

chasm/registry.go

@@ -168,6 +168,15 @@ func (r *Registry) TaskFqnByID(id uint32) (string, bool) {
 	return rt.fqType(), true
 }
 
+// TaskTelemetryNameByID returns the human-readable telemetry label for a task type ID.
+func (r *Registry) TaskTelemetryNameByID(id uint32) (string, bool) {
+	rt, ok := r.rtByID[id]
+	if !ok {
+		return "", false
+	}
+	return rt.telemetryType(), true
+}
+
 // TaskIDFor converts registered task instance to task type ID.
 // This method should only be used by CHASM framework internal code,
 // NOT CHASM library developers.

chasm/registry_test.go

@@ -147,6 +147,18 @@ func (s *RegistryTestSuite) TestRegistry_RegisterTasks_Success() {
 	require.True(s.T(), ok)
 	require.Equal(s.T(), "TestLibrary.Task2", rt2.FqType())
 
+	taskID1, ok := r.TaskIDFor(testTask1{})
+	require.True(s.T(), ok)
+	name1, ok := r.TaskTelemetryNameByID(taskID1)
+	require.True(s.T(), ok)
+	require.Equal(s.T(), "testTask1", name1)
+
+	taskID2, ok := r.TaskIDFor(testTask2{})
+	require.True(s.T(), ok)
+	name2, ok := r.TaskTelemetryNameByID(taskID2)
+	require.True(s.T(), ok)
+	require.Equal(s.T(), "testTask2", name2)
+
 	tInstance2 := "invalid task instance"
 	rt3, ok := r.TaskFor(tInstance2)
 	require.False(s.T(), ok)

chasm/statemachine.go

@@ -4,7 +4,10 @@ import (
 	"fmt"
 	"slices"
 
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/telemetry"
 )
 
 // ErrInvalidTransition is returned from [Transition.Apply] on an invalid state transition.
@@ -44,8 +47,24 @@ func (t Transition[S, SM, E]) Possible(sm SM) bool {
 
 // Apply applies a transition event to the given state machine changing the state machine's state to the transition's
 // Destination on success.
-func (t Transition[S, SM, E]) Apply(sm SM, ctx MutableContext, event E) error {
+func (t Transition[S, SM, E]) Apply(sm SM, ctx MutableContext, event E) (retErr error) {
 	prevState := sm.StateMachineState()
+
+	// Defer to always emit the transition telemetry event.
+	if telemetry.DebugMode() {
+		defer func() {
+			attrs := []attribute.KeyValue{
+				attribute.String("chasm.transition.source", fmt.Sprintf("%v", prevState)),
+				attribute.String("chasm.transition.destination", fmt.Sprintf("%v", t.Destination)),
+			}
+			if retErr != nil {
+				attrs = append(attrs, attribute.String("chasm.transition.error", retErr.Error()))
+			}
+			span := trace.SpanFromContext(ctx.goContext())
+			span.AddEvent("chasm.transition", trace.WithAttributes(attrs...))
+		}()
+	}
+
 	if !t.Possible(sm) {
 		return fmt.Errorf("%w from %v", ErrInvalidTransition, prevState)
 	}

cmd/tools/genrpcserverinterceptors/main.go

@@ -137,13 +137,6 @@ func workflowTagGetters(messageType reflect.Type, depth int) messageData {
 	case messageType.AssignableTo(workflowExecutionGetterT):
 		pd.WorkflowIDGetter = "GetWorkflowExecution().GetWorkflowId()"
 		pd.RunIDGetter = "GetWorkflowExecution().GetRunId()"
-	case messageType.AssignableTo(taskTokenGetterT):
-		for _, ert := range excludeTaskTokenTypes {
-			if messageType.AssignableTo(ert) {
-				return pd
-			}
-		}
-		pd.TaskTokenGetter = "GetTaskToken()"
 	default:
 		// Might have any combination of these, or none.
 		if messageType.AssignableTo(workflowIDGetterT) {
@@ -160,6 +153,25 @@ func workflowTagGetters(messageType reflect.Type, depth int) messageData {
 		}
 	}
 
+	if pd.ActivityIDGetter == "" && messageType.AssignableTo(activityIDGetterT) {
+		pd.ActivityIDGetter = "GetActivityId()"
+	}
+	if pd.OperationIDGetter == "" && messageType.AssignableTo(operationIDGetterT) {
+		pd.OperationIDGetter = "GetOperationId()"
+	}
+	if messageType.AssignableTo(taskTokenGetterT) {
+		excluded := false
+		for _, ert := range excludeTaskTokenTypes {
+			if messageType.AssignableTo(ert) {
+				excluded = true
+				break
+			}
+		}
+		if !excluded {
+			pd.TaskTokenGetter = "GetTaskToken()"
+		}
+	}
+
 	// Iterates over fields in order they defined in proto file, not proto index.
 	// Order is important because the first match wins.
 	for fieldNum := 0; fieldNum < messageType.Elem().NumField(); fieldNum++ {

cmd/tools/genrpcserverinterceptors/server_interceptors.tmpl

@@ -15,9 +15,7 @@ func (wt *WorkflowTags) extractFrom{{.Server}}Message(message any) []tag.Tag {
 	{{- range .Messages}}
 	case {{.Type}}:
 	{{- if or .TaskTokenGetter .WorkflowIDGetter .RunIDGetter .ActivityIDGetter .OperationIDGetter .ChasmRunIDGetter}}
-		{{- if .TaskTokenGetter}}
-		return wt.fromTaskToken(r.{{ .TaskTokenGetter}})
-		{{- else}}
+		{{- if or .WorkflowIDGetter .RunIDGetter .ActivityIDGetter .OperationIDGetter .ChasmRunIDGetter}}
 		return []tag.Tag{
 		{{if .WorkflowIDGetter}}	tag.WorkflowID(r.{{.WorkflowIDGetter}}),
 		{{end -}}
@@ -30,6 +28,8 @@ func (wt *WorkflowTags) extractFrom{{.Server}}Message(message any) []tag.Tag {
 		{{if .ChasmRunIDGetter}}	tag.ChasmRunID(r.{{.ChasmRunIDGetter}}),
 		{{end -}}
 		}
+		{{- else if .TaskTokenGetter}}
+		return wt.fromTaskToken(r.{{ .TaskTokenGetter}})
 		{{- end}}
 	{{- else}}
 		return nil

common/log/tag/tags.go

@@ -21,6 +21,8 @@ import (
 
 // LoggingCallAtKey is reserved tag
 const (
+	ActivityIDKey    = "activity-id"
+	ChasmRunIDKey    = "run-id"
 	LoggingCallAtKey = "logging-call-at"
 	WorkflowIDKey    = "wf-id"
 	WorkflowRunIDKey = "wf-run-id"
@@ -914,7 +916,7 @@ func ActivityInfo(activityInfo any) ZapTag {
 
 // ActivityID returns tag for an activity ID
 func ActivityID(id string) ZapTag {
-	return NewStringTag("activity-id", id)
+	return NewStringTag(ActivityIDKey, id)
 }
 
 // OperationID returns tag for a nexus operation ID
@@ -924,7 +926,7 @@ func OperationID(id string) ZapTag {
 
 // ChasmRunID returns tag for an entity run ID
 func ChasmRunID(id string) ZapTag {
-	return NewStringTag("run-id", id)
+	return NewStringTag(ChasmRunIDKey, id)
 }
 
 // ActivitySize returns a tag for a standalone activity size

common/rpc/interceptor/logtags/matching_service_server_gen.go

@@ -54,10 +54,22 @@ func (wt *WorkflowTags) fromTaskToken(taskTokenBytes []byte) []tag.Tag {
 	if len(taskTokenBytes) == 0 {
 		return nil
 	}
+
 	taskToken, err := wt.serializer.Deserialize(taskTokenBytes)
 	if err != nil {
 		wt.logger.Warn("unable to deserialize task token while getting workflow tags", tag.Error(err))
 		return nil
 	}
-	return []tag.Tag{tag.WorkflowID(taskToken.WorkflowId), tag.WorkflowRunID(taskToken.RunId)}
+
+	var tags []tag.Tag
+	if taskToken.WorkflowId != "" {
+		tags = append(tags, tag.WorkflowID(taskToken.WorkflowId))
+	}
+	if taskToken.RunId != "" {
+		tags = append(tags, tag.WorkflowRunID(taskToken.RunId))
+	}
+	if len(taskToken.ComponentRef) > 0 && taskToken.ActivityId != "" {
+		tags = append(tags, tag.ActivityID(taskToken.ActivityId))
+	}
+	return tags
 }