provide `RPCContext` to all RPC handlers (#15430)

Upcoming work to instrument the rate of RPC requests by consumer (and eventually
rate limit) requires that we thread the `RPCContext` through all RPC
handlers so that we can access the underlying connection. This changeset adds
the context to everywhere we intend to initially support it and intentionally
excludes streaming RPCs and client RPCs.

To improve the ergonomics of adding the context everywhere its needed and to
clarify the requirements of dynamic vs static handlers, I've also done a good
bit of refactoring here:

* canonicalized the RPC handler fields so they're as close to identical as
  possible without introducing unused fields (i.e. I didn't add loggers if the
  handler doesn't use them already).
* canonicalized the imports in the handler files.
* added a `NewExampleEndpoint` function for each handler that ensures we're
  constructing the handlers with the required arguments.
* reordered the registration in server.go to match the order of the files (to
  make it easier to see if we've missed one), and added a bunch of commentary
  there as to what the difference between static and dynamic handlers is.

nomad/acl_endpoint.go

@@ -9,10 +9,11 @@ import (
 	"strings"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-set"
+
 	policy "github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper"
 	"github.com/hashicorp/nomad/helper/uuid"
@@ -35,7 +36,12 @@ const (
 // ACL endpoint is used for manipulating ACL tokens and policies
 type ACL struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewACLEndpoint(srv *Server, ctx *RPCContext) *ACL {
+	return &ACL{srv: srv, ctx: ctx, logger: srv.logger.Named("acl")}
 }
 
 // UpsertPolicies is used to create or update a set of policies

nomad/alloc_endpoint.go

@@ -5,10 +5,10 @@ import (
 	"net/http"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
-	multierror "github.com/hashicorp/go-multierror"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-multierror"
 
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper/pointer"
@@ -21,10 +21,12 @@ import (
 // Alloc endpoint is used for manipulating allocations
 type Alloc struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
 
-	// ctx provides context regarding the underlying connection
-	ctx *RPCContext
+func NewAllocEndpoint(srv *Server, ctx *RPCContext) *Alloc {
+	return &Alloc{srv: srv, ctx: ctx, logger: srv.logger.Named("alloc")}
 }
 
 // List is used to list the allocations in the system

nomad/client_agent_endpoint.go

@@ -26,6 +26,10 @@ type Agent struct {
 	srv *Server
 }
 
+func NewAgentEndpoint(srv *Server) *Agent {
+	return &Agent{srv: srv}
+}
+
 func (a *Agent) register() {
 	a.srv.streamingRpcs.Register("Agent.Monitor", a.monitor)
 }

nomad/client_alloc_endpoint.go

@@ -7,9 +7,10 @@ import (
 	"net"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-msgpack/codec"
+
 	"github.com/hashicorp/nomad/acl"
 	cstructs "github.com/hashicorp/nomad/client/structs"
 	"github.com/hashicorp/nomad/helper/pointer"
@@ -20,7 +21,11 @@ import (
 // Allocation endpoint.
 type ClientAllocations struct {
 	srv    *Server
-	logger log.Logger
+	logger hclog.Logger
+}
+
+func NewClientAllocationsEndpoint(srv *Server) *ClientAllocations {
+	return &ClientAllocations{srv: srv, logger: srv.logger.Named("client_allocs")}
 }
 
 func (a *ClientAllocations) register() {

nomad/client_csi_endpoint.go

@@ -19,6 +19,10 @@ type ClientCSI struct {
 	logger log.Logger
 }
 
+func NewClientCSIEndpoint(srv *Server) *ClientCSI {
+	return &ClientCSI{srv: srv, logger: srv.logger.Named("client_csi")}
+}
+
 func (a *ClientCSI) ControllerAttachVolume(args *cstructs.ClientCSIControllerAttachVolumeRequest, reply *cstructs.ClientCSIControllerAttachVolumeResponse) error {
 	defer metrics.MeasureSince([]string{"nomad", "client_csi_controller", "attach_volume"}, time.Now())
 

nomad/client_fs_endpoint.go

@@ -25,6 +25,10 @@ type FileSystem struct {
 	logger log.Logger
 }
 
+func NewFileSystemEndpoint(srv *Server) *FileSystem {
+	return &FileSystem{srv: srv, logger: srv.logger.Named("client_fs")}
+}
+
 func (f *FileSystem) register() {
 	f.srv.streamingRpcs.Register("FileSystem.Logs", f.logs)
 	f.srv.streamingRpcs.Register("FileSystem.Stream", f.stream)

nomad/client_stats_endpoint.go

@@ -18,6 +18,10 @@ type ClientStats struct {
 	logger log.Logger
 }
 
+func NewClientStatsEndpoint(srv *Server) *ClientStats {
+	return &ClientStats{srv: srv, logger: srv.logger.Named("client_stats")}
+}
+
 func (s *ClientStats) Stats(args *nstructs.NodeSpecificRequest, reply *structs.ClientStatsResponse) error {
 	// We only allow stale reads since the only potentially stale information is
 	// the Node registration and the cost is fairly high for adding another hope

nomad/csi_endpoint.go

@@ -6,10 +6,11 @@ import (
 	"strings"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
-	multierror "github.com/hashicorp/go-multierror"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-multierror"
+
 	"github.com/hashicorp/nomad/acl"
 	cstructs "github.com/hashicorp/nomad/client/structs"
 	"github.com/hashicorp/nomad/nomad/state"
@@ -20,7 +21,12 @@ import (
 // CSIVolume wraps the structs.CSIVolume with request data and server context
 type CSIVolume struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewCSIVolumeEndpoint(srv *Server, ctx *RPCContext) *CSIVolume {
+	return &CSIVolume{srv: srv, ctx: ctx, logger: srv.logger.Named("csi_volume")}
 }
 
 // QueryACLObj looks up the ACL token in the request and returns the acl.ACL object
@@ -1428,7 +1434,12 @@ func (v *CSIVolume) ListSnapshots(args *structs.CSISnapshotListRequest, reply *s
 // CSIPlugin wraps the structs.CSIPlugin with request data and server context
 type CSIPlugin struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewCSIPluginEndpoint(srv *Server, ctx *RPCContext) *CSIPlugin {
+	return &CSIPlugin{srv: srv, ctx: ctx, logger: srv.logger.Named("csi_plugin")}
 }
 
 // List replies with CSIPlugins, filtered by ACL access

nomad/deployment_endpoint.go

@@ -5,9 +5,10 @@ import (
 	"net/http"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
+
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/state/paginator"
@@ -17,10 +18,12 @@ import (
 // Deployment endpoint is used for manipulating deployments
 type Deployment struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
 
-	// ctx provides context regarding the underlying connection
-	ctx *RPCContext
+func NewDeploymentEndpoint(srv *Server, ctx *RPCContext) *Deployment {
+	return &Deployment{srv: srv, ctx: ctx, logger: srv.logger.Named("deployment")}
 }
 
 // GetDeployment is used to request information about a specific deployment

nomad/endpoints_oss.go

@@ -10,7 +10,7 @@ type EnterpriseEndpoints struct{}
 
 // NewEnterpriseEndpoints returns a stub of the enterprise endpoints since there
 // are none in oss
-func NewEnterpriseEndpoints(s *Server) *EnterpriseEndpoints {
+func NewEnterpriseEndpoints(s *Server, ctx *RPCContext) *EnterpriseEndpoints {
 	return &EnterpriseEndpoints{}
 }
 

nomad/eval_endpoint.go

@@ -6,12 +6,12 @@ import (
 	"net/http"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
+	"github.com/armon/go-metrics"
 	"github.com/hashicorp/go-bexpr"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
-	multierror "github.com/hashicorp/go-multierror"
-	version "github.com/hashicorp/go-version"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/go-version"
 
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/nomad/state"
@@ -30,10 +30,12 @@ var minVersionEvalDeleteByFilter = version.Must(version.NewVersion("1.4.3"))
 // Eval endpoint is used for eval interactions
 type Eval struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
 
-	// ctx provides context regarding the underlying connection
-	ctx *RPCContext
+func NewEvalEndpoint(srv *Server, ctx *RPCContext) *Eval {
+	return &Eval{srv: srv, ctx: ctx, logger: srv.logger.Named("eval")}
 }
 
 // GetEval is used to request information about a specific evaluation

nomad/event_endpoint.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/go-msgpack/codec"
+
 	"github.com/hashicorp/nomad/helper/pointer"
 	"github.com/hashicorp/nomad/nomad/stream"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -16,6 +17,10 @@ type Event struct {
 	srv *Server
 }
 
+func NewEventEndpoint(srv *Server) *Event {
+	return &Event{srv: srv}
+}
+
 func (e *Event) register() {
 	e.srv.streamingRpcs.Register("Event.Stream", e.stream)
 }

nomad/job_endpoint.go

@@ -15,6 +15,7 @@ import (
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/go-set"
+
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper"
 	"github.com/hashicorp/nomad/helper/pointer"
@@ -50,6 +51,7 @@ var (
 // Job endpoint is used for job interactions
 type Job struct {
 	srv    *Server
+	ctx    *RPCContext
 	logger hclog.Logger
 
 	// builtin admission controllers
@@ -58,9 +60,10 @@ type Job struct {
 }
 
 // NewJobEndpoints creates a new job endpoint with builtin admission controllers
-func NewJobEndpoints(s *Server) *Job {
+func NewJobEndpoints(s *Server, ctx *RPCContext) *Job {
 	return &Job{
 		srv:    s,
+		ctx:    ctx,
 		logger: s.logger.Named("job"),
 		mutators: []jobMutator{
 			jobCanonicalizer{},

nomad/job_endpoint_hook_connect_test.go

@@ -341,7 +341,7 @@ func TestJobEndpointConnect_ConnectInterpolation(t *testing.T) {
 	ci.Parallel(t)
 
 	server := &Server{logger: testlog.HCLogger(t)}
-	jobEndpoint := NewJobEndpoints(server)
+	jobEndpoint := NewJobEndpoints(server, nil)
 
 	j := mock.ConnectJob()
 	j.TaskGroups[0].Services[0].Name = "${JOB}-api"

nomad/keyring_endpoint.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
+	"github.com/armon/go-metrics"
 	"github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-memdb"
 
 	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/state"
@@ -15,10 +15,15 @@ import (
 
 // Keyring endpoint serves RPCs for root key management
 type Keyring struct {
-	srv       *Server
-	logger    hclog.Logger
+	srv    *Server
+	ctx    *RPCContext
+	logger hclog.Logger
+
 	encrypter *Encrypter
-	ctx       *RPCContext // context for connection, to check TLS role
+}
+
+func NewKeyringEndpoint(srv *Server, ctx *RPCContext, enc *Encrypter) *Keyring {
+	return &Keyring{srv: srv, ctx: ctx, logger: srv.logger.Named("keyring"), encrypter: enc}
 }
 
 func (k *Keyring) Rotate(args *structs.KeyringRotateRootKeyRequest, reply *structs.KeyringRotateRootKeyResponse) error {

nomad/namespace_endpoint.go

@@ -4,9 +4,10 @@ import (
 	"fmt"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	memdb "github.com/hashicorp/go-memdb"
-	multierror "github.com/hashicorp/go-multierror"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-multierror"
+
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -14,6 +15,11 @@ import (
 // Namespace endpoint is used for manipulating namespaces
 type Namespace struct {
 	srv *Server
+	ctx *RPCContext
+}
+
+func NewNamespaceEndpoint(srv *Server, ctx *RPCContext) *Namespace {
+	return &Namespace{srv: srv, ctx: ctx}
 }
 
 // UpsertNamespaces is used to upsert a set of namespaces

nomad/node_endpoint.go

@@ -14,14 +14,15 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-multierror"
+	vapi "github.com/hashicorp/vault/api"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/state/paginator"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/hashicorp/raft"
-	vapi "github.com/hashicorp/vault/api"
-	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -77,6 +78,16 @@ type Node struct {
 	updatesLock sync.Mutex
 }
 
+func NewNodeEndpoint(srv *Server, ctx *RPCContext) *Node {
+	return &Node{
+		srv:     srv,
+		ctx:     ctx,
+		logger:  srv.logger.Named("client"),
+		updates: []*structs.Allocation{},
+		evals:   []*structs.Evaluation{},
+	}
+}
+
 // Register is used to upsert a client that is available for scheduling
 func (n *Node) Register(args *structs.NodeRegisterRequest, reply *structs.NodeUpdateResponse) error {
 	isForwarded := args.IsForwarded()

nomad/operator_endpoint.go

@@ -7,7 +7,7 @@ import (
 	"net"
 	"time"
 
-	log "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-msgpack/codec"
 	"github.com/hashicorp/raft"
 	"github.com/hashicorp/serf/serf"
@@ -20,7 +20,12 @@ import (
 // Operator endpoint is used to perform low-level operator tasks for Nomad.
 type Operator struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewOperatorEndpoint(srv *Server, ctx *RPCContext) *Operator {
+	return &Operator{srv: srv, ctx: ctx, logger: srv.logger.Named("operator")}
 }
 
 func (op *Operator) register() {

nomad/periodic_endpoint.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
 
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -15,7 +15,12 @@ import (
 // Periodic endpoint is used for periodic job interactions
 type Periodic struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewPeriodicEndpoint(srv *Server, ctx *RPCContext) *Periodic {
+	return &Periodic{srv: srv, ctx: ctx, logger: srv.logger.Named("periodic")}
 }
 
 // Force is used to force a new instance of a periodic job

nomad/plan_endpoint.go

@@ -4,8 +4,8 @@ import (
 	"fmt"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
 
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -13,10 +13,12 @@ import (
 // Plan endpoint is used for plan interactions
 type Plan struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
 
-	// ctx provides context regarding the underlying connection
-	ctx *RPCContext
+func NewPlanEndpoint(srv *Server, ctx *RPCContext) *Plan {
+	return &Plan{srv: srv, ctx: ctx, logger: srv.logger.Named("plan")}
 }
 
 // Submit is used to submit a plan to the leader

nomad/regions_endpoint.go

@@ -1,7 +1,7 @@
 package nomad
 
 import (
-	log "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-hclog"
 
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -9,7 +9,12 @@ import (
 // Region is used to query and list the known regions
 type Region struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewRegionEndpoint(srv *Server, ctx *RPCContext) *Region {
+	return &Region{srv: srv, ctx: ctx, logger: srv.logger.Named("region")}
 }
 
 // List is used to list all of the known regions. No leader forwarding is

nomad/scaling_endpoint.go

@@ -4,9 +4,9 @@ import (
 	"strings"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
-	log "github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-memdb"
 
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper"
@@ -17,7 +17,12 @@ import (
 // Scaling endpoint is used for listing and retrieving scaling policies
 type Scaling struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewScalingEndpoint(srv *Server, ctx *RPCContext) *Scaling {
+	return &Scaling{srv: srv, ctx: ctx, logger: srv.logger.Named("scaling")}
 }
 
 // ListPolicies is used to list the policies

nomad/search_endpoint.go

@@ -43,9 +43,14 @@ var (
 // Search endpoint is used to look up matches for a given prefix and context
 type Search struct {
 	srv    *Server
+	ctx    *RPCContext
 	logger hclog.Logger
 }
 
+func NewSearchEndpoint(srv *Server, ctx *RPCContext) *Search {
+	return &Search{srv: srv, ctx: ctx, logger: srv.logger.Named("search")}
+}
+
 // getPrefixMatches extracts matches for an iterator, and returns a list of ids for
 // these matches.
 func (s *Search) getPrefixMatches(iter memdb.ResultIterator, prefix string) ([]string, bool) {

nomad/server.go

@@ -1200,92 +1200,90 @@ func (s *Server) setupRPC(tlsWrap tlsutil.RegionWrapper) error {
 // setupRpcServer is used to populate an RPC server with endpoints
 func (s *Server) setupRpcServer(server *rpc.Server, ctx *RPCContext) error {
 
-	// Add the static endpoints to the RPC server.
+	// Add the static endpoints to the RPC server. These are the RPC handlers
+	// that get used when component on the server is making an internal RPC
+	// call, so we only need them to be initialized once and they have no RPC
+	// context.
 	if s.staticEndpoints.Status == nil {
-		// Initialize the list just once
-		s.staticEndpoints.ACL = &ACL{srv: s, logger: s.logger.Named("acl")}
-		s.staticEndpoints.Job = NewJobEndpoints(s)
-		s.staticEndpoints.CSIVolume = &CSIVolume{srv: s, logger: s.logger.Named("csi_volume")}
-		s.staticEndpoints.CSIPlugin = &CSIPlugin{srv: s, logger: s.logger.Named("csi_plugin")}
-		s.staticEndpoints.Operator = &Operator{srv: s, logger: s.logger.Named("operator")}
-		s.staticEndpoints.Operator.register()
+		// note: Alloc, Plan have only dynamic endpoints
+		s.staticEndpoints.ACL = NewACLEndpoint(s, nil)
+		s.staticEndpoints.CSIVolume = NewCSIVolumeEndpoint(s, nil)
+		s.staticEndpoints.CSIPlugin = NewCSIPluginEndpoint(s, nil)
+		s.staticEndpoints.Deployment = NewDeploymentEndpoint(s, nil)
+		s.staticEndpoints.Job = NewJobEndpoints(s, nil)
+		s.staticEndpoints.Keyring = NewKeyringEndpoint(s, nil, s.encrypter)
+		s.staticEndpoints.Namespace = NewNamespaceEndpoint(s, nil)
+		s.staticEndpoints.Node = NewNodeEndpoint(s, nil)
+		s.staticEndpoints.Operator = NewOperatorEndpoint(s, nil)
+		s.staticEndpoints.Operator.register() // register the streaming RPCs
+		s.staticEndpoints.Periodic = NewPeriodicEndpoint(s, nil)
+		s.staticEndpoints.Region = NewRegionEndpoint(s, nil)
+		s.staticEndpoints.Scaling = NewScalingEndpoint(s, nil)
+		s.staticEndpoints.Search = NewSearchEndpoint(s, nil)
+		s.staticEndpoints.ServiceRegistration = NewServiceRegistrationEndpoint(s, nil)
+		s.staticEndpoints.Status = NewStatusEndpoint(s, nil)
+		s.staticEndpoints.System = NewSystemEndpoint(s, nil)
+		s.staticEndpoints.Variables = NewVariablesEndpoint(s, nil, s.encrypter)
 
-		s.staticEndpoints.Periodic = &Periodic{srv: s, logger: s.logger.Named("periodic")}
-		s.staticEndpoints.Region = &Region{srv: s, logger: s.logger.Named("region")}
-		s.staticEndpoints.Scaling = &Scaling{srv: s, logger: s.logger.Named("scaling")}
-		s.staticEndpoints.Status = &Status{srv: s, logger: s.logger.Named("status")}
-		s.staticEndpoints.System = &System{srv: s, logger: s.logger.Named("system")}
-		s.staticEndpoints.Search = &Search{srv: s, logger: s.logger.Named("search")}
-		s.staticEndpoints.Namespace = &Namespace{srv: s}
-		s.staticEndpoints.Variables = &Variables{srv: s, logger: s.logger.Named("variables"), encrypter: s.encrypter}
-		s.staticEndpoints.Keyring = &Keyring{srv: s, logger: s.logger.Named("keyring"), encrypter: s.encrypter}
+		s.staticEndpoints.Enterprise = NewEnterpriseEndpoints(s, nil)
 
-		s.staticEndpoints.Enterprise = NewEnterpriseEndpoints(s)
-
-		// These endpoints are dynamic because they need access to the
-		// RPCContext, but they also need to be called directly in some cases,
-		// so store them into staticEndpoints for later access, but don't
-		// register them as static.
-		s.staticEndpoints.Deployment = &Deployment{srv: s, logger: s.logger.Named("deployment")}
-		s.staticEndpoints.Node = &Node{srv: s, logger: s.logger.Named("client")}
-		s.staticEndpoints.ServiceRegistration = &ServiceRegistration{srv: s}
+		// These endpoints don't have a dynamic counterpart, so they'll need to
+		// be re-registered per connection as well (see below)
 
 		// Client endpoints
-		s.staticEndpoints.ClientStats = &ClientStats{srv: s, logger: s.logger.Named("client_stats")}
-		s.staticEndpoints.ClientAllocations = &ClientAllocations{srv: s, logger: s.logger.Named("client_allocs")}
-		s.staticEndpoints.ClientAllocations.register()
-		s.staticEndpoints.ClientCSI = &ClientCSI{srv: s, logger: s.logger.Named("client_csi")}
+		s.staticEndpoints.ClientStats = NewClientStatsEndpoint(s)
+		s.staticEndpoints.ClientAllocations = NewClientAllocationsEndpoint(s)
+		s.staticEndpoints.ClientAllocations.register() // register the streaming RPCs
+		s.staticEndpoints.ClientCSI = NewClientCSIEndpoint(s)
 
 		// Streaming endpoints
-		s.staticEndpoints.FileSystem = &FileSystem{srv: s, logger: s.logger.Named("client_fs")}
+		s.staticEndpoints.FileSystem = NewFileSystemEndpoint(s)
 		s.staticEndpoints.FileSystem.register()
 
-		s.staticEndpoints.Agent = &Agent{srv: s}
+		s.staticEndpoints.Agent = NewAgentEndpoint(s)
 		s.staticEndpoints.Agent.register()
 
-		s.staticEndpoints.Event = &Event{srv: s}
+		s.staticEndpoints.Event = NewEventEndpoint(s)
 		s.staticEndpoints.Event.register()
-
 	}
 
-	// Register the static handlers
-	server.Register(s.staticEndpoints.ACL)
-	server.Register(s.staticEndpoints.Job)
-	server.Register(s.staticEndpoints.CSIVolume)
-	server.Register(s.staticEndpoints.CSIPlugin)
-	server.Register(s.staticEndpoints.Operator)
-	server.Register(s.staticEndpoints.Periodic)
-	server.Register(s.staticEndpoints.Region)
-	server.Register(s.staticEndpoints.Scaling)
-	server.Register(s.staticEndpoints.Status)
-	server.Register(s.staticEndpoints.System)
-	server.Register(s.staticEndpoints.Search)
-	s.staticEndpoints.Enterprise.Register(server)
+	// If an endpoint has any non-streaming RPCs doesn't have an RPC context,
+	// we'll register the static handler here instead of creating a new dynamic
+	// endpoint on each connection.
+
 	server.Register(s.staticEndpoints.ClientStats)
 	server.Register(s.staticEndpoints.ClientAllocations)
 	server.Register(s.staticEndpoints.ClientCSI)
 	server.Register(s.staticEndpoints.FileSystem)
 	server.Register(s.staticEndpoints.Agent)
-	server.Register(s.staticEndpoints.Namespace)
-	server.Register(s.staticEndpoints.Variables)
 
-	// Create new dynamic endpoints and add them to the RPC server.
-	alloc := &Alloc{srv: s, ctx: ctx, logger: s.logger.Named("alloc")}
-	deployment := &Deployment{srv: s, ctx: ctx, logger: s.logger.Named("deployment")}
-	eval := &Eval{srv: s, ctx: ctx, logger: s.logger.Named("eval")}
-	node := &Node{srv: s, ctx: ctx, logger: s.logger.Named("client")}
-	plan := &Plan{srv: s, ctx: ctx, logger: s.logger.Named("plan")}
-	serviceReg := &ServiceRegistration{srv: s, ctx: ctx}
-	keyringReg := &Keyring{srv: s, ctx: ctx, logger: s.logger.Named("keyring"), encrypter: s.encrypter}
+	// Dynamic endpoints are endpoints that include the connection context and
+	// are created on each connection. Register all the dynamic endpoints with
+	// the RPC server.
+
+	_ = server.Register(NewACLEndpoint(s, ctx))
+	_ = server.Register(NewAllocEndpoint(s, ctx))
+	_ = server.Register(NewCSIVolumeEndpoint(s, ctx))
+	_ = server.Register(NewCSIPluginEndpoint(s, ctx))
+	_ = server.Register(NewDeploymentEndpoint(s, ctx))
+	_ = server.Register(NewEvalEndpoint(s, ctx))
+	_ = server.Register(NewJobEndpoints(s, ctx))
+	_ = server.Register(NewKeyringEndpoint(s, ctx, s.encrypter))
+	_ = server.Register(NewNamespaceEndpoint(s, ctx))
+	_ = server.Register(NewNodeEndpoint(s, ctx))
+	_ = server.Register(NewOperatorEndpoint(s, ctx))
+	_ = server.Register(NewPeriodicEndpoint(s, ctx))
+	_ = server.Register(NewPlanEndpoint(s, ctx))
+	_ = server.Register(NewRegionEndpoint(s, ctx))
+	_ = server.Register(NewScalingEndpoint(s, ctx))
+	_ = server.Register(NewSearchEndpoint(s, ctx))
+	_ = server.Register(NewServiceRegistrationEndpoint(s, ctx))
+	_ = server.Register(NewStatusEndpoint(s, ctx))
+	_ = server.Register(NewSystemEndpoint(s, ctx))
+	_ = server.Register(NewVariablesEndpoint(s, ctx, s.encrypter))
+
+	_ = server.Register(NewEnterpriseEndpoints(s, ctx))
 
-	// Register the dynamic endpoints
-	server.Register(alloc)
-	server.Register(deployment)
-	server.Register(eval)
-	server.Register(node)
-	server.Register(plan)
-	_ = server.Register(serviceReg)
-	_ = server.Register(keyringReg)
 	return nil
 }
 

nomad/service_registration_endpoint.go

@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/go-set"
+
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper"
 	"github.com/hashicorp/nomad/nomad/state"
@@ -24,12 +25,13 @@ import (
 // "/v1/service{s}" HTTP API.
 type ServiceRegistration struct {
 	srv *Server
-
-	// ctx provides context regarding the underlying connection, so we can
-	// perform TLS certificate validation on internal only endpoints.
 	ctx *RPCContext
 }
 
+func NewServiceRegistrationEndpoint(srv *Server, ctx *RPCContext) *ServiceRegistration {
+	return &ServiceRegistration{srv: srv, ctx: ctx}
+}
+
 // Upsert creates or updates service registrations held within Nomad. This RPC
 // is only callable by Nomad nodes.
 func (s *ServiceRegistration) Upsert(

nomad/status_endpoint.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"strconv"
 
-	log "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-hclog"
 
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -13,7 +13,12 @@ import (
 // Status endpoint is used to check on server status
 type Status struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewStatusEndpoint(srv *Server, ctx *RPCContext) *Status {
+	return &Status{srv: srv, ctx: ctx, logger: srv.logger.Named("status")}
 }
 
 // Ping is used to just check for connectivity

nomad/system_endpoint.go

@@ -3,7 +3,7 @@ package nomad
 import (
 	"fmt"
 
-	log "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-hclog"
 
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -11,7 +11,12 @@ import (
 // System endpoint is used to call invoke system tasks.
 type System struct {
 	srv    *Server
-	logger log.Logger
+	ctx    *RPCContext
+	logger hclog.Logger
+}
+
+func NewSystemEndpoint(srv *Server, ctx *RPCContext) *System {
+	return &System{srv: srv, ctx: ctx, logger: srv.logger.Named("system")}
 }
 
 // GarbageCollect is used to trigger the system to immediately garbage collect nodes, evals

nomad/variables_endpoint.go

@@ -7,9 +7,9 @@ import (
 	"strings"
 	"time"
 
-	metrics "github.com/armon/go-metrics"
+	"github.com/armon/go-metrics"
 	"github.com/hashicorp/go-hclog"
-	memdb "github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-memdb"
 
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/helper"
@@ -22,11 +22,17 @@ import (
 // callable via the Variables RPCs and externally via the "/v1/var{s}"
 // HTTP API.
 type Variables struct {
-	srv       *Server
-	logger    hclog.Logger
+	srv    *Server
+	ctx    *RPCContext
+	logger hclog.Logger
+
 	encrypter *Encrypter
 }
 
+func NewVariablesEndpoint(srv *Server, ctx *RPCContext, enc *Encrypter) *Variables {
+	return &Variables{srv: srv, ctx: ctx, logger: srv.logger.Named("variables"), encrypter: enc}
+}
+
 // Apply is used to apply a SV update request to the data store.
 func (sv *Variables) Apply(args *structs.VariablesApplyRequest, reply *structs.VariablesApplyResponse) error {
 	if done, err := sv.srv.forward(structs.VariablesApplyRPCMethod, args, args, reply); done {