Skip to content

Commit 54b31d9

Browse files
committed
feat(sdk): add Go workspace service discovery client
Signed-off-by: Taylor Mutch <taylormutch@gmail.com>
1 parent d51549f commit 54b31d9

8 files changed

Lines changed: 377 additions & 0 deletions

File tree

‎sdk/go/docs/src/SUMMARY.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
- [Profiles](api/profiles.md)
2222
- [Refresh](api/refresh.md)
2323
- [Services](api/services.md)
24+
- [Service discovery](api/service-discovery.md)
2425
- [Files](api/files.md)
2526
- [Health](api/health.md)
2627
- [SSH](api/ssh.md)
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
# Service discovery
2+
3+
Accessor: `client.ServiceRoutes()` on the concrete SDK client. Consumers can
4+
accept the narrow `ServiceDiscoveryInterface`; existing `ClientInterface` mocks
5+
are unchanged. Discovery reads accepted supervisor installations for one explicit
6+
workspace and requires the same workspace read authorization as the gateway API.
7+
8+
```go
9+
snapshot, err := client.ServiceRoutes().Snapshot(ctx, "default")
10+
if err != nil {
11+
return err
12+
}
13+
for _, route := range snapshot.Routes {
14+
fmt.Println(route.Hostname, route.Endpoint, route.ExpiresAt)
15+
}
16+
17+
watch, err := client.ServiceRoutes().Watch(ctx, "default")
18+
if err != nil {
19+
return err
20+
}
21+
defer watch.Close()
22+
for {
23+
replacement, err := watch.Recv()
24+
if errors.Is(err, io.EOF) {
25+
break // reconnect and relist after normal bounded stream completion
26+
}
27+
if err != nil {
28+
return err // an error is not an empty replacement
29+
}
30+
_ = replacement // atomically replace the complete cached workspace view
31+
}
32+
```
33+
34+
The SDK returns complete replacement snapshots without background prefetch or
35+
implicit reconnect. A caller must expire cached entries at `ExpiresAt` even when
36+
discovery is unavailable, reconnect/relist after stream termination, and clear
37+
routes when authorization is revoked. `Close` cancels the RPC and is safe to call
38+
concurrently with one receiving goroutine. Derive a cancelable watch context and
39+
choose snapshot deadlines appropriate for the consumer.
40+
41+
Preserve `Hostname` as TLS SNI and HTTP authority when dialing `Endpoint`.
42+
`InstallationVersion` stays stable across lease-only renewal; `Revision` changes
43+
with expiry and is only an equality token. Installation is neither application
44+
health nor proof of external ingress acceptance. Platform-wide discovery-only
45+
identities and all-workspaces queries are separate work.

‎sdk/go/openshell/v1/client.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ type Client struct {
5656
templates SandboxTemplateInterface
5757
providers ProviderInterface
5858
services ServiceInterface
59+
serviceRoutes ServiceDiscoveryInterface
5960
exec ExecInterface
6061
files FileInterface
6162
health HealthInterface
@@ -102,6 +103,7 @@ func NewClient(cfg Config) (*Client, error) {
102103
c.templates = newSandboxTemplateClient(conn)
103104
c.providers = newProviderClient(conn)
104105
c.services = newServiceClient(conn)
106+
c.serviceRoutes = newServiceDiscoveryClient(conn)
105107
c.exec = newExecClient(conn, c.sandboxes)
106108
c.files = newFileClient(conn, c.sandboxes)
107109
c.health = newHealthClient(conn)
@@ -132,6 +134,10 @@ func (c *Client) Providers() ProviderInterface { return c.providers }
132134
// Services returns the service sub-client.
133135
func (c *Client) Services() ServiceInterface { return c.services }
134136

137+
// ServiceRoutes returns the read-only workspace discovery client. This additive
138+
// accessor has its own narrow consumer interface to preserve ClientInterface mocks.
139+
func (c *Client) ServiceRoutes() ServiceDiscoveryInterface { return c.serviceRoutes }
140+
135141
// Exec returns the exec sub-client.
136142
func (c *Client) Exec() ExecInterface { return c.exec }
137143

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package converter
5+
6+
import (
7+
"github.com/NVIDIA/OpenShell/sdk/go/openshell/v1/types"
8+
pb "github.com/NVIDIA/OpenShell/sdk/go/proto/openshellv1"
9+
)
10+
11+
// ServiceRouteSnapshotFromProto validates and copies a full discovery view.
12+
func ServiceRouteSnapshotFromProto(snapshot *pb.ServiceRouteSnapshot, workspace string) (*types.ServiceRouteSnapshot, error) {
13+
invalid := func() (*types.ServiceRouteSnapshot, error) {
14+
return nil, &types.StatusError{Code: types.ErrorInternal, Message: "invalid service discovery snapshot"}
15+
}
16+
if snapshot == nil || snapshot.GetRevision() == "" || len(snapshot.GetRoutes()) > 4096 {
17+
return invalid()
18+
}
19+
result := &types.ServiceRouteSnapshot{Revision: snapshot.GetRevision(), Routes: make([]types.DiscoveredServiceRoute, 0, len(snapshot.GetRoutes()))}
20+
hostnames := make(map[string]bool, len(snapshot.GetRoutes()))
21+
for _, route := range snapshot.GetRoutes() {
22+
if route == nil || route.GetWorkspace() != workspace || route.GetHostname() == "" || route.GetInstallationVersion() == "" || route.GetEndpoint() == nil || route.GetEndpoint().GetHost() == "" || route.GetEndpoint().GetPort() == 0 || route.GetEndpoint().GetPort() > 65535 || route.GetExpiresAt() == nil || route.GetExpiresAt().CheckValid() != nil || hostnames[route.GetHostname()] {
23+
return invalid()
24+
}
25+
hostnames[route.GetHostname()] = true
26+
result.Routes = append(result.Routes, types.DiscoveredServiceRoute{
27+
Workspace: route.GetWorkspace(), Sandbox: route.GetSandbox(), Name: route.GetName(), Hostname: route.GetHostname(),
28+
Endpoint: types.ServiceRouteEndpoint{Host: route.GetEndpoint().GetHost(), Port: route.GetEndpoint().GetPort()},
29+
InstallationVersion: route.GetInstallationVersion(), DeclarationRevision: route.GetDeclarationRevision(), ExpiresAt: route.GetExpiresAt().AsTime(),
30+
})
31+
}
32+
return result, nil
33+
}
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package converter
5+
6+
import (
7+
pb "github.com/NVIDIA/OpenShell/sdk/go/proto/openshellv1"
8+
"github.com/stretchr/testify/require"
9+
"google.golang.org/protobuf/types/known/timestamppb"
10+
"testing"
11+
"time"
12+
)
13+
14+
func TestDiscoverySnapshotCopiesAndRejectsIncompleteViews(t *testing.T) {
15+
wire := &pb.ServiceRouteSnapshot{Revision: "revision", Routes: []*pb.DiscoveredServiceRoute{{
16+
Workspace: "default", Sandbox: "demo", Hostname: "demo.example.com", InstallationVersion: "owner-1",
17+
Endpoint: &pb.ServiceRouteEndpoint{Host: "10.0.0.1", Port: 9443}, ExpiresAt: timestamppb.New(time.Now()),
18+
}}}
19+
snapshot, err := ServiceRouteSnapshotFromProto(wire, "default")
20+
require.NoError(t, err)
21+
snapshot.Routes[0].Endpoint.Host = "changed"
22+
require.Equal(t, "10.0.0.1", wire.Routes[0].Endpoint.Host)
23+
wire.Routes[0].Hostname = "different.example.com"
24+
require.Equal(t, "demo.example.com", snapshot.Routes[0].Hostname)
25+
_, err = ServiceRouteSnapshotFromProto(wire, "other")
26+
require.Error(t, err)
27+
wire.Routes[0].ExpiresAt = nil
28+
_, err = ServiceRouteSnapshotFromProto(wire, "default")
29+
require.Error(t, err)
30+
_, err = ServiceRouteSnapshotFromProto(nil, "default")
31+
require.Error(t, err)
32+
}
Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package v1
5+
6+
import (
7+
"context"
8+
"io"
9+
"sync"
10+
11+
"github.com/NVIDIA/OpenShell/sdk/go/openshell/v1/internal/converter"
12+
"github.com/NVIDIA/OpenShell/sdk/go/openshell/v1/types"
13+
pb "github.com/NVIDIA/OpenShell/sdk/go/proto/openshellv1"
14+
"google.golang.org/grpc"
15+
)
16+
17+
// ServiceRouteSnapshot is a complete replacement view of workspace installations.
18+
type ServiceRouteSnapshot = types.ServiceRouteSnapshot
19+
20+
// DiscoveredServiceRoute identifies an installed service and its current supervisor.
21+
type DiscoveredServiceRoute = types.DiscoveredServiceRoute
22+
23+
// ServiceRouteEndpoint is the trusted supervisor TLS endpoint.
24+
type ServiceRouteEndpoint = types.ServiceRouteEndpoint
25+
26+
// ServiceRouteWatch delivers full replacements. Call Recv from one goroutine;
27+
// Close is safe concurrently. EOF is normal bounded stream completion: reconnect
28+
// and relist. Errors are not empty replacements. Expire cached routes independently.
29+
type ServiceRouteWatch interface {
30+
Recv() (*ServiceRouteSnapshot, error)
31+
Close()
32+
}
33+
34+
// ServiceDiscoveryInterface is a narrow interface for external discovery consumers.
35+
// It supports one explicit workspace and does not request all-workspaces grants.
36+
type ServiceDiscoveryInterface interface {
37+
Snapshot(ctx context.Context, workspace string) (*ServiceRouteSnapshot, error)
38+
Watch(ctx context.Context, workspace string) (ServiceRouteWatch, error)
39+
}
40+
41+
type serviceDiscoveryClient struct{ client pb.OpenShellClient }
42+
43+
func newServiceDiscoveryClient(conn grpc.ClientConnInterface) *serviceDiscoveryClient {
44+
return &serviceDiscoveryClient{client: pb.NewOpenShellClient(conn)}
45+
}
46+
47+
func (s *serviceDiscoveryClient) Snapshot(ctx context.Context, workspace string) (*ServiceRouteSnapshot, error) {
48+
if workspace == "" {
49+
return nil, &StatusError{Code: ErrorInvalidArgument, Message: "explicit workspace is required"}
50+
}
51+
response, err := s.client.GetServiceRouteSnapshot(ctx, &pb.GetServiceRouteSnapshotRequest{WorkspaceScope: namedWorkspaceScope(workspace)})
52+
if err != nil {
53+
return nil, converter.FromGRPCError(err)
54+
}
55+
return converter.ServiceRouteSnapshotFromProto(response.GetSnapshot(), workspace)
56+
}
57+
58+
func (s *serviceDiscoveryClient) Watch(ctx context.Context, workspace string) (ServiceRouteWatch, error) {
59+
if workspace == "" {
60+
return nil, &StatusError{Code: ErrorInvalidArgument, Message: "explicit workspace is required"}
61+
}
62+
ctx, cancel := context.WithCancel(ctx)
63+
stream, err := s.client.WatchServiceRoutes(ctx, &pb.WatchServiceRoutesRequest{WorkspaceScope: namedWorkspaceScope(workspace)})
64+
if err != nil {
65+
cancel()
66+
return nil, converter.FromGRPCError(err)
67+
}
68+
return &serviceRouteWatch{stream: stream, cancel: cancel, workspace: workspace}, nil
69+
}
70+
71+
type serviceRouteWatch struct {
72+
stream grpc.ServerStreamingClient[pb.WatchServiceRoutesResponse]
73+
cancel context.CancelFunc
74+
workspace string
75+
once sync.Once
76+
}
77+
78+
func (w *serviceRouteWatch) Close() { w.once.Do(w.cancel) }
79+
80+
func (w *serviceRouteWatch) Recv() (*ServiceRouteSnapshot, error) {
81+
response, err := w.stream.Recv()
82+
if err != nil {
83+
w.Close()
84+
if err == io.EOF {
85+
return nil, io.EOF
86+
}
87+
return nil, converter.FromGRPCError(err)
88+
}
89+
result, err := converter.ServiceRouteSnapshotFromProto(response.GetSnapshot(), w.workspace)
90+
if err != nil {
91+
w.Close()
92+
}
93+
return result, err
94+
}
Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package v1
5+
6+
import (
7+
"context"
8+
"io"
9+
"net"
10+
"testing"
11+
"time"
12+
13+
dm "github.com/NVIDIA/OpenShell/sdk/go/proto/datamodelv1"
14+
pb "github.com/NVIDIA/OpenShell/sdk/go/proto/openshellv1"
15+
"github.com/stretchr/testify/require"
16+
"google.golang.org/grpc"
17+
"google.golang.org/grpc/codes"
18+
"google.golang.org/grpc/credentials/insecure"
19+
"google.golang.org/grpc/status"
20+
"google.golang.org/grpc/test/bufconn"
21+
"google.golang.org/protobuf/types/known/timestamppb"
22+
)
23+
24+
type discoveryServer struct {
25+
pb.UnimplementedOpenShellServer
26+
views []*pb.ServiceRouteSnapshot
27+
terminal error
28+
block bool
29+
canceled chan struct{}
30+
}
31+
32+
func discoveryView(revision string, expires time.Time) *pb.ServiceRouteSnapshot {
33+
return &pb.ServiceRouteSnapshot{Revision: revision, Routes: []*pb.DiscoveredServiceRoute{{
34+
Workspace: "default", Sandbox: "demo", Name: "web", Hostname: "demo.example.com",
35+
Endpoint: &pb.ServiceRouteEndpoint{Host: "10.0.0.1", Port: 9443},
36+
InstallationVersion: "owner-1", DeclarationRevision: 7, ExpiresAt: timestamppb.New(expires),
37+
}}}
38+
}
39+
40+
func (s *discoveryServer) GetServiceRouteSnapshot(_ context.Context, req *pb.GetServiceRouteSnapshotRequest) (*pb.GetServiceRouteSnapshotResponse, error) {
41+
if req.GetWorkspaceScope().GetWorkspace() != "default" {
42+
return nil, status.Error(codes.PermissionDenied, "workspace denied")
43+
}
44+
return &pb.GetServiceRouteSnapshotResponse{Snapshot: s.views[0]}, nil
45+
}
46+
func (s *discoveryServer) WatchServiceRoutes(req *pb.WatchServiceRoutesRequest, stream grpc.ServerStreamingServer[pb.WatchServiceRoutesResponse]) error {
47+
if _, ok := req.GetWorkspaceScope().GetSelection().(*dm.WorkspaceSelector_Workspace); !ok {
48+
return status.Error(codes.InvalidArgument, "explicit workspace required")
49+
}
50+
for _, view := range s.views {
51+
if err := stream.Send(&pb.WatchServiceRoutesResponse{Snapshot: view}); err != nil {
52+
return err
53+
}
54+
}
55+
if s.block {
56+
<-stream.Context().Done()
57+
close(s.canceled)
58+
return status.FromContextError(stream.Context().Err()).Err()
59+
}
60+
return s.terminal
61+
}
62+
63+
func discoveryClientForTest(t *testing.T, mock *discoveryServer) *serviceDiscoveryClient {
64+
t.Helper()
65+
listener := bufconn.Listen(bufSize)
66+
server := grpc.NewServer()
67+
pb.RegisterOpenShellServer(server, mock)
68+
go func() { _ = server.Serve(listener) }()
69+
conn, err := grpc.NewClient("passthrough:///bufconn", grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), grpc.WithTransportCredentials(insecure.NewCredentials()))
70+
require.NoError(t, err)
71+
t.Cleanup(func() { _ = conn.Close(); server.Stop(); _ = listener.Close() })
72+
return newServiceDiscoveryClient(conn)
73+
}
74+
75+
func TestDiscoverySnapshotAndWatchPreserveRenewalWithdrawalAndEOF(t *testing.T) {
76+
expiry := time.Now().Add(10 * time.Second).UTC()
77+
initial := discoveryView("revision-1", expiry)
78+
renewal := discoveryView("revision-2", expiry.Add(4*time.Second))
79+
empty := &pb.ServiceRouteSnapshot{Revision: "revision-3"}
80+
client := discoveryClientForTest(t, &discoveryServer{views: []*pb.ServiceRouteSnapshot{initial, renewal, empty}})
81+
snapshot, err := client.Snapshot(context.Background(), "default")
82+
require.NoError(t, err)
83+
require.Equal(t, "demo.example.com", snapshot.Routes[0].Hostname)
84+
require.Equal(t, expiry, snapshot.Routes[0].ExpiresAt)
85+
watch, err := client.Watch(context.Background(), "default")
86+
require.NoError(t, err)
87+
defer watch.Close()
88+
first, err := watch.Recv()
89+
require.NoError(t, err)
90+
second, err := watch.Recv()
91+
require.NoError(t, err)
92+
require.NotEqual(t, first.Revision, second.Revision)
93+
require.Equal(t, first.Routes[0].InstallationVersion, second.Routes[0].InstallationVersion)
94+
require.NotEqual(t, first.Routes[0].ExpiresAt, second.Routes[0].ExpiresAt)
95+
withdrawn, err := watch.Recv()
96+
require.NoError(t, err)
97+
require.Empty(t, withdrawn.Routes)
98+
none, err := watch.Recv()
99+
require.ErrorIs(t, err, io.EOF)
100+
require.Nil(t, none)
101+
_, err = client.Snapshot(context.Background(), "")
102+
require.True(t, IsInvalidArgument(err))
103+
_, err = client.Watch(context.Background(), "")
104+
require.True(t, IsInvalidArgument(err))
105+
_, err = client.Snapshot(context.Background(), "other")
106+
require.True(t, IsPermissionDenied(err))
107+
}
108+
109+
func TestDiscoveryErrorsAreNotEmptySnapshotsAndCloseCancelsRPC(t *testing.T) {
110+
view := discoveryView("revision", time.Now().Add(time.Second))
111+
client := discoveryClientForTest(t, &discoveryServer{views: []*pb.ServiceRouteSnapshot{view}, terminal: status.Error(codes.PermissionDenied, "revoked")})
112+
watch, err := client.Watch(context.Background(), "default")
113+
require.NoError(t, err)
114+
_, err = watch.Recv()
115+
require.NoError(t, err)
116+
next, err := watch.Recv()
117+
require.Nil(t, next)
118+
require.True(t, IsPermissionDenied(err))
119+
mock := &discoveryServer{views: []*pb.ServiceRouteSnapshot{view}, block: true, canceled: make(chan struct{})}
120+
client = discoveryClientForTest(t, mock)
121+
watch, err = client.Watch(context.Background(), "default")
122+
require.NoError(t, err)
123+
_, err = watch.Recv()
124+
require.NoError(t, err)
125+
watch.Close()
126+
watch.Close()
127+
_, err = watch.Recv()
128+
require.True(t, IsCancelled(err))
129+
select {
130+
case <-mock.canceled:
131+
case <-time.After(time.Second):
132+
t.Fatal("watch cancellation did not reach server")
133+
}
134+
}

0 commit comments

Comments
 (0)