Add cache-type for Internal.IntentionUpstreams

agent/cache-types/intention_upstreams.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 IntentionUpstreamsName = "intention-upstreams"
+
+// GatewayUpstreams supports fetching upstreams for a given gateway name.
+type IntentionUpstreams struct {
+	RegisterOptionsBlockingRefresh
+	RPC RPC
+}
+
+func (i *IntentionUpstreams) Fetch(opts cache.FetchOptions, req cache.Request) (cache.FetchResult, error) {
+	var result cache.FetchResult
+
+	// The request should be a ServiceSpecificRequest.
+	reqReal, ok := req.(*structs.ServiceSpecificRequest)
+	if !ok {
+		return result, fmt.Errorf(
+			"Internal cache failure: request wrong type: %T", req)
+	}
+
+	// Lightweight copy this object so that manipulating QueryOptions doesn't race.
+	dup := *reqReal
+	reqReal = &dup
+
+	// Set the minimum query index to our current index so we block
+	reqReal.QueryOptions.MinQueryIndex = opts.MinIndex
+	reqReal.QueryOptions.MaxQueryTime = opts.Timeout
+
+	// Always allow stale - there's no point in hitting leader if the request is
+	// going to be served from cache and end up arbitrarily stale anyway. This
+	// allows cached service-discover to automatically read scale across all
+	// servers too.
+	reqReal.AllowStale = true
+
+	// Fetch
+	var reply structs.IndexedServiceList
+	if err := i.RPC.RPC("Internal.IntentionUpstreams", reqReal, &reply); err != nil {
+		return result, err
+	}
+
+	result.Value = &reply
+	result.Index = reply.QueryMeta.Index
+	return result, nil
+}

agent/cache-types/intention_upstreams_test.go

@@ -0,0 +1,52 @@
+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 TestIntentionUpstreams(t *testing.T) {
+	rpc := TestRPC(t)
+	typ := &IntentionUpstreams{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.IndexedServiceList
+	rpc.On("RPC", "Internal.IntentionUpstreams", mock.Anything, mock.Anything).Return(nil).
+		Run(func(args mock.Arguments) {
+			req := args.Get(1).(*structs.ServiceSpecificRequest)
+			require.Equal(t, uint64(24), req.QueryOptions.MinQueryIndex)
+			require.Equal(t, 1*time.Second, req.QueryOptions.MaxQueryTime)
+			require.True(t, req.AllowStale)
+			require.Equal(t, "foo", req.ServiceName)
+
+			services := structs.ServiceList{
+				{Name: "foo"},
+			}
+			reply := args.Get(2).(*structs.IndexedServiceList)
+			reply.Services = services
+			reply.QueryMeta.Index = 48
+			resp = reply
+		})
+
+	// Fetch
+	resultA, err := typ.Fetch(cache.FetchOptions{
+		MinIndex: 24,
+		Timeout:  1 * time.Second,
+	}, &structs.ServiceSpecificRequest{
+		Datacenter:  "dc1",
+		ServiceName: "foo",
+	})
+	require.NoError(t, err)
+	require.Equal(t, cache.FetchResult{
+		Value: resp,
+		Index: 48,
+	}, resultA)
+
+	rpc.AssertExpectations(t)
+}