Flesh out relay_service configuration and heartbeats (#57645)

* proto: relay server heartbeat fields

* make grpc

* Update relay_service config section

* Fill in the Relay heartbeat

api/gen/proto/go/teleport/presence/v1/relay_server.pb.go

@@ -119,7 +119,24 @@ func (x *RelayServer) GetSpec() *RelayServer_Spec {
 
 // resource spec
 type RelayServer_Spec struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// host IDs of Proxy Service instances that this server is available on
+	// through a reverse tunnel
+	ProxyIds []string `protobuf:"bytes,1,rep,name=proxy_ids,json=proxyIds,proto3" json:"proxy_ids,omitempty"`
+	// configured hostname (or nodename) of the machine, for troubleshooting and
+	// debugging
+	Hostname string `protobuf:"bytes,2,opt,name=hostname,proto3" json:"hostname,omitempty"`
+	// the name of the Relay group this server belongs to
+	RelayGroup string `protobuf:"bytes,3,opt,name=relay_group,json=relayGroup,proto3" json:"relay_group,omitempty"`
+	// address and port that this server is reachable at by other Relay Service
+	// instance of the same group
+	PeerAddr string `protobuf:"bytes,4,opt,name=peer_addr,json=peerAddr,proto3" json:"peer_addr,omitempty"`
+	// random string chosen for the duration of the process, for troubleshooting
+	// and debugging
+	Nonce string `protobuf:"bytes,5,opt,name=nonce,proto3" json:"nonce,omitempty"`
+	// set after the Teleport instance has received a termination signal (but
+	// hasn't necessarily began shutting down)
+	Terminating   bool `protobuf:"varint,6,opt,name=terminating,proto3" json:"terminating,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -154,18 +171,67 @@ func (*RelayServer_Spec) Descriptor() ([]byte, []int) {
 	return file_teleport_presence_v1_relay_server_proto_rawDescGZIP(), []int{0, 0}
 }
 
+func (x *RelayServer_Spec) GetProxyIds() []string {
+	if x != nil {
+		return x.ProxyIds
+	}
+	return nil
+}
+
+func (x *RelayServer_Spec) GetHostname() string {
+	if x != nil {
+		return x.Hostname
+	}
+	return ""
+}
+
+func (x *RelayServer_Spec) GetRelayGroup() string {
+	if x != nil {
+		return x.RelayGroup
+	}
+	return ""
+}
+
+func (x *RelayServer_Spec) GetPeerAddr() string {
+	if x != nil {
+		return x.PeerAddr
+	}
+	return ""
+}
+
+func (x *RelayServer_Spec) GetNonce() string {
+	if x != nil {
+		return x.Nonce
+	}
+	return ""
+}
+
+func (x *RelayServer_Spec) GetTerminating() bool {
+	if x != nil {
+		return x.Terminating
+	}
+	return false
+}
+
 var File_teleport_presence_v1_relay_server_proto protoreflect.FileDescriptor
 
 const file_teleport_presence_v1_relay_server_proto_rawDesc = "" +
 	"\n" +
-	"'teleport/presence/v1/relay_server.proto\x12\x14teleport.presence.v1\x1a!teleport/header/v1/metadata.proto\"\xd4\x01\n" +
+	"'teleport/presence/v1/relay_server.proto\x12\x14teleport.presence.v1\x1a!teleport/header/v1/metadata.proto\"\x84\x03\n" +
 	"\vRelayServer\x12\x12\n" +
 	"\x04kind\x18\x01 \x01(\tR\x04kind\x12\x19\n" +
 	"\bsub_kind\x18\x02 \x01(\tR\asubKind\x12\x18\n" +
 	"\aversion\x18\x03 \x01(\tR\aversion\x128\n" +
 	"\bmetadata\x18\x04 \x01(\v2\x1c.teleport.header.v1.MetadataR\bmetadata\x12:\n" +
-	"\x04spec\x18\x05 \x01(\v2&.teleport.presence.v1.RelayServer.SpecR\x04spec\x1a\x06\n" +
-	"\x04SpecBTZRgithub.com/gravitational/teleport/api/gen/proto/go/teleport/presence/v1;presencev1b\x06proto3"
+	"\x04spec\x18\x05 \x01(\v2&.teleport.presence.v1.RelayServer.SpecR\x04spec\x1a\xb5\x01\n" +
+	"\x04Spec\x12\x1b\n" +
+	"\tproxy_ids\x18\x01 \x03(\tR\bproxyIds\x12\x1a\n" +
+	"\bhostname\x18\x02 \x01(\tR\bhostname\x12\x1f\n" +
+	"\vrelay_group\x18\x03 \x01(\tR\n" +
+	"relayGroup\x12\x1b\n" +
+	"\tpeer_addr\x18\x04 \x01(\tR\bpeerAddr\x12\x14\n" +
+	"\x05nonce\x18\x05 \x01(\tR\x05nonce\x12 \n" +
+	"\vterminating\x18\x06 \x01(\bR\vterminatingBTZRgithub.com/gravitational/teleport/api/gen/proto/go/teleport/presence/v1;presencev1b\x06proto3"
 
 var (
 	file_teleport_presence_v1_relay_server_proto_rawDescOnce sync.Once

api/proto/teleport/presence/v1/relay_server.proto

@@ -34,7 +34,28 @@ message RelayServer {
 
   // resource spec
   message Spec {
-    // TODO
+    // host IDs of Proxy Service instances that this server is available on
+    // through a reverse tunnel
+    repeated string proxy_ids = 1;
+
+    // configured hostname (or nodename) of the machine, for troubleshooting and
+    // debugging
+    string hostname = 2;
+
+    // the name of the Relay group this server belongs to
+    string relay_group = 3;
+
+    // address and port that this server is reachable at by other Relay Service
+    // instance of the same group
+    string peer_addr = 4;
+
+    // random string chosen for the duration of the process, for troubleshooting
+    // and debugging
+    string nonce = 5;
+
+    // set after the Teleport instance has received a termination signal (but
+    // hasn't necessarily began shutting down)
+    bool terminating = 6;
   }
   Spec spec = 5;
 }

lib/config/configuration.go

@@ -30,7 +30,9 @@ import (
 	"io"
 	"log/slog"
 	"maps"
+	"math"
 	"net"
+	"net/netip"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -3114,13 +3116,47 @@ func applyRelayConfig(fc *FileConfig, cfg *servicecfg.Config) error {
 	}
 	cfg.Relay.RelayGroup = fc.Relay.RelayGroup
 
-	if len(fc.Relay.APIPublicHostnames) < 1 {
-		return trace.BadParameter("missing relay_service.api_public_hostnames")
+	if fc.Relay.TargetConnectionCount < 1 || fc.Relay.TargetConnectionCount > math.MaxInt32 {
+		return trace.BadParameter("missing or invalid relay_service.target_connection_count")
 	}
-	if slices.Contains(fc.Relay.APIPublicHostnames, "") {
-		return trace.BadParameter("empty string in relay_service.api_public_hostnames")
+	cfg.Relay.TargetConnectionCount = int32(fc.Relay.TargetConnectionCount)
+
+	if len(fc.Relay.PublicHostnames) < 1 {
+		return trace.BadParameter("missing relay_service.public_hostnames")
 	}
-	cfg.Relay.APIPublicHostnames = slices.Clone(fc.Relay.APIPublicHostnames)
+	if slices.Contains(fc.Relay.PublicHostnames, "") {
+		return trace.BadParameter("empty string in relay_service.public_hostnames")
+	}
+	cfg.Relay.PublicHostnames = slices.Clone(fc.Relay.PublicHostnames)
+
+	if fc.Relay.TransportListenAddr == "" {
+		return trace.BadParameter("missing relay_service.transport_listen_addr")
+	}
+	transportListenAddr, err := netip.ParseAddrPort(fc.Relay.TransportListenAddr)
+	if err != nil {
+		return trace.Wrap(err, "parsing relay_service.transport_listen_addr")
+	}
+	cfg.Relay.TransportListenAddr = transportListenAddr
+	cfg.Relay.TransportPROXYProtocol = fc.Relay.TransportPROXYProtocol
+
+	if fc.Relay.PeerListenAddr == "" {
+		return trace.BadParameter("missing relay_service.peer_listen_addr")
+	}
+	peerListenAddr, err := netip.ParseAddrPort(fc.Relay.PeerListenAddr)
+	if err != nil {
+		return trace.Wrap(err, "parsing relay_service.peer_listen_addr")
+	}
+	cfg.Relay.PeerListenAddr = peerListenAddr
+
+	if fc.Relay.TunnelListenAddr == "" {
+		return trace.BadParameter("missing relay_service.tunnel_listen_addr")
+	}
+	tunnelListenAddr, err := netip.ParseAddrPort(fc.Relay.TunnelListenAddr)
+	if err != nil {
+		return trace.Wrap(err, "parsing relay_service.tunnel_listen_addr")
+	}
+	cfg.Relay.TunnelListenAddr = tunnelListenAddr
+	cfg.Relay.TunnelPROXYProtocol = fc.Relay.TunnelPROXYProtocol
 
 	return nil
 }

lib/config/fileconf.go

@@ -2936,9 +2936,33 @@ type Relay struct {
 	// enabled.
 	RelayGroup string `yaml:"relay_group"`
 
-	// APIPublicHostnames is the list of DNS names and IP addresses that the
+	// TargetConnectionCount is the connection count that agents are supposed to
+	// maintain when connecting to the Relay group of this instance.
+	TargetConnectionCount int `yaml:"target_connection_count"`
+
+	// PublicHostnames is the list of DNS names and IP addresses that the
 	// Relay service credentials should be authoritative for.
-	APIPublicHostnames []string `yaml:"api_public_hostnames"`
+	PublicHostnames []string `yaml:"public_hostnames"`
+
+	// TransportListenAddr is the listen address for the transport listener, in
+	// addr:port format.
+	TransportListenAddr string `yaml:"transport_listen_addr"`
+
+	// TransportPROXYProtocol is set if the transport listener should expect a
+	// PROXY protocol header in incoming connections.
+	TransportPROXYProtocol bool `yaml:"transport_proxy_protocol"`
+
+	// PeerListenAddr is the listen address for the peer listener, in addr:port
+	// format.
+	PeerListenAddr string `yaml:"peer_listen_addr"`
+
+	// TunnelListenAddr is the listen address for the tunnel listener, in
+	// addr:port format.
+	TunnelListenAddr string `yaml:"tunnel_listen_addr"`
+
+	// TunnelPROXYProtocol is set if the tunnel listener should expect a PROXY
+	// protocol header in incoming connections.
+	TunnelPROXYProtocol bool `yaml:"tunnel_proxy_protocol"`
 }
 
 // SessionRecordingEncryptionConfig is the session_recording_config.encryption

lib/service/relay.go

@@ -18,9 +18,12 @@ package service
 
 import (
 	"context"
+	"log/slog"
 	"sync/atomic"
 
+	"github.com/google/uuid"
 	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/gravitational/teleport"
 	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
@@ -36,6 +39,9 @@ func (process *TeleportProcess) initRelay() {
 
 func (process *TeleportProcess) runRelayService() error {
 	log := process.logger.With(teleport.ComponentKey, teleport.Component(teleport.ComponentRelay, process.id))
+	sublogger := func(subcomponent string) *slog.Logger {
+		return process.logger.With(teleport.ComponentKey, teleport.Component(teleport.ComponentRelay, process.id, subcomponent))
+	}
 
 	defer func() {
 		if err := process.closeImportedDescriptors(teleport.ComponentRelay); err != nil {
@@ -57,6 +63,7 @@ func (process *TeleportProcess) runRelayService() error {
 	// TODO(espadolini): use the access point
 	_ = accessPoint
 
+	nonce := uuid.NewString()
 	var relayServer atomic.Pointer[presencev1.RelayServer]
 	relayServer.Store(&presencev1.RelayServer{
 		Kind:    apitypes.KindRelayServer,
@@ -65,7 +72,11 @@ func (process *TeleportProcess) runRelayService() error {
 		Metadata: &headerv1.Metadata{
 			Name: conn.HostUUID(),
 		},
-		Spec: &presencev1.RelayServer_Spec{},
+		Spec: &presencev1.RelayServer_Spec{
+			Hostname:   process.Config.Hostname,
+			RelayGroup: process.Config.Relay.RelayGroup,
+			Nonce:      nonce,
+		},
 	})
 
 	hb, err := srv.NewRelayServerHeartbeat(srv.HeartbeatV2Config[*presencev1.RelayServer]{
@@ -79,15 +90,29 @@ func (process *TeleportProcess) runRelayService() error {
 		Announcer: nil,
 
 		OnHeartbeat: process.OnHeartbeat(teleport.ComponentRelay),
-	}, log)
+	}, sublogger("heartbeat"))
 	if err != nil {
 		return trace.Wrap(err)
 	}
 	go hb.Run()
 	defer hb.Close()
 
+	if err := process.closeImportedDescriptors(teleport.ComponentRelay); err != nil {
+		log.WarnContext(process.ExitContext(), "Failed closing imported file descriptors", "error", err)
+	}
+
 	process.BroadcastEvent(Event{Name: RelayReady})
-	log.InfoContext(process.ExitContext(), "The relay service has successfully started.")
+	log.InfoContext(process.ExitContext(), "The relay service has successfully started", "nonce", nonce)
+
+	_, _ = process.WaitForEvent(process.ExitContext(), TeleportTerminatingEvent)
+
+	log.InfoContext(process.ExitContext(), "Process is beginning shutdown, advertising terminating status and waiting for shutdown")
+
+	{
+		r := proto.CloneOf(relayServer.Load())
+		r.GetSpec().Terminating = true
+		relayServer.Store(r)
+	}
 
 	exitEvent, _ := process.WaitForEvent(process.ExitContext(), TeleportExitEvent)
 	ctx, _ := exitEvent.Payload.(context.Context)
@@ -96,15 +121,16 @@ func (process *TeleportProcess) runRelayService() error {
 		// WaitForEvent errored out because of the ungraceful shutdown; either
 		// way, process.ExitContext() is a done context and all operations
 		// should get canceled immediately
-		log.InfoContext(ctx, "Stopping the relay service ungracefully.")
 		ctx = process.ExitContext()
+		log.InfoContext(ctx, "Stopping the relay service ungracefully")
 	} else {
-		log.InfoContext(ctx, "Stopping the relay service.")
+		log.InfoContext(ctx, "Stopping the relay service")
 	}
 
+	warnOnErr(ctx, hb.Close(), log)
 	warnOnErr(ctx, conn.Close(), log)
 
-	log.InfoContext(ctx, "The relay service has stopped.")
+	log.InfoContext(ctx, "The relay service has stopped")
 
 	return nil
 }

lib/service/service.go

@@ -4119,7 +4119,7 @@ func (process *TeleportProcess) getAdditionalPrincipals(role types.SystemRole) (
 			}
 		}
 	case types.RoleRelay:
-		dnsNames = append(dnsNames, process.Config.Relay.APIPublicHostnames...)
+		dnsNames = append(dnsNames, process.Config.Relay.PublicHostnames...)
 	case types.RoleAuth, types.RoleAdmin:
 		addrs = process.Config.Auth.PublicAddrs
 	case types.RoleNode:

lib/service/servicecfg/relay.go

@@ -16,15 +16,38 @@
 
 package servicecfg
 
+import "net/netip"
+
 // RelayConfig contains the configuration for the Relay service.
 type RelayConfig struct {
 	Enabled bool
 
-	// RelayGroup is the Relay group name, required if the relay service is
+	// RelayGroup is the Relay group name, required if the Relay service is
 	// enabled.
 	RelayGroup string
 
-	// APIPublicHostnames is the list of DNS names and IP addresses that the
-	// Relay service credentials should be authoritative for.
-	APIPublicHostnames []string
+	// TargetConnectionCount is the connection count that agents are supposed to
+	// maintain when connecting to the Relay group of this instance.
+	TargetConnectionCount int32
+
+	// PublicHostnames is the list of DNS names and IP addresses that the Relay
+	// service credentials should be authoritative for.
+	PublicHostnames []string
+
+	// TransportListenAddr is the listen address for the transport listener.
+	TransportListenAddr netip.AddrPort
+
+	// TransportPROXYProtocol is set if the transport listener should expect a
+	// PROXY protocol header in incoming connections.
+	TransportPROXYProtocol bool
+
+	// PeerListenAddr is the listen address for the peer listener.
+	PeerListenAddr netip.AddrPort
+
+	// TunnelListenAddr is the listen address for the tunnel listener.
+	TunnelListenAddr netip.AddrPort
+
+	// TunnelPROXYProtocol is set if the tunnel listener should expect a PROXY
+	// protocol header in incoming connections.
+	TunnelPROXYProtocol bool
 }