Allow DNS interface to use agent cache

agent/agent.go

@@ -20,11 +20,11 @@ import (
 
 	"google.golang.org/grpc"
 
-	"github.com/armon/go-metrics"
+	metrics "github.com/armon/go-metrics"
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/ae"
 	"github.com/hashicorp/consul/agent/cache"
-	"github.com/hashicorp/consul/agent/cache-types"
+	cachetype "github.com/hashicorp/consul/agent/cache-types"
 	"github.com/hashicorp/consul/agent/checks"
 	"github.com/hashicorp/consul/agent/config"
 	"github.com/hashicorp/consul/agent/consul"
@@ -42,8 +42,8 @@ import (
 	"github.com/hashicorp/consul/logger"
 	"github.com/hashicorp/consul/types"
 	"github.com/hashicorp/consul/watch"
-	"github.com/hashicorp/go-multierror"
-	"github.com/hashicorp/go-uuid"
+	multierror "github.com/hashicorp/go-multierror"
+	uuid "github.com/hashicorp/go-uuid"
 	"github.com/hashicorp/memberlist"
 	"github.com/hashicorp/raft"
 	"github.com/hashicorp/serf/serf"
@@ -3468,6 +3468,15 @@ func (a *Agent) registerCache() {
 		// Prepared queries don't support blocking
 		Refresh: false,
 	})
+
+	a.cache.RegisterType(cachetype.NodeServicesName, &cachetype.NodeServices{
+		RPC: a,
+	}, &cache.RegisterOptions{
+		// Maintain a blocking query, retry dropped connections quickly
+		Refresh:        true,
+		RefreshTimer:   0 * time.Second,
+		RefreshTimeout: 10 * time.Minute,
+	})
 }
 
 // defaultProxyCommand returns the default Connect managed proxy command.

agent/cache-types/node_services.go

@@ -0,0 +1,52 @@
+package cachetype
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/consul/agent/cache"
+	"github.com/hashicorp/consul/agent/structs"
+)
+
+// Recommended name for registration.
+const NodeServicesName = "node-services"
+
+// NodeServices supports fetching discovering service instances via the
+// catalog.
+type NodeServices struct {
+	RPC RPC
+}
+
+func (c *NodeServices) Fetch(opts cache.FetchOptions, req cache.Request) (cache.FetchResult, error) {
+	var result cache.FetchResult
+
+	// The request should be a DCSpecificRequest.
+	reqReal, ok := req.(*structs.NodeSpecificRequest)
+	if !ok {
+		return result, fmt.Errorf(
+			"Internal cache failure: request wrong type: %T", req)
+	}
+
+	// Set the minimum query index to our current index so we block
+	reqReal.QueryOptions.MinQueryIndex = opts.MinIndex
+	reqReal.QueryOptions.MaxQueryTime = opts.Timeout
+
+	// Allways allow stale - there's no point in hitting leader if the request is
+	// going to be served from cache and endup arbitrarily stale anyway. This
+	// allows cached service-discover to automatically read scale across all
+	// servers too.
+	reqReal.AllowStale = true
+
+	// Fetch
+	var reply structs.IndexedNodeServices
+	if err := c.RPC.RPC("Catalog.NodeServices", reqReal, &reply); err != nil {
+		return result, err
+	}
+
+	result.Value = &reply
+	result.Index = reply.QueryMeta.Index
+	return result, nil
+}
+
+func (c *NodeServices) SupportsBlocking() bool {
+	return true
+}

agent/cache-types/node_services_test.go

@@ -0,0 +1,71 @@
+package cachetype
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hashicorp/consul/agent/cache"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNodeServices(t *testing.T) {
+	require := require.New(t)
+	rpc := TestRPC(t)
+	defer rpc.AssertExpectations(t)
+	typ := &NodeServices{RPC: rpc}
+
+	// Expect the proper RPC call. This also sets the expected value
+	// since that is return-by-pointer in the arguments.
+	var resp *structs.IndexedNodeServices
+	rpc.On("RPC", "Catalog.NodeServices", mock.Anything, mock.Anything).Return(nil).
+		Run(func(args mock.Arguments) {
+			req := args.Get(1).(*structs.NodeSpecificRequest)
+			require.Equal(uint64(24), req.QueryOptions.MinQueryIndex)
+			require.Equal(1*time.Second, req.QueryOptions.MaxQueryTime)
+			require.Equal("node-01", req.Node)
+			require.True(req.AllowStale)
+
+			reply := args.Get(2).(*structs.IndexedNodeServices)
+			reply.NodeServices = &structs.NodeServices{
+				Node: &structs.Node{
+					ID:         "abcdef",
+					Node:       "node-01",
+					Address:    "127.0.0.5",
+					Datacenter: "dc1",
+				},
+			}
+
+			reply.QueryMeta.Index = 48
+			resp = reply
+		})
+
+	// Fetch
+	resultA, err := typ.Fetch(cache.FetchOptions{
+		MinIndex: 24,
+		Timeout:  1 * time.Second,
+	}, &structs.NodeSpecificRequest{
+		Datacenter: "dc1",
+		Node:       "node-01",
+	})
+	require.NoError(err)
+	require.Equal(cache.FetchResult{
+		Value: resp,
+		Index: 48,
+	}, resultA)
+}
+
+func TestNodeServices_badReqType(t *testing.T) {
+	require := require.New(t)
+	rpc := TestRPC(t)
+	defer rpc.AssertExpectations(t)
+	typ := &NodeServices{RPC: rpc}
+
+	// Fetch
+	_, err := typ.Fetch(cache.FetchOptions{}, cache.TestRequest(
+		t, cache.RequestInfo{Key: "foo", MinIndex: 64}))
+	require.Error(err)
+	require.Contains(err.Error(), "wrong type")
+
+}

agent/config/builder.go

@@ -715,6 +715,8 @@ func (b *Builder) Build() (rt RuntimeConfig, err error) {
 		DNSSOA:                soa,
 		DNSUDPAnswerLimit:     b.intVal(c.DNS.UDPAnswerLimit),
 		DNSNodeMetaTXT:        b.boolValWithDefault(c.DNS.NodeMetaTXT, true),
+		DNSUseCache:           b.boolVal(c.DNS.UseCache),
+		DNSCacheMaxAge:        b.durationVal("dns_config.cache_max_age", c.DNS.CacheMaxAge),
 
 		// HTTP
 		HTTPPort:            httpPort,

agent/config/config.go

@@ -555,6 +555,8 @@ type DNS struct {
 	UDPAnswerLimit     *int              `json:"udp_answer_limit,omitempty" hcl:"udp_answer_limit" mapstructure:"udp_answer_limit"`
 	NodeMetaTXT        *bool             `json:"enable_additional_node_meta_txt,omitempty" hcl:"enable_additional_node_meta_txt" mapstructure:"enable_additional_node_meta_txt"`
 	SOA                *SOA              `json:"soa,omitempty" hcl:"soa" mapstructure:"soa"`
+	UseCache           *bool             `json:"use_cache,omitempty" hcl:"use_cache" mapstructure:"use_cache"`
+	CacheMaxAge        *string           `json:"cache_max_age,omitempty" hcl:"cache_max_age" mapstructure:"cache_max_age"`
 }
 
 type HTTPConfig struct {

agent/config/runtime.go

@@ -314,6 +314,16 @@ type RuntimeConfig struct {
 	// flag: -recursor string [-recursor string]
 	DNSRecursors []string
 
+	// DNSUseCache wether or not to use cache for dns queries
+	//
+	// hcl: dns_config { use_cache = (true|false) }
+	DNSUseCache bool
+
+	// DNSUseCache wether or not to use cache for dns queries
+	//
+	// hcl: dns_config { cache_max_age = "duration" }
+	DNSCacheMaxAge time.Duration
+
 	// HTTPBlockEndpoints is a list of endpoint prefixes to block in the
 	// HTTP API. Any requests to these will get a 403 response.
 	//

agent/config/runtime_test.go

@@ -3066,7 +3066,9 @@ func TestFullConfig(t *testing.T) {
 				"service_ttl": {
 					"*": "32030s"
 				},
-				"udp_answer_limit": 29909
+				"udp_answer_limit": 29909,
+				"use_cache": true,
+				"cache_max_age": "5m"
 			},
 			"enable_acl_replication": true,
 			"enable_agent_tls_for_checks": true,
@@ -3620,6 +3622,8 @@ func TestFullConfig(t *testing.T) {
 					"*" = "32030s"
 				}
 				udp_answer_limit = 29909
+				use_cache = true
+				cache_max_age = "5m"
 			}
 			enable_acl_replication = true
 			enable_agent_tls_for_checks = true
@@ -4249,6 +4253,8 @@ func TestFullConfig(t *testing.T) {
 		DNSServiceTTL:                    map[string]time.Duration{"*": 32030 * time.Second},
 		DNSUDPAnswerLimit:                29909,
 		DNSNodeMetaTXT:                   true,
+		DNSUseCache:                      true,
+		DNSCacheMaxAge:                   5 * time.Minute,
 		DataDir:                          dataDir,
 		Datacenter:                       "rzo029wg",
 		DevMode:                          true,
@@ -5043,6 +5049,8 @@ func TestSanitize(t *testing.T) {
 			"Minttl": 0
 		},
 		"DNSUDPAnswerLimit": 0,
+		"DNSUseCache": false,
+		"DNSCacheMaxAge": "0s",
 		"DataDir": "",
 		"Datacenter": "",
 		"DevMode": false,