RSDK-2880: Add timestamp information to the replay camera #2371
Conversation
|
Warning your change may break code samples. If your change modifies any of the following functions please contact @viamrobotics/fleet-management. Thanks!
|
viambot
added
the
safe to test
robot/client/client.go
Outdated
| @@ -278,6 +279,7 @@ func New(ctx context.Context, address string, logger golog.Logger, opts ...Robot | |||
| // interceptors are applied in order from first to last | |||
| rc.dialOptions = append( | |||
| rc.dialOptions, | |||
| rpc.WithUnaryClientInterceptor(rdkutils.ContextWithMetadataUnaryClientInterceptor), | |||
There was a problem hiding this comment.
@edaniels Do we need to register interceptors anywhere else? This is the only place that seemed necessary for this to work.
utils/context.go
Outdated
| func ContextWithMetadataUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { | ||
| var header metadata.MD | ||
| opts = append(opts, grpc.Header(&header)) | ||
| invoker(ctx, method, req, reply, cc, opts...) |
There was a problem hiding this comment.
[q] The grpc.ClientConn includes an invoke function. Should we be using this instead of the passed-in invoker? Or do we need this need invoker due to the ctx with metadata?
There was a problem hiding this comment.
Hm honestly I'm not sure. This is the pattern I've seen in most other gRPC interceptors. Probably either would work. The comment above the ClientConnInterface says:
// ClientConnInterface defines the functions clients need to perform unary and
// streaming RPCs. It is implemented by *ClientConn, and is only intended to
// be referenced by generated code.
so probably better practice to use invoker.
There was a problem hiding this comment.
@edaniels for sanity check
There was a problem hiding this comment.
the interceptors require you call that invoker since you don't know where in the middleware pipeline you are
components/camera/camera.go
Outdated
| // SubtypeName is a constant that identifies the camera resource subtype string. | ||
| SubtypeName = "camera" | ||
|
|
||
| // TimeRequestedMetadataKey is optional metadata in the gRPC response header that correlates |
There was a problem hiding this comment.
[opt] Move this to replaypcd. We can always import replaypcd into viam-carto and it would be good to indicate that this is currently the only model that support/uses these fields
There was a problem hiding this comment.
I actually like having this in camera so it can be used for other camera implementations if needed. I think as long as it's well-documented, it's okay. Thoughts @edaniels ?
There was a problem hiding this comment.
I think we should move this to some context utils or something. we can always move it in the future as well
utils/context.go
Outdated
|
|
||
| const MetadataKey = "viam-metadata" | ||
|
|
||
| func ContextWithMetadata(ctx context.Context) (context.Context, map[string][]string) { |
There was a problem hiding this comment.
Do we need to return md here? Can it not be accessed via the ctx directly? https://pkg.go.dev/google.golang.org/grpc@v1.55.0/metadata? i.e. FromIncomingContext/FromOutgoingContext
There was a problem hiding this comment.
The caller could call context.Value(utils.MetadataKey) instead to get the metadata. I think either design is fine; I thought this was cleaner because they don't have to check for whether or not the metadata exists (they know it does). I don't think they could call those grpc methods since at the point where this is used, the context does not have any gRPC stream in it afaik.
components/camera/camera.go
Outdated
| // SubtypeName is a constant that identifies the camera resource subtype string. | ||
| const SubtypeName = "camera" | ||
| const ( | ||
| // SubtypeName is a constant that identifies the camera resource subtype string. |
There was a problem hiding this comment.
can revert this to what it was since it's one const
|
|
||
| // TimeRequestedMetadataKey is optional metadata in the gRPC response header that correlates | ||
| // to the time right before the point cloud was captured. | ||
| TimeRequestedMetadataKey = "viam-time-requested" |
There was a problem hiding this comment.
these other consts need to be contextKey wrapped as well
There was a problem hiding this comment.
Ah yes, I wasn't overly concerned with making this a custom type because they're keys in the metadata itself, not in the context. I don't think there's a concern that a third party would extract the viam metadata using the contextKey type but then overwrite the time requested or time received metadata keys without meaning to. The keys are also used in the gRPC header metadata, where we might be more concerned about collisions, but unfortunately gRPC requires the string type for the key so we wouldn't even be able to use a custom type if we wanted to. Lmk if you disagree with that line of reasoning though!
There was a problem hiding this comment.
Oh whoops that's my bad. I assumed this const block was all context keys. will add another comment to clear up confusion
| @@ -0,0 +1,65 @@ | |||
| // Package contextutils provides utility for adding and retrieving metadata to/from a context. | |||
There was a problem hiding this comment.
we'll want unit tests for this file
There was a problem hiding this comment.
Ah okay, I added a test for the interceptor, do you mean unit tests just for ContextWithMetadata? Showing how it works when you have metadata on the context vs not? Or is there something else you think we're missing coverage for on top of that
There was a problem hiding this comment.
yep just simple tests of putting data in and getting it out as well as calling the Context function with a context has nothing or the wrong value paired with the context key
| @@ -64,7 +64,7 @@ func TestSafetyMonitor(t *testing.T) { | |||
| } | |||
|
|
|||
| func TestSafetyMonitorForMetadata(t *testing.T) { | |||
| stream1 := &myStream{} | |||
| stream1 := testutils.NewServerTransportStream() | |||
| } | ||
|
|
||
| md := ctx.Value(MetadataKey) | ||
| if mdMap, ok := md.(map[string][]string); ok { |
There was a problem hiding this comment.
Should an error be returned if ContextWithMetadataUnaryClientInterceptor and no md in the map[string][]string format is available?
There was a problem hiding this comment.
This interceptor is used for every single gRPC unary call from the gRPC client. Most of the gRPC requests will not have metadata present in the context, so we wouldn't want to error out.
testutils/rpc.go
Outdated
| return nil | ||
| } | ||
|
|
||
| // Value returns the value in the metadata map corresponding stored for the given key. |
There was a problem hiding this comment.
[opt/nit]
| // Value returns the value in the metadata map corresponding stored for the given key. | |
| // Value returns the value in the metadata map corresponding to a given key. |
| return &ServerTransportStream{} | ||
| } | ||
|
|
||
| // SetHeader implements grpc.ServerTransportStream. |
There was a problem hiding this comment.
[nit] might be worth explaining how this is supposed to differ from SendHeader and why in this mock up the code is the same
There was a problem hiding this comment.
Very good question, I actually meant to follow up on this because I wasn't sure myself, and I forgot. Turns out the different between SetHeader and SendHeader is that SetHeader is supposed to merge the new metadata fields with the existing metadata fields, whereas SendHeader overwrites the metadata and is only supposed to be called once. So I think there were two things that were slightly wrong here: our implementation of SetHeader, which completely overwrites the metadata rather than merging, and the fact that in replaypcd camera I'm calling SendHeader, which afaik will overwrite any metadata already set by SetHeader and error if anyone else tries to call SendHeader or SetHeader after that point. I've changed both; thanks for calling this out! cc @edaniels for awareness of this change (Jeremy's out so can't re-request his review)
| @@ -64,7 +64,7 @@ func TestSafetyMonitor(t *testing.T) { | |||
| } | |||
|
|
|||
| func TestSafetyMonitorForMetadata(t *testing.T) { | |||
| stream1 := &myStream{} | |||
| stream1 := testutils.NewServerTransportStream() | |||
| @@ -28,6 +30,8 @@ import ( | |||
| "go.viam.com/rdk/testutils/inject" | |||
| ) | |||
|
|
|||
| var testTime = "2000-01-01T12:00:%02dZ" | |||
|
|
||
| // Repeatedly call NextPointCloud, checking for timestamps in the gRPC header. | ||
| for i := 0; i < numPCDFiles; i++ { | ||
| serverStream := testutils.NewServerTransportStream() |
There was a problem hiding this comment.
[q] Does the metadata context need to be created every time? If not we could move lines 443-444 to 435 and call:
ctx = grpc.NewContextWithServerTransportStream(context.Background(), serverStream)
There was a problem hiding this comment.
It doesn't need to be, but I felt this more closely mimicked how this works in prod with consecutive gRPC calls using separate ServerTransportStreams and gRPC contexts. So I'm going to leave this as-is unless someone else feels strongly that this should change.
| @@ -60,9 +64,17 @@ func (mDServer *mockDataServiceServer) BinaryDataByFilter(ctx context.Context, r | |||
| gz.Close() | |||
|
|
|||
| // Construct response | |||
| timeReq, err := time.Parse(time.RFC3339, fmt.Sprintf(testTime, newFileNum)) | |||
There was a problem hiding this comment.
[q] When should/do we need to use time.PFC3339 and time.RFC339Nano?
There was a problem hiding this comment.
Right now, in production code paths, we're using RFC3339 for parsing the requested time ranges from the config for the replay sensor. This means that we're limiting "users" (Cloud Story) from configuring anything here with higher granularity than 1 second. This matches what we do in the Data tab and I think this is ok!
We then use RFC3339Nano whenever we're stringifying or parsing one of the timestamps correlated with the images. These have nanosecond granularity so we need to use a time format that supports that so we don't lose that information.
Here in the tests I'm using RFC3339 because I'm only changing the second of the timestamp depending on the file, so I hard-coded something in the RFC3339 format.
components/camera/client_test.go
Outdated
| test.That(t, err, test.ShouldBeNil) | ||
| _, got := pcB.At(5, 5, 5) | ||
| test.That(t, got, test.ShouldBeTrue) | ||
| test.That(t, md["hello"][0], test.ShouldEqual, "world") |
There was a problem hiding this comment.
Can you link this to the input on line 414 so they reference to the same variable?
viambot
added
safe to test
viambot
added
safe to test
viambot
added
safe to test
viambot
added
safe to test
utils/contextutils/context.go
Outdated
|
|
||
| const ( | ||
| // MetadataKey is the key used to access metadata from a context with metadata. | ||
| MetadataKey = contextKey("viam-metadata") |
There was a problem hiding this comment.
the convention is to call this MetadataContextKey
viambot
added
safe to test
viambot
added
safe to test
|
|
Manually tested once again with final changes, still works. Merging |
Summary
This PR modifies the replay camera to attach timestamps in the gRPC response if it's being called over gRPC. It also introduces an interceptor that can be used to inject metadata from the gRPC response header into the context provided by the client.
Manual Testing
I tested by configuring a robot (my Mac) with a replay camera and providing that as a sensor for SLAM. Along with the changes in the corresponding viam-cartographer PR, I confirmed that the files get written to disk with the correct timestamps correlated with the time they were captured (last week):