process gtps as soon they are applied (#223)

admiral/pkg/clusters/types.go

@@ -3,17 +3,19 @@ package clusters
 import (
 	"context"
 	"errors"
+	"fmt"
+	"sync"
+	"time"
+
 	argo "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
-	"github.com/istio-ecosystem/admiral/admiral/pkg/apis/admiral/v1"
+	v1 "github.com/istio-ecosystem/admiral/admiral/pkg/apis/admiral/v1"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/admiral"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/common"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/istio"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/secret"
 	log "github.com/sirupsen/logrus"
 	k8sAppsV1 "k8s.io/api/apps/v1"
 	k8s "k8s.io/client-go/kubernetes"
-	"sync"
-	"time"
 )
 
 type RemoteController struct {
@@ -180,14 +182,26 @@ func (dh *DependencyHandler) Deleted(obj *v1.Dependency) {
 
 func (gtp *GlobalTrafficHandler) Added(obj *v1.GlobalTrafficPolicy) {
 	log.Infof(LogFormat, "Added", "globaltrafficpolicy", obj.Name, gtp.ClusterID, "received")
+	err := HandleEventForGlobalTrafficPolicy(admiral.Add, obj, gtp.RemoteRegistry, gtp.ClusterID)
+	if err != nil {
+		log.Infof(err.Error())
+	}
 }
 
 func (gtp *GlobalTrafficHandler) Updated(obj *v1.GlobalTrafficPolicy) {
 	log.Infof(LogFormat, "Updated", "globaltrafficpolicy", obj.Name, gtp.ClusterID, "received")
+	err := HandleEventForGlobalTrafficPolicy(admiral.Update, obj, gtp.RemoteRegistry, gtp.ClusterID)
+	if err != nil {
+		log.Infof(err.Error())
+	}
 }
 
 func (gtp *GlobalTrafficHandler) Deleted(obj *v1.GlobalTrafficPolicy) {
 	log.Infof(LogFormat, "Deleted", "globaltrafficpolicy", obj.Name, gtp.ClusterID, "received")
+	err := HandleEventForGlobalTrafficPolicy(admiral.Delete, obj, gtp.RemoteRegistry, gtp.ClusterID)
+	if err != nil {
+		log.Infof(err.Error())
+	}
 }
 
 func (pc *DeploymentHandler) Added(obj *k8sAppsV1.Deployment) {
@@ -243,3 +257,19 @@ func HandleEventForDeployment(event admiral.EventType, obj *k8sAppsV1.Deployment
 	// Use the same function as added deployment function to update and put new service entry in place to replace old one
 	modifyServiceEntryForNewServiceOrPod(event, env, globalIdentifier, remoteRegistry)
 }
+
+// HandleEventForGlobalTrafficPolicy processes all the events related to GTPs
+func HandleEventForGlobalTrafficPolicy(event admiral.EventType, gtp *v1.GlobalTrafficPolicy, remoteRegistry *RemoteRegistry, clusterName string) error {
+
+	globalIdentifier := common.GetGtpIdentity(gtp)
+
+	if len(globalIdentifier) == 0 {
+		return fmt.Errorf(LogFormat, "Event", "globaltrafficpolicy", gtp.Name, clusterName, "Skipped as '"+common.GetWorkloadIdentifier()+" was not found', namespace="+gtp.Namespace)
+	}
+
+	env := common.GetGtpEnv(gtp)
+
+	// Use the same function as added deployment function to update and put new service entry in place to replace old one
+	modifyServiceEntryForNewServiceOrPod(event, env, globalIdentifier, remoteRegistry)
+	return nil
+}

admiral/pkg/clusters/types_test.go

@@ -12,6 +12,7 @@ import (
 	admiralFake "github.com/istio-ecosystem/admiral/admiral/pkg/client/clientset/versioned/fake"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/admiral"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/common"
+	"github.com/stretchr/testify/assert"
 	v12 "k8s.io/api/apps/v1"
 	v13 "k8s.io/api/core/v1"
 	time2 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -197,10 +198,64 @@ func TestRolloutHandler(t *testing.T) {
 			gtpCache.identityCache = make(map[string]*v1.GlobalTrafficPolicy)
 			gtpCache.mutex = &sync.Mutex{}
 			handler.RemoteRegistry.AdmiralCache.GlobalTrafficCache = gtpCache
-
 			handler.Added(c.addedRolout)
 			handler.Deleted(c.addedRolout)
 			handler.Updated(c.addedRolout)
 		})
 	}
 }
+
+func TestHandleEventForGlobalTrafficPolicy(t *testing.T) {
+	event := admiral.EventType("Add")
+	p := common.AdmiralParams{
+		KubeconfigPath: "testdata/fake.config",
+	}
+	registry, _ := InitAdmiral(context.Background(), p)
+
+	testcases := []struct {
+		name      string
+		gtp       *v1.GlobalTrafficPolicy
+		doesError bool
+	}{
+		{
+			name: "missing identity label in GTP should result in error being returned by the handler",
+			gtp: &v1.GlobalTrafficPolicy{
+				ObjectMeta: time2.ObjectMeta{
+					Name:        "testgtp",
+					Annotations: map[string]string{"admiral.io/env": "testenv"},
+				},
+			},
+			doesError: true,
+		},
+		{
+			name: "empty identity label in GTP should result in error being returned by the handler",
+			gtp: &v1.GlobalTrafficPolicy{
+				ObjectMeta: time2.ObjectMeta{
+					Name:        "testgtp",
+					Labels:      map[string]string{"identity": ""},
+					Annotations: map[string]string{"admiral.io/env": "testenv"},
+				},
+			},
+			doesError: true,
+		},
+		{
+			name: "valid GTP config which is expected to pass",
+			gtp: &v1.GlobalTrafficPolicy{
+				ObjectMeta: time2.ObjectMeta{
+					Name:        "testgtp",
+					Labels:      map[string]string{"identity": "testapp"},
+					Annotations: map[string]string{"admiral.io/env": "testenv"},
+				},
+			},
+			doesError: false,
+		},
+	}
+
+	for _, c := range testcases {
+		t.Run(c.name, func(t *testing.T) {
+			err := HandleEventForGlobalTrafficPolicy(event, c.gtp, registry, "testcluster")
+			assert.Equal(t, err != nil, c.doesError)
+		})
+	}
+
+}