status-im/consul
2
0
Fork 0
mirror of https://github.com/status-im/consul.git synced 2025-01-22 03:29:43 +00:00
Code Issues Projects Releases Wiki Activity

end to end changes to pass gatewayservices to /ui/services/

Browse Source
This commit is contained in:
freddygv 2020-07-30 10:21:11 -06:00
parent 51255eede7
commit aa6c59dbfc
8 changed files with 209 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
agent/cache-types/service_dump.go
View File

@ -41,7 +41,7 @@ func (c *InternalServiceDump) Fetch(opts cache.FetchOptions, req cache.Request)
reqReal.AllowStale = true
// Fetch
var reply structs.IndexedCheckServiceNodes
var reply structs.IndexedNodesWithGateways
if err := c.RPC.RPC("Internal.ServiceDump", reqReal, &reply); err != nil {
return result, err
}

2
agent/cache-types/service_dump_test.go
View File

@ -16,7 +16,7 @@ func TestInternalServiceDump(t *testing.T) {
// Expect the proper RPC call. This also sets the expected value
// since that is return-by-pointer in the arguments.
var resp *structs.IndexedCheckServiceNodes
var resp *structs.IndexedNodesWithGateways
rpc.On("RPC", "Internal.ServiceDump", mock.Anything, mock.Anything).Return(nil).
Run(func(args mock.Arguments) {
req := args.Get(1).(*structs.ServiceDumpRequest)

25
agent/consul/internal_endpoint.go
View File

@ -88,7 +88,7 @@ func (m *Internal) NodeDump(args *structs.DCSpecificRequest,
})
}
func (m *Internal) ServiceDump(args *structs.ServiceDumpRequest, reply *structs.IndexedCheckServiceNodes) error {
func (m *Internal) ServiceDump(args *structs.ServiceDumpRequest, reply *structs.IndexedNodesWithGateways) error {
if done, err := m.srv.ForwardRPC("Internal.ServiceDump", args, args, reply); done {
return err
}
@ -107,13 +107,30 @@ func (m *Internal) ServiceDump(args *structs.ServiceDumpRequest, reply *structs.
&args.QueryOptions,
&reply.QueryMeta,
func(ws memdb.WatchSet, state *state.Store) error {
index, nodes, err := state.ServiceDump(ws, args.ServiceKind, args.UseServiceKind, &args.EnterpriseMeta)
// Get, store, and filter nodes
maxIdx, nodes, err := state.ServiceDump(ws, args.ServiceKind, args.UseServiceKind, &args.EnterpriseMeta)
if err != nil {
return err
}
reply.Nodes = nodes
reply.Index, reply.Nodes = index, nodes
if err := m.srv.filterACL(args.Token, reply); err != nil {
if err := m.srv.filterACL(args.Token, reply.Nodes); err != nil {
return err
}
// Get, store, and filter gateway services
idx, gatewayServices, err := state.DumpGatewayServices(ws)
if err != nil {
return err
}
reply.Gateways = gatewayServices
if idx > maxIdx {
maxIdx = idx
}
reply.Index = maxIdx
if err := m.srv.filterACL(args.Token, reply.Gateways); err != nil {
return err
}

4
agent/consul/internal_endpoint_test.go
View File

@ -575,7 +575,7 @@ func TestInternal_ServiceDump(t *testing.T) {
QueryOptions: structs.QueryOptions{Filter: filter},
}
var out structs.IndexedCheckServiceNodes
var out structs.IndexedNodesWithGateways
require.NoError(t, msgpackrpc.CallWithCodec(codec, "Internal.ServiceDump", &args, &out))
return out.Nodes
}
@ -622,7 +622,7 @@ func TestInternal_ServiceDump_Kind(t *testing.T) {
UseServiceKind: true,
}
var out structs.IndexedCheckServiceNodes
var out structs.IndexedNodesWithGateways
require.NoError(t, msgpackrpc.CallWithCodec(codec, "Internal.ServiceDump", &args, &out))
return out.Nodes
}

31
agent/consul/state/catalog.go
View File

@ -2711,6 +2711,37 @@ func gatewayServices(tx *txn, name string, entMeta *structs.EnterpriseMeta) (mem
return tx.Get(gatewayServicesTableName, "gateway", structs.NewServiceName(name, entMeta))
}
func (s *Store) DumpGatewayServices(ws memdb.WatchSet) (uint64, structs.GatewayServices, error) {
tx := s.db.ReadTxn()
defer tx.Abort()
gatewayServices, err := tx.Get(gatewayServicesTableName, "id")
if err != nil {
return 0, nil, fmt.Errorf("failed to dump gateway-services: %s", err)
}
ws.Add(gatewayServices.WatchCh())
var maxIdx uint64
var results structs.GatewayServices
for obj := gatewayServices.Next(); obj != nil; obj = gatewayServices.Next() {
gs := obj.(*structs.GatewayService)
if gs.Service.Name != structs.WildcardSpecifier {
idx, matches, err := s.checkProtocolMatch(tx, ws, gs)
if err != nil {
return 0, nil, fmt.Errorf("failed checking protocol: %s", err)
}
maxIdx = lib.MaxUint64(maxIdx, idx)
if matches {
results = append(results, gs)
}
}
}
return maxIdx, results, nil
}
// TODO(ingress): How to handle index rolling back when a config entry is
// deleted that references a service?
// We might need something like the service_last_extinction index?

18
agent/structs/structs.go
View File

@ -1774,8 +1774,16 @@ func (n *ServiceName) Matches(o *ServiceName) bool {
return true
}
func (si *ServiceName) ToServiceID() ServiceID {
return ServiceID{ID: si.Name, EnterpriseMeta: si.EnterpriseMeta}
func (n *ServiceName) ToServiceID() ServiceID {
return ServiceID{ID: n.Name, EnterpriseMeta: n.EnterpriseMeta}
}
func (n *ServiceName) LessThan(other *ServiceName) bool {
if n.EnterpriseMeta.LessThan(&other.EnterpriseMeta) {
return true
}
return n.Name < other.Name
}
type ServiceList []ServiceName
@ -1812,6 +1820,12 @@ type IndexedCheckServiceNodes struct {
QueryMeta
}
type IndexedNodesWithGateways struct {
Nodes CheckServiceNodes
Gateways GatewayServices
QueryMeta
}
type DatacenterIndexedCheckServiceNodes struct {
DatacenterNodes map[string]CheckServiceNodes
QueryMeta

51
agent/ui_endpoint.go
View File

@ -17,7 +17,8 @@ import (
const metaExternalSource = "external-source"
type GatewayConfig struct {
Addresses []string `json:",omitempty"`
AssociatedServiceCount int `json:",omitempty"`
Addresses []string `json:",omitempty"`
// internal to track uniqueness
addressesSet map[string]struct{}
}
@ -150,7 +151,7 @@ func (s *HTTPServer) UIServices(resp http.ResponseWriter, req *http.Request) (in
s.parseFilter(req, &args.Filter)
// Make the RPC request
var out structs.IndexedCheckServiceNodes
var out structs.IndexedNodesWithGateways
defer setMeta(resp, &out.QueryMeta)
RPC:
if err := s.agent.RPC("Internal.ServiceDump", &args, &out); err != nil {
@ -164,7 +165,7 @@ RPC:
// Generate the summary
// TODO (gateways) (freddy) Have Internal.ServiceDump return ServiceDump instead. Need to add bexpr filtering for type.
return summarizeServices(out.Nodes.ToServiceDump(), s.agent.config), nil
return summarizeServices(out.Nodes.ToServiceDump(), out.Gateways, s.agent.config), nil
}
// UIGatewayServices is used to query all the nodes for services associated with a gateway along with their gateway config
@ -199,22 +200,22 @@ RPC:
return nil, err
}
return summarizeServices(out.Dump, s.agent.config), nil
return summarizeServices(out.Dump, nil, s.agent.config), nil
}
func summarizeServices(dump structs.ServiceDump, cfg *config.RuntimeConfig) []*ServiceSummary {
func summarizeServices(dump structs.ServiceDump, gateways structs.GatewayServices, cfg *config.RuntimeConfig) []*ServiceSummary {
// Collect the summary information
var services []structs.ServiceID
summary := make(map[structs.ServiceID]*ServiceSummary)
var services []structs.ServiceName
summary := make(map[structs.ServiceName]*ServiceSummary)
hasGateway := make(map[structs.ServiceID]bool)
hasProxy := make(map[structs.ServiceID]bool)
linkedGateways := make(map[structs.ServiceName][]structs.ServiceName)
hasProxy := make(map[structs.ServiceName]bool)
getService := func(service structs.ServiceID) *ServiceSummary {
getService := func(service structs.ServiceName) *ServiceSummary {
serv, ok := summary[service]
if !ok {
serv = &ServiceSummary{
Name: service.ID,
Name: service.Name,
EnterpriseMeta: service.EnterpriseMeta,
// the other code will increment this unconditionally so we
// shouldn't initialize it to 1
@ -226,10 +227,19 @@ func summarizeServices(dump structs.ServiceDump, cfg *config.RuntimeConfig) []*S
return serv
}
// Collect the list of services linked to each gateway up front
// THis also allows tracking whether a service name is associated with a gateway
gsCount := make(map[structs.ServiceName]int)
for _, gs := range gateways {
gsCount[gs.Gateway] += 1
linkedGateways[gs.Service] = append(linkedGateways[gs.Service], gs.Gateway)
}
for _, csn := range dump {
if csn.GatewayService != nil {
gwsvc := csn.GatewayService
sum := getService(gwsvc.Service.ToServiceID())
sum := getService(gwsvc.Service)
modifySummaryForGatewayService(cfg, sum, gwsvc)
}
@ -237,7 +247,7 @@ func summarizeServices(dump structs.ServiceDump, cfg *config.RuntimeConfig) []*S
if csn.Service == nil {
continue
}
sid := structs.NewServiceID(csn.Service.Service, &csn.Service.EnterpriseMeta)
sid := structs.NewServiceName(csn.Service.Service, &csn.Service.EnterpriseMeta)
sum := getService(sid)
svc := csn.Service
@ -245,7 +255,7 @@ func summarizeServices(dump structs.ServiceDump, cfg *config.RuntimeConfig) []*S
sum.Kind = svc.Kind
sum.InstanceCount += 1
if svc.Kind == structs.ServiceKindConnectProxy {
hasProxy[structs.NewServiceID(svc.Proxy.DestinationServiceName, &svc.EnterpriseMeta)] = true
hasProxy[structs.NewServiceName(svc.Proxy.DestinationServiceName, &svc.EnterpriseMeta)] = true
}
for _, tag := range svc.Tags {
found := false
@ -292,16 +302,23 @@ func summarizeServices(dump structs.ServiceDump, cfg *config.RuntimeConfig) []*S
sort.Slice(services, func(i, j int) bool {
return services[i].LessThan(&services[j])
})
output := make([]*ServiceSummary, len(summary))
for idx, service := range services {
// Sort the nodes and tags
sum := summary[service]
if hasProxy[service] {
sum.ConnectedWithProxy = true
}
if hasGateway[service] {
sum.ConnectedWithGateway = true
// Verify that at least one of the gateways linked by config entry has an instance registered in the catalog
for _, gw := range linkedGateways[service] {
if s := summary[gw]; s != nil && s.InstanceCount > 0 {
sum.ConnectedWithGateway = true
}
}
sum.GatewayConfig.AssociatedServiceCount = gsCount[service]
// Sort the nodes and tags
sort.Strings(sum.Nodes)
sort.Strings(sum.Tags)
output[idx] = sum

133
agent/ui_endpoint_test.go
View File

@ -300,6 +300,64 @@ func TestUiServices(t *testing.T) {
require.NoError(t, a.RPC("Catalog.Register", args, &out))
}
// Register a terminating gateway associated with api and cache
{
arg := structs.RegisterRequest{
Datacenter: "dc1",
Node: "foo",
Address: "127.0.0.1",
Service: &structs.NodeService{
ID: "terminating-gateway",
Service: "terminating-gateway",
Kind: structs.ServiceKindTerminatingGateway,
Port: 443,
},
}
var regOutput struct{}
require.NoError(t, a.RPC("Catalog.Register", &arg, &regOutput))
args := &structs.TerminatingGatewayConfigEntry{
Name: "terminating-gateway",
Kind: structs.TerminatingGateway,
Services: []structs.LinkedService{
{
Name: "api",
},
{
Name: "cache",
},
},
}
req := structs.ConfigEntryRequest{
Op: structs.ConfigEntryUpsert,
Datacenter: "dc1",
Entry: args,
}
var configOutput bool
require.NoError(t, a.RPC("ConfigEntry.Apply", &req, &configOutput))
require.True(t, configOutput)
// Web should not show up as ConnectedWithGateway since this one does not have any instances
args = &structs.TerminatingGatewayConfigEntry{
Name: "other-terminating-gateway",
Kind: structs.TerminatingGateway,
Services: []structs.LinkedService{
{
Name: "web",
},
},
}
req = structs.ConfigEntryRequest{
Op: structs.ConfigEntryUpsert,
Datacenter: "dc1",
Entry: args,
}
require.NoError(t, a.RPC("ConfigEntry.Apply", &req, &configOutput))
require.True(t, configOutput)
}
t.Run("No Filter", func(t *testing.T) {
t.Parallel()
req, _ := http.NewRequest("GET", "/v1/internal/ui/services/dc1", nil)
@ -310,7 +368,7 @@ func TestUiServices(t *testing.T) {
// Should be 2 nodes, and all the empty lists should be non-nil
summary := obj.([]*ServiceSummary)
require.Len(t, summary, 4)
require.Len(t, summary, 5)
// internal accounting that users don't see can be blown away
for _, sum := range summary {
@ -319,27 +377,29 @@ func TestUiServices(t *testing.T) {
expected := []*ServiceSummary{
{
Kind: structs.ServiceKindTypical,
Name: "api",
Tags: []string{"tag1", "tag2"},
Nodes: []string{"foo"},
InstanceCount: 1,
ChecksPassing: 2,
ChecksWarning: 1,
ChecksCritical: 0,
ConnectedWithProxy: true,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
Kind: structs.ServiceKindTypical,
Name: "api",
Tags: []string{"tag1", "tag2"},
Nodes: []string{"foo"},
InstanceCount: 1,
ChecksPassing: 2,
ChecksWarning: 1,
ChecksCritical: 0,
ConnectedWithProxy: true,
ConnectedWithGateway: true,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
},
{
Kind: structs.ServiceKindTypical,
Name: "cache",
Tags: nil,
Nodes: []string{"zip"},
InstanceCount: 1,
ChecksPassing: 0,
ChecksWarning: 0,
ChecksCritical: 0,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
Kind: structs.ServiceKindTypical,
Name: "cache",
Tags: nil,
Nodes: []string{"zip"},
InstanceCount: 1,
ChecksPassing: 0,
ChecksWarning: 0,
ChecksCritical: 0,
ConnectedWithGateway: true,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
},
{
Kind: structs.ServiceKindConnectProxy,
@ -364,7 +424,19 @@ func TestUiServices(t *testing.T) {
ChecksCritical: 0,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
},
{
Kind: structs.ServiceKindTerminatingGateway,
Name: "terminating-gateway",
Tags: nil,
Nodes: []string{"foo"},
InstanceCount: 1,
ChecksPassing: 2,
ChecksWarning: 1,
GatewayConfig: GatewayConfig{AssociatedServiceCount: 2},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
},
}
require.ElementsMatch(t, expected, summary)
})
@ -387,16 +459,17 @@ func TestUiServices(t *testing.T) {
expected := []*ServiceSummary{
{
Kind: structs.ServiceKindTypical,
Name: "api",
Tags: []string{"tag1", "tag2"},
Nodes: []string{"foo"},
InstanceCount: 1,
ChecksPassing: 2,
ChecksWarning: 1,
ChecksCritical: 0,
ConnectedWithProxy: true,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
Kind: structs.ServiceKindTypical,
Name: "api",
Tags: []string{"tag1", "tag2"},
Nodes: []string{"foo"},
InstanceCount: 1,
ChecksPassing: 2,
ChecksWarning: 1,
ChecksCritical: 0,
ConnectedWithProxy: true,
ConnectedWithGateway: false,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
},
{
Kind: structs.ServiceKindConnectProxy,