497df1ca3b
This is the OSS portion of enterprise PR 2056. This commit provides server-local implementations of the proxycfg.ConfigEntry and proxycfg.ConfigEntryList interfaces, that source data from streaming events. It makes use of the LocalMaterializer type introduced for peering replication, adding the necessary support for authorization. It also adds support for "wildcard" subscriptions (within a topic) to the event publisher, as this is needed to fetch service-resolvers for all services when configuring mesh gateways. Currently, events will be emitted for just the ingress-gateway, service-resolver, and mesh config entry types, as these are the only entries required by proxycfg — the events will be emitted on topics named IngressGateway, ServiceResolver, and MeshConfig topics respectively. Though these events will only be consumed "locally" for now, they can also be consumed via the gRPC endpoint (confirmed using grpcurl) so using them from client agents should be a case of swapping the LocalMaterializer for an RPCMaterializer.
59 lines
1.7 KiB
Go
59 lines
1.7 KiB
Go
package pbsubscribe
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
)
|
|
|
|
// RequestDatacenter implements structs.RPCInfo
|
|
func (req *SubscribeRequest) RequestDatacenter() string {
|
|
return req.Datacenter
|
|
}
|
|
|
|
// IsRead implements structs.RPCInfo
|
|
func (req *SubscribeRequest) IsRead() bool {
|
|
return true
|
|
}
|
|
|
|
// AllowStaleRead implements structs.RPCInfo
|
|
func (req *SubscribeRequest) AllowStaleRead() bool {
|
|
return true
|
|
}
|
|
|
|
// TokenSecret implements structs.RPCInfo
|
|
func (req *SubscribeRequest) TokenSecret() string {
|
|
return req.Token
|
|
}
|
|
|
|
// SetTokenSecret implements structs.RPCInfo
|
|
func (req *SubscribeRequest) SetTokenSecret(token string) {
|
|
req.Token = token
|
|
}
|
|
|
|
// HasTimedOut implements structs.RPCInfo
|
|
func (req *SubscribeRequest) HasTimedOut(start time.Time, rpcHoldTimeout, maxQueryTime, defaultQueryTime time.Duration) (bool, error) {
|
|
return time.Since(start) > req.Timeout(rpcHoldTimeout, maxQueryTime, defaultQueryTime), nil
|
|
}
|
|
|
|
// Timeout implements structs.RPCInfo
|
|
func (req *SubscribeRequest) Timeout(rpcHoldTimeout, maxQueryTime, defaultQueryTime time.Duration) time.Duration {
|
|
return rpcHoldTimeout
|
|
}
|
|
|
|
// EnterpriseMeta returns the EnterpriseMeta encoded in the request's Subject.
|
|
func (req *SubscribeRequest) EnterpriseMeta() acl.EnterpriseMeta {
|
|
if req.GetWildcardSubject() {
|
|
// Note: EnterpriseMeta is ignored for the wildcard subject (as it will
|
|
// receive all events in the topic regardless of partition, namespace etc).
|
|
return acl.EnterpriseMeta{}
|
|
}
|
|
|
|
if sub := req.GetNamedSubject(); sub != nil {
|
|
return acl.NewEnterpriseMetaWithPartition(sub.Partition, sub.Namespace)
|
|
}
|
|
|
|
// Deprecated top-level Partition and Namespace fields.
|
|
return acl.NewEnterpriseMetaWithPartition(req.Partition, req.Namespace)
|
|
}
|