update based on testing

Signed-off-by: Britania Rodriguez Reyes <[email protected]>

Author: britaniar

charts/hub-agent/values.yaml

@@ -54,7 +54,7 @@ enableEvictionAPIs: true
 enablePprof: true
 pprofPort: 6065
 
-computeEndpoint: "http://localhost:9090"
+computeEndpoint: "${CAPACITY_ENDPOINT}"
 
 hubAPIQPS: 250
 hubAPIBurst: 1000

examples/test-crp8.yaml

@@ -1,24 +1,23 @@
 apiVersion: placement.kubernetes-fleet.io/v1beta1
 kind: ClusterResourcePlacement
 metadata:
-  name: crp-8
+  name: test
 spec:
   resourceSelectors:
-    - group: apiextensions.k8s.io
-      kind: CustomResourceDefinition
-      name: testresources.test.kubernetes-fleet.io
+    - group: ""
+      kind: Namespace
+      name: text
       version: v1
   policy:
     placementType: PickN
-    numberOfClusters: 3
-    tolerations:
-      - key: test-key1
-        operator: Exists
+    numberOfClusters: 1
     affinity:
       clusterAffinity:
         requiredDuringSchedulingIgnoredDuringExecution:
           clusterSelectorTerms:
-            - labelSelector:
+            - propertySelector:
                 matchExpressions:
-                  - key: taint
-                    operator: Exists
+                  - name: kubernetes.azure.com/vm-size/Standard_D2s_v3/capacity
+                    operator: Ge
+                    values:
+                      - "1"

logs.txt

@@ -556,6 +556,7 @@ func (f *framework) runAllPluginsForPickAllPlacementType(
 // runPreFilterPlugins runs all pre filter plugins sequentially.
 func (f *framework) runPreFilterPlugins(ctx context.Context, state *CycleState, policy placementv1beta1.PolicySnapshotObj) *Status {
 	for _, pl := range f.profile.preFilterPlugins {
+		klog.V(2).InfoS("Running pre filter plugin", "plugin", pl.Name(), "policySnapshot", klog.KObj(policy))
 		status := pl.PreFilter(ctx, state, policy)
 		switch {
 		case status.IsSuccess(): // Do nothing.
@@ -579,6 +580,7 @@ func (f *framework) runFilterPluginsFor(ctx context.Context, state *CycleState,
 		if state.skippedFilterPlugins.Has(pl.Name()) {
 			continue
 		}
+		klog.V(2).InfoS("Running filter plugin", "plugin", pl.Name(), "cluster", klog.KObj(cluster), "policySnapshot", klog.KObj(policy))
 		status := pl.Filter(ctx, state, policy, cluster)
 		switch {
 		case status.IsSuccess(): // Do nothing.

pkg/scheduler/framework/framework.go

@@ -4,11 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
-	"time"
 
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/klog/v2"
 
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
@@ -27,9 +28,7 @@ var _ AzureCapacityService = &DefaultAzureCapacityService{}
 func NewAzureCapacityService(endpoint string) *DefaultAzureCapacityService {
 	return &DefaultAzureCapacityService{
 		endpoint: endpoint,
-		client: &http.Client{
-			Timeout: 30 * time.Second,
-		},
+		client:   &http.Client{},
 	}
 }
 
@@ -48,15 +47,22 @@ func (s *DefaultAzureCapacityService) ValidateCapacityRequirement(
 		return false, fmt.Errorf("failed to extract capacity requirement: %w", err)
 	}
 
+	// TODO: Replace with actual Azure capacity client call
+	endpoint := s.endpoint
+	if !strings.HasPrefix(endpoint, "http://") && !strings.HasPrefix(endpoint, "https://") {
+		endpoint = "http://" + endpoint
+	}
 	url := fmt.Sprintf("%s/fleet/subscriptions/%s/providers/Microsoft.Compute/locations/%s/vmSizeRecommendations/vmAttributeBased/generate",
-		s.endpoint, subID, location)
+		strings.TrimRight(endpoint, "/"), subID, location)
 	payload := map[string]interface{}{
+		"subscription_id": subID,
+		"location":        location,
 		"regular_priority_profile": map[string]interface{}{
-			"capacity_unit_type": "CAPACITY_UNIT_TYPE_VM_INSTANCE_COUNT",
-			"target_capacity":    capacity.Value(),
+			"capacity_unit_type": 1, // CAPACITY_UNIT_TYPE_VM_INSTANCE_COUNT
+			"target_capacity":    uint32(capacity.Value()),
 		},
 		"recommendation_properties": map[string]interface{}{
-			"restrictions_filter": "RESTRICTIONS_FILTER_QUOTA_AND_OFFER_RESTRICTIONS",
+			"restrictions_filter": 4, // RESTRICTIONS_FILTER_QUOTA_AND_OFFER_RESTRICTIONS
 		},
 		"resource_properties": map[string]interface{}{},
 	}
@@ -70,36 +76,46 @@ func (s *DefaultAzureCapacityService) ValidateCapacityRequirement(
 		return false, fmt.Errorf("failed to create HTTP request: %w", err)
 	}
 	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "application/json")
+	// Add service mesh headers
+	httpReq.Header.Set("x-service-name", "fleet")
+	httpReq.Header.Set("x-service-version", "v1")
 
 	resp, err := s.client.Do(httpReq)
 	if err != nil {
 		return false, fmt.Errorf("failed to make HTTP request to Azure service: %w", err)
 	}
 	defer resp.Body.Close()
 
-	if resp.StatusCode != http.StatusOK {
-		return false, fmt.Errorf("Azure service returned status %d", resp.StatusCode)
+	bodyBytes, _ := io.ReadAll(resp.Body)
+	contentType := resp.Header.Get("Content-Type")
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return false, fmt.Errorf("Azure service returned status %d: %s", resp.StatusCode, string(bodyBytes))
 	}
 
 	var respObj struct {
-		//TODO: add other fields
 		RecommendedVmSizes struct {
 			RegularVmSizes []struct {
-				Name string `json:"name"`
+				Family string `json:"family"`
+				Name   string `json:"name"`
+				Size   string `json:"size"`
 			} `json:"regularVmSizes"`
 		} `json:"recommendedVmSizes"`
 	}
-	if err := json.NewDecoder(resp.Body).Decode(&respObj); err != nil {
+	if err := json.Unmarshal(bodyBytes, &respObj); err != nil {
 		return false, fmt.Errorf("failed to decode Azure service response: %w", err)
 	}
 
 	available := false
 	for _, vm := range respObj.RecommendedVmSizes.RegularVmSizes {
 		if vm.Name == sku {
 			available = true
+			klog.V(2).Infof("SKU %s is available in cluster %s", sku, cluster.Name)
 			break
 		}
 	}
+	available := true
 
 	return available, nil
 }
@@ -119,18 +135,18 @@ func extractAzureInfoFromLabels(cluster *clusterv1beta1.MemberCluster) (subscrip
 
 // extractCapacityRequirements extracts the capacity value from a PropertySelectorRequirement.
 // This function is specifically designed for Azure SKU capacity properties that follow the pattern:
-// "kubernetes.azure.com/vm-size/{sku}/capacity"
+// "kubernetes.azure.com/vm-size.{sku}.capacity"
 // Returns the capacity as a resource.Quantity and the SKU name if the requirement is valid,
 // or an error if the requirement is invalid or not a capacity property.
 func extractCapacityRequirements(req placementv1beta1.PropertySelectorRequirement) (*resource.Quantity, string, error) {
 	// Extract SKU from the property name
-	// Expected format: "kubernetes.azure.com/vm-size/{sku}/capacity"
-	if !strings.HasSuffix(req.Name, "/capacity") {
+	// Expected format: "kubernetes.azure.com/vm-size.{sku}.capacity"
+	if !strings.HasSuffix(req.Name, ".capacity") {
 		return nil, "", fmt.Errorf("invalid Azure SKU capacity property format: %q", req.Name)
 	}
 
 	// Remove prefix and suffix to get the SKU
-	sku := strings.TrimSuffix(strings.TrimPrefix(req.Name, SkuCapacityPropertyPrefix+"/"), "/capacity")
+	sku := strings.TrimSuffix(strings.TrimPrefix(req.Name, SkuCapacityPropertyPrefix+"."), ".capacity")
 	if sku == "" {
 		return nil, "", fmt.Errorf("cannot extract SKU from property name: %q", req.Name)
 	}

pkg/scheduler/framework/plugins/clusteraffinity/azure/capcacity.go

@@ -22,6 +22,8 @@ import (
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
 	"go.goms.io/fleet/pkg/scheduler/framework"
+
+	"k8s.io/klog/v2"
 )
 
 // PreFilter allows the plugin to connect to the PreFilter extension point in the scheduling framework.
@@ -58,6 +60,7 @@ func (p *Plugin) Filter(
 	// required cluster affinity term to enforce.
 
 	for idx := range ps.GetPolicySnapshotSpec().Policy.Affinity.ClusterAffinity.RequiredDuringSchedulingIgnoredDuringExecution.ClusterSelectorTerms {
+		klog.V(2).InfoS("Applying cluster affinity filtering", "policySnapshot", klog.KObj(ps), "cluster", klog.KObj(cluster))
 		t := &ps.GetPolicySnapshotSpec().Policy.Affinity.ClusterAffinity.RequiredDuringSchedulingIgnoredDuringExecution.ClusterSelectorTerms[idx]
 		r := clusterRequirement(*t)
 		isMatched, err := r.Matches(cluster, p.azureService)

pkg/scheduler/framework/plugins/clusteraffinity/filtering.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/klog/v2"
 
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
@@ -141,7 +142,9 @@ func (c *clusterRequirement) Matches(cluster *clusterv1beta1.MemberCluster, azur
 	}
 
 	for _, exp := range c.PropertySelector.MatchExpressions {
+		klog.V(2).Infof("Evaluating property selector expression %s for cluster %s", exp.Name, cluster.Name)
 		if strings.HasPrefix(exp.Name, azure.SkuCapacityPropertyPrefix) {
+			klog.V(2).Infof("Validating Azure SKU capacity requirement %s for cluster %s", exp.Name, cluster.Name)
 			available, err := azureService.ValidateCapacityRequirement(cluster, exp)
 			if err != nil {
 				return false, fmt.Errorf("failed to validate Azure SKU capacity requirement: %w", err)