Implement filtering of some endpoints

agent/agent_endpoint.go

@@ -31,6 +31,7 @@ import (
 	"github.com/hashicorp/consul/lib/file"
 	"github.com/hashicorp/consul/logger"
 	"github.com/hashicorp/consul/types"
+	bexpr "github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/logutils"
 	"github.com/hashicorp/serf/coordinate"
 	"github.com/hashicorp/serf/serf"
@@ -219,6 +220,9 @@ func (s *HTTPServer) AgentServices(resp http.ResponseWriter, req *http.Request)
 	var token string
 	s.parseToken(req, &token)
 
+	var filterExpression string
+	s.parseFilter(req, &filterExpression)
+
 	services := s.agent.State.Services()
 	if err := s.agent.filterServices(token, &services); err != nil {
 		return nil, err
@@ -238,7 +242,12 @@ func (s *HTTPServer) AgentServices(resp http.ResponseWriter, req *http.Request)
 		agentSvcs[id] = &agentService
 	}
 
-	return agentSvcs, nil
+	filter, err := bexpr.CreateFilter(filterExpression, nil, agentSvcs)
+	if err != nil {
+		return nil, err
+	}
+
+	return filter.Execute(agentSvcs)
 }
 
 // GET /v1/agent/service/:service_id
@@ -403,6 +412,13 @@ func (s *HTTPServer) AgentChecks(resp http.ResponseWriter, req *http.Request) (i
 	var token string
 	s.parseToken(req, &token)
 
+	var filterExpression string
+	s.parseFilter(req, &filterExpression)
+	filter, err := bexpr.CreateFilter(filterExpression, nil, nil)
+	if err != nil {
+		return nil, err
+	}
+
 	checks := s.agent.State.Checks()
 	if err := s.agent.filterChecks(token, &checks); err != nil {
 		return nil, err
@@ -417,14 +433,17 @@ func (s *HTTPServer) AgentChecks(resp http.ResponseWriter, req *http.Request) (i
 		}
 	}
 
-	return checks, nil
+	return filter.Execute(checks)
 }
 
 func (s *HTTPServer) AgentMembers(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
 	// Fetch the ACL token, if any.
 	var token string
 	s.parseToken(req, &token)
 
+	var filterExpression string
+	s.parseFilter(req, &filterExpression)
+
 	// Check if the WAN is being queried
 	wan := false
 	if other := req.URL.Query().Get("wan"); other != "" {

agent/catalog_endpoint.go

@@ -92,6 +92,7 @@ func (s *HTTPServer) CatalogNodes(resp http.ResponseWriter, req *http.Request) (
 	// Setup the request
 	args := structs.DCSpecificRequest{}
 	s.parseSource(req, &args.Source)
+	s.parseFilter(req, &args.Filter)
 	args.NodeMetaFilters = s.parseMetaFilter(req)
 	if done := s.parse(resp, req, &args.Datacenter, &args.QueryOptions); done {
 		metrics.IncrCounterWithLabels([]string{"client", "rpc", "error", "catalog_nodes"}, 1,
@@ -180,6 +181,7 @@ func (s *HTTPServer) catalogServiceNodes(resp http.ResponseWriter, req *http.Req
 	// Set default DC
 	args := structs.ServiceSpecificRequest{Connect: connect}
 	s.parseSource(req, &args.Source)
+	s.parseFilter(req, &args.Filter)
 	args.NodeMetaFilters = s.parseMetaFilter(req)
 	if done := s.parse(resp, req, &args.Datacenter, &args.QueryOptions); done {
 		return nil, nil
@@ -261,6 +263,8 @@ func (s *HTTPServer) CatalogNodeServices(resp http.ResponseWriter, req *http.Req
 		return nil, nil
 	}
 
+	s.parseFilter(req, &args.Filter)
+
 	// Pull out the node name
 	args.Node = strings.TrimPrefix(req.URL.Path, "/v1/catalog/node/")
 	if args.Node == "" {

agent/consul/catalog_endpoint.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/ipaddr"
 	"github.com/hashicorp/consul/types"
+	bexpr "github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-uuid"
 )
@@ -219,6 +220,11 @@ func (c *Catalog) ListNodes(args *structs.DCSpecificRequest, reply *structs.Inde
 		return err
 	}
 
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.Nodes)
+	if err != nil {
+		return err
+	}
+
 	return c.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -239,6 +245,13 @@ func (c *Catalog) ListNodes(args *structs.DCSpecificRequest, reply *structs.Inde
 			if err := c.srv.filterACL(args.Token, reply); err != nil {
 				return err
 			}
+
+			raw, err := filter.Execute(reply.Nodes)
+			if err != nil {
+				return err
+			}
+			reply.Nodes = raw.(structs.Nodes)
+
 			return c.srv.sortNodesByDistanceFrom(args.Source, reply.Nodes)
 		})
 }
@@ -327,7 +340,12 @@ func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *stru
 		}
 	}
 
-	err := c.srv.blockingQuery(
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.ServiceNodes)
+	if err != nil {
+		return err
+	}
+
+	err = c.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
 		func(ws memdb.WatchSet, state *state.Store) error {
@@ -346,9 +364,19 @@ func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *stru
 				}
 				reply.ServiceNodes = filtered
 			}
+
 			if err := c.srv.filterACL(args.Token, reply); err != nil {
 				return err
 			}
+
+			// This is safe to do even when the filter is nil - its just a no-op then
+			raw, err := filter.Execute(reply.ServiceNodes)
+			if err != nil {
+				return err
+			}
+
+			reply.ServiceNodes = raw.(structs.ServiceNodes)
+
 			return c.srv.sortNodesByDistanceFrom(args.Source, reply.ServiceNodes)
 		})
 
@@ -400,6 +428,12 @@ func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs
 		return fmt.Errorf("Must provide node")
 	}
 
+	var filterType map[string]*structs.NodeService
+	filter, err := bexpr.CreateFilter(args.Filter, nil, filterType)
+	if err != nil {
+		return err
+	}
+
 	return c.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -410,6 +444,18 @@ func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs
 			}
 
 			reply.Index, reply.NodeServices = index, services
-			return c.srv.filterACL(args.Token, reply)
+			if err := c.srv.filterACL(args.Token, reply); err != nil {
+				return err
+			}
+
+			if reply.NodeServices != nil {
+				raw, err := filter.Execute(reply.NodeServices.Services)
+				if err != nil {
+					return err
+				}
+				reply.NodeServices.Services = raw.(map[string]*structs.NodeService)
+			}
+
+			return nil
 		})
 }

agent/consul/health_endpoint.go

@@ -7,6 +7,7 @@ import (
 	"github.com/armon/go-metrics"
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
+	bexpr "github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/go-memdb"
 )
 
@@ -22,6 +23,11 @@ func (h *Health) ChecksInState(args *structs.ChecksInStateRequest,
 		return err
 	}
 
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.HealthChecks)
+	if err != nil {
+		return err
+	}
+
 	return h.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -41,6 +47,13 @@ func (h *Health) ChecksInState(args *structs.ChecksInStateRequest,
 			if err := h.srv.filterACL(args.Token, reply); err != nil {
 				return err
 			}
+
+			raw, err := filter.Execute(reply.HealthChecks)
+			if err != nil {
+				return err
+			}
+			reply.HealthChecks = raw.(structs.HealthChecks)
+
 			return h.srv.sortNodesByDistanceFrom(args.Source, reply.HealthChecks)
 		})
 }
@@ -52,6 +65,11 @@ func (h *Health) NodeChecks(args *structs.NodeSpecificRequest,
 		return err
 	}
 
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.HealthChecks)
+	if err != nil {
+		return err
+	}
+
 	return h.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -61,7 +79,16 @@ func (h *Health) NodeChecks(args *structs.NodeSpecificRequest,
 				return err
 			}
 			reply.Index, reply.HealthChecks = index, checks
-			return h.srv.filterACL(args.Token, reply)
+			if err := h.srv.filterACL(args.Token, reply); err != nil {
+				return err
+			}
+
+			raw, err := filter.Execute(reply.HealthChecks)
+			if err != nil {
+				return err
+			}
+			reply.HealthChecks = raw.(structs.HealthChecks)
+			return nil
 		})
 }
 
@@ -78,6 +105,11 @@ func (h *Health) ServiceChecks(args *structs.ServiceSpecificRequest,
 		return err
 	}
 
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.HealthChecks)
+	if err != nil {
+		return err
+	}
+
 	return h.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -97,6 +129,13 @@ func (h *Health) ServiceChecks(args *structs.ServiceSpecificRequest,
 			if err := h.srv.filterACL(args.Token, reply); err != nil {
 				return err
 			}
+
+			raw, err := filter.Execute(reply.HealthChecks)
+			if err != nil {
+				return err
+			}
+			reply.HealthChecks = raw.(structs.HealthChecks)
+
 			return h.srv.sortNodesByDistanceFrom(args.Source, reply.HealthChecks)
 		})
 }
@@ -138,7 +177,12 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 		}
 	}
 
-	err := h.srv.blockingQuery(
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.Nodes)
+	if err != nil {
+		return err
+	}
+
+	err = h.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
 		func(ws memdb.WatchSet, state *state.Store) error {
@@ -151,9 +195,17 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 			if len(args.NodeMetaFilters) > 0 {
 				reply.Nodes = nodeMetaFilter(args.NodeMetaFilters, reply.Nodes)
 			}
+
 			if err := h.srv.filterACL(args.Token, reply); err != nil {
 				return err
 			}
+
+			raw, err := filter.Execute(reply.Nodes)
+			if err != nil {
+				return err
+			}
+			reply.Nodes = raw.(structs.CheckServiceNodes)
+
 			return h.srv.sortNodesByDistanceFrom(args.Source, reply.Nodes)
 		})
 

agent/consul/internal_endpoint.go

@@ -6,6 +6,7 @@ import (
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
+	bexpr "github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/serf/serf"
@@ -46,6 +47,11 @@ func (m *Internal) NodeDump(args *structs.DCSpecificRequest,
 		return err
 	}
 
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.Dump)
+	if err != nil {
+		return err
+	}
+
 	return m.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -56,7 +62,51 @@ func (m *Internal) NodeDump(args *structs.DCSpecificRequest,
 			}
 
 			reply.Index, reply.Dump = index, dump
-			return m.srv.filterACL(args.Token, reply)
+			if err := m.srv.filterACL(args.Token, reply); err != nil {
+				return err
+			}
+
+			raw, err := filter.Execute(reply.Dump)
+			if err != nil {
+				return err
+			}
+
+			reply.Dump = raw.(structs.NodeDump)
+			return nil
+		})
+}
+
+func (m *Internal) ServiceDump(args *structs.DCSpecificRequest, reply *structs.IndexedCheckServiceNodes) error {
+	if done, err := m.srv.forward("Internal.ServiceDump", args, args, reply); done {
+		return err
+	}
+
+	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.Nodes)
+	if err != nil {
+		return err
+	}
+
+	return m.srv.blockingQuery(
+		&args.QueryOptions,
+		&reply.QueryMeta,
+		func(ws memdb.WatchSet, state *state.Store) error {
+			index, nodes, err := state.ServiceDump(ws)
+			if err != nil {
+				return err
+			}
+
+			reply.Index, reply.Nodes = index, nodes
+			if err := m.srv.filterACL(args.Token, reply); err != nil {
+				return err
+			}
+
+			raw, err := filter.Execute(reply.Nodes)
+			if err != nil {
+				return err
+			}
+
+			reply.Nodes = raw.(structs.CheckServiceNodes)
+			return nil
 		})
 }