Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion cmd/ateapi/internal/controlapi/create_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,11 +33,12 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
if err := validateCreateActorRequest(req); err != nil {
return nil, err
}

in := req.GetActor()
templateNamespace := in.GetActorTemplateNamespace()
templateName := in.GetActorTemplateName()

setSpanActorRefIdentity(ctx, in.GetMetadata().GetAtespace(), in.GetMetadata().GetName())

_, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName)
if err != nil {
if k8serrors.IsNotFound(err) {
Expand Down Expand Up @@ -76,6 +77,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
return nil, fmt.Errorf("while recording actor: %w", err)
}

setSpanActorIdentity(ctx, stored)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: rename this tosetSpanActorAttributes to distinguish from setSpanActorRefIdentity?

return stored, nil
}

Expand Down
1 change: 1 addition & 0 deletions cmd/ateapi/internal/controlapi/delete_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ
if err := validateDeleteActorRequest(req); err != nil {
return nil, err
}
setSpanActorRefIdentity(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())

deleted, err := s.persistence.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
if err != nil {
Expand Down
2 changes: 2 additions & 0 deletions cmd/ateapi/internal/controlapi/pause_actor.go
Comment thread
zoez7 marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorReques
if err := validatePauseActorRequest(req); err != nil {
return nil, err
}
setSpanActorRefIdentity(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

out of curiosity, what will be happened if instead of setSpanActorRefIdentity, the "setSpanActorIdentity" will be called.
The only value that is changing between line 33 to 46 is version.
I guess the intention is to keep the latest version right? Is it possible to call to setSpanActorIdentity twice. before s.actorWorkflow.PauseActor and after? The latest version will be submitted in successful case and previous version will be submitted in case of error.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't call the full one early. At line 33 we only have the ObjectRef from the request and setSpanActorIdentity needs the resolved *Actor that we don't have until the workflow returns it.

Attributes overwrite by key, so the later wins and we get the latest version on success. On error we return early with only the ref, so no version.

I guess we could also add the version on mid-workflow failures if we move it inside the workflow. Maybe that's better as a follow-up if it's worth it. What do you think?

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 without looking at the code, it's not obvious which "version" is in the attribute (is it the version before or after the pause?) we can update the key to "new_version" or "updated_version" for clarification.

Image

@krisztianfekete Krisztian F (krisztianfekete) Jul 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd keep it as ate.actor.version. It means the same thing on every op, which is the version the op resulted in. Forking it to new_version on pause but version on create breaks querying for spans that have actor A at version X.

Happy to add a line to the attr doc spelling out "resulting version" or similar if that makes it better?


actor, err := s.actorWorkflow.PauseActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
if err != nil {
Expand All @@ -42,6 +43,7 @@ func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorReques
return nil, err
}

setSpanActorIdentity(ctx, actor)
return &ateapipb.PauseActorResponse{Actor: actor}, nil
}

Expand Down
2 changes: 2 additions & 0 deletions cmd/ateapi/internal/controlapi/resume_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequ
if err := validateResumeActorRequest(req); err != nil {
return nil, err
}
setSpanActorRefIdentity(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())

actor, err := s.actorWorkflow.ResumeActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName(), req.GetBoot())
if err != nil {
Expand All @@ -42,6 +43,7 @@ func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequ
return nil, err
}

setSpanActorIdentity(ctx, actor)
return &ateapipb.ResumeActorResponse{Actor: actor}, nil
}

Expand Down
36 changes: 36 additions & 0 deletions cmd/ateapi/internal/controlapi/span_identity.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package controlapi

import (
"context"

"go.opentelemetry.io/otel/trace"

"github.com/agent-substrate/substrate/internal/ateattr"
"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
)

// setSpanActorIdentity annotates the RPC's server span (from ctx) with the
// actor's full identity. A no-op when ctx carries no recording span.
func setSpanActorIdentity(ctx context.Context, a *ateapipb.Actor) {
trace.SpanFromContext(ctx).SetAttributes(ateattr.ActorIdentity(a)...)
}

// setSpanActorRefIdentity is setSpanActorIdentity for the identity subset known
// before the Actor record resolves, so a failed lookup still carries who/where.
func setSpanActorRefIdentity(ctx context.Context, atespace, actorID string) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the parameter name is actorID but the callsites are passing in actor name, can we change it to actorName?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair, but this turned out to be a bit deeper when started looking into it. ate.actor.id actually had the name while we also have a real uid. Went full k8s-style as per OTel semconv: the attr is now ate.actor.name and I added ate.actor.uid.

trace.SpanFromContext(ctx).SetAttributes(ateattr.ActorRefIdentity(atespace, actorID)...)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similarly here, can we rename ActorRefIdentity to ActorObjectReference to be consistent with pkg/proto/ateapipb/ateapi.proto

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I went with ActorRefAttributes instead of ActorObjectReference, as it returns attributes, not a reference object, and it matches ActorAttributes. What do you think?

}
149 changes: 149 additions & 0 deletions cmd/ateapi/internal/controlapi/span_identity_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package controlapi

import (
"context"
"testing"

"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"

"github.com/agent-substrate/substrate/internal/ateattr"
"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
)

// The functional harness registers no otelgrpc StatsHandler, so there is no
// server span on the client path. These tests instead call the Service methods
// in-process under a self-provided recording root span, which stands in for the
// span the otelgrpc handler injects in production and exercises the same
// trace.SpanFromContext(ctx).SetAttributes path.
func installSpanRecorder(t *testing.T) *tracetest.SpanRecorder {
t.Helper()
sr := tracetest.NewSpanRecorder()
tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
prev := otel.GetTracerProvider()
otel.SetTracerProvider(tp)
t.Cleanup(func() { otel.SetTracerProvider(prev) })
return sr
}

func rootSpanAttrs(t *testing.T, sr *tracetest.SpanRecorder, fn func(ctx context.Context)) map[attribute.Key]attribute.Value {
t.Helper()
ctx, root := otel.Tracer("test").Start(context.Background(), "root")
fn(ctx)
root.End()
for _, s := range sr.Ended() {
if s.Name() == "root" {
m := make(map[attribute.Key]attribute.Value, len(s.Attributes()))
for _, kv := range s.Attributes() {
m[kv.Key] = kv.Value
}
return m
}
}
t.Fatal("root span not recorded")
return nil
}

func assertSpanStr(t *testing.T, attrs map[attribute.Key]attribute.Value, key attribute.Key, want string) {
t.Helper()
v, ok := attrs[key]
if !ok {
t.Errorf("missing %s", key)
return
}
if v.AsString() != want {
t.Errorf("%s = %q, want %q", key, v.AsString(), want)
}
}

func TestCreateActor_StampsFullSpanIdentity(t *testing.T) {
Comment thread
krisztianfekete marked this conversation as resolved.
Outdated
ns := namespaceForTest("ns-span-create")
tc := setupTest(t, ns)
defer tc.cleanup()
createTemplate(t, tc, ns)

sr := installSpanRecorder(t)
attrs := rootSpanAttrs(t, sr, func(ctx context.Context) {
if _, err := tc.service.CreateActor(ctx, &ateapipb.CreateActorRequest{
Actor: &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
},
}); err != nil {
t.Fatalf("CreateActor: %v", err)
}
})

assertSpanStr(t, attrs, ateattr.AtespaceKey, testAtespace)
assertSpanStr(t, attrs, ateattr.ActorIDKey, "id1")
assertSpanStr(t, attrs, ateattr.ActorTemplateNameKey, "tmpl1")
assertSpanStr(t, attrs, ateattr.ActorTemplateNamespaceKey, ns)
if v, ok := attrs[ateattr.ActorVersionKey]; !ok || v.Type() != attribute.INT64 || v.AsInt64() != 1 {
t.Errorf("%s = %v, want int64 1", ateattr.ActorVersionKey, v.Emit())
}
}

func TestDeleteActor_StampsRefSpanIdentity(t *testing.T) {
ns := namespaceForTest("ns-span-delete")
tc := setupTest(t, ns)
defer tc.cleanup()
createTemplate(t, tc, ns)
if _, err := tc.service.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Actor: &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
},
}); err != nil {
t.Fatalf("seed CreateActor: %v", err)
}

sr := installSpanRecorder(t)
attrs := rootSpanAttrs(t, sr, func(ctx context.Context) {
if _, err := tc.service.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"},
}); err != nil {
t.Fatalf("DeleteActor: %v", err)
}
})

assertSpanStr(t, attrs, ateattr.AtespaceKey, testAtespace)
assertSpanStr(t, attrs, ateattr.ActorIDKey, "id1")
}

// The early ref stamp must land on the span even when the operation fails, so a
// failed resume is still attributable to who/where.
func TestResumeActor_ErrorStillStampsRefSpanIdentity(t *testing.T) {
ns := namespaceForTest("ns-span-resume-err")
tc := setupTest(t, ns)
defer tc.cleanup()

sr := installSpanRecorder(t)
attrs := rootSpanAttrs(t, sr, func(ctx context.Context) {
if _, err := tc.service.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "missing"},
}); err == nil {
t.Fatal("expected error resuming missing actor")
}
})

assertSpanStr(t, attrs, ateattr.AtespaceKey, testAtespace)
assertSpanStr(t, attrs, ateattr.ActorIDKey, "missing")
}
2 changes: 2 additions & 0 deletions cmd/ateapi/internal/controlapi/suspend_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRe
if err := validateSuspendActorRequest(req); err != nil {
return nil, err
}
setSpanActorRefIdentity(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())

actor, err := s.actorWorkflow.SuspendActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
if err != nil {
Expand All @@ -42,6 +43,7 @@ func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRe
return nil, err
}

setSpanActorIdentity(ctx, actor)
return &ateapipb.SuspendActorResponse{Actor: actor}, nil
}

Expand Down
54 changes: 54 additions & 0 deletions internal/ateattr/ateattr.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package ateattr projects an Actor onto substrate's ate.* identity attributes.
// Identity is a span-level subject attribute (the producer is the substrate
// component, the actor is the subject), so it belongs on spans rather than the
// resource, and uses substrate's own ate.* namespace rather than service.*.
package ateattr

import (
"go.opentelemetry.io/otel/attribute"

"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
)

// Dotted ate.* matches the metric-instrument naming (atenet.*, atelet.*), not the
// ate.dev/ slash form used for k8s labels and stdout log fields.
const (
AtespaceKey = attribute.Key("ate.atespace")

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ate.actor.atespace?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I prefer ate.atespace. Atespace is the tenant/isolation boundary, where its own resource, templates is too), so if it's a top level attr you can filter spans by tenants regardless of resource type. It's the same reason OTel uses k8s.namespace.name, not k8s.pod.namespace. What do you think?

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the ActorTemplateNameKey on line 37 should be ate.template.name instead?

ActorIDKey = attribute.Key("ate.actor.id")
ActorTemplateNameKey = attribute.Key("ate.actor.template.name")
ActorTemplateNamespaceKey = attribute.Key("ate.actor.template.namespace")
ActorVersionKey = attribute.Key("ate.actor.version")
)

// ActorRefIdentity returns the subset knowable before the Actor record resolves.
func ActorRefIdentity(atespace, actorID string) []attribute.KeyValue {
return []attribute.KeyValue{
AtespaceKey.String(atespace),
ActorIDKey.String(actorID),
}
}

// ActorIdentity is nil-safe; a nil Actor yields zero-valued attributes.
func ActorIdentity(a *ateapipb.Actor) []attribute.KeyValue {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: I think this should be ActorAttributes since it logs several fields that I would not consider as the "identity" of the actor.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is now done.

return []attribute.KeyValue{
AtespaceKey.String(a.GetMetadata().GetAtespace()),
ActorIDKey.String(a.GetMetadata().GetName()),
ActorTemplateNameKey.String(a.GetActorTemplateName()),
ActorTemplateNamespaceKey.String(a.GetActorTemplateNamespace()),
ActorVersionKey.Int64(a.GetMetadata().GetVersion()),
}
}
Loading