-
Notifications
You must be signed in to change notification settings - Fork 79
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Client (mesh egress) discovery for Jobs and Numaflow (vertices and mo…
…novertices) (#339) ### Description What does this change do and why? This change introduces client discovery where there is no ingress but only mesh egress required for types k8s Jobs, numaflow Vertices and Monovertices. Signed-off-by: Anil Attuluri <[email protected]>
- Loading branch information
Showing
48 changed files
with
3,357 additions
and
746 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| package clusters | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/istio-ecosystem/admiral/admiral/pkg/controller/admiral" | ||
| "github.com/istio-ecosystem/admiral/admiral/pkg/controller/common" | ||
| commonUtil "github.com/istio-ecosystem/admiral/admiral/pkg/util" | ||
| log "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| type ClientDiscoveryHandler struct { | ||
| RemoteRegistry *RemoteRegistry | ||
| ClusterID string | ||
| } | ||
|
|
||
| func (cdh *ClientDiscoveryHandler) Added(ctx context.Context, obj *common.K8sObject) error { | ||
| err := HandleEventForClientDiscovery(ctx, admiral.Add, obj, cdh.RemoteRegistry, cdh.ClusterID) | ||
| if err != nil { | ||
| return fmt.Errorf(LogErrFormat, common.Add, common.JobResourceType, obj.Name, cdh.ClusterID, err) | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| func HandleEventForClientDiscovery(ctx context.Context, event admiral.EventType, obj *common.K8sObject, | ||
| remoteRegistry *RemoteRegistry, clusterName string) error { | ||
| log.Infof(LogFormat, event, obj.Type, obj.Name, clusterName, common.ReceivedStatus) | ||
| globalIdentifier := common.GetGlobalIdentifier(obj.Annotations, obj.Labels) | ||
| originalIdentifier := common.GetOriginalIdentifier(obj.Annotations, obj.Labels) | ||
| if len(globalIdentifier) == 0 { | ||
| log.Infof(LogFormat, event, obj.Type, obj.Name, clusterName, "Skipped as '"+common.GetWorkloadIdentifier()+" was not found', namespace="+obj.Namespace) | ||
| return nil | ||
| } | ||
| ctxLogger := common.GetCtxLogger(ctx, globalIdentifier, "") | ||
|
|
||
| ctx = context.WithValue(ctx, "clusterName", clusterName) | ||
| ctx = context.WithValue(ctx, "eventResourceType", obj.Type) | ||
|
|
||
| if remoteRegistry.AdmiralCache != nil { | ||
|
|
||
| UpdateIdentityClusterCache(remoteRegistry, globalIdentifier, clusterName) | ||
|
|
||
| if common.EnableSWAwareNSCaches() { | ||
| if remoteRegistry.AdmiralCache.IdentityClusterNamespaceCache != nil { | ||
| remoteRegistry.AdmiralCache.IdentityClusterNamespaceCache.Put(globalIdentifier, clusterName, obj.Namespace, obj.Namespace) | ||
| } | ||
| if remoteRegistry.AdmiralCache.PartitionIdentityCache != nil && len(common.GetIdentityPartition(obj.Annotations, obj.Labels)) > 0 { | ||
| remoteRegistry.AdmiralCache.PartitionIdentityCache.Put(globalIdentifier, originalIdentifier) | ||
| } | ||
| } | ||
| } else { | ||
| log.Warnf(LogFormatAdv, "Process", obj.Type, obj.Name, obj.Namespace, clusterName, "Skipping client discovery as Admiral cache is not initialized for identity="+globalIdentifier) | ||
| return fmt.Errorf(common.CtxLogFormat, event, obj.Name, obj.Namespace, clusterName, "processing skipped as Admiral cache is not initialized for identity="+globalIdentifier) | ||
| } | ||
|
|
||
| if commonUtil.IsAdmiralReadOnly() { | ||
| ctxLogger.Infof(common.CtxLogFormat, event, "", "", "", "processing skipped as Admiral is in Read-only mode") | ||
| return nil | ||
| } | ||
|
|
||
| if IsCacheWarmupTime(remoteRegistry) { | ||
| ctxLogger.Infof(common.CtxLogFormat, event, "", "", "", "processing skipped during cache warm up state") | ||
| return fmt.Errorf(common.CtxLogFormat, event, obj.Name, obj.Namespace, clusterName, "processing skipped during cache warm up state for env="+" identity="+globalIdentifier) | ||
| } | ||
|
|
||
| //if we have a deployment/rollout in this namespace skip processing to save some cycles | ||
| if DeploymentOrRolloutExistsInNamespace(remoteRegistry, globalIdentifier, clusterName, obj.Namespace) { | ||
| log.Infof(LogFormatAdv, "Process", obj.Type, obj.Name, obj.Namespace, clusterName, "Skipping client discovery as Deployment/Rollout already present in namespace for client="+globalIdentifier) | ||
| return nil | ||
| } | ||
|
|
||
| //write SEs required for this client | ||
| depRecord := remoteRegistry.DependencyController.Cache.Get(globalIdentifier) | ||
|
|
||
| if depRecord == nil { | ||
| log.Warnf(LogFormatAdv, "Process", obj.Type, obj.Name, obj.Namespace, clusterName, "Skipping client discovery as no dependency record found for client="+globalIdentifier) | ||
| return nil | ||
| } | ||
| err := remoteRegistry.DependencyController.DepHandler.Added(ctx, depRecord) | ||
|
|
||
| if err != nil { | ||
| return fmt.Errorf(LogFormatAdv, "Process", obj.Type, obj.Name, obj.Namespace, clusterName, "Error processing client discovery") | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,179 @@ | ||
| package clusters | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/istio-ecosystem/admiral/admiral/pkg/apis/admiral/model" | ||
| "github.com/istio-ecosystem/admiral/admiral/pkg/apis/admiral/v1alpha1" | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| "time" | ||
|
|
||
| argo "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1" | ||
| "github.com/istio-ecosystem/admiral/admiral/pkg/client/loader" | ||
| "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/test" | ||
| coreV1 "k8s.io/api/core/v1" | ||
| metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/client-go/rest" | ||
| ) | ||
|
|
||
| func TestClientDiscoveryHandler_Added(t *testing.T) { | ||
| //Struct of test case info. Name is required. | ||
| const ( | ||
| namespace = "namespace" | ||
| namespace_no_deps = "namespace-no-deps" | ||
| namespaceForRollout = "rollout-namespace" | ||
| cluster1 = "cluster1" | ||
| cluster2 = "cluster2" | ||
| cluster3 = "cluster3" | ||
| identity1 = "identity1" | ||
| identity2 = "identity2" | ||
| identity3 = "identity3" | ||
| identity4 = "identity4" | ||
| ) | ||
| var ( | ||
| stop = make(chan struct{}) | ||
| config = rest.Config{ | ||
| Host: "localhost", | ||
| } | ||
| matchLabel = map[string]string{ | ||
| "app": "test", | ||
| } | ||
| labelSelector = metaV1.LabelSelector{ | ||
| MatchLabels: matchLabel, | ||
| } | ||
| dep = v1alpha1.Dependency{ | ||
| Spec: model.Dependency{ | ||
| Source: identity1, | ||
| Destinations: []string{identity2}, | ||
| }, | ||
| } | ||
| rollout = argo.Rollout{ | ||
| Spec: argo.RolloutSpec{ | ||
| Selector: &labelSelector, | ||
| Strategy: argo.RolloutStrategy{ | ||
| BlueGreen: &argo.BlueGreenStrategy{}, | ||
| }, | ||
| Template: coreV1.PodTemplateSpec{ | ||
| ObjectMeta: metaV1.ObjectMeta{Annotations: map[string]string{ | ||
| "sidecar.istio.io/inject": "true", | ||
| "identity": identity1, | ||
| }, Labels: map[string]string{ | ||
| "env": "stage", | ||
| }}, | ||
| }, | ||
| }, | ||
| ObjectMeta: metaV1.ObjectMeta{ | ||
| Namespace: namespaceForRollout, | ||
| }, | ||
| } | ||
| objRolloutNamespace = &common.K8sObject{ | ||
| Namespace: namespaceForRollout, | ||
| Name: "obj-rollout-namespace", | ||
| Annotations: map[string]string{"identity": identity1}, | ||
| } | ||
| objValid = &common.K8sObject{ | ||
| Namespace: namespace, | ||
| Name: "obj", | ||
| Annotations: map[string]string{"identity": identity1}, | ||
| } | ||
| objValidWithNoDeps = &common.K8sObject{ | ||
| Namespace: namespace_no_deps, | ||
| Name: "obj-no-deps", | ||
| Annotations: map[string]string{"identity": identity4}, | ||
| } | ||
| ) | ||
|
|
||
| setupForHandlerTests(true) | ||
|
|
||
| depController, err := admiral.NewDependencyController(stop, &test.MockDependencyHandler{}, loader.FakeKubeconfigPath, "", time.Second*time.Duration(300), loader.GetFakeClientLoader()) | ||
| if err != nil { | ||
| t.Fatalf("failed to initialize dependency controller, err: %v", err) | ||
| } | ||
| r, err := admiral.NewRolloutsController(stop, &test.MockRolloutHandler{}, &config, time.Second*time.Duration(300), loader.GetFakeClientLoader()) | ||
| if err != nil { | ||
| t.Fatalf("failed to initialize rollout controller, err: %v", err) | ||
| } | ||
| d, err := admiral.NewDeploymentController(stop, &test.MockDeploymentHandler{}, &config, time.Second*time.Duration(300), loader.GetFakeClientLoader()) | ||
| if err != nil { | ||
| t.Fatalf("failed to initialize rollout controller, err: %v", err) | ||
| } | ||
|
|
||
| rcCluster1 := &RemoteController{ | ||
| RolloutController: r, | ||
| DeploymentController: d, | ||
| } | ||
|
|
||
| rcCluster2 := &RemoteController{ | ||
| RolloutController: r, | ||
| DeploymentController: d, | ||
| } | ||
|
|
||
| ctx := context.Background() | ||
|
|
||
| remoteRegistry := NewRemoteRegistry(context.TODO(), common.AdmiralParams{}) | ||
| remoteRegistry.DependencyController = depController | ||
| remoteRegistry.remoteControllers[cluster1] = rcCluster1 | ||
| remoteRegistry.remoteControllers[cluster2] = rcCluster2 | ||
|
|
||
| remoteRegistry.AdmiralCache = &AdmiralCache{ | ||
| IdentityClusterCache: common.NewMapOfMaps(), | ||
| IdentityClusterNamespaceCache: common.NewMapOfMapOfMaps(), | ||
| PartitionIdentityCache: common.NewMap(), | ||
| } | ||
|
|
||
| err = r.Added(ctx, &rollout) | ||
| assert.NoError(t, err) | ||
| rolloutFromCache := r.Cache.Get(identity1, "stage") | ||
| assert.NotNil(t, rolloutFromCache) | ||
| err = depController.Added(ctx, &dep) | ||
| assert.NoError(t, err) | ||
| depFromCache := depController.Cache.Get(identity1) | ||
| assert.NotNil(t, depFromCache) | ||
|
|
||
| testCases := []struct { | ||
| name string | ||
| obj *common.K8sObject | ||
| rr *RemoteRegistry | ||
| clusterName string | ||
| expectedError error | ||
| }{ | ||
| { | ||
| name: "No global identifier on obj results in no errors", | ||
| obj: &common.K8sObject{Annotations: map[string]string{}, Labels: map[string]string{}}, | ||
| rr: remoteRegistry, | ||
| clusterName: cluster1, | ||
| expectedError: nil, | ||
| }, { | ||
| name: "Rollout with same identifier present in the namespace should " + | ||
| "return without processing and no errors", | ||
| obj: objRolloutNamespace, | ||
| rr: remoteRegistry, | ||
| clusterName: cluster1, | ||
| expectedError: nil, | ||
| }, { | ||
| name: "Valid client with no dependency record returns without processing", | ||
| obj: objValidWithNoDeps, | ||
| rr: remoteRegistry, | ||
| clusterName: cluster1, | ||
| expectedError: nil, | ||
| }, { | ||
| name: "Valid client with valid dependency record results in writing all dependencies endpoints to this cluster", | ||
| obj: objValid, | ||
| rr: remoteRegistry, | ||
| clusterName: cluster1, | ||
| expectedError: nil, | ||
| }, | ||
| } | ||
|
|
||
| //Run the test for every provided case | ||
| for _, c := range testCases { | ||
| t.Run(c.name, func(t *testing.T) { | ||
| result := HandleEventForClientDiscovery(ctx, admiral.Add, c.obj, remoteRegistry, c.clusterName) | ||
| if c.expectedError == nil && result != nil { | ||
| t.Fatalf("Expected error %v got %v", c.expectedError, result) | ||
| } | ||
| }) | ||
| } | ||
| } |
Oops, something went wrong.