backport of commit 817d85d (#16344)

Co-authored-by: Derek Menteer <derek.menteer@hashicorp.com>

.changelog/16339.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+peering: Fix bug where services were incorrectly imported as connect-enabled.
+```

agent/consul/state/catalog.go

@@ -900,12 +900,17 @@ func ensureServiceTxn(tx WriteTxn, idx uint64, node string, preserveIndexes bool
 			return fmt.Errorf("failed updating gateway mapping: %s", err)
 		}
 
+		if svc.PeerName == "" && sn.Name != "" {
+			if err := upsertKindServiceName(tx, idx, structs.ServiceKindConnectEnabled, sn); err != nil {
+				return fmt.Errorf("failed to persist service name as connect-enabled: %v", err)
+			}
+		}
+
+		// Update the virtual IP for the service
 		supported, err := virtualIPsSupported(tx, nil)
 		if err != nil {
 			return err
 		}
-
-		// Update the virtual IP for the service
 		if supported {
 			psn := structs.PeeredServiceName{Peer: svc.PeerName, ServiceName: sn}
 			vip, err := assignServiceVirtualIP(tx, idx, psn)
@@ -1964,6 +1969,24 @@ func (s *Store) deleteServiceTxn(tx WriteTxn, idx uint64, nodeName, serviceID st
 		return fmt.Errorf("Could not find any service %s: %s", svc.ServiceName, err)
 	}
 
+	// Cleanup ConnectEnabled for this service if none exist.
+	if svc.PeerName == "" && (svc.ServiceKind == structs.ServiceKindConnectProxy || svc.ServiceConnect.Native) {
+		service := svc.ServiceName
+		if svc.ServiceKind == structs.ServiceKindConnectProxy {
+			service = svc.ServiceProxy.DestinationServiceName
+		}
+		sn := structs.ServiceName{Name: service, EnterpriseMeta: svc.EnterpriseMeta}
+		connectEnabled, err := serviceHasConnectEnabledInstances(tx, sn.Name, &sn.EnterpriseMeta)
+		if err != nil {
+			return fmt.Errorf("failed to search for connect instances for service %q: %w", sn.Name, err)
+		}
+		if !connectEnabled {
+			if err := cleanupKindServiceName(tx, idx, sn, structs.ServiceKindConnectEnabled); err != nil {
+				return fmt.Errorf("failed to cleanup connect-enabled service name: %v", err)
+			}
+		}
+	}
+
 	if svc.PeerName == "" {
 		sn := structs.ServiceName{Name: svc.ServiceName, EnterpriseMeta: svc.EnterpriseMeta}
 		if err := cleanupGatewayWildcards(tx, idx, sn, false); err != nil {
@@ -3731,6 +3754,27 @@ func serviceHasConnectInstances(tx WriteTxn, serviceName string, entMeta *acl.En
 	return hasConnectInstance, hasNonConnectInstance, nil
 }
 
+// serviceHasConnectEnabledInstances returns whether the given service name
+// has a corresponding connect-proxy or connect-native instance.
+// This function is mostly a clone of `serviceHasConnectInstances`, but it has
+// an early return to improve performance and returns true if at least one
+// connect-native instance exists.
+func serviceHasConnectEnabledInstances(tx WriteTxn, serviceName string, entMeta *acl.EnterpriseMeta) (bool, error) {
+	query := Query{
+		Value:          serviceName,
+		EnterpriseMeta: *entMeta,
+	}
+
+	svc, err := tx.First(tableServices, indexConnect, query)
+	if err != nil {
+		return false, fmt.Errorf("failed service lookup: %w", err)
+	}
+	if svc != nil {
+		return true, nil
+	}
+	return false, nil
+}
+
 // updateGatewayService associates services with gateways after an eligible event
 // ie. Registering a service in a namespace targeted by a gateway
 func updateGatewayService(tx WriteTxn, idx uint64, mapping *structs.GatewayService) error {

agent/consul/state/catalog_test.go

@@ -8664,7 +8664,7 @@ func TestStateStore_EnsureService_ServiceNames(t *testing.T) {
 		},
 	}
 
-	var idx uint64
+	var idx, connectEnabledIdx uint64
 	testRegisterNode(t, s, idx, "node1")
 
 	for _, svc := range services {
@@ -8678,7 +8678,28 @@ func TestStateStore_EnsureService_ServiceNames(t *testing.T) {
 		require.Len(t, gotNames, 1)
 		require.Equal(t, svc.CompoundServiceName(), gotNames[0].Service)
 		require.Equal(t, svc.Kind, gotNames[0].Kind)
+		if svc.Kind == structs.ServiceKindConnectProxy {
+			connectEnabledIdx = idx
+		}
+	}
+
+	// A ConnectEnabled service should exist if a corresponding ConnectProxy or ConnectNative service exists.
+	verifyConnectEnabled := func(expectIdx uint64) {
+		gotIdx, gotNames, err := s.ServiceNamesOfKind(nil, structs.ServiceKindConnectEnabled)
+		require.NoError(t, err)
+		require.Equal(t, expectIdx, gotIdx)
+		require.Equal(t, []*KindServiceName{
+			{
+				Kind:    structs.ServiceKindConnectEnabled,
+				Service: structs.NewServiceName("foo", entMeta),
+				RaftIndex: structs.RaftIndex{
+					CreateIndex: connectEnabledIdx,
+					ModifyIndex: connectEnabledIdx,
+				},
+			},
+		}, gotNames)
 	}
+	verifyConnectEnabled(connectEnabledIdx)
 
 	// Register another ingress gateway and there should be two names under the kind index
 	newIngress := structs.NodeService{
@@ -8749,6 +8770,38 @@ func TestStateStore_EnsureService_ServiceNames(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, idx, gotIdx)
 	require.Empty(t, got)
+
+	// A ConnectEnabled entry should not be removed until all corresponding services are removed.
+	{
+		verifyConnectEnabled(connectEnabledIdx)
+		// Add a connect-native service.
+		idx++
+		require.NoError(t, s.EnsureService(idx, "node1", &structs.NodeService{
+			Kind:           structs.ServiceKindTypical,
+			ID:             "foo",
+			Service:        "foo",
+			Address:        "5.5.5.5",
+			Port:           5555,
+			EnterpriseMeta: *entMeta,
+			Connect: structs.ServiceConnect{
+				Native: true,
+			},
+		}))
+		verifyConnectEnabled(connectEnabledIdx)
+
+		// Delete the proxy. This should not clean up the entry, because we still have a
+		// connect-native service registered.
+		idx++
+		require.NoError(t, s.DeleteService(idx, "node1", "connect-proxy", entMeta, ""))
+		verifyConnectEnabled(connectEnabledIdx)
+
+		// Remove the connect-native service to clear out the connect-enabled entry.
+		require.NoError(t, s.DeleteService(idx, "node1", "foo", entMeta, ""))
+		gotIdx, gotNames, err := s.ServiceNamesOfKind(nil, structs.ServiceKindConnectEnabled)
+		require.NoError(t, err)
+		require.Equal(t, idx, gotIdx)
+		require.Empty(t, gotNames)
+	}
 }
 
 func assertMaxIndexes(t *testing.T, tx ReadTxn, expect map[string]uint64, skip ...string) {