8cd6c9f95e
This is the OSS portion of enterprise PR 2460. Introduces a server-local implementation of the proxycfg.ResolvedServiceConfig interface that sources data from a blocking query against the server's state store. It moves the service config resolution logic into the agent/configentry package so that it can be used in both the RPC handler and data source. I've also done a little re-arranging and adding comments to call out data sources for which there is to be no server-local equivalent.
71 lines
2.4 KiB
Go
71 lines
2.4 KiB
Go
package proxycfgglue
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
"github.com/hashicorp/go-memdb"
|
|
|
|
"github.com/hashicorp/consul/agent/cache"
|
|
cachetype "github.com/hashicorp/consul/agent/cache-types"
|
|
"github.com/hashicorp/consul/agent/configentry"
|
|
"github.com/hashicorp/consul/agent/consul/watch"
|
|
"github.com/hashicorp/consul/agent/proxycfg"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
)
|
|
|
|
// CacheResolvedServiceConfig satisfies the proxycfg.ResolvedServiceConfig
|
|
// interface by sourcing data from the agent cache.
|
|
func CacheResolvedServiceConfig(c *cache.Cache) proxycfg.ResolvedServiceConfig {
|
|
return &cacheProxyDataSource[*structs.ServiceConfigRequest]{c, cachetype.ResolvedServiceConfigName}
|
|
}
|
|
|
|
// ServerResolvedServiceConfig satisfies the proxycfg.ResolvedServiceConfig
|
|
// interface by sourcing data from a blocking query against the server's state
|
|
// store.
|
|
func ServerResolvedServiceConfig(deps ServerDataSourceDeps, remoteSource proxycfg.ResolvedServiceConfig) proxycfg.ResolvedServiceConfig {
|
|
return &serverResolvedServiceConfig{deps, remoteSource}
|
|
}
|
|
|
|
type serverResolvedServiceConfig struct {
|
|
deps ServerDataSourceDeps
|
|
remoteSource proxycfg.ResolvedServiceConfig
|
|
}
|
|
|
|
func (s *serverResolvedServiceConfig) Notify(ctx context.Context, req *structs.ServiceConfigRequest, correlationID string, ch chan<- proxycfg.UpdateEvent) error {
|
|
if req.Datacenter != s.deps.Datacenter {
|
|
return s.remoteSource.Notify(ctx, req, correlationID, ch)
|
|
}
|
|
|
|
if len(req.Upstreams) != 0 {
|
|
return errors.New("ServerResolvedServiceConfig does not support the legacy Upstreams parameter")
|
|
}
|
|
|
|
return watch.ServerLocalNotify(ctx, correlationID, s.deps.GetStore,
|
|
func(ws memdb.WatchSet, store Store) (uint64, *structs.ServiceConfigResponse, error) {
|
|
authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, &req.EnterpriseMeta, nil)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
if err := authz.ToAllowAuthorizer().ServiceReadAllowed(req.Name, nil); err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
idx, entries, err := store.ReadResolvedServiceConfigEntries(ws, req.Name, &req.EnterpriseMeta, req.UpstreamIDs, req.Mode)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
reply, err := configentry.ComputeResolvedServiceConfig(req, req.UpstreamIDs, false, entries, s.deps.Logger)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
reply.Index = idx
|
|
|
|
return idx, reply, nil
|
|
},
|
|
dispatchBlockingQueryUpdate[*structs.ServiceConfigResponse](ch),
|
|
)
|
|
}
|