Avoid accumulating synthetic upstreams

Synthetic upstreams from service-defaults config are stored locally in
the Upstreams list. Since these come from service-defaults they should
be cleaned up locally when no longer present in the service config
response.

agent/service_manager.go

@@ -326,7 +326,7 @@ func makeConfigRequest(bd BaseDeps, addReq AddServiceRequest) *structs.ServiceCo
 		// Also if we have any upstreams defined, add them to the request so we can
 		// learn about their configs.
 		for _, us := range ns.Proxy.Upstreams {
-			if us.DestinationType == "" || us.DestinationType == structs.UpstreamDestTypeService {
+			if us.DestinationType == "" || us.DestinationType == structs.UpstreamDestTypeService && !us.CentrallyConfigured {
 				sid := us.DestinationID()
 				sid.EnterpriseMeta.Merge(&ns.EnterpriseMeta)
 				upstreams = append(upstreams, sid)
@@ -440,14 +440,25 @@ func mergeServiceConfig(defaults *structs.ServiceConfigResponse, service *struct
 	// This does not apply outside of TransparentProxy mode because in that situation every possible upstream already exists
 	// inside of ns.Proxy.Upstreams.
 	if ns.Proxy.TransparentProxy {
+		var upstreams structs.Upstreams
+
+		for _, us := range ns.Proxy.Upstreams {
+			if _, ok := remoteUpstreams[us.DestinationID()]; !ok && us.CentrallyConfigured {
+				// If a centrally configured upstream is only present locally then that means it was
+				// removed from central config and should be removed from the local list as well.
+				continue
+			}
+			upstreams = append(upstreams, us)
+		}
+
 		for id, remote := range remoteUpstreams {
 			if _, ok := localUpstreams[id]; ok {
 				// Remote upstream is already present locally
 				continue
 			}
-
+			upstreams = append(upstreams, remote)
-			ns.Proxy.Upstreams = append(ns.Proxy.Upstreams, remote)
 		}
+		ns.Proxy.Upstreams = upstreams
 	}
 
 	return ns, err

agent/service_manager_test.go

@@ -1047,6 +1047,64 @@ func Test_mergeServiceConfig_UpstreamOverrides(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "synthetic local upstream is cleaned up if not in config response",
+			args: args{
+				defaults: &structs.ServiceConfigResponse{
+					UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{
+						{
+							Upstream: structs.ServiceID{
+								ID:             "zap",
+								EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+							},
+							Config: map[string]interface{}{
+								"protocol": "grpc",
+							},
+						},
+					},
+				},
+				service: &structs.NodeService{
+					ID:      "foo-proxy",
+					Service: "foo-proxy",
+					Proxy: structs.ConnectProxyConfig{
+						DestinationServiceName: "foo",
+						DestinationServiceID:   "foo",
+						TransparentProxy:       true,
+						Upstreams: structs.Upstreams{
+							structs.Upstream{
+								DestinationNamespace: "default",
+								DestinationName:      "zip",
+								LocalBindPort:        8080,
+								Config: map[string]interface{}{
+									"protocol": "http",
+								},
+								// Should get zip cleaned up
+								CentrallyConfigured: true,
+							},
+						},
+					},
+				},
+			},
+			want: &structs.NodeService{
+				ID:      "foo-proxy",
+				Service: "foo-proxy",
+				Proxy: structs.ConnectProxyConfig{
+					DestinationServiceName: "foo",
+					DestinationServiceID:   "foo",
+					TransparentProxy:       true,
+					Upstreams: structs.Upstreams{
+						structs.Upstream{
+							DestinationNamespace: "default",
+							DestinationName:      "zap",
+							Config: map[string]interface{}{
+								"protocol": "grpc",
+							},
+							CentrallyConfigured: true,
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "upstream mode from remote defaults overrides local default",
 			args: args{