Extract clients from onebox.go (#10575)

## What changed

`onebox.go` now creates its frontend/admin/operator, history/scheduler,
and matching test clients directly from the configured test hosts.

## Why?

This removes the `fx.Populate` references. One step closer to the goal
of deleting the separate `onebox.go` fx graph all-together and making
use of the production fx graph from the `temporal` package instead.

Author: stephanos

common/testing/await/require_ctx_test.go

@@ -478,6 +478,7 @@ func newRecordingTB() *recordingTB {
 
 func (r *recordingTB) Helper()      {}
 func (r *recordingTB) Failed() bool { return r.failed.Load() }
+func (r *recordingTB) Name() string { return "recordingTB" }
 func (r *recordingTB) Logf(format string, args ...any) {
 	r.mu.Lock()
 	defer r.mu.Unlock()

common/testing/grpcinject/fx.go

@@ -8,9 +8,13 @@ import (
 	"time"
 
 	"go.temporal.io/server/common/debug"
+	"google.golang.org/grpc/metadata"
 )
 
-const defaultTimeout = 90 * time.Second
+const (
+	defaultTimeout      = 90 * time.Second
+	testNameMetadataKey = "temporal-test-name"
+)
 
 type contextStore struct {
 	sync.Mutex
@@ -97,6 +101,10 @@ func getContextState(tb testing.TB, timeout time.Duration) *contextState {
 	}
 
 	ctx, cancel := context.WithTimeout(tb.Context(), timeout)
+
+	// Annotate gRPC requests with the test name for OTEL tracing.
+	ctx = metadata.AppendToOutgoingContext(ctx, testNameMetadataKey, tb.Name())
+
 	st := &contextState{
 		ctx:        ctx,
 		cancel:     cancel,

common/testing/grpcinject/grpcinject.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/metadata"
 )
 
 func TestWithTimeout(t *testing.T) {
@@ -18,6 +19,15 @@ func TestWithTimeout(t *testing.T) {
 	require.WithinDuration(t, time.Now().Add(time.Second), deadline, 50*time.Millisecond)
 }
 
+func TestNameMetadata(t *testing.T) {
+	t.Parallel()
+
+	ctx := New(t)
+	md, ok := metadata.FromOutgoingContext(ctx)
+	require.True(t, ok)
+	require.Equal(t, []string{t.Name()}, md.Get(testNameMetadataKey))
+}
+
 func TestContextDecorators(t *testing.T) {
 	t.Parallel()
 

common/testing/testcontext/context.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"go.temporal.io/api/serviceerror"
 	sdkclient "go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/worker"
 	"go.temporal.io/sdk/workflow"
@@ -130,5 +131,6 @@ func (s *AddTasksSuite) TestAddTasks_ErrGetShardByID() {
 		ShardId: 0,
 	})
 	s.Error(err)
-	s.Contains(strings.ToLower(err.Error()), "invalid shardid")
+	s.ErrorAs(err, new(*serviceerror.InvalidArgument))
+	s.Contains(strings.ToLower(err.Error()), "shard id cannot be equal or lower than zero")
 }

common/testing/testcontext/context_test.go

@@ -0,0 +1,163 @@
+package testcore
+
+import (
+	"crypto/tls"
+	"fmt"
+	"sync"
+
+	"go.temporal.io/api/operatorservice/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/api/adminservice/v1"
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
+	schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/membership/static"
+	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/primitives"
+	"go.temporal.io/server/common/rpc"
+	"go.temporal.io/server/common/rpc/encryption"
+	"google.golang.org/grpc"
+)
+
+type clients struct {
+	logger            log.Logger
+	hostsByService    map[primitives.ServiceName]static.Hosts
+	tlsConfigProvider *encryption.FixedTLSConfigProvider
+
+	frontend frontendClients
+	history  historyClients
+	matching matchingClient
+}
+
+type frontendClients struct {
+	once     sync.Once
+	conn     *grpc.ClientConn
+	admin    adminservice.AdminServiceClient
+	frontend workflowservice.WorkflowServiceClient
+	operator operatorservice.OperatorServiceClient
+}
+
+type historyClients struct {
+	once      sync.Once
+	conn      *grpc.ClientConn
+	history   historyservice.HistoryServiceClient
+	scheduler schedulerpb.SchedulerServiceClient
+}
+
+type matchingClient struct {
+	client matchingservice.MatchingServiceClient
+}
+
+func newClients(
+	logger log.Logger,
+	hostsByService map[primitives.ServiceName]static.Hosts,
+	tlsConfigProvider *encryption.FixedTLSConfigProvider,
+) clients {
+	return clients{
+		logger:            logger,
+		hostsByService:    hostsByService,
+		tlsConfigProvider: tlsConfigProvider,
+	}
+}
+
+func (c *clients) AdminClient() adminservice.AdminServiceClient {
+	c.ensureFrontend()
+	return c.frontend.admin
+}
+
+func (c *clients) OperatorClient() operatorservice.OperatorServiceClient {
+	c.ensureFrontend()
+	return c.frontend.operator
+}
+
+func (c *clients) FrontendClient() workflowservice.WorkflowServiceClient {
+	c.ensureFrontend()
+	return c.frontend.frontend
+}
+
+func (c *clients) ensureFrontend() {
+	c.frontend.once.Do(func() {
+		conn, err := c.newConn(primitives.FrontendService)
+		if err != nil {
+			c.logger.Fatal("unable to create frontend test client", tag.Error(err))
+		}
+		c.frontend.conn = conn
+		c.frontend.admin = adminservice.NewAdminServiceClient(conn)
+		c.frontend.frontend = workflowservice.NewWorkflowServiceClient(conn)
+		c.frontend.operator = operatorservice.NewOperatorServiceClient(conn)
+	})
+}
+
+func (c *clients) HistoryClient() historyservice.HistoryServiceClient {
+	c.ensureHistory()
+	return c.history.history
+}
+
+func (c *clients) SchedulerClient() schedulerpb.SchedulerServiceClient {
+	c.ensureHistory()
+	return c.history.scheduler
+}
+
+func (c *clients) ensureHistory() {
+	c.history.once.Do(func() {
+		conn, err := c.newConn(primitives.HistoryService)
+		if err != nil {
+			c.logger.Fatal("unable to create history test client", tag.Error(err))
+		}
+		c.history.conn = conn
+		c.history.history = historyservice.NewHistoryServiceClient(conn)
+		c.history.scheduler = schedulerpb.NewSchedulerServiceClient(conn)
+	})
+}
+
+func (c *clients) MatchingClient() matchingservice.MatchingServiceClient {
+	return c.matching.client
+}
+
+func (c *clients) close() []error {
+	var errs []error
+	for _, conn := range []*grpc.ClientConn{
+		c.frontend.conn,
+		c.history.conn,
+	} {
+		if conn != nil {
+			errs = append(errs, conn.Close())
+		}
+	}
+	c.frontend.conn = nil
+	c.history.conn = nil
+	return errs
+}
+
+func (c *clients) newConn(serviceName primitives.ServiceName) (*grpc.ClientConn, error) {
+	address, err := c.grpcAddress(serviceName)
+	if err != nil {
+		return nil, err
+	}
+	tlsConfig, err := c.tlsConfig(serviceName)
+	if err != nil {
+		return nil, err
+	}
+
+	return rpc.Dial(address, tlsConfig, c.logger, metrics.NoopMetricsHandler)
+}
+
+func (c *clients) grpcAddress(serviceName primitives.ServiceName) (string, error) {
+	hosts := c.hostsByService[serviceName].All
+	if len(hosts) == 0 {
+		return "", fmt.Errorf("no %s gRPC hosts configured", serviceName)
+	}
+	return hosts[0], nil
+}
+
+func (c *clients) tlsConfig(serviceName primitives.ServiceName) (*tls.Config, error) {
+	if c.tlsConfigProvider == nil {
+		return nil, nil
+	}
+	if serviceName == primitives.FrontendService {
+		return c.tlsConfigProvider.GetFrontendClientConfig()
+	}
+	return c.tlsConfigProvider.GetInternodeClientConfig()
+}

tests/add_tasks_test.go

@@ -48,8 +48,6 @@ import (
 	"go.temporal.io/server/common/testing/updateutils"
 	"go.temporal.io/server/components/nexusoperations"
 	"go.uber.org/fx"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/metadata"
 )
 
 type (
@@ -378,11 +376,6 @@ func (s *FunctionalTestBase) SetupTest() {
 	s.initAssertions()
 	s.setupSdk()
 	s.taskPoller = taskpoller.New(s.T(), s.FrontendClient(), s.Namespace().String())
-
-	// Annotate gRPC requests with the test name for OTEL tracing.
-	s.testCluster.host.grpcClientInterceptor.Set(func(ctx context.Context) context.Context {
-		return metadata.AppendToOutgoingContext(ctx, "temporal-test-name", s.T().Name())
-	})
 }
 
 func (s *FunctionalTestBase) SetupSubTest() {
@@ -435,13 +428,6 @@ func (s *FunctionalTestBase) setupSdk() {
 		clientOptions.ConnectionOptions.TLS = provider.FrontendClientConfig
 	}
 
-	if interceptor := s.testCluster.host.grpcClientInterceptor; interceptor != nil {
-		clientOptions.ConnectionOptions.DialOptions = []grpc.DialOption{
-			grpc.WithUnaryInterceptor(interceptor.Unary()),
-			grpc.WithStreamInterceptor(interceptor.Stream()),
-		}
-	}
-
 	var err error
 	s.sdkClient, err = sdkclient.Dial(clientOptions)
 	s.NoError(err)
@@ -499,7 +485,6 @@ func (s *FunctionalTestBase) tearDownTestCluster() error {
 func (s *FunctionalTestBase) TearDownTest() {
 	s.exportOTELTraces()
 	s.tearDownSdk()
-	s.testCluster.host.grpcClientInterceptor.Set(nil)
 }
 
 // **IMPORTANT**: When overridding this, make sure to invoke `s.FunctionalTestBase.TearDownSubTest()`.