Skip to content

feat: update scheduler for capacity-based property scheduling #1223

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
britaniar merged 1 commit into from
Nov 20, 2025
Merged

feat: update scheduler for capacity-based property scheduling #1223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ require (
github.com/gofrs/uuid v4.4.0+incompatible
github.com/google/go-cmp v0.7.0
github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0
github.com/kubefleet-dev/kubefleet v0.0.1
github.com/onsi/ginkgo/v2 v2.23.4
github.com/onsi/gomega v1.37.0
github.com/prometheus/client_golang v1.22.0
Expand Down
2 changes: 2 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kubefleet-dev/kubefleet v0.0.1 h1:Xo/DvRlKq7YLvzjGp8GFD8G6SLAyI41l4Sw4+df534g=
github.com/kubefleet-dev/kubefleet v0.0.1/go.mod h1:EXicokCzLknCzcpQ/8ehtXqBokr0/xUfzsSsU++rReY=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
Expand Down
71 changes: 7 additions & 64 deletions pkg/clients/azure/compute/vmsizerecommenderclient_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,22 +7,15 @@ package compute

import (
"context"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"

"github.com/google/go-cmp/cmp"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"

computev1 "go.goms.io/fleet/apis/protos/azure/compute/v1"
)

const (
testTenantID = "test-tenant-id"
"go.goms.io/fleet/test/utils/azure/compute"
)

func TestNewAttributeBasedVMSizeRecommenderClient(t *testing.T) {
Expand All @@ -43,27 +36,27 @@ func TestNewAttributeBasedVMSizeRecommenderClient(t *testing.T) {
},
{
name: "with empty server address",
tenantID: testTenantID,
tenantID: compute.TestTenantID,
serverAddress: "",
httpClient: http.DefaultClient,
wantClient: nil,
wantErr: true,
},
{
name: "with nil HTTP client",
tenantID: testTenantID,
tenantID: compute.TestTenantID,
serverAddress: "http://localhost:8080",
httpClient: nil,
wantClient: nil,
wantErr: true,
},
{
name: "with all fields properly set",
tenantID: testTenantID,
tenantID: compute.TestTenantID,
serverAddress: "https://example.com",
httpClient: http.DefaultClient,
wantClient: &AttributeBasedVMSizeRecommenderClient{
tenantID: testTenantID,
tenantID: compute.TestTenantID,
baseURL: "https://example.com",
httpClient: http.DefaultClient,
},
Expand Down Expand Up @@ -222,59 +215,9 @@ func TestClient_GenerateAttributeBasedRecommendations(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Set tenant ID environment variable to create client.
t.Setenv(tenantIDEnvVarName, testTenantID)
t.Setenv(tenantIDEnvVarName, compute.TestTenantID)
// Create mock server.
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Verify request method.
if r.Method != http.MethodPost {
t.Errorf("got %s, want POST request", r.Method)
}

// Verify headers.
if r.Header.Get("Content-Type") != "application/json" {
t.Errorf("got %s, want Content-Type: application/json", r.Header.Get("Content-Type"))
}
if r.Header.Get("Accept") != "application/json" {
t.Errorf("got %s, want Accept: application/json", r.Header.Get("Accept"))
}
if r.Header.Get("Grpc-Metadata-subscriptionTenantID") != testTenantID {
t.Errorf("got %s, want Grpc-Metadata-subscriptionTenantID: %s",
r.Header.Get("Grpc-Metadata-subscriptionTenantID"), testTenantID)
}
if r.Header.Get("Grpc-Metadata-clientRequestID") == "" {
t.Error("Grpc-Metadata-clientRequestID header is missing")
}

// Verify URL path if request is not nil.
if tt.request != nil && tt.request.SubscriptionId != "" && tt.request.Location != "" {
wantPath := fmt.Sprintf(recommendationsPathTemplate, tt.request.SubscriptionId, tt.request.Location)
if r.URL.Path != wantPath {
t.Errorf("got %s, want path %s", r.URL.Path, wantPath)
}

// Verify request body using protojson for proper proto3 oneof support.
body, err := io.ReadAll(r.Body)
if err != nil {
t.Fatalf("failed to read request body: %v", err)
}
var req computev1.GenerateAttributeBasedRecommendationsRequest
unmarshaler := protojson.UnmarshalOptions{
DiscardUnknown: true,
}
if err := unmarshaler.Unmarshal(body, &req); err != nil {
t.Fatalf("failed to unmarshal request body: %v", err)
}
if !proto.Equal(tt.request, &req) {
t.Errorf("request body mismatch: got %+v, want %+v", &req, tt.request)
}
}

// Write mock response.
w.WriteHeader(tt.mockStatusCode)
if _, err := w.Write([]byte(tt.mockResponse)); err != nil {
t.Fatalf("failed to write response: %v", err)
}
}))
server := compute.CreateMockAttributeBasedVMSizeRecommenderServer(t, tt.request, compute.TestTenantID, tt.mockResponse, tt.mockStatusCode)
defer server.Close()

// Create client.
Expand Down
117 changes: 32 additions & 85 deletions pkg/propertychecker/azure/checker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,22 +6,18 @@ package azure

import (
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"

"google.golang.org/protobuf/encoding/protojson"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
computev1 "go.goms.io/fleet/apis/protos/azure/compute/v1"
"go.goms.io/fleet/pkg/clients/azure/compute"
"go.goms.io/fleet/pkg/clients/httputil"
"go.goms.io/fleet/pkg/propertyprovider/azure"
"go.goms.io/fleet/pkg/utils/labels"
testcompute "go.goms.io/fleet/test/utils/azure/compute"
)

func TestValidateCapacity(t *testing.T) {
Expand Down Expand Up @@ -305,6 +301,7 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
name string
cluster *clusterv1beta1.MemberCluster
sku string
targetCapacity uint32
req placementv1beta1.PropertySelectorRequirement
mockStatusCode int
wantAvailable bool
Expand All @@ -315,15 +312,17 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
name: "valid capacity request",
cluster: cluster,
sku: validSKU,
targetCapacity: 3,
req: validPropertySelectorRequirement,
mockStatusCode: http.StatusOK,
wantAvailable: true,
wantError: false,
},
{
name: "unavailable SKU request",
cluster: cluster,
sku: "Standard_D2s_v4",
name: "unavailable SKU request",
cluster: cluster,
sku: "Standard_D2s_v4",
targetCapacity: 1,
req: placementv1beta1.PropertySelectorRequirement{
Name: fmt.Sprintf(azure.CapacityPerSKUPropertyTmpl, "Standard_D2s_v4"),
Operator: placementv1beta1.PropertySelectorGreaterThanOrEqualTo,
Expand All @@ -343,6 +342,7 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
},
},
sku: validSKU,
targetCapacity: 3,
req: validPropertySelectorRequirement,
wantError: true,
errorSubstring: "failed to extract Azure location label from cluster : label \"fleet.azure.com/location\" not found in cluster",
Expand All @@ -357,6 +357,7 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
},
},
sku: validSKU,
targetCapacity: 3,
req: validPropertySelectorRequirement,
wantError: true,
errorSubstring: "failed to extract Azure location label from cluster",
Expand All @@ -371,6 +372,7 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
},
},
sku: validSKU,
targetCapacity: 3,
req: validPropertySelectorRequirement,
wantError: true,
errorSubstring: "failed to extract Azure subscription ID label from cluster",
Expand All @@ -379,28 +381,31 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
name: "Azure API returns error",
cluster: cluster,
sku: validSKU,
targetCapacity: 3,
req: validPropertySelectorRequirement,
mockStatusCode: http.StatusInternalServerError,
wantError: true,
errorSubstring: "failed to generate VM size recommendations from Azure",
},
{
name: "invalid operator in requirement",
cluster: cluster,
sku: validSKU,
name: "invalid operator in requirement",
cluster: cluster,
sku: validSKU,
targetCapacity: 2,
req: placementv1beta1.PropertySelectorRequirement{
Name: fmt.Sprintf(azure.CapacityPerSKUPropertyTmpl, validSKU),
Operator: placementv1beta1.PropertySelectorEqualTo,
Values: []string{"3"},
Values: []string{"2"},
},
mockStatusCode: http.StatusOK,
wantError: true,
errorSubstring: "unsupported operator \"Eq\" for SKU capacity property, only GreaterThan (Gt) and GreaterThanOrEqualTo (Ge) are supported",
},
{
name: "unsupported operator in requirement",
cluster: cluster,
sku: validSKU,
name: "unsupported operator in requirement",
cluster: cluster,
sku: validSKU,
targetCapacity: 0,
req: placementv1beta1.PropertySelectorRequirement{
Name: fmt.Sprintf(azure.CapacityPerSKUPropertyTmpl, validSKU),
Operator: placementv1beta1.PropertySelectorGreaterThanOrEqualTo,
Expand All @@ -411,9 +416,10 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
errorSubstring: "capacity value cannot be zero for operator",
},
{
name: "cases-insensitive request - unavailable SKU",
cluster: cluster,
sku: "STANDARD_D2S_V3",
name: "cases-insensitive request - available SKU",
cluster: cluster,
sku: "STANDARD_D2S_V3",
targetCapacity: 1,
req: placementv1beta1.PropertySelectorRequirement{
Name: fmt.Sprintf(azure.CapacityPerSKUPropertyTmpl, "STANDARD_D2S_V3"),
Operator: placementv1beta1.PropertySelectorGreaterThanOrEqualTo,
Expand All @@ -426,11 +432,14 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Create mock server
server := createMockAttributeBasedVMSizeRecommenderServer(t, tt.mockStatusCode)
// Set tenant ID environment variable to create client.
testTenantId := "test-tenant-id"
t.Setenv("AZURE_TENANT_ID", testTenantId)
// Create mock server.
mockRequest := testcompute.GenerateAttributeBasedVMSizeRecommenderRequest(tt.cluster.Labels[labels.AzureSubscriptionIDLabel], tt.cluster.Labels[labels.AzureLocationLabel], tt.sku, tt.targetCapacity)
server := testcompute.CreateMockAttributeBasedVMSizeRecommenderServer(t, mockRequest, testTenantId, testcompute.MockAttributeBasedVMSizeRecommenderResponse, tt.mockStatusCode)
defer server.Close()

t.Setenv("AZURE_TENANT_ID", "test-tenant-id")
client, err := compute.NewAttributeBasedVMSizeRecommenderClient(server.URL, http.DefaultClient)
if err != nil {
t.Fatalf("failed to create VM size recommender client: %v", err)
Expand All @@ -442,12 +451,11 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
if err == nil {
t.Fatalf("CheckIfMeetSKUCapacityRequirement error () = nil, want error")
} else if tt.errorSubstring != "" && !strings.Contains(err.Error(), tt.errorSubstring) {
t.Errorf("CheckIfMeetSKUCapacityRequirement error () = %s, want %v", err, tt.errorSubstring)
t.Fatalf("CheckIfMeetSKUCapacityRequirement error () = %s, want %v", err, tt.errorSubstring)
}
return
}

if err != nil {
if !tt.wantError && err != nil {
t.Fatalf("CheckIfMeetSKUCapacityRequirement error () = %v, want nil", err)
}

Expand All @@ -457,64 +465,3 @@ func TestCheckIfMeetSKUCapacityRequirement(t *testing.T) {
})
}
}

// createMockAttributeBasedVMSizeRecommenderServer creates a mock HTTP server for testing AttributeBasedVMSizeRecommenderClient.
func createMockAttributeBasedVMSizeRecommenderServer(t *testing.T, httpStatusCode int) *httptest.Server {
// Create mock server
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Verify request method
if r.Method != http.MethodPost {
t.Errorf("Mock PropertyChecker method () = %s, want POST request", r.Method)
}

// Verify headers
if r.Header.Get(httputil.HeaderContentTypeKey) != httputil.HeaderContentTypeJSON {
t.Errorf("Mock PropertyChecker content () = %s, want %s", r.Header.Get(httputil.HeaderContentTypeKey), httputil.HeaderContentTypeJSON)
}
if r.Header.Get(httputil.HeaderAcceptKey) != httputil.HeaderContentTypeJSON {
t.Errorf("Mock PropertyChecker accept () = %s, want %s", r.Header.Get(httputil.HeaderAcceptKey), httputil.HeaderContentTypeJSON)
}

// Verify request body using proto json for proper proto3 one of support
body, err := io.ReadAll(r.Body)
if err != nil {
t.Fatalf("failed to read request body: %v", err)
}
var req computev1.GenerateAttributeBasedRecommendationsRequest
unmarshaler := protojson.UnmarshalOptions{
DiscardUnknown: true,
}
if err := unmarshaler.Unmarshal(body, &req); err != nil {
t.Fatalf("failed to unmarshal request body: %v", err)
}

// Write mock response with status code from test case
if httpStatusCode == 0 {
httpStatusCode = http.StatusOK
}
w.Header().Set(httputil.HeaderContentTypeKey, httputil.HeaderContentTypeJSON)
w.WriteHeader(httpStatusCode)

// Mock the expected response from the Azure API.
mockAzureResponse := `{
"recommendedVmSizes": {
"regularVmSizes": [
{
"family": "Dsv3",
"name": "Standard_D2s_v3",
"size": "D2"
},
{
"family": "Standard",
"name": "Standard_B1s",
"size": "Standard_B1s"
}
]
}
}`

if _, err := w.Write([]byte(mockAzureResponse)); err != nil {
t.Fatalf("failed to write mock response: %v", err)
}
}))
}
1 change: 1 addition & 0 deletions pkg/scheduler/framework/plugins/clusteraffinity/filtering.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func (p *Plugin) Filter(
t := &ps.GetPolicySnapshotSpec().Policy.Affinity.ClusterAffinity.RequiredDuringSchedulingIgnoredDuringExecution.ClusterSelectorTerms[idx]
r := clusterRequirement{
ClusterSelectorTerm: *t,
PropertyChecker: p.PropertyChecker,
}
isMatched, err := r.Matches(cluster)
if err != nil {
Expand Down
Loading
Loading