HTTP API support (#4543)

common/config/config.go

@@ -85,7 +85,7 @@ type (
 
 	// RPC contains the rpc config items
 	RPC struct {
-		// GRPCPort is the port  on which gRPC will listen
+		// GRPCPort is the port on which gRPC will listen
 		GRPCPort int `yaml:"grpcPort"`
 		// Port used for membership listener
 		MembershipPort int `yaml:"membershipPort"`
@@ -95,6 +95,12 @@ type (
 		// check net.ParseIP for supported syntax, only IPv4 is supported,
 		// mutually exclusive with `BindOnLocalHost` option
 		BindOnIP string `yaml:"bindOnIP"`
+		// HTTPPort is the port on which HTTP will listen. If unset/0, HTTP will be
+		// disabled. This setting only applies to the frontend service.
+		HTTPPort int `yaml:"httpPort"`
+		// HTTPAdditionalForwardedHeaders adds additional headers to the default set
+		// forwarded from HTTP to gRPC.
+		HTTPAdditionalForwardedHeaders []string `yaml:"httpAdditionalForwardedHeaders"`
 	}
 
 	// Global contains config items that apply process-wide to all services

common/headers/version_checker.go

@@ -38,6 +38,7 @@ import (
 
 const (
 	ClientNameServer        = "temporal-server"
+	ClientNameServerHTTP    = "temporal-server-http"
 	ClientNameGoSDK         = "temporal-go"
 	ClientNameJavaSDK       = "temporal-java"
 	ClientNamePHPSDK        = "temporal-php"

common/metrics/metric_defs.go

@@ -1163,7 +1163,7 @@ const (
 var (
 	ServiceRequests = NewCounterDef(
 		"service_requests",
-		WithDescription("The number of gRPC requests received by the service."),
+		WithDescription("The number of RPC requests received by the service."),
 	)
 	ServicePendingRequests                   = NewGaugeDef("service_pending_requests")
 	ServiceFailures                          = NewCounterDef("service_errors")
@@ -1231,6 +1231,10 @@ var (
 	VersionCheckFailedCount                  = NewCounterDef("version_check_failed")
 	VersionCheckRequestFailedCount           = NewCounterDef("version_check_request_failed")
 	VersionCheckLatency                      = NewTimerDef("version_check_latency")
+	HTTPServiceRequests                      = NewCounterDef(
+		"http_service_requests",
+		WithDescription("The number of HTTP requests received by the service."),
+	)
 
 	// History
 	CacheRequests                                = NewCounterDef("cache_requests")

common/metrics/metricstest/capture_handler.go

@@ -0,0 +1,146 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// 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 metricstest
+
+import (
+	"sync"
+	"time"
+
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/metrics"
+)
+
+// CapturedRecording is a single recording. Fields here should not be mutated.
+type CapturedRecording struct {
+	Value any
+	Tags  map[string]string
+	Unit  metrics.MetricUnit
+}
+
+// Capture is a specific capture instance.
+type Capture struct {
+	recordings     map[string][]*CapturedRecording
+	recordingsLock sync.RWMutex
+}
+
+// Snapshot returns a copy of all metrics recorded, keyed by name.
+func (c *Capture) Snapshot() map[string][]*CapturedRecording {
+	c.recordingsLock.RLock()
+	defer c.recordingsLock.RUnlock()
+	ret := make(map[string][]*CapturedRecording, len(c.recordings))
+	for k, v := range c.recordings {
+		recs := make([]*CapturedRecording, len(v))
+		copy(recs, v)
+		ret[k] = recs
+	}
+	return ret
+}
+
+func (c *Capture) record(name string, r *CapturedRecording) {
+	c.recordingsLock.Lock()
+	defer c.recordingsLock.Unlock()
+	c.recordings[name] = append(c.recordings[name], r)
+}
+
+// CaptureHandler is a [metrics.Handler] that captures each metric recording.
+type CaptureHandler struct {
+	tags         []metrics.Tag
+	captures     map[*Capture]struct{}
+	capturesLock *sync.RWMutex
+}
+
+var _ metrics.Handler = (*CaptureHandler)(nil)
+
+// NewCaptureHandler creates a new [metrics.Handler] that captures.
+func NewCaptureHandler() *CaptureHandler {
+	return &CaptureHandler{
+		captures:     map[*Capture]struct{}{},
+		capturesLock: &sync.RWMutex{},
+	}
+}
+
+// StartCapture returns a started capture. StopCapture should be called on
+// complete.
+func (c *CaptureHandler) StartCapture() *Capture {
+	capture := &Capture{recordings: map[string][]*CapturedRecording{}}
+	c.capturesLock.Lock()
+	defer c.capturesLock.Unlock()
+	c.captures[capture] = struct{}{}
+	return capture
+}
+
+// StopCapture stops capturing metrics for the given capture instance.
+func (c *CaptureHandler) StopCapture(capture *Capture) {
+	c.capturesLock.Lock()
+	defer c.capturesLock.Unlock()
+	delete(c.captures, capture)
+}
+
+// WithTags implements [metrics.Handler.WithTags].
+func (c *CaptureHandler) WithTags(tags ...metrics.Tag) metrics.Handler {
+	return &CaptureHandler{
+		tags:         append(append(make([]metrics.Tag, 0, len(c.tags)+len(tags)), c.tags...), tags...),
+		captures:     c.captures,
+		capturesLock: c.capturesLock,
+	}
+}
+
+func (c *CaptureHandler) record(name string, v any, unit metrics.MetricUnit, tags ...metrics.Tag) {
+	rec := &CapturedRecording{Value: v, Tags: make(map[string]string, len(c.tags)+len(tags)), Unit: unit}
+	for _, tag := range c.tags {
+		rec.Tags[tag.Key()] = tag.Value()
+	}
+	for _, tag := range tags {
+		rec.Tags[tag.Key()] = tag.Value()
+	}
+	c.capturesLock.RLock()
+	defer c.capturesLock.RUnlock()
+	for c := range c.captures {
+		c.record(name, rec)
+	}
+}
+
+// Counter implements [metrics.Handler.Counter].
+func (c *CaptureHandler) Counter(name string) metrics.CounterIface {
+	return metrics.CounterFunc(func(v int64, tags ...metrics.Tag) { c.record(name, v, "", tags...) })
+}
+
+// Gauge implements [metrics.Handler.Gauge].
+func (c *CaptureHandler) Gauge(name string) metrics.GaugeIface {
+	return metrics.GaugeFunc(func(v float64, tags ...metrics.Tag) { c.record(name, v, "", tags...) })
+}
+
+// Timer implements [metrics.Handler.Timer].
+func (c *CaptureHandler) Timer(name string) metrics.TimerIface {
+	return metrics.TimerFunc(func(v time.Duration, tags ...metrics.Tag) { c.record(name, v, "", tags...) })
+}
+
+// Histogram implements [metrics.Handler.Histogram].
+func (c *CaptureHandler) Histogram(name string, unit metrics.MetricUnit) metrics.HistogramIface {
+	return metrics.HistogramFunc(func(v int64, tags ...metrics.Tag) { c.record(name, v, unit, tags...) })
+}
+
+// Stop implements [metrics.Handler.Stop].
+func (*CaptureHandler) Stop(log.Logger) {}

common/rpc/encryption/fixedTLSConfigProvider.go

@@ -0,0 +1,79 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// 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 encryption
+
+import (
+	"crypto/tls"
+	"time"
+)
+
+// FixedTLSConfigProvider is a [TLSConfigProvider] that is for fixed sets of TLS
+// configs. This is usually only used for testing.
+
+type FixedTLSConfigProvider struct {
+	InternodeServerConfig      *tls.Config
+	InternodeClientConfig      *tls.Config
+	FrontendServerConfig       *tls.Config
+	FrontendClientConfig       *tls.Config
+	RemoteClusterClientConfigs map[string]*tls.Config
+	CertExpirationChecker      CertExpirationChecker
+}
+
+var _ TLSConfigProvider = (*FixedTLSConfigProvider)(nil)
+
+// GetInternodeServerConfig implements [TLSConfigProvider.GetInternodeServerConfig].
+func (f *FixedTLSConfigProvider) GetInternodeServerConfig() (*tls.Config, error) {
+	return f.InternodeServerConfig, nil
+}
+
+// GetInternodeClientConfig implements [TLSConfigProvider.GetInternodeClientConfig].
+func (f *FixedTLSConfigProvider) GetInternodeClientConfig() (*tls.Config, error) {
+	return f.InternodeClientConfig, nil
+}
+
+// GetFrontendServerConfig implements [TLSConfigProvider.GetFrontendServerConfig].
+func (f *FixedTLSConfigProvider) GetFrontendServerConfig() (*tls.Config, error) {
+	return f.FrontendServerConfig, nil
+}
+
+// GetFrontendClientConfig implements [TLSConfigProvider.GetFrontendClientConfig].
+func (f *FixedTLSConfigProvider) GetFrontendClientConfig() (*tls.Config, error) {
+	return f.FrontendClientConfig, nil
+}
+
+// GetRemoteClusterClientConfig implements [TLSConfigProvider.GetRemoteClusterClientConfig].
+func (f *FixedTLSConfigProvider) GetRemoteClusterClientConfig(hostname string) (*tls.Config, error) {
+	return f.RemoteClusterClientConfigs[hostname], nil
+}
+
+// GetExpiringCerts implements [TLSConfigProvider.GetExpiringCerts].
+func (f *FixedTLSConfigProvider) GetExpiringCerts(
+	timeWindow time.Duration,
+) (expiring CertExpirationMap, expired CertExpirationMap, err error) {
+	if f.CertExpirationChecker != nil {
+		return f.CertExpirationChecker.GetExpiringCerts(timeWindow)
+	}
+	return nil, nil, nil
+}

common/rpc/grpc.go

@@ -52,6 +52,10 @@ const (
 	// MaxBackoffDelay is a maximum interval between reconnect attempts.
 	MaxBackoffDelay = 10 * time.Second
 
+	// MaxHTTPAPIRequestBytes is the maximum number of bytes an HTTP API request
+	// can have. This is currently set to the max gRPC request size.
+	MaxHTTPAPIRequestBytes = 4 * 1024 * 1024
+
 	// minConnectTimeout is the minimum amount of time we are willing to give a connection to complete.
 	minConnectTimeout = 20 * time.Second
 

config/development-cass-archival.yaml

@@ -35,6 +35,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-cass-es.yaml

@@ -42,6 +42,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-cass-s3.yaml

@@ -33,6 +33,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-cass.yaml

@@ -59,6 +59,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-cluster-a.yaml

@@ -42,6 +42,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-mysql-es.yaml

@@ -48,6 +48,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-mysql.yaml

@@ -50,6 +50,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-mysql8.yaml

@@ -50,6 +50,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-postgres-es.yaml

@@ -50,6 +50,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-postgres.yaml

@@ -50,6 +50,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-postgres12.yaml

@@ -50,6 +50,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-sqlite-file.yaml

@@ -74,6 +74,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

config/development-sqlite.yaml

@@ -70,6 +70,7 @@ services:
       grpcPort: 7233
       membershipPort: 6933
       bindOnLocalHost: true
+      httpPort: 7243
 
   matching:
     rpc:

docker/config_template.yaml

@@ -304,6 +304,7 @@ services:
             grpcPort: {{ $temporalGrpcPort }}
             membershipPort: {{ default .Env.FRONTEND_MEMBERSHIP_PORT "6933" }}
             bindOnIP: {{ default .Env.BIND_ON_IP "127.0.0.1" }}
+            httpPort: {{ default .Env.FRONTEND_HTTP_PORT "7243" }}
 
     {{- if .Env.USE_INTERNAL_FRONTEND }}
     internal-frontend: