Add path based primary write forwarding (PBPWF) - OSS (#18735)

* Add WriteForwardedStorage to sdk's plugin, logical in OSS

This should allow backends to specify paths to forward write
(storage.Put(...) and storage.Delete(...)) operations for.

Notably, these semantics are subject to change and shouldn't yet be
relied on.

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Collect paths for write forwarding in OSS

This adds a path manager to Core, allowing tracking across all Vault
versions of paths which could use write forwarding if available. In
particular, even on OSS offerings, we'll need to template {{clusterId}}
into the paths, in the event of later upgrading to Enterprise. If we
didn't, we'd end up writing paths which will no longer be accessible
post-migration, due to write forwarding now replacing the sentinel with
the actual cluster identifier.

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Add forwarded writer implementation to OSS

Here, for paths given to us, we determine if we need to do cluster
translation and perform local writing. This is the OSS variant.

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Wire up mount-specific request forwarding in OSS

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Clarify that state lock needs to be held to call HAState in OSS

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Move cluster sentinel constant to sdk/logical

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Expose ClusterID to Plugins via SystemView

This will let plugins learn what the Cluster's ID is, without having to
resort to hacks like writing a random string to its cluster-prefixed
namespace and then reading it once it has replicated.

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

* Add GRPC ClusterID implementation

For any external plugins which wish to use it.

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

Signed-off-by: Alexander Scheel <alex.scheel@hashicorp.com>

sdk/logical/logical.go

@@ -131,6 +131,28 @@ type Paths struct {
 	// should be seal wrapped with extra encryption. It is exact matching
 	// unless it ends with '/' in which case it will be treated as a prefix.
 	SealWrapStorage []string
+
+	// WriteForwardedStorage are storage paths that, when running on a PR
+	// Secondary cluster, cause a GRPC call up to the PR Primary cluster's
+	// active node to handle storage.Put(...) and storage.Delete(...) events.
+	// These paths MUST include a {{clusterId}} literal, which the write layer
+	// will resolve to this cluster's UUID ("replication set" identifier).
+	// storage.List(...) and storage.Get(...) operations occur from the
+	// locally replicated data set, but can use path template expansion to be
+	// identifier agnostic.
+	//
+	// These paths require careful considerations by developers to use. In
+	// particular, writes on secondary clusters will not appear (when a
+	// corresponding read is issued immediately after a write) until the
+	// replication from primary->secondary has occurred. This replication
+	// triggers an InvalidateKey(...) call on the secondary, which can be
+	// used to detect the write has finished syncing. However, this will
+	// likely occur after the request has finished, so it is important to
+	// not block on this occurring.
+	//
+	// On standby nodes, like all storage write operations, this will trigger
+	// an ErrReadOnly return.
+	WriteForwardedStorage []string
 }
 
 type Auditor interface {

sdk/logical/storage.go

@@ -20,6 +20,11 @@ var ErrReadOnly = errors.New("cannot write to readonly storage")
 // storage while the backend is still being setup.
 var ErrSetupReadOnly = errors.New("cannot write to storage during setup")
 
+// Plugins using Paths.WriteForwardedStorage will need to use this sentinel
+// in their path to write cross-cluster. See the description of that parameter
+// for more information.
+const PBPWFClusterSentinel = "{{clusterId}}"
+
 // Storage is the way that logical backends are able read/write data.
 type Storage interface {
 	List(context.Context, string) ([]string, error)

sdk/logical/system_view.go

@@ -89,6 +89,11 @@ type SystemView interface {
 	// GeneratePasswordFromPolicy generates a password from the policy referenced.
 	// If the policy does not exist, this will return an error.
 	GeneratePasswordFromPolicy(ctx context.Context, policyName string) (password string, err error)
+
+	// ClusterID returns the replication ClusterID, for use with path-based
+	// write forwarding (WriteForwardedPaths). This value will be templated
+	// in for the {{cluterId}} sentinel.
+	ClusterID(ctx context.Context) (string, error)
 }
 
 type PasswordPolicy interface {
@@ -119,6 +124,7 @@ type StaticSystemView struct {
 	PluginEnvironment   *PluginEnvironment
 	PasswordPolicies    map[string]PasswordGenerator
 	VersionString       string
+	ClusterUUID         string
 }
 
 type noopAuditor struct{}
@@ -240,3 +246,7 @@ func (d *StaticSystemView) DeletePasswordPolicy(name string) (existed bool) {
 	delete(d.PasswordPolicies, name)
 	return existed
 }
+
+func (d StaticSystemView) ClusterID(ctx context.Context) (string, error) {
+	return d.ClusterUUID, nil
+}

sdk/plugin/grpc_backend_client.go

@@ -128,6 +128,7 @@ func (b *backendGRPCPluginClient) SpecialPaths() *logical.Paths {
 		Unauthenticated:       reply.Paths.Unauthenticated,
 		LocalStorage:          reply.Paths.LocalStorage,
 		SealWrapStorage:       reply.Paths.SealWrapStorage,
+		WriteForwardedStorage: reply.Paths.WriteForwardedStorage,
 	}
 }
 

sdk/plugin/grpc_backend_server.go

@@ -190,6 +190,7 @@ func (b *backendGRPCPluginServer) SpecialPaths(ctx context.Context, args *pb.Emp
 			Unauthenticated:       paths.Unauthenticated,
 			LocalStorage:          paths.LocalStorage,
 			SealWrapStorage:       paths.SealWrapStorage,
+			WriteForwardedStorage: paths.WriteForwardedStorage,
 		},
 	}, nil
 }

sdk/plugin/grpc_system.go

@@ -199,6 +199,15 @@ func (s *gRPCSystemViewClient) GeneratePasswordFromPolicy(ctx context.Context, p
 	return resp.Password, nil
 }
 
+func (s gRPCSystemViewClient) ClusterID(ctx context.Context) (string, error) {
+	reply, err := s.client.ClusterInfo(ctx, &pb.Empty{})
+	if err != nil {
+		return "", err
+	}
+
+	return reply.ClusterID, nil
+}
+
 type gRPCSystemViewServer struct {
 	pb.UnimplementedSystemViewServer
 
@@ -367,3 +376,18 @@ func (s *gRPCSystemViewServer) GeneratePasswordFromPolicy(ctx context.Context, r
 	}
 	return resp, nil
 }
+
+func (s *gRPCSystemViewServer) ClusterInfo(ctx context.Context, _ *pb.Empty) (*pb.ClusterInfoReply, error) {
+	if s.impl == nil {
+		return nil, errMissingSystemView
+	}
+
+	clusterId, err := s.impl.ClusterID(ctx)
+	if err != nil {
+		return &pb.ClusterInfoReply{}, status.Errorf(codes.Internal, "failed to fetch cluster id")
+	}
+
+	return &pb.ClusterInfoReply{
+		ClusterID: clusterId,
+	}, nil
+}

sdk/plugin/pb/backend.proto

@@ -47,6 +47,13 @@ message Paths {
 	// should be seal wrapped with extra encryption. It is exact matching
 	// unless it ends with '/' in which case it will be treated as a prefix.
 	repeated string seal_wrap_storage = 4;
+
+	// WriteForwardedStorage are storage paths that, when running on a PR
+	// Secondary cluster, cause a GRPC call up to the PR Primary cluster's
+	// active node to handle storage.Put(...) and storage.Delete(...) events.
+	//
+	// See extended note in /sdk/logical/logical.go.
+	repeated string write_forwarded_storage = 5;
 }
 
 message Request {
@@ -566,6 +573,12 @@ message GeneratePasswordFromPolicyReply {
 	string password = 1;
 }
 
+message ClusterInfoReply {
+	string cluster_name = 1;
+	string cluster_id = 2;
+	string err = 3;
+}
+
 // SystemView exposes system configuration information in a safe way for plugins
 // to consume. Plugins should implement the client for this service.
 service SystemView {
@@ -618,6 +631,9 @@ service SystemView {
 
 	// GeneratePasswordFromPolicy generates a password from an existing password policy
 	rpc GeneratePasswordFromPolicy(GeneratePasswordFromPolicyRequest) returns (GeneratePasswordFromPolicyReply);
+
+	// ClusterInfo returns the ClusterID information; may be reused if ClusterName is also exposed.
+	rpc ClusterInfo(Empty) returns (ClusterInfoReply);
 }
 
 message Connection {

sdk/plugin/pb/backend_grpc.pb.go

@@ -642,6 +642,8 @@ type SystemViewClient interface {
 	GroupsForEntity(ctx context.Context, in *EntityInfoArgs, opts ...grpc.CallOption) (*GroupsForEntityReply, error)
 	// GeneratePasswordFromPolicy generates a password from an existing password policy
 	GeneratePasswordFromPolicy(ctx context.Context, in *GeneratePasswordFromPolicyRequest, opts ...grpc.CallOption) (*GeneratePasswordFromPolicyReply, error)
+	// ClusterInfo returns the ClusterID information; may be reused if ClusterName is also exposed.
+	ClusterInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ClusterInfoReply, error)
 }
 
 type systemViewClient struct {
@@ -760,6 +762,15 @@ func (c *systemViewClient) GeneratePasswordFromPolicy(ctx context.Context, in *G
 	return out, nil
 }
 
+func (c *systemViewClient) ClusterInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ClusterInfoReply, error) {
+	out := new(ClusterInfoReply)
+	err := c.cc.Invoke(ctx, "/pb.SystemView/ClusterInfo", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // SystemViewServer is the server API for SystemView service.
 // All implementations must embed UnimplementedSystemViewServer
 // for forward compatibility
@@ -802,6 +813,8 @@ type SystemViewServer interface {
 	GroupsForEntity(context.Context, *EntityInfoArgs) (*GroupsForEntityReply, error)
 	// GeneratePasswordFromPolicy generates a password from an existing password policy
 	GeneratePasswordFromPolicy(context.Context, *GeneratePasswordFromPolicyRequest) (*GeneratePasswordFromPolicyReply, error)
+	// ClusterInfo returns the ClusterID information; may be reused if ClusterName is also exposed.
+	ClusterInfo(context.Context, *Empty) (*ClusterInfoReply, error)
 	mustEmbedUnimplementedSystemViewServer()
 }
 
@@ -845,6 +858,9 @@ func (UnimplementedSystemViewServer) GroupsForEntity(context.Context, *EntityInf
 func (UnimplementedSystemViewServer) GeneratePasswordFromPolicy(context.Context, *GeneratePasswordFromPolicyRequest) (*GeneratePasswordFromPolicyReply, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GeneratePasswordFromPolicy not implemented")
 }
+func (UnimplementedSystemViewServer) ClusterInfo(context.Context, *Empty) (*ClusterInfoReply, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ClusterInfo not implemented")
+}
 func (UnimplementedSystemViewServer) mustEmbedUnimplementedSystemViewServer() {}
 
 // UnsafeSystemViewServer may be embedded to opt out of forward compatibility for this service.
@@ -1074,6 +1090,24 @@ func _SystemView_GeneratePasswordFromPolicy_Handler(srv interface{}, ctx context
 	return interceptor(ctx, in, info, handler)
 }
 
+func _SystemView_ClusterInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(Empty)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SystemViewServer).ClusterInfo(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/pb.SystemView/ClusterInfo",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SystemViewServer).ClusterInfo(ctx, req.(*Empty))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // SystemView_ServiceDesc is the grpc.ServiceDesc for SystemView service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -1129,6 +1163,10 @@ var SystemView_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GeneratePasswordFromPolicy",
 			Handler:    _SystemView_GeneratePasswordFromPolicy_Handler,
 		},
+		{
+			MethodName: "ClusterInfo",
+			Handler:    _SystemView_ClusterInfo_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "sdk/plugin/pb/backend.proto",

vault/core.go

@@ -49,6 +49,7 @@ import (
 	"github.com/hashicorp/vault/sdk/helper/consts"
 	"github.com/hashicorp/vault/sdk/helper/jsonutil"
 	"github.com/hashicorp/vault/sdk/helper/logging"
+	"github.com/hashicorp/vault/sdk/helper/pathmanager"
 	"github.com/hashicorp/vault/sdk/logical"
 	"github.com/hashicorp/vault/sdk/physical"
 	sr "github.com/hashicorp/vault/serviceregistration"
@@ -681,8 +682,15 @@ type Core struct {
 	expirationRevokeRetryBase   time.Duration
 
 	events *eventbus.EventBus
+
+	// writeForwardedPaths are a set of storage paths which are GRPC forwarded
+	// to the active node of the primary cluster, when present. This PathManager
+	// contains absolute paths that we intend to forward (and template) when
+	// we're on a secondary cluster.
+	writeForwardedPaths *pathmanager.PathManager
 }
 
+// c.stateLock needs to be held in read mode before calling this function.
 func (c *Core) HAState() consts.HAState {
 	switch {
 	case c.perfStandby:
@@ -1068,6 +1076,10 @@ func CreateCore(conf *CoreConfig) (*Core, error) {
 		Enabled: new(uint32),
 	}
 
+	// Load write-forwarded path manager.
+	c.writeForwardedPaths = pathmanager.New()
+
+	// Load seal information.
 	if c.seal == nil {
 		wrapper := aeadwrapper.NewShamirWrapper()
 		wrapper.SetConfig(context.Background(), awskms.WithLogger(c.logger.Named("shamir")))

vault/dynamic_system_view.go

@@ -415,3 +415,12 @@ func (d dynamicSystemView) GeneratePasswordFromPolicy(ctx context.Context, polic
 
 	return passPolicy.Generate(ctx, nil)
 }
+
+func (d dynamicSystemView) ClusterID(ctx context.Context) (string, error) {
+	clusterInfo, err := d.core.Cluster(ctx)
+	if err != nil || clusterInfo.ID == "" {
+		return "", fmt.Errorf("unable to retrieve cluster info or empty ID: %w", err)
+	}
+
+	return clusterInfo.ID, nil
+}

vault/forwarded_writer_oss.go

@@ -0,0 +1,117 @@
+//go:build !enterprise
+
+package vault
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/vault/sdk/logical"
+)
+
+// Our forwarded writer has two components: a reference to Core, allowing
+// us to tap into the GRPC client and resolved paths, and lower storage
+// layer to call upon when we don't wish to forward our writes.
+//
+// This implementation lives in OSS: while the GRPC connection isn't present
+// on OSS, we need to ensure paths written to these forwarded nodes correctly
+// template {{clusterId}} if they are later upgraded to Enterprise, and don't
+// just write with the template sentinel still there.
+//
+// XXX: In the future, we'll need to support wrapping transactional storage.
+type ForwardedWriter struct {
+	core      *Core
+	lower     logical.Storage
+	clusterID string
+}
+
+func (c *Core) NewForwardedWriter(ctx context.Context, wrapped logical.Storage, _ bool /* local */) (logical.Storage, error) {
+	// local is unused above on this OSS implementation: local mounts only
+	// exist on Vault Enterprise.
+
+	// Cache the cluster id; we assume we'll be recreated when plugins reload
+	// if this changes, and should not change without reloading plugins.
+	cluster, err := c.Cluster(ctx)
+	if err != nil || cluster.ID == "" {
+		return nil, fmt.Errorf("failed to fetch local cluster info: %v", err)
+	}
+
+	return &ForwardedWriter{
+		core:      c,
+		lower:     wrapped,
+		clusterID: cluster.ID,
+	}, nil
+}
+
+func (w *ForwardedWriter) List(ctx context.Context, path string) ([]string, error) {
+	// storage.List(...) operations are always handled locally. However, we
+	// may need to resolve any {{clusterId}} template sentinels if given to us
+	// and we'd otherwise consider this a forwarded write operation.
+	var err error
+	path, err = w.resolvePathIfNecessary(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to do local cross-cluster list: failed to resolve path: %w", err)
+	}
+
+	return w.lower.List(ctx, path)
+}
+
+func (w *ForwardedWriter) Get(ctx context.Context, path string) (*logical.StorageEntry, error) {
+	// See note in List(...)above.
+	var err error
+	path, err = w.resolvePathIfNecessary(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to do local cross-cluster read: failed to resolve path: %w", err)
+	}
+
+	return w.lower.Get(ctx, path)
+}
+
+func (w *ForwardedWriter) Put(ctx context.Context, entry *logical.StorageEntry) error {
+	// See note above about List(...).
+	var err error
+	entry.Key, err = w.resolvePathIfNecessary(entry.Key)
+	if err != nil {
+		return fmt.Errorf("failed to do local cross-cluster write: failed to resolve path: %w", err)
+	}
+
+	return w.lower.Put(ctx, entry)
+}
+
+func (w *ForwardedWriter) Delete(ctx context.Context, path string) error {
+	// See note above about List(...).
+	var err error
+	path, err = w.resolvePathIfNecessary(path)
+	if err != nil {
+		return fmt.Errorf("failed to do local cross-cluster delete: failed to resolve path: %w", err)
+	}
+	return w.lower.Delete(ctx, path)
+}
+
+func (w *ForwardedWriter) resolvePathIfNecessary(path string) (string, error) {
+	// We should only resolve this path when we're going to be servicing
+	// it locally.
+	//
+	// We don't bother checking if we're a perf primary or not, as even
+	// perf secondaries could use locally serviced operations on these paths
+	// (e.g., a storage.List(...)).
+	forwardablePath := w.core.writeForwardedPaths.HasPath(path)
+
+	if forwardablePath {
+		return w.resolvePath(path)
+	}
+
+	return path, nil
+}
+
+func (w *ForwardedWriter) resolvePath(path string) (string, error) {
+	// This is the source-agnostic path resolution helper. Here we ensure
+	// we've got a forwarded path (one that contains the proper UUID
+	// sentinel) and we fetch this cluster's UUID and update the path.
+	if !strings.Contains(path, logical.PBPWFClusterSentinel) {
+		return "", fmt.Errorf("invalid path: lacks '%v' sentinel for expansion", logical.PBPWFClusterSentinel)
+	}
+
+	return strings.Replace(path, logical.PBPWFClusterSentinel, w.clusterID, 1), nil
+}

vault/mount.go

@@ -621,8 +621,16 @@ func (c *Core) mountInternal(ctx context.Context, entry *MountEntry, updateStora
 	// Sync values to the cache
 	entry.SyncCache()
 
+	// Resolution to absolute storage paths (versus uuid-relative) needs
+	// to happen prior to calling into the forwarded writer. Thus we
+	// intercept writes just before they hit barrier storage.
+	forwarded, err := c.NewForwardedWriter(ctx, c.barrier, entry.Local)
+	if err != nil {
+		return fmt.Errorf("error creating forwarded writer: %v", err)
+	}
+
 	viewPath := entry.ViewPath()
-	view := NewBarrierView(c.barrier, viewPath)
+	view := NewBarrierView(forwarded, viewPath)
 
 	// Singleton mounts cannot be filtered manually on a per-secondary basis
 	// from replication.
@@ -1424,8 +1432,16 @@ func (c *Core) setupMounts(ctx context.Context) error {
 		// Initialize the backend, special casing for system
 		barrierPath := entry.ViewPath()
 
-		// Create a barrier view using the UUID
-		view := NewBarrierView(c.barrier, barrierPath)
+		// Resolution to absolute storage paths (versus uuid-relative) needs
+		// to happen prior to calling into the forwarded writer. Thus we
+		// intercept writes just before they hit barrier storage.
+		forwarded, err := c.NewForwardedWriter(ctx, c.barrier, entry.Local)
+		if err != nil {
+			return fmt.Errorf("error creating forwarded writer: %v", err)
+		}
+
+		// Create a barrier storage view using the UUID
+		view := NewBarrierView(forwarded, barrierPath)
 
 		// Singleton mounts cannot be filtered manually on a per-secondary basis
 		// from replication

vault/mount_util.go

@@ -10,8 +10,14 @@ import (
 	"github.com/hashicorp/vault/sdk/logical"
 )
 
-func addPathCheckers(*Core, *MountEntry, logical.Backend, string)             {}
-func removePathCheckers(*Core, *MountEntry, string)                           {}
+func addPathCheckers(c *Core, entry *MountEntry, backend logical.Backend, viewPath string) {
+	c.addBackendWriteForwardedPaths(backend, viewPath)
+}
+
+func removePathCheckers(c *Core, entry *MountEntry, viewPath string) {
+	c.writeForwardedPaths.RemovePathPrefix(viewPath)
+}
+
 func addAuditPathChecker(*Core, *MountEntry, *BarrierView, string)            {}
 func removeAuditPathChecker(*Core, *MountEntry)                               {}
 func addFilterablePath(*Core, string)                                         {}

vault/mount_util_shared.go

@@ -0,0 +1,28 @@
+package vault
+
+import (
+	"github.com/hashicorp/vault/sdk/logical"
+)
+
+func (c *Core) addBackendWriteForwardedPaths(backend logical.Backend, viewPath string) {
+	paths := collectBackendSpecialPaths(backend, viewPath, func(specialPaths *logical.Paths) []string {
+		return specialPaths.WriteForwardedStorage
+	})
+
+	c.logger.Trace("adding write forwarded paths", "paths", paths)
+	c.writeForwardedPaths.AddPaths(paths)
+}
+
+func collectBackendSpecialPaths(backend logical.Backend, viewPath string, accessor func(specialPaths *logical.Paths) []string) []string {
+	if backend == nil || backend.SpecialPaths() == nil {
+		return nil
+	}
+	paths := accessor(backend.SpecialPaths())
+
+	var ret []string
+	for _, path := range paths {
+		ret = append(ret, viewPath+path)
+	}
+
+	return ret
+}