update gateway-services table with endpoints

agent/consul/catalog_endpoint_test.go

@@ -3054,6 +3054,7 @@ func TestCatalog_GatewayServices_TerminatingGateway(t *testing.T) {
 				CertFile:    "api/client.crt",
 				KeyFile:     "api/client.key",
 				SNI:         "my-domain",
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 			{
 				Service:     structs.NewServiceName("db", nil),
@@ -3062,6 +3063,7 @@ func TestCatalog_GatewayServices_TerminatingGateway(t *testing.T) {
 				CAFile:      "",
 				CertFile:    "",
 				KeyFile:     "",
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 			{
 				Service:      structs.NewServiceName("redis", nil),
@@ -3206,6 +3208,7 @@ func TestCatalog_GatewayServices_BothGateways(t *testing.T) {
 				Service:     structs.NewServiceName("api", nil),
 				Gateway:     structs.NewServiceName("gateway", nil),
 				GatewayKind: structs.ServiceKindTerminatingGateway,
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 		}
 
@@ -3428,11 +3431,13 @@ service "gateway" {
 				Service:     structs.NewServiceName("db", nil),
 				Gateway:     structs.NewServiceName("gateway", nil),
 				GatewayKind: structs.ServiceKindTerminatingGateway,
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 			{
 				Service:     structs.NewServiceName("db_replica", nil),
 				Gateway:     structs.NewServiceName("gateway", nil),
 				GatewayKind: structs.ServiceKindTerminatingGateway,
+				ServiceKind: structs.GatewayServiceKindUnknown,
 			},
 		}
 

agent/consul/internal_endpoint_test.go

@@ -1122,6 +1122,7 @@ func TestInternal_GatewayServiceDump_Terminating(t *testing.T) {
 				Gateway:     structs.NewServiceName("terminating-gateway", nil),
 				Service:     structs.NewServiceName("db", nil),
 				GatewayKind: "terminating-gateway",
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 		},
 		{
@@ -1155,6 +1156,7 @@ func TestInternal_GatewayServiceDump_Terminating(t *testing.T) {
 				Gateway:     structs.NewServiceName("terminating-gateway", nil),
 				Service:     structs.NewServiceName("db", nil),
 				GatewayKind: "terminating-gateway",
+				ServiceKind: structs.GatewayServiceKindService,
 			},
 		},
 		{

agent/consul/state/catalog.go

@@ -846,9 +846,16 @@ func ensureServiceTxn(tx WriteTxn, idx uint64, node string, preserveIndexes bool
 	}
 
 	if svc.PeerName == "" {
-		// Check if this service is covered by a gateway's wildcard specifier
-		if err = checkGatewayWildcardsAndUpdate(tx, idx, svc); err != nil {
-			return fmt.Errorf("failed updating gateway mapping: %s", err)
+		// Do not associate non-typical services with gateways or consul services
+		if svc.Kind == structs.ServiceKindTypical && svc.Service != "consul" {
+			// Check if this service is covered by a gateway's wildcard specifier, we force the service kind to a gateway-service here as that take precedence
+			sn := structs.NewServiceName(svc.Service, &svc.EnterpriseMeta)
+			if err = checkGatewayWildcardsAndUpdate(tx, idx, &sn, structs.GatewayServiceKindService); err != nil {
+				return fmt.Errorf("failed updating gateway mapping: %s", err)
+			}
+			if err = checkGatewayAndUpdate(tx, idx, &sn, structs.GatewayServiceKindService); err != nil {
+				return fmt.Errorf("failed updating gateway mapping: %s", err)
+			}
 		}
 	}
 	if err := upsertKindServiceName(tx, idx, svc.Kind, svc.CompoundServiceName()); err != nil {
@@ -3483,6 +3490,10 @@ func terminatingConfigGatewayServices(
 
 	var gatewayServices structs.GatewayServices
 	for _, svc := range entry.Services {
+		kind, err := GatewayServiceKind(tx, svc.Name, &svc.EnterpriseMeta)
+		if err != nil {
+			return false, nil, fmt.Errorf("failed to get gateway service kind for service %s: %v", svc.Name, err)
+		}
 		mapping := &structs.GatewayService{
 			Gateway:     gateway,
 			Service:     structs.NewServiceName(svc.Name, &svc.EnterpriseMeta),
@@ -3491,13 +3502,42 @@ func terminatingConfigGatewayServices(
 			CertFile:    svc.CertFile,
 			CAFile:      svc.CAFile,
 			SNI:         svc.SNI,
+			ServiceKind: kind,
 		}
 
 		gatewayServices = append(gatewayServices, mapping)
 	}
 	return false, gatewayServices, nil
 }
 
+func GatewayServiceKind(tx ReadTxn, name string, entMeta *acl.EnterpriseMeta) (structs.GatewayServiceKind, error) {
+	serviceIter, err := tx.First(tableServices, indexService, Query{
+		Value:          name,
+		EnterpriseMeta: *entMeta,
+	})
+	if err != nil {
+		return structs.GatewayServiceKindUnknown, err
+	}
+	if serviceIter != nil {
+		return structs.GatewayServiceKindService, err
+	}
+
+	_, entry, err := configEntryTxn(tx, nil, structs.ServiceDefaults, name, entMeta)
+	if err != nil {
+		return structs.GatewayServiceKindUnknown, err
+	}
+	if entry != nil {
+		sd, ok := entry.(*structs.ServiceConfigEntry)
+		if !ok {
+			return structs.GatewayServiceKindUnknown, fmt.Errorf("invalid config entry type %T", entry)
+		}
+		if sd.Destination != nil {
+			return structs.GatewayServiceKindDestination, nil
+		}
+	}
+	return structs.GatewayServiceKindUnknown, nil
+}
+
 // updateGatewayNamespace is used to target all services within a namespace
 func updateGatewayNamespace(tx WriteTxn, idx uint64, service *structs.GatewayService, entMeta *acl.EnterpriseMeta) error {
 	if entMeta == nil {
@@ -3538,6 +3578,41 @@ func updateGatewayNamespace(tx WriteTxn, idx uint64, service *structs.GatewaySer
 			return err
 		}
 	}
+	entries, err := tx.Get(tableConfigEntries, indexID+"_prefix", ConfigEntryKindQuery{Kind: structs.ServiceDefaults, EnterpriseMeta: *entMeta})
+	if err != nil {
+		return fmt.Errorf("failed querying entries: %s", err)
+	}
+	for entry := entries.Next(); entry != nil; entry = entries.Next() {
+		e := entry.(*structs.ServiceConfigEntry)
+		if e.Destination == nil {
+			continue
+		}
+
+		sn := structs.ServiceName{
+			Name:           e.Name,
+			EnterpriseMeta: e.EnterpriseMeta,
+		}
+		existing, err := tx.First(tableGatewayServices, indexID, service.Gateway, sn, service.Port)
+		if err != nil {
+			return fmt.Errorf("gateway service lookup failed: %s", err)
+		}
+		if existing != nil {
+			// If there's an existing service associated with this gateway then we skip it.
+			// This means the service was specified on its own, and the service entry overrides the wildcard entry.
+			continue
+		}
+
+		mapping := service.Clone()
+
+		mapping.Service = structs.NewServiceName(e.Name, &service.Service.EnterpriseMeta)
+		mapping.ServiceKind = structs.GatewayServiceKindDestination
+		mapping.FromWildcard = true
+
+		err = updateGatewayService(tx, idx, mapping)
+		if err != nil {
+			return err
+		}
+	}
 
 	// Also store a mapping for the wildcard so that the TLS creds can be pulled
 	// for new services registered in its namespace
@@ -3585,25 +3660,21 @@ func updateGatewayService(tx WriteTxn, idx uint64, mapping *structs.GatewayServi
 // checkWildcardForGatewaysAndUpdate checks whether a service matches a
 // wildcard definition in gateway config entries and if so adds it the the
 // gateway-services table.
-func checkGatewayWildcardsAndUpdate(tx WriteTxn, idx uint64, svc *structs.NodeService) error {
-	// Do not associate non-typical services with gateways or consul services
-	if svc.Kind != structs.ServiceKindTypical || svc.Service == "consul" {
-		return nil
-	}
-
+func checkGatewayWildcardsAndUpdate(tx WriteTxn, idx uint64, svc *structs.ServiceName, kind structs.GatewayServiceKind) error {
 	sn := structs.ServiceName{Name: structs.WildcardSpecifier, EnterpriseMeta: svc.EnterpriseMeta}
 	svcGateways, err := tx.Get(tableGatewayServices, indexService, sn)
 	if err != nil {
-		return fmt.Errorf("failed gateway lookup for %q: %s", svc.Service, err)
+		return fmt.Errorf("failed gateway lookup for %q: %s", svc.Name, err)
 	}
 	for service := svcGateways.Next(); service != nil; service = svcGateways.Next() {
 		if wildcardSvc, ok := service.(*structs.GatewayService); ok && wildcardSvc != nil {
 
 			// Copy the wildcard mapping and modify it
 			gatewaySvc := wildcardSvc.Clone()
 
-			gatewaySvc.Service = structs.NewServiceName(svc.Service, &svc.EnterpriseMeta)
+			gatewaySvc.Service = structs.NewServiceName(svc.Name, &svc.EnterpriseMeta)
 			gatewaySvc.FromWildcard = true
+			gatewaySvc.ServiceKind = kind
 
 			if err = updateGatewayService(tx, idx, gatewaySvc); err != nil {
 				return fmt.Errorf("Failed to associate service %q with gateway %q", gatewaySvc.Service.String(), gatewaySvc.Gateway.String())
@@ -3613,6 +3684,31 @@ func checkGatewayWildcardsAndUpdate(tx WriteTxn, idx uint64, svc *structs.NodeSe
 	return nil
 }
 
+// checkGatewayAndUpdate checks whether a service matches a
+// wildcard definition in gateway config entries and if so adds it the the
+// gateway-services table.
+func checkGatewayAndUpdate(tx WriteTxn, idx uint64, svc *structs.ServiceName, kind structs.GatewayServiceKind) error {
+	sn := structs.ServiceName{Name: svc.Name, EnterpriseMeta: svc.EnterpriseMeta}
+	svcGateways, err := tx.First(tableGatewayServices, indexService, sn)
+	if err != nil {
+		return fmt.Errorf("failed gateway lookup for %q: %s", svc.Name, err)
+	}
+
+	if service, ok := svcGateways.(*structs.GatewayService); ok && service != nil {
+		// Copy the wildcard mapping and modify it
+		gatewaySvc := service.Clone()
+
+		gatewaySvc.Service = structs.NewServiceName(svc.Name, &svc.EnterpriseMeta)
+		gatewaySvc.ServiceKind = kind
+
+		if err = updateGatewayService(tx, idx, gatewaySvc); err != nil {
+			return fmt.Errorf("Failed to associate service %q with gateway %q", gatewaySvc.Service.String(), gatewaySvc.Gateway.String())
+		}
+	}
+
+	return nil
+}
+
 func cleanupGatewayWildcards(tx WriteTxn, idx uint64, svc *structs.ServiceNode) error {
 	// Clean up association between service name and gateways if needed
 	sn := structs.ServiceName{Name: svc.ServiceName, EnterpriseMeta: svc.EnterpriseMeta}
@@ -3643,6 +3739,12 @@ func cleanupGatewayWildcards(tx WriteTxn, idx uint64, svc *structs.ServiceNode)
 			if err := deleteGatewayServiceTopologyMapping(tx, idx, m); err != nil {
 				return fmt.Errorf("failed to reconcile mesh topology for gateway: %v", err)
 			}
+		} else {
+			kind, err := GatewayServiceKind(tx, m.Service.Name, &m.Service.EnterpriseMeta)
+			if err != nil {
+				return fmt.Errorf("failed to get gateway service kind for service %s: %v", svc.ServiceName, err)
+			}
+			checkGatewayAndUpdate(tx, idx, &structs.ServiceName{Name: m.Service.Name, EnterpriseMeta: m.Service.EnterpriseMeta}, kind)
 		}
 	}
 	return nil

agent/consul/state/catalog_events_test.go

@@ -1473,6 +1473,18 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
 		},
 		WantEvents: []stream.Event{
 			testServiceHealthEvent(t, "srv1", evNodeUnchanged),
+			testServiceHealthDeregistrationEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evServiceTermingGateway("srv1"),
+				evTerminatingGatewayVirtualIPs("srv1"),
+			),
+			testServiceHealthEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evNodeUnchanged,
+				evServiceUnchanged,
+				evServiceTermingGateway("srv1")),
 		},
 	})
 	run(t, eventsTestCase{
@@ -1505,6 +1517,18 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
 		},
 		WantEvents: []stream.Event{
 			testServiceHealthDeregistrationEvent(t, "srv1"),
+			testServiceHealthDeregistrationEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evServiceTermingGateway("srv1"),
+				evTerminatingGatewayVirtualIPs("srv1"),
+			),
+			testServiceHealthEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evNodeUnchanged,
+				evServiceUnchanged,
+				evServiceTermingGateway("srv1")),
 		},
 	})
 	run(t, eventsTestCase{
@@ -1625,6 +1649,92 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
 				evTerminatingGatewayVirtualIPs("srv1", "srv2")),
 		},
 	})
+	run(t, eventsTestCase{
+		Name: "terminating gateway destination service-defaults",
+		Setup: func(s *Store, tx *txn) error {
+			configEntry := &structs.TerminatingGatewayConfigEntry{
+				Kind: structs.TerminatingGateway,
+				Name: "tgate1",
+				Services: []structs.LinkedService{
+					{
+						Name:           "destination1",
+						EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
+					},
+				},
+				EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
+			}
+			err := ensureConfigEntryTxn(tx, tx.Index, configEntry)
+			if err != nil {
+				return err
+			}
+			return s.ensureRegistrationTxn(tx, tx.Index, false,
+				testServiceRegistration(t, "tgate1", regTerminatingGateway), false)
+		},
+		Mutate: func(s *Store, tx *txn) error {
+			configEntryDest := &structs.ServiceConfigEntry{
+				Kind:        structs.ServiceDefaults,
+				Name:        "destination1",
+				Destination: &structs.DestinationConfig{Port: 9000, Address: "kafka.test.com"},
+			}
+			return ensureConfigEntryTxn(tx, tx.Index, configEntryDest)
+		},
+		WantEvents: []stream.Event{
+			testServiceHealthDeregistrationEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evServiceTermingGateway("destination1"),
+				evTerminatingGatewayVirtualIPs("destination1")),
+			testServiceHealthEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evNodeUnchanged,
+				evServiceUnchanged,
+				evServiceTermingGateway("destination1"),
+				evTerminatingGatewayVirtualIPs("destination1"),
+			),
+		},
+	})
+
+	run(t, eventsTestCase{
+		Name: "terminating gateway destination service-defaults wildcard",
+		Setup: func(s *Store, tx *txn) error {
+			configEntry := &structs.TerminatingGatewayConfigEntry{
+				Kind: structs.TerminatingGateway,
+				Name: "tgate1",
+				Services: []structs.LinkedService{
+					{
+						Name:           "*",
+						EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
+					},
+				},
+				EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
+			}
+			err := ensureConfigEntryTxn(tx, tx.Index, configEntry)
+			if err != nil {
+				return err
+			}
+			return s.ensureRegistrationTxn(tx, tx.Index, false,
+				testServiceRegistration(t, "tgate1", regTerminatingGateway), false)
+		},
+		Mutate: func(s *Store, tx *txn) error {
+			configEntryDest := &structs.ServiceConfigEntry{
+				Kind:        structs.ServiceDefaults,
+				Name:        "destination1",
+				Destination: &structs.DestinationConfig{Port: 9000, Address: "kafka.test.com"},
+			}
+			return ensureConfigEntryTxn(tx, tx.Index, configEntryDest)
+		},
+		WantEvents: []stream.Event{
+			testServiceHealthEvent(t,
+				"tgate1",
+				evConnectTopic,
+				evNodeUnchanged,
+				evServiceUnchanged,
+				evServiceTermingGateway("destination1"),
+				evTerminatingGatewayVirtualIPs("*"),
+			),
+		},
+	})
 }
 
 func (tc eventsTestCase) run(t *testing.T) {