Add support for forwarding Nexus HTTP requests

common/cluster/frontend_http_client.go

@@ -0,0 +1,114 @@
+// The MIT License
+//
+// Copyright (c) 2024 Temporal Technologies Inc.  All rights reserved.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package cluster
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net/http"
+	"net/url"
+
+	"go.temporal.io/api/serviceerror"
+
+	"go.temporal.io/server/common/collection"
+)
+
+type tlsConfigProvider interface {
+	GetRemoteClusterClientConfig(hostname string) (*tls.Config, error)
+}
+
+type FrontendHTTPClient struct {
+	http.Client
+	Address string
+}
+
+type FrontendHTTPClientCache struct {
+	metadata    Metadata
+	tlsProvider tlsConfigProvider
+	clients     *collection.FallibleOnceMap[string, *FrontendHTTPClient]
+}
+
+func NewFrontendHTTPClientCache(
+	metadata Metadata,
+	tlsProvider tlsConfigProvider,
+) *FrontendHTTPClientCache {
+	cache := &FrontendHTTPClientCache{
+		metadata:    metadata,
+		tlsProvider: tlsProvider,
+	}
+	cache.clients = collection.NewFallibleOnceMap(cache.newClientForCluster)
+	metadata.RegisterMetadataChangeCallback(cache, cache.evictionCallback)
+	return cache
+}
+
+// Get returns a cached HttpClient if available, or constructs a new one for the given cluster name.
+func (c *FrontendHTTPClientCache) Get(targetClusterName string) (*FrontendHTTPClient, error) {
+	return c.clients.Get(targetClusterName)
+}
+
+func (c *FrontendHTTPClientCache) newClientForCluster(targetClusterName string) (*FrontendHTTPClient, error) {
+	targetInfo, ok := c.metadata.GetAllClusterInfo()[targetClusterName]
+	if !ok {
+		return nil, serviceerror.NewNotFound(fmt.Sprintf("could not find cluster metadata for cluster %s", targetClusterName))
+	}
+
+	address, err := url.Parse(targetInfo.HTTPAddress)
+	if err != nil {
+		return nil, err
+	}
+
+	client := http.Client{}
+
+	if c.tlsProvider != nil {
+		tlsClientConfig, err := c.tlsProvider.GetRemoteClusterClientConfig(address.Hostname())
+		if err != nil {
+			return nil, err
+		}
+		client.Transport = &http.Transport{TLSClientConfig: tlsClientConfig}
+	}
+
+	return &FrontendHTTPClient{
+		Address: targetInfo.HTTPAddress,
+		Client:  client,
+	}, nil
+}
+
+// evictionCallback is invoked by cluster.Metadata when cluster information changes.
+// It invalidates clients which are either no longer present or have had their HTTP address changed.
+// It is assumed that TLS information has not changed for clusters that are unmodified.
+func (c *FrontendHTTPClientCache) evictionCallback(oldClusterMetadata map[string]*ClusterInformation, newClusterMetadata map[string]*ClusterInformation) {
+	for oldClusterName, oldClusterInfo := range oldClusterMetadata {
+		if oldClusterName == c.metadata.GetCurrentClusterName() || oldClusterInfo == nil {
+			continue
+		}
+
+		newClusterInfo, exists := newClusterMetadata[oldClusterName]
+		if !exists || oldClusterInfo.HTTPAddress != newClusterInfo.HTTPAddress {
+			// Cluster was removed or had its HTTP address changed, so invalidate the cached client for that cluster.
+			client, ok := c.clients.Pop(oldClusterName)
+			if ok {
+				client.CloseIdleConnections()
+			}
+		}
+	}
+}

common/collection/oncemap.go

@@ -95,3 +95,13 @@ func (p *FallibleOnceMap[K, T]) Get(key K) (T, error) {
 
 	return value, nil
 }
+
+func (p *FallibleOnceMap[K, T]) Pop(key K) (T, bool) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	val, ok := p.inner[key]
+	if ok {
+		delete(p.inner, key)
+	}
+	return val, ok
+}

common/nexus/failure.go

@@ -24,6 +24,7 @@ package nexus
 
 import (
 	"errors"
+	"net/http"
 
 	"github.com/nexus-rpc/sdk-go/nexus"
 	commonpb "go.temporal.io/api/common/v1"
@@ -192,3 +193,40 @@ func AdaptAuthorizeError(err error) error {
 	}
 	return nexus.HandlerErrorf(nexus.HandlerErrorTypeUnauthorized, "permission denied")
 }
+
+func HandlerErrorFromClientError(err error) error {
+	var unexpectedRespErr *nexus.UnexpectedResponseError
+	if errors.As(err, &unexpectedRespErr) {
+		handlerErr := &nexus.HandlerError{
+			Failure: unexpectedRespErr.Failure,
+		}
+
+		switch unexpectedRespErr.Response.StatusCode {
+		case http.StatusBadRequest:
+			handlerErr.Type = nexus.HandlerErrorTypeBadRequest
+		case http.StatusUnauthorized:
+			handlerErr.Type = nexus.HandlerErrorTypeUnauthenticated
+		case http.StatusForbidden:
+			handlerErr.Type = nexus.HandlerErrorTypeUnauthorized
+		case http.StatusNotFound:
+			handlerErr.Type = nexus.HandlerErrorTypeNotFound
+		case http.StatusTooManyRequests:
+			handlerErr.Type = nexus.HandlerErrorTypeResourceExhausted
+		case http.StatusInternalServerError:
+			handlerErr.Type = nexus.HandlerErrorTypeInternal
+		case http.StatusNotImplemented:
+			handlerErr.Type = nexus.HandlerErrorTypeNotImplemented
+		case http.StatusServiceUnavailable:
+			handlerErr.Type = nexus.HandlerErrorTypeUnavailable
+		case nexus.StatusDownstreamError:
+			handlerErr.Type = nexus.HandlerErrorTypeDownstreamError
+		case nexus.StatusDownstreamTimeout:
+			handlerErr.Type = nexus.HandlerErrorTypeDownstreamTimeout
+		}
+
+		return handlerErr
+	}
+
+	// Let the nexus SDK handle this for us (log and convert to an internal error).
+	return err
+}

common/resource/fx.go

@@ -122,6 +122,7 @@ var Module = fx.Options(
 	fx.Provide(MatchingRawClientProvider),
 	fx.Provide(MatchingClientProvider),
 	membership.GRPCResolverModule,
+	fx.Provide(ClusterHttpClientCacheProvider),
 	fx.Invoke(RegisterBootstrapContainer),
 	fx.Provide(PersistenceConfigProvider),
 	fx.Provide(health.NewServer),
@@ -408,6 +409,13 @@ func RPCFactoryProvider(
 	), nil
 }
 
+func ClusterHttpClientCacheProvider(
+	metadata cluster.Metadata,
+	tlsConfigProvider encryption.TLSConfigProvider,
+) *cluster.FrontendHTTPClientCache {
+	return cluster.NewFrontendHTTPClientCache(metadata, tlsConfigProvider)
+}
+
 func getFrontendConnectionDetails(
 	cfg *config.Config,
 	tlsConfigProvider encryption.TLSConfigProvider,

service/frontend/fx.go

@@ -710,9 +710,12 @@ func RegisterNexusHTTPHandler(
 	serviceName primitives.ServiceName,
 	matchingClient resource.MatchingClient,
 	metricsHandler metrics.Handler,
+	clusterMetadata cluster.Metadata,
+	clientCache *cluster.FrontendHTTPClientCache,
 	namespaceRegistry namespace.Registry,
 	endpointRegistry *nexus.EndpointRegistry,
 	authInterceptor *authorization.Interceptor,
+	redirectionInterceptor *RedirectionInterceptor,
 	namespaceRateLimiterInterceptor *interceptor.NamespaceRateLimitInterceptor,
 	namespaceCountLimiterInterceptor *interceptor.ConcurrentRequestLimitInterceptor,
 	namespaceValidatorInterceptor *interceptor.NamespaceValidatorInterceptor,
@@ -724,9 +727,12 @@ func RegisterNexusHTTPHandler(
 		serviceConfig,
 		matchingClient,
 		metricsHandler,
+		clusterMetadata,
+		clientCache,
 		namespaceRegistry,
 		endpointRegistry,
 		authInterceptor,
+		redirectionInterceptor,
 		namespaceValidatorInterceptor,
 		namespaceRateLimiterInterceptor,
 		namespaceCountLimiterInterceptor,