34140ff3e0
Previously, public referred to gRPC services that are both exposed on the dedicated gRPC port and have their definitions in the proto-public directory (so were considered usable by 3rd parties). Whereas private referred to services on the multiplexed server port that are only usable by agents and other servers. Now, we're splitting these definitions, such that external/internal refers to the port and public/private refers to whether they can be used by 3rd parties. This is necessary because the peering replication API needs to be exposed on the dedicated port, but is not (yet) suitable for use by 3rd parties.
121 lines
3.1 KiB
Go
121 lines
3.1 KiB
Go
package subscribe
|
|
|
|
import (
|
|
"errors"
|
|
|
|
"github.com/hashicorp/go-hclog"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
"github.com/hashicorp/consul/agent/consul/stream"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/proto/pbsubscribe"
|
|
)
|
|
|
|
// Server implements a StateChangeSubscriptionServer for accepting SubscribeRequests,
|
|
// and sending events to the subscription topic.
|
|
type Server struct {
|
|
Backend Backend
|
|
Logger Logger
|
|
}
|
|
|
|
func NewServer(backend Backend, logger Logger) *Server {
|
|
return &Server{Backend: backend, Logger: logger}
|
|
}
|
|
|
|
type Logger interface {
|
|
Trace(msg string, args ...interface{})
|
|
With(args ...interface{}) hclog.Logger
|
|
}
|
|
|
|
var _ pbsubscribe.StateChangeSubscriptionServer = (*Server)(nil)
|
|
|
|
type Backend interface {
|
|
ResolveTokenAndDefaultMeta(token string, entMeta *acl.EnterpriseMeta, authzContext *acl.AuthorizerContext) (acl.Authorizer, error)
|
|
Forward(info structs.RPCInfo, f func(*grpc.ClientConn) error) (handled bool, err error)
|
|
Subscribe(req *stream.SubscribeRequest) (*stream.Subscription, error)
|
|
}
|
|
|
|
func (h *Server) Subscribe(req *pbsubscribe.SubscribeRequest, serverStream pbsubscribe.StateChangeSubscription_SubscribeServer) error {
|
|
logger := newLoggerForRequest(h.Logger, req)
|
|
handled, err := h.Backend.Forward(req, forwardToDC(req, serverStream, logger))
|
|
if handled || err != nil {
|
|
return err
|
|
}
|
|
|
|
logger.Trace("new subscription")
|
|
defer logger.Trace("subscription closed")
|
|
|
|
entMeta := req.EnterpriseMeta()
|
|
authz, err := h.Backend.ResolveTokenAndDefaultMeta(req.Token, &entMeta, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
subReq, err := state.PBToStreamSubscribeRequest(req, entMeta)
|
|
if err != nil {
|
|
return status.Error(codes.InvalidArgument, err.Error())
|
|
}
|
|
|
|
sub, err := h.Backend.Subscribe(subReq)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer sub.Unsubscribe()
|
|
|
|
ctx := serverStream.Context()
|
|
elog := &eventLogger{logger: logger}
|
|
for {
|
|
event, err := sub.Next(ctx)
|
|
switch {
|
|
case errors.Is(err, stream.ErrSubForceClosed):
|
|
logger.Trace("subscription reset by server")
|
|
return status.Error(codes.Aborted, err.Error())
|
|
case err != nil:
|
|
return err
|
|
}
|
|
|
|
if !event.Payload.HasReadPermission(authz) {
|
|
continue
|
|
}
|
|
|
|
elog.Trace(event)
|
|
|
|
// TODO: This conversion could be cached if needed
|
|
e := event.Payload.ToSubscriptionEvent(event.Index)
|
|
if err := serverStream.Send(e); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
func forwardToDC(
|
|
req *pbsubscribe.SubscribeRequest,
|
|
serverStream pbsubscribe.StateChangeSubscription_SubscribeServer,
|
|
logger Logger,
|
|
) func(conn *grpc.ClientConn) error {
|
|
return func(conn *grpc.ClientConn) error {
|
|
logger.Trace("forwarding to another DC")
|
|
defer logger.Trace("forwarded stream closed")
|
|
|
|
client := pbsubscribe.NewStateChangeSubscriptionClient(conn)
|
|
streamHandle, err := client.Subscribe(serverStream.Context(), req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for {
|
|
event, err := streamHandle.Recv()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := serverStream.Send(event); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|