jiazhizhong/higress
1
0
Fork 0
mirror of https://github.com/alibaba/higress.git synced 2026-03-02 23:51:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: Support only watching key resources in one namespace (#1821)

Browse Source
This commit is contained in:
Kent Dong
2025-03-03 15:40:44 +08:00
committed by GitHub
parent 988e2c1fa7
commit 188914a16b
17 changed files with 116 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
helm/core/templates/controller-deployment.yaml
View File

@@ -139,6 +139,10 @@ spec:
periodSeconds: 3
timeoutSeconds: 5
env:
{{- if .Values.global.watchNamespace }}
- name: ISTIO_WATCH_NAMESPACE
value: "{{ .Values.global.watchNamespace }}"
{{- end }}
- name: ENABLE_PUSH_ALL_MCP_CLUSTERS
value: "{{ .Values.global.enablePushAllMCPClusters }}"
- name: PILOT_ENABLE_LDS_CACHE

2
pkg/bootstrap/server.go
View File

@@ -235,7 +235,7 @@ func (s *Server) initConfigController() error {
options.ClusterId = ""
}
ingressConfig := translation.NewIngressTranslation(s.kubeClient, s.xdsServer, ns, options.ClusterId)
ingressConfig := translation.NewIngressTranslation(s.kubeClient, s.xdsServer, ns, options)
ingressConfig.AddLocalCluster(options)
s.configStores = append(s.configStores, ingressConfig)

11
pkg/ingress/config/ingress_config.go
View File

@@ -153,7 +153,8 @@ type IngressConfig struct {
httpsConfigMgr *cert.ConfigMgr
}
func NewIngressConfig(localKubeClient kube.Client, xdsUpdater istiomodel.XDSUpdater, namespace string, clusterId cluster.ID) *IngressConfig {
func NewIngressConfig(localKubeClient kube.Client, xdsUpdater istiomodel.XDSUpdater, namespace string, options common.Options) *IngressConfig {
clusterId := options.ClusterId
if clusterId == "Kubernetes" {
clusterId = ""
}
@@ -170,17 +171,17 @@ func NewIngressConfig(localKubeClient kube.Client, xdsUpdater istiomodel.XDSUpda
wasmPlugins: make(map[string]*extensions.WasmPlugin),
http2rpcs: make(map[string]*higressv1.Http2Rpc),
}
mcpbridgeController := mcpbridge.NewController(localKubeClient, clusterId)
mcpbridgeController := mcpbridge.NewController(localKubeClient, options)
mcpbridgeController.AddEventHandler(config.AddOrUpdateMcpBridge, config.DeleteMcpBridge)
config.mcpbridgeController = mcpbridgeController
config.mcpbridgeLister = mcpbridgeController.Lister()
wasmPluginController := wasmplugin.NewController(localKubeClient, clusterId)
wasmPluginController := wasmplugin.NewController(localKubeClient, options)
wasmPluginController.AddEventHandler(config.AddOrUpdateWasmPlugin, config.DeleteWasmPlugin)
config.wasmPluginController = wasmPluginController
config.wasmPluginLister = wasmPluginController.Lister()
http2rpcController := http2rpc.NewController(localKubeClient, clusterId)
http2rpcController := http2rpc.NewController(localKubeClient, options)
http2rpcController.AddEventHandler(config.AddOrUpdateHttp2Rpc, config.DeleteHttp2Rpc)
config.http2rpcController = http2rpcController
config.http2rpcLister = http2rpcController.Lister()
@@ -225,7 +226,7 @@ func (m *IngressConfig) RegisterEventHandler(kind config.GroupVersionKind, f ist
}
func (m *IngressConfig) AddLocalCluster(options common.Options) {
secretController := secret.NewController(m.localKubeClient, options.ClusterId)
secretController := secret.NewController(m.localKubeClient, options)
secretController.AddEventHandler(m.ReflectSecretChanges)
var ingressController common.IngressController

9
pkg/ingress/config/ingress_config_test.go
View File

@@ -127,7 +127,14 @@ func TestConvertGatewaysForIngress(t *testing.T) {
}
ingressV1Beta1Controller := controllerv1beta1.NewController(fake, fake, v1Beta1Options, nil)
ingressV1Controller := controllerv1.NewController(fake, fake, v1Options, nil)
m := NewIngressConfig(fake, nil, "wakanda", "gw-123-istio")
options := common.Options{
Enable: true,
ClusterId: "gw-123-istio",
RawClusterId: "gw-123-istio__",
GatewayHttpPort: 80,
GatewayHttpsPort: 443,
}
m := NewIngressConfig(fake, nil, "wakanda", options)
m.remoteIngressControllers = map[cluster.ID]common.IngressController{
"ingress-v1beta1": ingressV1Beta1Controller,
"ingress-v1": ingressV1Controller,

5
pkg/ingress/config/kingress_config.go
View File

@@ -75,10 +75,11 @@ type KIngressConfig struct {
clusterId cluster.ID
}
func NewKIngressConfig(localKubeClient kube.Client, XDSUpdater istiomodel.XDSUpdater, namespace string, clusterId cluster.ID) *KIngressConfig {
func NewKIngressConfig(localKubeClient kube.Client, XDSUpdater istiomodel.XDSUpdater, namespace string, options common.Options) *KIngressConfig {
if localKubeClient.KIngressInformer() == nil {
return nil
}
clusterId := options.ClusterId
if clusterId == "Kubernetes" {
clusterId = ""
}
@@ -114,7 +115,7 @@ func (m *KIngressConfig) RegisterEventHandler(kind config.GroupVersionKind, f is
}
func (m *KIngressConfig) AddLocalCluster(options common.Options) common.KIngressController {
secretController := secret.NewController(m.localKubeClient, options.ClusterId)
secretController := secret.NewController(m.localKubeClient, options)
secretController.AddEventHandler(m.ReflectSecretChanges)
var ingressController common.KIngressController

9
pkg/ingress/config/kingress_config_test.go
View File

@@ -118,7 +118,14 @@ func TestConvertGatewaysForKIngress(t *testing.T) {
RawClusterId: "kingress__",
}
kingressV1Controller := kcontrollerv1.NewController(fake, fake, v1Options, nil)
m := NewKIngressConfig(fake, nil, "wakanda", "gw-123-istio")
options := common.Options{
Enable: true,
ClusterId: "gw-123-istio",
RawClusterId: "gw-123-istio__",
GatewayHttpPort: 80,
GatewayHttpsPort: 443,
}
m := NewKIngressConfig(fake, nil, "wakanda", options)
m.remoteIngressControllers = map[cluster.ID]common.KIngressController{
"kingress": kingressV1Controller,
}

22
pkg/ingress/kube/http2rpc/controller.go
View File

@@ -15,21 +15,33 @@
package http2rpc
import (
"istio.io/istio/pkg/cluster"
"time"
"istio.io/istio/pkg/kube/controllers"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/cache"
v1 "github.com/alibaba/higress/client/pkg/apis/networking/v1"
"github.com/alibaba/higress/client/pkg/clientset/versioned"
informersv1 "github.com/alibaba/higress/client/pkg/informers/externalversions/networking/v1"
listersv1 "github.com/alibaba/higress/client/pkg/listers/networking/v1"
"github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/pkg/ingress/kube/controller"
kubeclient "github.com/alibaba/higress/pkg/kube"
)
type Http2RpcController controller.Controller[listersv1.Http2RpcLister]
func NewController(client kubeclient.Client, clusterId cluster.ID) Http2RpcController {
informer := client.HigressInformer().Networking().V1().Http2Rpcs().Informer()
return controller.NewCommonController("http2rpc", client.HigressInformer().Networking().V1().Http2Rpcs().Lister(),
informer, GetHttp2Rpc, clusterId)
func NewController(client kubeclient.Client, options common.Options) Http2RpcController {
var informer cache.SharedIndexInformer
if options.WatchNamespace == "" {
informer = client.HigressInformer().Networking().V1().Http2Rpcs().Informer()
} else {
informer = client.HigressInformer().InformerFor(&v1.Http2Rpc{}, func(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return informersv1.NewHttp2RpcInformer(client, options.WatchNamespace, resyncPeriod, nil)
})
}
return controller.NewCommonController("http2rpc", listersv1.NewHttp2RpcLister(informer.GetIndexer()), informer, GetHttp2Rpc, options.ClusterId)
}
func GetHttp2Rpc(lister listersv1.Http2RpcLister, namespacedName types.NamespacedName) (controllers.Object, error) {

2
pkg/ingress/kube/ingress/controller.go
View File

@@ -100,7 +100,7 @@ type controller struct {
// NewController creates a new Kubernetes controller
func NewController(localKubeClient, client kubeclient.Client, options common.Options,
secretController secret.SecretController) common.IngressController {
opts := ktypes.InformerOptions{}
opts := ktypes.InformerOptions{Namespace: options.WatchNamespace}
ingressInformer := util.GetInformerFiltered(client, opts, gvrIngressV1Beta1, &ingress.Ingress{},
func(options metav1.ListOptions) (runtime.Object, error) {
return client.Kube().NetworkingV1beta1().Ingresses(opts.Namespace).List(context.Background(), options)

6
pkg/ingress/kube/ingress/controller_test.go
View File

@@ -54,7 +54,7 @@ func TestIngressControllerApplies(t *testing.T) {
options := common.Options{IngressClass: "mse", ClusterId: ""}
secretController := secret.NewController(localKubeClient, options.ClusterId)
secretController := secret.NewController(localKubeClient, options)
ingressController := NewController(localKubeClient, client, options, secretController)
testcases := map[string]func(*testing.T, common.IngressController){
@@ -253,7 +253,7 @@ func TestIngressControllerConventions(t *testing.T) {
options := common.Options{IngressClass: "mse", ClusterId: "", EnableStatus: true}
secretController := secret.NewController(localKubeClient, options.ClusterId)
secretController := secret.NewController(localKubeClient, options)
ingressController := NewController(localKubeClient, client, options, secretController)
testcases := map[string]func(*testing.T, common.IngressController){
@@ -1142,7 +1142,7 @@ func TestIngressControllerProcessing(t *testing.T) {
options := common.Options{IngressClass: "mse", ClusterId: "", EnableStatus: true}
secretController := secret.NewController(localKubeClient, options.ClusterId)
secretController := secret.NewController(localKubeClient, options)
opts := ktypes.InformerOptions{}
ingressInformer := util.GetInformerFiltered(fakeClient, opts, gvrIngressV1Beta1, &ingress.Ingress{},

2
pkg/ingress/kube/ingressv1/controller.go
View File

@@ -92,7 +92,7 @@ type controller struct {
// NewController creates a new Kubernetes controller
func NewController(localKubeClient, client kubeclient.Client, options common.Options, secretController secret.SecretController) common.IngressController {
opts := ktypes.InformerOptions{}
opts := ktypes.InformerOptions{Namespace: options.WatchNamespace}
ingressInformer := schemakubeclient.GetInformerFilteredFromGVR(client, opts, gvr.Ingress)
ingressLister := networkinglister.NewIngressLister(ingressInformer.Informer.GetIndexer())
serviceInformer := schemakubeclient.GetInformerFilteredFromGVR(client, opts, gvr.Service)

24
pkg/ingress/kube/kingress/controller.go
View File

@@ -21,6 +21,7 @@ import (
"sort"
"strings"
"sync"
"time"
"github.com/hashicorp/go-multierror"
networking "istio.io/api/networking/v1alpha3"
@@ -43,7 +44,9 @@ import (
listerv1 "k8s.io/client-go/listers/core/v1"
"k8s.io/client-go/tools/cache"
ingress "knative.dev/networking/pkg/apis/networking/v1alpha1"
networkingv1alpha1 "knative.dev/networking/pkg/client/listers/networking/v1alpha1"
"knative.dev/networking/pkg/client/clientset/versioned"
informernetworkingv1alpha1 "knative.dev/networking/pkg/client/informers/externalversions/networking/v1alpha1"
listernetworkingv1alpha1 "knative.dev/networking/pkg/client/listers/networking/v1alpha1"
"github.com/alibaba/higress/pkg/ingress/kube/annotations"
"github.com/alibaba/higress/pkg/ingress/kube/common"
@@ -76,7 +79,7 @@ type controller struct {
ingresses map[string]*ingress.Ingress
ingressInformer cache.SharedInformer
ingressLister networkingv1alpha1.IngressLister
ingressLister listernetworkingv1alpha1.IngressLister
serviceInformer informerfactory.StartableInformer
serviceLister listerv1.ServiceLister
secretController secret.SecretController
@@ -86,16 +89,23 @@ type controller struct {
// NewController creates a new Kubernetes controller
func NewController(localKubeClient, client kube.Client, options common.Options,
secretController secret.SecretController) common.KIngressController {
//var namespace string = "default"
ingressInformer := client.KIngressInformer().Networking().V1alpha1().Ingresses()
serviceInformer := schemakubeclient.GetInformerFilteredFromGVR(client, ktypes.InformerOptions{}, gvr.Service)
var ingressInformer cache.SharedIndexInformer
if options.WatchNamespace == "" {
ingressInformer = client.KIngressInformer().Networking().V1alpha1().Ingresses().Informer()
} else {
ingressInformer = client.KIngressInformer().InformerFor(&ingress.Ingress{}, func(c versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return informernetworkingv1alpha1.NewIngressInformer(c, options.WatchNamespace, resyncPeriod, nil)
})
}
ingressLister := listernetworkingv1alpha1.NewIngressLister(ingressInformer.GetIndexer())
serviceInformer := schemakubeclient.GetInformerFilteredFromGVR(client, ktypes.InformerOptions{Namespace: options.WatchNamespace}, gvr.Service)
serviceLister := listerv1.NewServiceLister(serviceInformer.Informer.GetIndexer())
c := &controller{
options: options,
ingresses: make(map[string]*ingress.Ingress),
ingressInformer: ingressInformer.Informer(),
ingressLister: ingressInformer.Lister(),
ingressInformer: ingressInformer,
ingressLister: ingressLister,
serviceInformer: serviceInformer,
serviceLister: serviceLister,
secretController: secretController,

2
pkg/ingress/kube/kingress/controller_test.go
View File

@@ -154,7 +154,7 @@ func TestKIngressControllerConventions(t *testing.T) {
options := common.Options{IngressClass: "mse", ClusterId: "", EnableStatus: true}
secretController := secret.NewController(localKubeClient, options.ClusterId)
secretController := secret.NewController(localKubeClient, options)
ingressController := NewController(localKubeClient, client, options, secretController)
testcases := map[string]func(*testing.T, common.KIngressController){

22
pkg/ingress/kube/mcpbridge/controller.go
View File

@@ -15,21 +15,33 @@
package mcpbridge
import (
"istio.io/istio/pkg/cluster"
"time"
"istio.io/istio/pkg/kube/controllers"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/cache"
v1 "github.com/alibaba/higress/client/pkg/apis/networking/v1"
"github.com/alibaba/higress/client/pkg/clientset/versioned"
informersv1 "github.com/alibaba/higress/client/pkg/informers/externalversions/networking/v1"
listersv1 "github.com/alibaba/higress/client/pkg/listers/networking/v1"
"github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/pkg/ingress/kube/controller"
kubeclient "github.com/alibaba/higress/pkg/kube"
)
type McpBridgeController controller.Controller[listersv1.McpBridgeLister]
func NewController(client kubeclient.Client, clusterId cluster.ID) McpBridgeController {
informer := client.HigressInformer().Networking().V1().McpBridges().Informer()
return controller.NewCommonController("mcpbridge", client.HigressInformer().Networking().V1().McpBridges().Lister(),
informer, GetMcpBridge, clusterId)
func NewController(client kubeclient.Client, options common.Options) McpBridgeController {
var informer cache.SharedIndexInformer
if options.WatchNamespace == "" {
informer = client.HigressInformer().Networking().V1().McpBridges().Informer()
} else {
informer = client.HigressInformer().InformerFor(&v1.McpBridge{}, func(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return informersv1.NewMcpBridgeInformer(client, options.WatchNamespace, resyncPeriod, nil)
})
}
return controller.NewCommonController("mcpbridge", listersv1.NewMcpBridgeLister(informer.GetIndexer()), informer, GetMcpBridge, options.ClusterId)
}
func GetMcpBridge(lister listersv1.McpBridgeLister, namespacedName types.NamespacedName) (controllers.Object, error) {

11
pkg/ingress/kube/secret/controller.go
View File

@@ -15,15 +15,14 @@
package secret
import (
"github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/pkg/ingress/kube/controller"
"istio.io/istio/pkg/cluster"
"istio.io/istio/pkg/config/schema/gvr"
schemakubeclient "istio.io/istio/pkg/config/schema/kubeclient"
kubeclient "istio.io/istio/pkg/kube"
"istio.io/istio/pkg/kube/controllers"
ktypes "istio.io/istio/pkg/kube/kubetypes"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/types"
listersv1 "k8s.io/client-go/listers/core/v1"
@@ -31,17 +30,17 @@ import (
type SecretController controller.Controller[listersv1.SecretLister]
func NewController(client kubeclient.Client, clusterId cluster.ID) SecretController {
func NewController(client kubeclient.Client, options common.Options) SecretController {
opts := ktypes.InformerOptions{
Namespace: metav1.NamespaceAll,
Cluster: clusterId,
Namespace: options.WatchNamespace,
Cluster: options.ClusterId,
FieldSelector: fields.AndSelectors(
fields.OneTermNotEqualSelector("type", "helm.sh/release.v1"),
fields.OneTermNotEqualSelector("type", string(v1.SecretTypeServiceAccountToken)),
).String(),
}
informer := schemakubeclient.GetInformerFilteredFromGVR(client, opts, gvr.Secret)
return controller.NewCommonController("secret", listersv1.NewSecretLister(informer.Informer.GetIndexer()), informer.Informer, GetSecret, clusterId)
return controller.NewCommonController("secret", listersv1.NewSecretLister(informer.Informer.GetIndexer()), informer.Informer, GetSecret, options.ClusterId)
}
func GetSecret(lister listersv1.SecretLister, namespacedName types.NamespacedName) (controllers.Object, error) {

3
pkg/ingress/kube/secret/controller_test.go
View File

@@ -16,6 +16,7 @@ package secret
import (
"context"
"github.com/alibaba/higress/pkg/ingress/kube/common"
"reflect"
"sync"
"testing"
@@ -43,7 +44,7 @@ var period = time.Second
func TestController(t *testing.T) {
client := kubeclient.NewFakeClient()
ctrl := NewController(client, "fake-cluster")
ctrl := NewController(client, common.Options{ClusterId: "fake-cluster"})
stop := make(chan struct{})
t.Cleanup(func() {

22
pkg/ingress/kube/wasmplugin/controller.go
View File

@@ -15,21 +15,33 @@
package wasmplugin
import (
"istio.io/istio/pkg/cluster"
"time"
"istio.io/istio/pkg/kube/controllers"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/cache"
v1 "github.com/alibaba/higress/client/pkg/apis/extensions/v1alpha1"
"github.com/alibaba/higress/client/pkg/clientset/versioned"
informersv1 "github.com/alibaba/higress/client/pkg/informers/externalversions/extensions/v1alpha1"
listersv1 "github.com/alibaba/higress/client/pkg/listers/extensions/v1alpha1"
"github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/pkg/ingress/kube/controller"
kubeclient "github.com/alibaba/higress/pkg/kube"
)
type WasmPluginController controller.Controller[listersv1.WasmPluginLister]
func NewController(client kubeclient.Client, clusterId cluster.ID) WasmPluginController {
informer := client.HigressInformer().Extensions().V1alpha1().WasmPlugins().Informer()
return controller.NewCommonController("wasmplugin", client.HigressInformer().Extensions().V1alpha1().WasmPlugins().Lister(),
informer, GetWasmPlugin, clusterId)
func NewController(client kubeclient.Client, options common.Options) WasmPluginController {
var informer cache.SharedIndexInformer
if options.WatchNamespace == "" {
informer = client.HigressInformer().Extensions().V1alpha1().WasmPlugins().Informer()
} else {
informer = client.HigressInformer().InformerFor(&v1.WasmPlugin{}, func(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return informersv1.NewWasmPluginInformer(client, options.WatchNamespace, resyncPeriod, nil)
})
}
return controller.NewCommonController("wasmplugin", listersv1.NewWasmPluginLister(informer.GetIndexer()), informer, GetWasmPlugin, options.ClusterId)
}
func GetWasmPlugin(lister listersv1.WasmPluginLister, namespacedName types.NamespacedName) (controllers.Object, error) {

11
pkg/ingress/translation/translation.go
View File

@@ -19,7 +19,6 @@ import (
"istio.io/istio/pilot/pkg/model"
istiomodel "istio.io/istio/pilot/pkg/model"
"istio.io/istio/pkg/cluster"
"istio.io/istio/pkg/config"
"istio.io/istio/pkg/config/schema/collection"
"istio.io/istio/pkg/config/schema/gvk"
@@ -45,13 +44,13 @@ type IngressTranslation struct {
higressDomainCache model.IngressDomainCollection
}
func NewIngressTranslation(localKubeClient kube.Client, xdsUpdater istiomodel.XDSUpdater, namespace string, clusterId cluster.ID) *IngressTranslation {
if clusterId == "Kubernetes" {
clusterId = ""
func NewIngressTranslation(localKubeClient kube.Client, xdsUpdater istiomodel.XDSUpdater, namespace string, options common.Options) *IngressTranslation {
if options.ClusterId == "Kubernetes" {
options.ClusterId = ""
}
Config := &IngressTranslation{
ingressConfig: ingressconfig.NewIngressConfig(localKubeClient, xdsUpdater, namespace, clusterId),
kingressConfig: ingressconfig.NewKIngressConfig(localKubeClient, xdsUpdater, namespace, clusterId),
ingressConfig: ingressconfig.NewIngressConfig(localKubeClient, xdsUpdater, namespace, options),
kingressConfig: ingressconfig.NewKIngressConfig(localKubeClient, xdsUpdater, namespace, options),
}
return Config
}