Skip to content

Fix dispatcher capacity and add ipCapacity field to count for maxNodes #5346

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

Open
deepsm007 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix dispatcher capacity and add ipCapacity field to count for maxNodes #5346

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
32 changes: 23 additions & 9 deletions cmd/prow-job-dispatcher/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,9 +208,11 @@ type clusterVolume struct {
clusterMap dispatcher.ClusterMap
}

func clusterStressScore(volume float64, info dispatcher.ClusterInfo, maxIP int) float64 {
return volume / dispatcher.LoadWeight(info, maxIP)
}

// findClusterForJobConfig finds a cluster running on a preferred cloud provider for the jobs in a Prow job config.
// The chosen cluster will be the one with minimal workload with the given cloud provider.
// If the cluster provider is empty string, it will choose the one with minimal workload across all cloud providers.
func (cv *clusterVolume) findClusterForJobConfig(cloudProvider string, jc *prowconfig.JobConfig, path string, config *dispatcher.Config, jobVolumes map[string]float64) (string, error) {
if _, ok := cv.clusterVolumeMap[cloudProvider]; !ok {
cloudProvider = ""
Expand All @@ -224,26 +226,38 @@ func (cv *clusterVolume) findClusterForJobConfig(cloudProvider string, jc *prowc
mostUsedCluster := dispatcher.FindMostUsedCluster(jc)
// TODO: 75% as we still have manual assignments and these are affecting even distribution, re-evaluate when manual assignments are gone
if determinedCloudProvider := config.IsInBuildFarm(api.Cluster(mostUsedCluster)); determinedCloudProvider != "" &&
!cv.blocked.Has(mostUsedCluster) &&
cv.clusterVolumeMap[string(determinedCloudProvider)][mostUsedCluster] < cv.volumeDistribution[mostUsedCluster]*0.75 {
cluster = mostUsedCluster
} else {
min := float64(-1)
maxIP := dispatcher.MaxIPCapacity(cv.clusterMap)
minScore := float64(-1)
for _, cp := range sets.List(cv.cloudProviders) {
m := cv.clusterVolumeMap[cp]
for c, v := range m {
if cv.clusterMap[c].Capacity != 100 {
if cloudProvider != "" && cloudProvider != cp {
continue
}
if cloudProvider == "" || cloudProvider == cp {
if min < 0 || min > v {
min = v
cluster = c
}
if cv.blocked.Has(c) {
continue
}
info, ok := cv.clusterMap[c]
if !ok || info.Capacity <= 0 {
continue
}
score := clusterStressScore(v, info, maxIP)
if minScore < 0 || score < minScore {
minScore = score
cluster = c
}
}
}
Comment thread
deepsm007 marked this conversation as resolved.
}

if cluster == "" {
return "", fmt.Errorf("no eligible cluster found for job config %q (cloud provider: %q)", path, cloudProvider)
}

var errs []error
for k := range jc.PresubmitsStatic {
for _, job := range jc.PresubmitsStatic[k] {
Expand Down
200 changes: 200 additions & 0 deletions cmd/prow-job-dispatcher/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,159 @@ func TestDispatchJobConfig(t *testing.T) {
},
expected: "build02",
},
{
name: "capacity-weighted: prefers lower volume/capacity even when capacity is not 100",
cv: &clusterVolume{
clusterVolumeMap: map[string]map[string]float64{
"aws": {
"build01": 90,
"build09": 20,
},
},
cloudProviders: sets.New[string]("aws"),
pjs: map[string]dispatcher.ProwJobData{},
blocked: sets.New[string](),
specialClusters: map[string]float64{},
volumeDistribution: map[string]float64{
"build01": 50,
"build09": 30,
},
clusterMap: dispatcher.ClusterMap{
"build01": dispatcher.ClusterInfo{Capacity: 100},
"build09": dispatcher.ClusterInfo{Capacity: 75},
},
},
config: &c,
jc: &prowconfig.JobConfig{
PresubmitsStatic: map[string][]prowconfig.Presubmit{
"repo": {{JobBase: prowconfig.JobBase{Name: "job",
Spec: &corev1.PodSpec{
Containers: []corev1.Container{
{Env: []corev1.EnvVar{{Name: "CLUSTER_TYPE", Value: "openstack"}}},
},
}}}},
},
},
path: "repo-presubmits.yaml",
jobVolumes: map[string]float64{
"job": 1,
},
expected: "build09",
},
{
name: "capacity-weighted: skips blocked clusters",
cv: &clusterVolume{
clusterVolumeMap: map[string]map[string]float64{
"aws": {
"build01": 10,
"build09": 0,
},
},
cloudProviders: sets.New[string]("aws"),
pjs: map[string]dispatcher.ProwJobData{},
blocked: sets.New[string]("build09"),
specialClusters: map[string]float64{},
volumeDistribution: map[string]float64{
"build01": 50,
"build09": 30,
},
clusterMap: dispatcher.ClusterMap{
"build01": dispatcher.ClusterInfo{Capacity: 100},
"build09": dispatcher.ClusterInfo{Capacity: 75},
},
},
config: &c,
jc: &prowconfig.JobConfig{
PresubmitsStatic: map[string][]prowconfig.Presubmit{
"repo": {{JobBase: prowconfig.JobBase{Name: "job",
Spec: &corev1.PodSpec{
Containers: []corev1.Container{
{Env: []corev1.EnvVar{{Name: "CLUSTER_TYPE", Value: "openstack"}}},
},
}}}},
},
},
path: "repo-presubmits.yaml",
jobVolumes: map[string]float64{
"job": 1,
},
expected: "build01",
},
{
name: "ipCapacity: prefers roomy /24 cluster over tight /26 at equal load capacity",
cv: &clusterVolume{
clusterVolumeMap: map[string]map[string]float64{
"aws": {
"build05": 40,
"build09": 40,
},
},
cloudProviders: sets.New[string]("aws"),
pjs: map[string]dispatcher.ProwJobData{},
blocked: sets.New[string](),
specialClusters: map[string]float64{},
volumeDistribution: map[string]float64{
"build05": 50,
"build09": 50,
},
clusterMap: dispatcher.ClusterMap{
"build05": dispatcher.ClusterInfo{Capacity: 50, IPCapacity: 743},
"build09": dispatcher.ClusterInfo{Capacity: 50, IPCapacity: 167},
},
},
config: &c,
jc: &prowconfig.JobConfig{
PresubmitsStatic: map[string][]prowconfig.Presubmit{
"repo": {{JobBase: prowconfig.JobBase{Name: "job",
Spec: &corev1.PodSpec{
Containers: []corev1.Container{
{Env: []corev1.EnvVar{{Name: "CLUSTER_TYPE", Value: "openstack"}}},
},
}}}},
},
},
path: "repo-presubmits.yaml",
jobVolumes: map[string]float64{
"job": 1,
},
expected: "build05",
},
{
name: "no eligible cluster returns error",
cv: &clusterVolume{
clusterVolumeMap: map[string]map[string]float64{
"aws": {
"build01": 0,
},
},
cloudProviders: sets.New[string]("aws"),
pjs: map[string]dispatcher.ProwJobData{},
blocked: sets.New[string]("build01"),
specialClusters: map[string]float64{},
volumeDistribution: map[string]float64{
"build01": 50,
},
clusterMap: dispatcher.ClusterMap{
"build01": dispatcher.ClusterInfo{Capacity: 100},
},
},
config: &c,
jc: &prowconfig.JobConfig{
PresubmitsStatic: map[string][]prowconfig.Presubmit{
"repo": {{JobBase: prowconfig.JobBase{Name: "job",
Spec: &corev1.PodSpec{
Containers: []corev1.Container{
{Env: []corev1.EnvVar{{Name: "CLUSTER_TYPE", Value: "openstack"}}},
},
}}}},
},
},
path: "repo-presubmits.yaml",
jobVolumes: map[string]float64{
"job": 1,
},
expectedErr: fmt.Errorf("fail to find cluster for job config: no eligible cluster found for job config %q (cloud provider: %q)", "repo-presubmits.yaml", ""),
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
Expand All @@ -257,6 +410,53 @@ func TestDispatchJobConfig(t *testing.T) {
}
}

func TestClusterStressScore(t *testing.T) {
testCases := []struct {
name string
volume float64
info dispatcher.ClusterInfo
maxIP int
expected float64
}{
{
name: "capacity only when ipCapacity omitted",
volume: 75,
info: dispatcher.ClusterInfo{Capacity: 100},
maxIP: 743,
expected: 0.75,
},
{
name: "capacity 100 at farm max IP keeps full weight",
volume: 50,
info: dispatcher.ClusterInfo{Capacity: 100, IPCapacity: 743},
maxIP: 743,
expected: 0.5,
},
{
name: "capacity 50 halves load weight",
volume: 50,
info: dispatcher.ClusterInfo{Capacity: 50, IPCapacity: 743},
maxIP: 743,
expected: 1.0,
},
{
name: "tight CIDR reduces weight vs roomy maxIP",
volume: 50,
info: dispatcher.ClusterInfo{Capacity: 100, IPCapacity: 167},
maxIP: 743,
expected: 50.0 / (100.0 * 167.0 / 743.0),
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
actual := clusterStressScore(tc.volume, tc.info, tc.maxIP)
if diff := cmp.Diff(tc.expected, actual); diff != "" {
t.Errorf("%s: actual does not match expected, diff: %s", tc.name, diff)
}
})
}
}

func TestGetCloudProvidersForE2ETests(t *testing.T) {
testCases := []struct {
name string
Expand Down
26 changes: 26 additions & 0 deletions pkg/dispatcher/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,38 @@ import (
type ClusterInfo struct {
Provider string
Capacity int
IPCapacity int // max usable node IPs (0=omit); with Capacity forms load weight
Capabilities []string
}

// ClusterMap maps a cluster name to its corresponding ClusterInfo.
type ClusterMap map[string]ClusterInfo

// MaxIPCapacity returns the largest IPCapacity in m.
func MaxIPCapacity(m ClusterMap) int {
max := 0
for _, info := range m {
if info.IPCapacity > max {
max = info.IPCapacity
}
}
return max
}

// LoadWeight is Capacity scaled by ipCapacity/maxIP when any cluster sets ipCapacity.
// Omitted ipCapacity uses maxIP as baseline so weights stay stable across partial config.
func LoadWeight(info ClusterInfo, maxIP int) float64 {
w := float64(info.Capacity)
if maxIP <= 0 {
return w
}
ip := info.IPCapacity
if ip <= 0 {
ip = maxIP
}
return w * float64(ip) / float64(maxIP)
}

// Config is the configuration file of this tools, which defines the cluster parameter for each Prow job, i.e., where it runs
type Config struct {
// the job will be run on the same cloud as the one for the e2e test
Expand Down
21 changes: 21 additions & 0 deletions pkg/dispatcher/helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
package dispatcher

import (
"fmt"
"os"
"reflect"

utilerrors "k8s.io/apimachinery/pkg/util/errors"
"k8s.io/apimachinery/pkg/util/sets"
prowconfig "sigs.k8s.io/prow/pkg/config"
"sigs.k8s.io/yaml"
Expand All @@ -13,6 +15,7 @@ func loadClusterConfigFromBytes(data []byte) (ClusterMap, sets.Set[string], erro
var clusters map[string][]struct {
Name string `yaml:"name"`
Capacity int `yaml:"capacity"`
IPCapacity int `yaml:"ipCapacity,omitempty"`
Comment thread
deepsm007 marked this conversation as resolved.
Capabilities []string `yaml:"capabilities"`
Blocked bool `yaml:"blocked"`
}
Expand All @@ -36,14 +39,29 @@ func loadClusterConfigFromBytes(data []byte) (ClusterMap, sets.Set[string], erro
clusterMap[cluster.Name] = ClusterInfo{
Provider: provider,
Capacity: cluster.Capacity,
IPCapacity: cluster.IPCapacity,
Capabilities: cluster.Capabilities,
}
}
}

if err := validateClusterMapIPCapacity(clusterMap); err != nil {
return nil, nil, err
}

return clusterMap, blockedClusters, nil
}

func validateClusterMapIPCapacity(clusterMap ClusterMap) error {
var errs []error
for name, info := range clusterMap {
if info.IPCapacity < 0 {
errs = append(errs, fmt.Errorf("cluster %q has negative ipCapacity: %d", name, info.IPCapacity))
}
}
return utilerrors.NewAggregate(errs)
}

// LoadClusterConfig loads cluster configuration from a YAML file, returning a ClusterMap and a set of blocked clusters.
func LoadClusterConfig(filePath string) (ClusterMap, sets.Set[string], error) {
data, err := os.ReadFile(filePath)
Expand Down Expand Up @@ -109,6 +127,9 @@ func HasCapacityOrCapabilitiesChanged(prev, next ClusterMap) bool {
if info1.Capacity != info2.Capacity {
return true
}
if info1.IPCapacity != info2.IPCapacity {
return true
}
if !reflect.DeepEqual(info1.Capabilities, info2.Capabilities) {
return true
}
Expand Down
Loading