Add context across layers [POC]

ako-clean/cleanup.go

@@ -412,7 +412,7 @@ func sanitzeAviCloud() error {
 		Model:   "cloud",
 	}
 	restLayer := rest.NewRestOperations(nil, true)
-	return restLayer.AviRestOperateWrapper(aviClient, []*utils.RestOp{&restOp}, "aviCleanup")
+	return restLayer.AviRestOperateWrapper(context.Background(), aviClient, []*utils.RestOp{&restOp}, "aviCleanup")
 }
 
 func checkAndUpdateIPAM() error {
@@ -449,7 +449,7 @@ func checkAndUpdateIPAM() error {
 		Model:  "ipamdnsproviderprofile",
 	}
 	restLayer := rest.NewRestOperations(nil, true)
-	return restLayer.AviRestOperateWrapper(aviClient, []*utils.RestOp{&restOp}, "aviCleanup")
+	return restLayer.AviRestOperateWrapper(context.Background(), aviClient, []*utils.RestOp{&restOp}, "aviCleanup")
 }
 
 func convertPemToDer(cert string) string {

ako-gateway-api/k8s/ako_init.go

@@ -229,7 +229,6 @@ func (c *GatewayController) addIndexers() {
 }
 
 func (c *GatewayController) FullSyncK8s(sync bool) error {
-
 	if c.DisableSync {
 		utils.AviLog.Infof("Sync disabled, skipping full sync")
 		return nil
@@ -256,6 +255,7 @@ func (c *GatewayController) FullSyncK8s(sync bool) error {
 	}
 	for _, filteredGatewayClass := range filteredGatewayClasses {
 		key := lib.GatewayClass + "/" + utils.ObjKey(filteredGatewayClass)
+
 		akogatewayapinodes.DequeueIngestion(key, true)
 	}
 
@@ -286,6 +286,7 @@ func (c *GatewayController) FullSyncK8s(sync bool) error {
 	})
 	for _, filteredGateway := range filteredGateways {
 		key := lib.Gateway + "/" + utils.ObjKey(filteredGateway)
+
 		akogatewayapinodes.DequeueIngestion(key, true)
 	}
 
@@ -316,6 +317,7 @@ func (c *GatewayController) FullSyncK8s(sync bool) error {
 	})
 	for _, filteredHTTPRoute := range filteredHTTPRoutes {
 		key := lib.HTTPRoute + "/" + utils.ObjKey(filteredHTTPRoute)
+
 		akogatewayapinodes.DequeueIngestion(key, true)
 	}
 
@@ -345,6 +347,7 @@ func (c *GatewayController) FullSyncK8s(sync bool) error {
 		for _, podObj := range podObjs {
 			podLabel := utils.ObjKey(podObj)
 			key := utils.Pod + "/" + podLabel
+
 			if _, ok := podObj.GetAnnotations()[lib.NPLPodAnnotation]; !ok {
 				utils.AviLog.Warnf("key : %s, msg: 'nodeportlocal.antrea.io' annotation not found, ignoring the pod", key)
 				continue
@@ -382,14 +385,14 @@ func (c *GatewayController) publishAllParentVSKeysToRestLayer() {
 		modelName := vsCacheKey.Namespace + "/" + vsCacheKey.Name
 		delete(allModels, modelName)
 		utils.AviLog.Infof("Model published in full sync %s", modelName)
-		nodes.PublishKeyToRestLayer(modelName, "fullsync", sharedQueue)
+		nodes.PublishKeyToRestLayer(context.Background(), "fullsync", modelName, sharedQueue)
 
 	}
 	// Now also publish the newly generated models (if any)
 	// Publish all the models to REST layer.
 	utils.AviLog.Debugf("Newly generated models that do not exist in cache %s", utils.Stringify(allModels))
 	for modelName := range allModels {
-		nodes.PublishKeyToRestLayer(modelName, "fullsync", sharedQueue)
+		nodes.PublishKeyToRestLayer(context.Background(), "fullsync", modelName, sharedQueue)
 	}
 }
 
@@ -425,14 +428,14 @@ func (c *GatewayController) FullSync() {
 }
 
 func SyncFromNodesLayer(key interface{}, wg *sync.WaitGroup) error {
-	keyStr, ok := key.(string)
+	keyStr, ok := key.(akogatewayapilib.KeyContext)
 	if !ok {
 		utils.AviLog.Warnf("Unexpected object type: expected string, got %T", key)
 		return nil
 	}
 	cache := avicache.SharedAviObjCache()
 	restlayer := rest.NewRestOperations(cache)
-	restlayer.DequeueNodes(keyStr)
+	restlayer.DequeueNodes(keyStr.Ctx, keyStr.KeyStr)
 	return nil
 }
 
@@ -454,6 +457,7 @@ func SyncFromIngestionLayer(key interface{}, wg *sync.WaitGroup) error {
 	akogatewayapinodes.DequeueIngestion(keyStr, false)
 	return nil
 }
+
 func SyncFromFastRetryLayer(key interface{}, wg *sync.WaitGroup) error {
 	keyStr, ok := key.(string)
 	if !ok {

ako-gateway-api/k8s/gateway_controller.go

@@ -15,6 +15,7 @@
 package k8s
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"sort"
@@ -50,6 +51,19 @@ type GatewayController struct {
 	DisableSync bool
 }
 
+type keyData struct {
+	key string
+	ctx context.Context
+}
+
+func InitKeyData(key string) keyData {
+	keyCtx := keyData{}
+	keyCtx.key = key
+	keyCtx.ctx = context.Background()
+	// infralog.GetInfraLogger(infralog.WithInitService())
+	return keyCtx
+}
+
 func SharedGatewayController() *GatewayController {
 	ctrlonce.Do(func() {
 		controllerInstance = &GatewayController{
@@ -126,6 +140,7 @@ func (c *GatewayController) SetupEventHandlers(k8sinfo k8s.K8sinformers) {
 					utils.AviLog.Debugf("Endpointslice Add event: Endpointslice does not have backing svc")
 					return
 				}
+
 				key := utils.Endpointslices + "/" + namespace + "/" + svcName
 				bkt := utils.Bkt(namespace, numWorkers)
 				c.workqueue[bkt].AddRateLimited(key)
@@ -155,6 +170,7 @@ func (c *GatewayController) SetupEventHandlers(k8sinfo k8s.K8sinformers) {
 					utils.AviLog.Debugf("Endpointslice Delete event: Endpointslice does not have backing svc")
 					return
 				}
+
 				key := utils.Endpointslices + "/" + namespace + "/" + svcName
 				bkt := utils.Bkt(namespace, numWorkers)
 				c.workqueue[bkt].AddRateLimited(key)
@@ -177,6 +193,7 @@ func (c *GatewayController) SetupEventHandlers(k8sinfo k8s.K8sinformers) {
 						}
 						svcName = svcNameOld
 					}
+
 					key := utils.Endpointslices + "/" + namespace + "/" + svcName
 					bkt := utils.Bkt(namespace, numWorkers)
 					c.workqueue[bkt].AddRateLimited(key)
@@ -513,6 +530,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 				return
 			}
 			gw := obj.(*gatewayv1.Gateway)
+
 			key := lib.Gateway + "/" + utils.ObjKey(gw)
 			ok, resVer := objects.SharedResourceVerInstanceLister().Get(key)
 			if ok && resVer.(string) == gw.ResourceVersion {
@@ -555,6 +573,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 					return
 				}
 			}
+
 			key := lib.Gateway + "/" + utils.ObjKey(gw)
 			objects.SharedResourceVerInstanceLister().Delete(key)
 			namespace, _, _ := cache.SplitMetaNamespaceKey(utils.ObjKey(gw))
@@ -570,7 +589,9 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 			oldGw := old.(*gatewayv1.Gateway)
 			gw := obj.(*gatewayv1.Gateway)
 			if IsGatewayUpdated(oldGw, gw) {
+
 				key := lib.Gateway + "/" + utils.ObjKey(gw)
+
 				valid, allowedRoutesAll := IsValidGateway(key, gw)
 				if !valid {
 					return
@@ -599,6 +620,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 				return
 			}
 			gwClass := obj.(*gatewayv1.GatewayClass)
+
 			key := lib.GatewayClass + "/" + utils.ObjKey(gwClass)
 			ok, resVer := objects.SharedResourceVerInstanceLister().Get(key)
 			if ok && resVer.(string) == gwClass.ResourceVersion {
@@ -635,6 +657,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 			if !akogatewayapilib.CheckGatewayClassController(controllerName) {
 				return
 			}
+
 			key := lib.GatewayClass + "/" + utils.ObjKey(gwClass)
 			objects.SharedResourceVerInstanceLister().Delete(key)
 			namespace, _, _ := cache.SplitMetaNamespaceKey(utils.ObjKey(gwClass))
@@ -649,6 +672,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 			oldGwClass := old.(*gatewayv1.GatewayClass)
 			gwClass := obj.(*gatewayv1.GatewayClass)
 			if !reflect.DeepEqual(oldGwClass.Spec, gwClass.Spec) || gwClass.GetDeletionTimestamp() != nil {
+
 				key := lib.GatewayClass + "/" + utils.ObjKey(gwClass)
 				if !IsGatewayClassValid(key, gwClass) {
 					return
@@ -668,6 +692,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 				return
 			}
 			httpRoute := obj.(*gatewayv1.HTTPRoute)
+
 			key := lib.HTTPRoute + "/" + utils.ObjKey(httpRoute)
 			ok, resVer := objects.SharedResourceVerInstanceLister().Get(key)
 			if ok && resVer.(string) == httpRoute.ResourceVersion {
@@ -700,6 +725,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 					return
 				}
 			}
+
 			key := lib.HTTPRoute + "/" + utils.ObjKey(httpRoute)
 			objects.SharedResourceVerInstanceLister().Delete(key)
 			namespace, _, _ := cache.SplitMetaNamespaceKey(utils.ObjKey(httpRoute))
@@ -715,6 +741,7 @@ func (c *GatewayController) SetupGatewayApiEventHandlers(numWorkers uint32) {
 			oldHTTPRoute := old.(*gatewayv1.HTTPRoute)
 			newHTTPRoute := obj.(*gatewayv1.HTTPRoute)
 			if IsHTTPRouteUpdated(oldHTTPRoute, newHTTPRoute) {
+
 				key := lib.HTTPRoute + "/" + utils.ObjKey(newHTTPRoute)
 				if !IsHTTPRouteConfigValid(key, newHTTPRoute) {
 					return

ako-gateway-api/lib/lib.go

@@ -16,16 +16,17 @@ package lib
 
 import (
 	"fmt"
+	"sort"
 
 	"os"
 	"strings"
 
+	cache2 "github.com/vmware/load-balancer-and-ingress-services-for-kubernetes/internal/cache"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/kubernetes"
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 
 	"github.com/vmware/load-balancer-and-ingress-services-for-kubernetes/internal/lib"
-	"github.com/vmware/load-balancer-and-ingress-services-for-kubernetes/internal/nodes"
 	"github.com/vmware/load-balancer-and-ingress-services-for-kubernetes/pkg/utils"
 )
 
@@ -160,7 +161,7 @@ func IsListenerInvalid(gwStatus *gatewayv1.GatewayStatus, listenerIndex int) boo
 func VerifyHostnameSubdomainMatch(hostname string) bool {
 	// Check if a hostname is valid or not by verifying if it has a prefix that
 	// matches any of the sub-domains.
-	subDomains := nodes.GetDefaultSubDomain()
+	subDomains := GetDefaultSubDomain()
 	if len(subDomains) == 0 {
 		// No IPAM DNS configured, we simply pass the hostname
 		return true
@@ -187,3 +188,24 @@ func ProtocolToRoute(proto string) string {
 	return innerMap[proto]
 
 }
+
+func GetDefaultSubDomain() []string {
+	cache := cache2.SharedAviObjCache()
+	cloud, ok := cache.CloudKeyCache.AviCacheGet(utils.CloudName)
+	if !ok || cloud == nil {
+		utils.AviLog.Warnf("Cloud object %s not found in cache", utils.CloudName)
+		return nil
+	}
+	cloudProperty, ok := cloud.(*cache2.AviCloudPropertyCache)
+	if !ok {
+		utils.AviLog.Warnf("Cloud property object not found")
+		return nil
+	}
+
+	if len(cloudProperty.NSIpamDNS) > 0 {
+		sort.Strings(cloudProperty.NSIpamDNS)
+	} else {
+		return nil
+	}
+	return cloudProperty.NSIpamDNS
+}

ako-gateway-api/lib/types.go

@@ -15,12 +15,41 @@
 package lib
 
 import (
+	"context"
+	"strings"
+
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 
+	"github.com/google/uuid"
+
+	"github.com/vmware/alb-sdk/go/logger"
+
 	"github.com/vmware/load-balancer-and-ingress-services-for-kubernetes/internal/lib"
 )
 
 var SupportedKinds = map[gatewayv1.ProtocolType][]gatewayv1.RouteGroupKind{
 	gatewayv1.HTTPProtocolType:  {{Kind: lib.HTTPRoute}},
 	gatewayv1.HTTPSProtocolType: {{Kind: lib.HTTPRoute}},
 }
+
+type traceID string
+type KeyContext struct {
+	KeyStr string
+	Ctx    context.Context
+}
+
+func getNewTraceId() string {
+	traceID := uuid.New().String()
+	traceID = strings.Replace(traceID, "-", "", -1) // default value
+	return traceID
+}
+
+func NewKeyContextWithTraceID(key string, ctx context.Context) KeyContext {
+	return KeyContext{KeyStr: key,
+		Ctx: logger.SetTraceID(ctx, getNewTraceId())}
+}
+
+func NewKeyContext(key string, ctx context.Context) KeyContext {
+	return KeyContext{KeyStr: key,
+		Ctx: ctx}
+}

ako-gateway-api/nodes/dequeue_ingestion.go

@@ -29,7 +29,9 @@ import (
 )
 
 func DequeueIngestion(key string, fullsync bool) {
-	utils.AviLog.Infof("key: %s, msg: starting graph Sync", key)
+	keyCtx := akogatewayapilib.NewKeyContextWithTraceID(key, context.Background())
+	ctx := keyCtx.Ctx
+	utils.AviLog.WithContext(ctx).Infof("key: %s, msg: starting graph Sync", key)
 	objType, namespace, name := lib.ExtractTypeNameNamespace(key)
 
 	schema, valid := ConfigDescriptor().GetByType(objType)
@@ -113,7 +115,7 @@ func DequeueIngestion(key string, fullsync bool) {
 				objects.SharedAviGraphLister().Save(modelName, nil)
 				if !fullsync {
 					sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-					nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+					nodes.PublishKeyToRestLayer(ctx, key, modelName, sharedQueue)
 					return
 				}
 			}
@@ -148,7 +150,7 @@ func DequeueIngestion(key string, fullsync bool) {
 		modelChanged := saveAviModel(modelName, model.AviObjectGraph, key)
 		if modelChanged && !fullsync {
 			sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-			nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+			nodes.PublishKeyToRestLayer(ctx, key, modelName, sharedQueue)
 		}
 	}
 }
@@ -207,7 +209,7 @@ func handleGateway(namespace, name string, fullsync bool, key string) {
 			objects.SharedAviGraphLister().Save(modelName, nil)
 			if !fullsync {
 				sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-				nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+				nodes.PublishKeyToRestLayer(context.Background(), key, modelName, sharedQueue)
 			}
 		}
 		return
@@ -221,7 +223,7 @@ func handleGateway(namespace, name string, fullsync bool, key string) {
 		objects.SharedAviGraphLister().Save(modelName, nil)
 		if !fullsync {
 			sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-			nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+			nodes.PublishKeyToRestLayer(context.Background(), key, modelName, sharedQueue)
 		}
 		return
 	}
@@ -237,7 +239,7 @@ func handleGateway(namespace, name string, fullsync bool, key string) {
 	modelChanged := saveAviModel(modelName, aviModelGraph.AviObjectGraph, key)
 	if modelChanged && !fullsync {
 		sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-		nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+		nodes.PublishKeyToRestLayer(context.Background(), key, modelName, sharedQueue)
 	}
 }
 
@@ -289,7 +291,7 @@ func (o *AviObjectGraph) ProcessRouteDeletion(key, parentNsName string, routeMod
 	ok := saveAviModel(modelName, o.AviObjectGraph, key)
 	if ok && len(o.AviObjectGraph.GetOrderedNodes()) != 0 && !fullsync {
 		sharedQueue := utils.SharedWorkQueue().GetQueueByName(utils.GraphLayer)
-		nodes.PublishKeyToRestLayer(modelName, key, sharedQueue)
+		nodes.PublishKeyToRestLayer(context.Background(), key, modelName, sharedQueue)
 	}
 
 }