Fix subscription burst control

Author: sgayangi

apim-apk-agent/internal/k8sClient/k8s_client.go

@@ -476,6 +476,10 @@ func DeploySubscriptionRateLimitPolicyCR(policy eventhubTypes.SubscriptionPolicy
 							RequestsPerUnit: uint32(policy.DefaultLimit.RequestCount.RequestCount),
 							Unit:            policy.DefaultLimit.RequestCount.TimeUnit,
 						},
+						BurstControl: &dpv1alpha3.BurstControl{
+							RequestsPerUnit: uint32(policy.RateLimitCount),
+							Unit:            policy.RateLimitTimeUnit,
+						},
 					},
 				},
 				TargetRef: gwapiv1b1.NamespacedPolicyTargetReference{Group: constants.GatewayGroup, Kind: "Subscription", Name: "default"},
@@ -489,8 +493,16 @@ func DeploySubscriptionRateLimitPolicyCR(policy eventhubTypes.SubscriptionPolicy
 	} else {
 		crRateLimitPolicy.Spec.Override.Subscription.StopOnQuotaReach = policy.StopOnQuotaReach
 		crRateLimitPolicy.Spec.Override.Subscription.Organization = policy.TenantDomain
+		if crRateLimitPolicy.Spec.Override.Subscription.RequestCount == nil {
+			crRateLimitPolicy.Spec.Override.Subscription.RequestCount = &dpv1alpha3.RequestCount{}
+		}
 		crRateLimitPolicy.Spec.Override.Subscription.RequestCount.RequestsPerUnit = uint32(policy.DefaultLimit.RequestCount.RequestCount)
 		crRateLimitPolicy.Spec.Override.Subscription.RequestCount.Unit = policy.DefaultLimit.RequestCount.TimeUnit
+		if crRateLimitPolicy.Spec.Override.Subscription.BurstControl == nil {
+			crRateLimitPolicy.Spec.Override.Subscription.BurstControl = &dpv1alpha3.BurstControl{}
+		}
+		crRateLimitPolicy.Spec.Override.Subscription.BurstControl.RequestsPerUnit = uint32(policy.RateLimitCount)
+		crRateLimitPolicy.Spec.Override.Subscription.BurstControl.Unit = policy.RateLimitTimeUnit
 		if err := k8sClient.Update(context.Background(), &crRateLimitPolicy); err != nil {
 			loggers.LoggerK8sClient.Error("Unable to update RateLimitPolicies CR: " + err.Error())
 		} else {

apim-apk-agent/internal/synchronizer/ratelimit_policy_fetcher.go

@@ -141,12 +141,25 @@ func FetchRateLimitPoliciesOnEvent(ratelimitName string, organization string, c
 		logger.LoggerSynchronizer.Debugf("Policies received: %v", rateLimitPolicyList.List)
 		var rateLimitPolicies []eventhubTypes.RateLimitPolicy = rateLimitPolicyList.List
 		for _, policy := range rateLimitPolicies {
-			if policy.DefaultLimit.RequestCount.TimeUnit == "min" {
+			switch policy.DefaultLimit.RequestCount.TimeUnit {
+			case "min":
 				policy.DefaultLimit.RequestCount.TimeUnit = "Minute"
-			} else if policy.DefaultLimit.RequestCount.TimeUnit == "hour" {
+			case "hour":
 				policy.DefaultLimit.RequestCount.TimeUnit = "Hour"
-			} else if policy.DefaultLimit.RequestCount.TimeUnit == "day" {
+			case "day":
 				policy.DefaultLimit.RequestCount.TimeUnit = "Day"
+			default:
+				logger.LoggerSynchronizer.Errorf("Unsupported timeunit %s", policy.DefaultLimit.RequestCount.TimeUnit)
+				continue
+			}
+			switch policy.RateLimitTimeUnit {
+			case "min":
+				policy.RateLimitTimeUnit = "Minute"
+			case "sec":
+				policy.RateLimitTimeUnit = "Second"
+			default:
+				logger.LoggerSynchronizer.Errorf("Unsupported timeunit %s", policy.RateLimitTimeUnit)
+				continue
 			}
 			managementserver.AddRateLimitPolicy(policy)
 			logger.LoggerSynchronizer.Infof("RateLimit Policy added to internal map: %v", policy)
@@ -317,12 +330,25 @@ func FetchSubscriptionRateLimitPoliciesOnEvent(ratelimitName string, organizatio
 					logger.LoggerSynchronizer.Errorf("AIQuota type response recieved but no data found. %+v", policy.DefaultLimit)
 				}
 			} else {
-				if policy.DefaultLimit.RequestCount.TimeUnit == "min" {
+				switch policy.DefaultLimit.RequestCount.TimeUnit {
+				case "min":
 					policy.DefaultLimit.RequestCount.TimeUnit = "Minute"
-				} else if policy.DefaultLimit.RequestCount.TimeUnit == "hours" {
+				case "hours":
 					policy.DefaultLimit.RequestCount.TimeUnit = "Hour"
-				} else if policy.DefaultLimit.RequestCount.TimeUnit == "days" {
+				case "days":
 					policy.DefaultLimit.RequestCount.TimeUnit = "Day"
+				default:
+					logger.LoggerSynchronizer.Errorf("Unsupported timeunit %s", policy.DefaultLimit.RequestCount.TimeUnit)
+					continue
+				}
+				switch policy.RateLimitTimeUnit {
+				case "min":
+					policy.RateLimitTimeUnit = "Minute"
+				case "sec":
+					policy.RateLimitTimeUnit = "Second"
+				default:
+					logger.LoggerSynchronizer.Errorf("Unsupported timeunit %s", policy.RateLimitTimeUnit)
+					continue
 				}
 				managementserver.AddSubscriptionPolicy(policy)
 				logger.LoggerSynchronizer.Infof("RateLimit Policy added to internal map: %v", policy)

apim-apk-agent/pkg/eventhub/types/types.go

@@ -178,12 +178,14 @@ type RateLimitPolicyList struct {
 
 // RateLimitPolicy for struct RateLimitPolicy Info events
 type RateLimitPolicy struct {
-	TenantDomain    string           `json:"tenantDomain"`
-	Name            string           `json:"name"`
-	QuotaType       string           `json:"quotaType"`
-	ConditionGroups []ConditionGroup `json:"conditionGroups"`
-	ApplicableLevel string           `json:"applicableLevel"`
-	DefaultLimit    DefaultLimit     `json:"defaultLimit"`
+	TenantDomain      string           `json:"tenantDomain"`
+	Name              string           `json:"name"`
+	QuotaType         string           `json:"quotaType"`
+	ConditionGroups   []ConditionGroup `json:"conditionGroups"`
+	ApplicableLevel   string           `json:"applicableLevel"`
+	DefaultLimit      DefaultLimit     `json:"defaultLimit"`
+	RateLimitCount    int              `json:"rateLimitCount"`
+	RateLimitTimeUnit string           `json:"rateLimitTimeUnit"`
 }
 
 // ConditionGroup represents the condition group within the response.

helm-charts/values.yaml

@@ -56,9 +56,9 @@ serviceAccount:
   enableClusterRoleCreation: true
   serviceAccountName: wso2agent-platform
   roleName: wso2agent-role
-configs:
-  tls:
-    secretName: "keys-certs"
-    certKeyFilename: "tls.key"
-    certFilename: "tls.crt"
-    certCAFilename: "ca.crt"
+# configs:
+#   tls:
+#     secretName: "apk-agent-tls-secret"
+#     certKeyFilename: "tls.key"
+#     certFilename: "tls.crt"
+#     certCAFilename: "ca.crt"