server: when the v2 catalog experiment is enabled reject api and rpc requests that are for the v1 catalog

agent/ae/ae.go

@@ -81,8 +81,9 @@ type StateSyncer struct {
 	SyncChanges *Trigger
 
 	// paused stores whether sync runs are temporarily disabled.
-	pauseLock sync.Mutex
-	paused    int
+	pauseLock    sync.Mutex
+	paused       int
+	hardDisabled bool
 
 	// serverUpInterval is the max time after which a full sync is
 	// performed when a server has been added to the cluster.
@@ -151,9 +152,20 @@ const (
 	retryFullSyncState fsmState = "retryFullSync"
 )
 
+// HardDisableSync is like PauseSync but is one-way. It causes other
+// Pause/Resume/Start operations to be completely ignored.
+func (s *StateSyncer) HardDisableSync() {
+	s.pauseLock.Lock()
+	s.hardDisabled = true
+	s.pauseLock.Unlock()
+}
+
 // Run is the long running method to perform state synchronization
 // between local and remote servers.
 func (s *StateSyncer) Run() {
+	if s.Disabled() {
+		return
+	}
 	if s.ClusterSize == nil {
 		panic("ClusterSize not set")
 	}
@@ -329,7 +341,14 @@ func (s *StateSyncer) Pause() {
 func (s *StateSyncer) Paused() bool {
 	s.pauseLock.Lock()
 	defer s.pauseLock.Unlock()
-	return s.paused != 0
+	return s.paused != 0 || s.hardDisabled
+}
+
+// Disabled returns whether sync runs are permanently disabled.
+func (s *StateSyncer) Disabled() bool {
+	s.pauseLock.Lock()
+	defer s.pauseLock.Unlock()
+	return s.hardDisabled
 }
 
 // Resume re-enables sync runs. It returns true if it was the last pause/resume
@@ -340,7 +359,7 @@ func (s *StateSyncer) Resume() bool {
 	if s.paused < 0 {
 		panic("unbalanced pause/resume")
 	}
-	trigger := s.paused == 0
+	trigger := s.paused == 0 && !s.hardDisabled
 	s.pauseLock.Unlock()
 	if trigger {
 		s.SyncChanges.Trigger()

agent/agent.go

@@ -22,8 +22,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/hashicorp/consul/lib/stringslice"
-
 	"github.com/armon/go-metrics"
 	"github.com/armon/go-metrics/prometheus"
 	"github.com/hashicorp/go-connlimit"
@@ -623,6 +621,9 @@ func (a *Agent) Start(ctx context.Context) error {
 	// create the state synchronization manager which performs
 	// regular and on-demand state synchronizations (anti-entropy).
 	a.sync = ae.NewStateSyncer(a.State, c.AEInterval, a.shutdownCh, a.logger)
+	if a.baseDeps.UseV2Resources() {
+		a.sync.HardDisableSync()
+	}
 
 	err = validateFIPSConfig(a.config)
 	if err != nil {
@@ -724,7 +725,7 @@ func (a *Agent) Start(ctx context.Context) error {
 		)
 
 		var pt *proxytracker.ProxyTracker
-		if a.useV2Resources() {
+		if a.baseDeps.UseV2Resources() {
 			pt = proxyWatcher.(*proxytracker.ProxyTracker)
 		}
 		server, err := consul.NewServer(consulCfg, a.baseDeps.Deps, a.externalGRPCServer, incomingRPCLimiter, serverLogger, pt)
@@ -911,20 +912,11 @@ func (a *Agent) Failed() <-chan struct{} {
 	return a.apiServers.failed
 }
 
-// useV2Resources returns true if "resource-apis" is present in the Experiments
-// array of the agent config.
-func (a *Agent) useV2Resources() bool {
-	if stringslice.Contains(a.baseDeps.Experiments, consul.CatalogResourceExperimentName) {
-		return true
-	}
-	return false
-}
-
 // getProxyWatcher returns the proper implementation of the ProxyWatcher interface.
 // It will return a ProxyTracker if "resource-apis" experiment is active.  Otherwise,
 // it will return a ConfigSource.
 func (a *Agent) getProxyWatcher() xds.ProxyWatcher {
-	if a.useV2Resources() {
+	if a.baseDeps.UseV2Resources() {
 		a.logger.Trace("returning proxyTracker for getProxyWatcher")
 		return proxytracker.NewProxyTracker(proxytracker.ProxyTrackerConfig{
 			Logger:         a.logger.Named("proxy-tracker"),

agent/agent_endpoint_test.go

@@ -79,6 +79,34 @@ func createACLTokenWithAgentReadPolicy(t *testing.T, srv *HTTPHandlers) string {
 	return svcToken.SecretID
 }
 
+func TestAgentEndpointsFailInV2(t *testing.T) {
+	t.Parallel()
+
+	a := NewTestAgent(t, `experiments = ["resource-apis"]`)
+
+	checkRequest := func(method, url string) {
+		t.Run(method+" "+url, func(t *testing.T) {
+			assertV1CatalogEndpointDoesNotWorkWithV2(t, a, method, url, `{}`)
+		})
+	}
+
+	checkRequest("PUT", "/v1/agent/maintenance")
+	checkRequest("GET", "/v1/agent/services")
+	checkRequest("GET", "/v1/agent/service/web")
+	checkRequest("GET", "/v1/agent/checks")
+	checkRequest("GET", "/v1/agent/health/service/id/web")
+	checkRequest("GET", "/v1/agent/health/service/name/web")
+	checkRequest("PUT", "/v1/agent/check/register")
+	checkRequest("PUT", "/v1/agent/check/deregister/web")
+	checkRequest("PUT", "/v1/agent/check/pass/web")
+	checkRequest("PUT", "/v1/agent/check/warn/web")
+	checkRequest("PUT", "/v1/agent/check/fail/web")
+	checkRequest("PUT", "/v1/agent/check/update/web")
+	checkRequest("PUT", "/v1/agent/service/register")
+	checkRequest("PUT", "/v1/agent/service/deregister/web")
+	checkRequest("PUT", "/v1/agent/service/maintenance/web")
+}
+
 func TestAgent_Services(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")

agent/catalog_endpoint_test.go

@@ -6,24 +6,68 @@ package agent
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 	"testing"
 	"time"
 
-	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/api"
-
 	"github.com/hashicorp/serf/coordinate"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
 	"github.com/hashicorp/consul/testrpc"
 )
 
+func TestCatalogEndpointsFailInV2(t *testing.T) {
+	t.Parallel()
+
+	a := NewTestAgent(t, `experiments = ["resource-apis"]`)
+
+	checkRequest := func(method, url string) {
+		t.Run(method+" "+url, func(t *testing.T) {
+			assertV1CatalogEndpointDoesNotWorkWithV2(t, a, method, url, "{}")
+		})
+	}
+
+	checkRequest("PUT", "/v1/catalog/register")
+	checkRequest("GET", "/v1/catalog/connect/")
+	checkRequest("PUT", "/v1/catalog/deregister")
+	checkRequest("GET", "/v1/catalog/datacenters")
+	checkRequest("GET", "/v1/catalog/nodes")
+	checkRequest("GET", "/v1/catalog/services")
+	checkRequest("GET", "/v1/catalog/service/")
+	checkRequest("GET", "/v1/catalog/node/")
+	checkRequest("GET", "/v1/catalog/node-services/")
+	checkRequest("GET", "/v1/catalog/gateway-services/")
+}
+
+func assertV1CatalogEndpointDoesNotWorkWithV2(t *testing.T, a *TestAgent, method, url string, requestBody string) {
+	var body io.Reader
+	switch method {
+	case http.MethodPost, http.MethodPut:
+		body = strings.NewReader(requestBody + "\n")
+	}
+
+	req, err := http.NewRequest(method, url, body)
+	require.NoError(t, err)
+
+	resp := httptest.NewRecorder()
+	a.srv.h.ServeHTTP(resp, req)
+	require.Equal(t, http.StatusBadRequest, resp.Code)
+
+	got, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	require.Contains(t, string(got), structs.ErrUsingV2CatalogExperiment.Error())
+}
+
 func TestCatalogRegister_PeeringRegistration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")

agent/config_endpoint_test.go

@@ -19,6 +19,23 @@ import (
 	"github.com/hashicorp/consul/testrpc"
 )
 
+func TestConfigEndpointsFailInV2(t *testing.T) {
+	t.Parallel()
+
+	a := NewTestAgent(t, `experiments = ["resource-apis"]`)
+
+	checkRequest := func(method, url string) {
+		t.Run(method+" "+url, func(t *testing.T) {
+			assertV1CatalogEndpointDoesNotWorkWithV2(t, a, method, url, `{"kind":"service-defaults", "name":"web"}`)
+		})
+	}
+
+	checkRequest("GET", "/v1/config/service-defaults")
+	checkRequest("GET", "/v1/config/service-defaults/web")
+	checkRequest("DELETE", "/v1/config/service-defaults/web")
+	checkRequest("PUT", "/v1/config")
+}
+
 func TestConfig_Get(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")

agent/consul/options.go

@@ -4,6 +4,7 @@
 package consul
 
 import (
+	"github.com/hashicorp/consul/lib/stringslice"
 	"google.golang.org/grpc"
 
 	"github.com/hashicorp/consul-net-rpc/net/rpc"
@@ -48,6 +49,15 @@ type Deps struct {
 	EnterpriseDeps
 }
 
+// useV2Resources returns true if "resource-apis" is present in the Experiments
+// array of the agent config.
+func (d Deps) UseV2Resources() bool {
+	if stringslice.Contains(d.Experiments, CatalogResourceExperimentName) {
+		return true
+	}
+	return false
+}
+
 type GRPCClientConner interface {
 	ClientConn(datacenter string) (*grpc.ClientConn, error)
 	ClientConnLeader() (*grpc.ClientConn, error)

agent/consul/server.go

@@ -19,12 +19,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/hashicorp/consul/internal/auth"
-	"github.com/hashicorp/consul/internal/mesh"
-	"github.com/hashicorp/consul/internal/resource"
-
 	"github.com/armon/go-metrics"
-	"github.com/hashicorp/consul-net-rpc/net/rpc"
 	"github.com/hashicorp/go-connlimit"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
@@ -41,6 +36,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/reflection"
 
+	"github.com/hashicorp/consul-net-rpc/net/rpc"
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/acl/resolver"
 	"github.com/hashicorp/consul/agent/blockingquery"
@@ -74,9 +70,12 @@ import (
 	"github.com/hashicorp/consul/agent/rpc/peering"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/agent/token"
+	"github.com/hashicorp/consul/internal/auth"
 	"github.com/hashicorp/consul/internal/catalog"
 	"github.com/hashicorp/consul/internal/controller"
+	"github.com/hashicorp/consul/internal/mesh"
 	proxysnapshot "github.com/hashicorp/consul/internal/mesh/proxy-snapshot"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/internal/resource/reaper"
 	raftstorage "github.com/hashicorp/consul/internal/storage/raft"
@@ -466,6 +465,8 @@ type Server struct {
 	reportingManager *reporting.ReportingManager
 
 	registry resource.Registry
+
+	useV2Resources bool
 }
 
 func (s *Server) DecrementBlockingQueries() uint64 {
@@ -554,6 +555,7 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 		incomingRPCLimiter:      incomingRPCLimiter,
 		routineManager:          routine.NewManager(logger.Named(logging.ConsulServer)),
 		registry:                flat.Registry,
+		useV2Resources:          flat.UseV2Resources(),
 	}
 	incomingRPCLimiter.Register(s)
 
@@ -589,7 +591,17 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 	}
 
 	rpcServerOpts := []func(*rpc.Server){
-		rpc.WithPreBodyInterceptor(middleware.GetNetRPCRateLimitingInterceptor(s.incomingRPCLimiter, middleware.NewPanicHandler(s.logger))),
+		rpc.WithPreBodyInterceptor(
+			middleware.ChainedRPCPreBodyInterceptor(
+				func(reqServiceMethod string, sourceAddr net.Addr) error {
+					if s.useV2Resources && isV1CatalogRequest(reqServiceMethod) {
+						return structs.ErrUsingV2CatalogExperiment
+					}
+					return nil
+				},
+				middleware.GetNetRPCRateLimitingInterceptor(s.incomingRPCLimiter, middleware.NewPanicHandler(s.logger)),
+			),
+		),
 	}
 
 	if flat.GetNetRPCInterceptorFunc != nil {
@@ -898,8 +910,27 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 	return s, nil
 }
 
+func isV1CatalogRequest(rpcName string) bool {
+	switch {
+	case strings.HasPrefix(rpcName, "Catalog."),
+		strings.HasPrefix(rpcName, "Health."),
+		strings.HasPrefix(rpcName, "ConfigEntry."):
+		return true
+	}
+
+	switch rpcName {
+	case "Internal.EventFire", "Internal.KeyringOperation", "Internal.OIDCAuthMethods":
+		return false
+	default:
+		if strings.HasPrefix(rpcName, "Internal.") {
+			return true
+		}
+		return false
+	}
+}
+
 func (s *Server) registerControllers(deps Deps, proxyUpdater ProxyUpdater) error {
-	if stringslice.Contains(deps.Experiments, CatalogResourceExperimentName) {
+	if s.useV2Resources {
 		catalog.RegisterControllers(s.controllerManager, catalog.DefaultControllerDependencies())
 
 		defaultAllow, err := s.config.ACLResolverSettings.IsDefaultAllow()

agent/health_endpoint_test.go

@@ -28,6 +28,25 @@ import (
 	"github.com/hashicorp/consul/types"
 )
 
+func TestHealthEndpointsFailInV2(t *testing.T) {
+	t.Parallel()
+
+	a := NewTestAgent(t, `experiments = ["resource-apis"]`)
+
+	checkRequest := func(method, url string) {
+		t.Run(method+" "+url, func(t *testing.T) {
+			assertV1CatalogEndpointDoesNotWorkWithV2(t, a, method, url, "{}")
+		})
+	}
+
+	checkRequest("GET", "/v1/health/node/web")
+	checkRequest("GET", "/v1/health/checks/web")
+	checkRequest("GET", "/v1/health/state/web")
+	checkRequest("GET", "/v1/health/service/web")
+	checkRequest("GET", "/v1/health/connect/web")
+	checkRequest("GET", "/v1/health/ingress/web")
+}
+
 func TestHealthChecksInState(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")