Merge pull request #1728 from tariq1890/istio-informers

add event handler for istio gateway and virtualservice source

source/gateway.go

@@ -28,6 +28,8 @@ import (
 	log "github.com/sirupsen/logrus"
 	networkingv1alpha3 "istio.io/client-go/pkg/apis/networking/v1alpha3"
 	istioclient "istio.io/client-go/pkg/clientset/versioned"
+	istioinformers "istio.io/client-go/pkg/informers/externalversions"
+	networkingv1alpha3informer "istio.io/client-go/pkg/informers/externalversions/networking/v1alpha3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -51,6 +53,7 @@ type gatewaySource struct {
 	combineFQDNAnnotation    bool
 	ignoreHostnameAnnotation bool
 	serviceInformer          coreinformers.ServiceInformer
+	gatewayInformer          networkingv1alpha3informer.GatewayInformer
 }
 
 // NewIstioGatewaySource creates a new gatewaySource with the given config.
@@ -81,6 +84,8 @@ func NewIstioGatewaySource(
 	// Set resync period to 0, to prevent processing when nothing has changed
 	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0, kubeinformers.WithNamespace(namespace))
 	serviceInformer := informerFactory.Core().V1().Services()
+	istioInformerFactory := istioinformers.NewSharedInformerFactory(istioClient, 0)
+	gatewayInformer := istioInformerFactory.Networking().V1alpha3().Gateways()
 
 	// Add default resource event handlers to properly initialize informer.
 	serviceInformer.Informer().AddEventHandler(
@@ -91,8 +96,17 @@ func NewIstioGatewaySource(
 		},
 	)
 
+	gatewayInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				log.Debug("gateway added")
+			},
+		},
+	)
+
 	// TODO informer is not explicitly stopped since controller is not passing in its channel.
 	informerFactory.Start(wait.NeverStop)
+	istioInformerFactory.Start(wait.NeverStop)
 
 	// wait for the local cache to be populated.
 	err = poll(time.Second, 60*time.Second, func() (bool, error) {
@@ -102,6 +116,14 @@ func NewIstioGatewaySource(
 		return nil, fmt.Errorf("failed to sync cache: %v", err)
 	}
 
+	// wait for the local cache to be populated.
+	err = poll(time.Second, 60*time.Second, func() (bool, error) {
+		return gatewayInformer.Informer().HasSynced(), nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to sync cache: %v", err)
+	}
+
 	return &gatewaySource{
 		kubeClient:               kubeClient,
 		istioClient:              istioClient,
@@ -111,6 +133,7 @@ func NewIstioGatewaySource(
 		combineFQDNAnnotation:    combineFQDNAnnotation,
 		ignoreHostnameAnnotation: ignoreHostnameAnnotation,
 		serviceInformer:          serviceInformer,
+		gatewayInformer:          gatewayInformer,
 	}, nil
 }
 
@@ -180,9 +203,23 @@ func (sc *gatewaySource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, e
 	return endpoints, nil
 }
 
-// TODO(tariq1890): Implement this once we have evaluated and tested GatewayInformers
 // AddEventHandler adds an event handler that should be triggered if the watched Istio Gateway changes.
 func (sc *gatewaySource) AddEventHandler(ctx context.Context, handler func()) {
+	log.Debug("Adding event handler for Istio Gateway")
+
+	sc.gatewayInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				handler()
+			},
+			UpdateFunc: func(old interface{}, new interface{}) {
+				handler()
+			},
+			DeleteFunc: func(obj interface{}) {
+				handler()
+			},
+		},
+	)
 }
 
 // filterByAnnotations filters a list of configs by a given annotation selector.

source/virtualservice.go

@@ -29,6 +29,8 @@ import (
 
 	log "github.com/sirupsen/logrus"
 	istioclient "istio.io/client-go/pkg/clientset/versioned"
+	istioinformers "istio.io/client-go/pkg/informers/externalversions"
+	networkingv1alpha3informer "istio.io/client-go/pkg/informers/externalversions/networking/v1alpha3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -55,6 +57,7 @@ type virtualServiceSource struct {
 	combineFQDNAnnotation    bool
 	ignoreHostnameAnnotation bool
 	serviceInformer          coreinformers.ServiceInformer
+	virtualserviceInformer   networkingv1alpha3informer.VirtualServiceInformer
 }
 
 // NewIstioVirtualServiceSource creates a new virtualServiceSource with the given config.
@@ -85,6 +88,8 @@ func NewIstioVirtualServiceSource(
 	// Set resync period to 0, to prevent processing when nothing has changed
 	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0, kubeinformers.WithNamespace(namespace))
 	serviceInformer := informerFactory.Core().V1().Services()
+	istioInformerFactory := istioinformers.NewSharedInformerFactory(istioClient, 0)
+	virtualServiceInformer := istioInformerFactory.Networking().V1alpha3().VirtualServices()
 
 	// Add default resource event handlers to properly initialize informer.
 	serviceInformer.Informer().AddEventHandler(
@@ -95,8 +100,17 @@ func NewIstioVirtualServiceSource(
 		},
 	)
 
+	virtualServiceInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				log.Debug("virtual service added")
+			},
+		},
+	)
+
 	// TODO informer is not explicitly stopped since controller is not passing in its channel.
 	informerFactory.Start(wait.NeverStop)
+	istioInformerFactory.Start(wait.NeverStop)
 
 	// wait for the local cache to be populated.
 	err = wait.Poll(time.Second, 60*time.Second, func() (bool, error) {
@@ -106,6 +120,13 @@ func NewIstioVirtualServiceSource(
 		return nil, fmt.Errorf("failed to sync cache: %v", err)
 	}
 
+	err = wait.Poll(time.Second, 60*time.Second, func() (bool, error) {
+		return virtualServiceInformer.Informer().HasSynced(), nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to sync cache: %v", err)
+	}
+
 	return &virtualServiceSource{
 		kubeClient:               kubeClient,
 		istioClient:              istioClient,
@@ -115,6 +136,7 @@ func NewIstioVirtualServiceSource(
 		combineFQDNAnnotation:    combineFQDNAnnotation,
 		ignoreHostnameAnnotation: ignoreHostnameAnnotation,
 		serviceInformer:          serviceInformer,
+		virtualserviceInformer:   virtualServiceInformer,
 	}, nil
 }
 
@@ -179,9 +201,23 @@ func (sc *virtualServiceSource) Endpoints(ctx context.Context) ([]*endpoint.Endp
 	return endpoints, nil
 }
 
-// TODO(tariq1890): Implement this once we have evaluated and tested VirtualServiceInformers
 // AddEventHandler adds an event handler that should be triggered if the watched Istio VirtualService changes.
 func (sc *virtualServiceSource) AddEventHandler(ctx context.Context, handler func()) {
+	log.Debug("Adding event handler for Istio VirtualService")
+
+	sc.virtualserviceInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				handler()
+			},
+			UpdateFunc: func(old interface{}, new interface{}) {
+				handler()
+			},
+			DeleteFunc: func(obj interface{}) {
+				handler()
+			},
+		},
+	)
 }
 
 func (sc *virtualServiceSource) getGateway(ctx context.Context, gatewayStr string, virtualService networkingv1alpha3.VirtualService) *networkingv1alpha3.Gateway {