update gateway-services table with endpoints

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.ServiceName{Name: svc.Service, EnterpriseMeta: 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 {
@@ -3469,6 +3476,7 @@ func terminatingConfigGatewayServices(
 		return false, nil, fmt.Errorf("unexpected config entry type: %T", conf)
 	}
 
+	// TODO(egress-tproxy): Check if it's an endpoint by quering the service and the service defaut config entry.
 	// Check if service list matches the last known list for the config entry, if it does, skip the update
 	_, c, err := configEntryTxn(tx, nil, conf.GetKind(), conf.GetName(), entMeta)
 	if err != nil {
@@ -3483,6 +3491,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 query endpoints: %v", err)
+		}
 		mapping := &structs.GatewayService{
 			Gateway:     gateway,
 			Service:     structs.NewServiceName(svc.Name, &svc.EnterpriseMeta),
@@ -3491,13 +3503,36 @@ func terminatingConfigGatewayServices(
 			CertFile:    svc.CertFile,
 			CAFile:      svc.CAFile,
 			SNI:         svc.SNI,
+			Kind:        kind,
 		}
 
 		gatewayServices = append(gatewayServices, mapping)
 	}
 	return false, gatewayServices, nil
 }
 
+func GatewayServiceKind(tx ReadTxn, name string, entMeta *acl.EnterpriseMeta) (structs.GatewayServiceKind, error) {
+	serviceIter, err := tx.Get(tableServices, indexService, Query{
+		Value:          name,
+		EnterpriseMeta: *entMeta,
+	})
+	if err != nil {
+		return structs.GatewayservicekindService, err
+	}
+	for service := serviceIter.Next(); service != nil; service = serviceIter.Next() {
+		return structs.GatewayservicekindUnknown, nil
+	}
+
+	_, entry, err := configEntryTxn(tx, nil, structs.ServiceDefaults, name, entMeta)
+	if err != nil {
+		return structs.GatewayservicekindUnknown, err
+	}
+	if entry != 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 +3573,41 @@ func updateGatewayNamespace(tx WriteTxn, idx uint64, service *structs.GatewaySer
 			return err
 		}
 	}
+	endpoints, err := tx.Get(tableConfigEntries, indexID+"_prefix", ConfigEntryKindQuery{Kind: structs.ServiceDefaults, EnterpriseMeta: *entMeta})
+	if err != nil {
+		return fmt.Errorf("failed querying endpoints: %s", err)
+	}
+	for endpoint := endpoints.Next(); endpoint != nil; endpoint = endpoints.Next() {
+		e := endpoint.(*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.Kind = 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 +3655,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.Kind = 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 +3679,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.Kind = 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 +3734,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 reconcile mesh topology for gateway: %v", 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) {

agent/consul/state/catalog_test.go

@@ -5170,6 +5170,7 @@ func TestStateStore_GatewayServices_Terminating(t *testing.T) {
 				CreateIndex: 23,
 				ModifyIndex: 23,
 			},
+			Kind: structs.GatewayservicekindService,
 		},
 	}
 	assert.Equal(t, expect, out)
@@ -6421,6 +6422,7 @@ func TestStateStore_DumpGatewayServices(t *testing.T) {
 					CreateIndex: 22,
 					ModifyIndex: 22,
 				},
+				Kind: structs.GatewayservicekindService,
 			},
 		}
 		assert.Equal(t, expect, out)