dataplane management (#2624)

internal/daemon/worker/addressreceiver.go

@@ -16,9 +16,9 @@ func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) {
 type receiverType uint
 
 const (
-	UnknownReceiverType         receiverType = 0
-	grpcResolverReceiverType    receiverType = 1
-	dialingListenerReceiverType receiverType = 2
+	UnknownReceiverType receiverType = iota
+	grpcResolverReceiverType
+	secondaryConnectionReceiverType
 )
 
 // String returns a string representation of the receiverType
@@ -26,7 +26,7 @@ func (s receiverType) String() string {
 	return [...]string{
 		"unknown",
 		"grpcResolver",
-		"dialingListener",
+		"secondaryConnections",
 	}[s]
 }
 

internal/daemon/worker/controller_connection.go

@@ -33,6 +33,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud:9202"
@@ -77,8 +78,14 @@ func (w *Worker) StartControllerConnections() error {
 	return nil
 }
 
-func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
-	const op = "worker.(Worker).controllerDialerFunc"
+// upstreamDialerFunc dials an upstream server. extraAlpnProtos can be provided
+// to kms and pki connections and are used for identifying, on the server side,
+// the intended purpose of the connection.  upstreamDialerFunc takes an optional
+// state struct which, if the connection is made using the node enrolment
+// library, sends the state to the server and can be retrieved from the
+// resulting protocol.Conn.
+func (w *Worker) upstreamDialerFunc(state *structpb.Struct, extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
+	const op = "worker.(Worker).upstreamDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		var conn net.Conn
 		var err error
@@ -90,6 +97,7 @@ func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Co
 				ctx,
 				w.WorkerAuthStorage,
 				addr,
+				nodeenrollment.WithState(state),
 				nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
 				nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos),
 				// If the activation token hasn't been populated, this won't do
@@ -202,7 +210,7 @@ func (w *Worker) createClientConn(addr string) error {
 		grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()),
 		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)),
 		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32)),
-		grpc.WithContextDialer(w.controllerDialerFunc()),
+		grpc.WithContextDialer(w.upstreamDialerFunc(nil)),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(defServiceConfig),
 		// Don't have the resolver reach out for a service config from the

internal/daemon/worker/listeners.go

@@ -171,14 +171,14 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	// Connections coming in here are authed by nodeenrollment and are for the
 	// reverse grpc purpose
-	reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue)
+	reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating reverse grpc split listener: %w", err)
 	}
 
 	// Connections coming in here are authed by nodeenrollment and are for the
 	// multi-hop session-proxying
-	websocketProxyingListener, err := w.workerAuthSplitListener.GetListener(common.WebsocketProxyingAlpnValue)
+	websocketProxyingListener, err := w.workerAuthSplitListener.GetListener(common.WebsocketProxyingAlpnValue, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating websocket proxying split listener: %w", err)
 	}

internal/daemon/worker/status.go

@@ -261,10 +261,10 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	}
 
 	// regardless of whether or not it's a new address, we need to set
-	// them for dialingListeners
+	// them for secondaryConnections
 	for _, as := range *addressReceivers {
 		switch {
-		case as.Type() == dialingListenerReceiverType:
+		case as.Type() == secondaryConnectionReceiverType:
 			tmpAddrs := make([]string, len(addrs))
 			copy(tmpAddrs, addrs)
 			if len(tmpAddrs) == 0 {