Merge pull request #4562 from jmguzik/rebalancer2

Cluster profiles rebalancer as part of autoconfigbrancher + fix for dispatcher

Author: openshift-merge-robot

cmd/autoconfigbrancher/main.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"gopkg.in/robfig/cron.v2"
 
 	"sigs.k8s.io/prow/cmd/generic-autobumper/bumper"
 	"sigs.k8s.io/prow/pkg/config/secret"
@@ -35,12 +36,13 @@ const (
 type options struct {
 	selfApprove bool
 
-	githubLogin string
-	gitName     string
-	gitEmail    string
-	targetDir   string
-	assign      string
-	whitelist   string
+	githubLogin    string
+	gitName        string
+	gitEmail       string
+	targetDir      string
+	assign         string
+	whitelist      string
+	rebalancerCron string
 
 	promotion.FutureOptions
 	flagutil.GitHubOptions
@@ -56,7 +58,7 @@ func parseOptions() options {
 	fs.StringVar(&o.targetDir, "target-dir", "", "The directory containing the target repo.")
 	fs.StringVar(&o.assign, "assign", githubTeam, "The github username or group name to assign the created pull request to.")
 	fs.StringVar(&o.whitelist, "whitelist-file", "", "The path of the whitelisted repositories file.")
-
+	fs.StringVar(&o.rebalancerCron, "rebalancer-cron", "", "Cron expression defining how often rebalancer should run (plus/minus 1h time window). If not specified, rebalancer will not run.")
 	fs.BoolVar(&o.selfApprove, "self-approve", false, "Self-approve the PR by adding the `approved` and `lgtm` labels. Requires write permissions on the repo.")
 	o.AddFlags(fs)
 	o.AllowAnonymous = true
@@ -142,6 +144,11 @@ func main() {
 		logrus.WithError(err).Fatal("error getting GitHub client")
 	}
 
+	rebalance, err := withinWindow(o.rebalancerCron)
+	if err != nil {
+		logrus.WithError(err).Fatal("failed to parse cron")
+	}
+
 	logrus.Infof("Changing working directory to '%s' ...", o.targetDir)
 	if err := os.Chdir(o.targetDir); err != nil {
 		logrus.WithError(err).Fatal("Failed to change to root dir")
@@ -238,6 +245,17 @@ func main() {
 		},
 	}
 
+	if rebalance {
+		steps = append([]step{{
+			command: "/usr/bin/rebalancer",
+			arguments: []string{
+				"--profiles=aws,aws-2,aws-3",
+				"--profiles=gcp-openshift-gce-devel-ci-2,gcp,gcp-3",
+				"--prometheus-bearer-token-path=/etc/prometheus/token",
+			},
+		}}, steps...)
+	}
+
 	stdout := bumper.HideSecretsWriter{Delegate: os.Stdout, Censor: secret.Censor}
 	stderr := bumper.HideSecretsWriter{Delegate: os.Stderr, Censor: secret.Censor}
 	author := fmt.Sprintf("%s <%s>", o.gitName, o.gitEmail)
@@ -300,3 +318,26 @@ func runSteps(steps []step, author string, stdout, stderr io.Writer) (needsPushi
 
 	return true, nil
 }
+
+// withinWindow returns true if the schedule fires at any time
+// between now-1h and now+1h.
+func withinWindow(cronExpr string) (bool, error) {
+	if cronExpr == "" {
+		return false, nil
+	}
+	schedule, err := cron.Parse(cronExpr)
+	if err != nil {
+		return false, err
+	}
+
+	now := time.Now()
+	windowStart := now.Add(-1 * time.Hour)
+	windowEnd := now.Add(+1 * time.Hour)
+
+	firstFire := schedule.Next(windowStart)
+
+	if firstFire.Before(windowEnd) {
+		return true, nil
+	}
+	return false, nil
+}

cmd/prow-job-dispatcher/main.go

@@ -84,7 +84,7 @@ func gatherOptions() options {
 	fs.StringVar(&o.configPath, "config-path", "", "Path to the config file (core-services/sanitize-prow-jobs/_config.yaml in openshift/release)")
 	fs.StringVar(&o.clusterConfigPath, "cluster-config-path", "core-services/sanitize-prow-jobs/_clusters.yaml", "Path to the config file (core-services/sanitize-prow-jobs/_clusters.yaml in openshift/release)")
 	fs.StringVar(&o.jobsStoragePath, "jobs-storage-path", "", "Path to the file holding only job assignments in Gob format")
-	fs.IntVar(&o.prometheusDaysBefore, "prometheus-days-before", 1, "Number [1,15] of days before. Time 00-00-00 of that day will be used as time to query Prometheus. E.g., 1 means 00-00-00 of yesterday.")
+	fs.IntVar(&o.prometheusDaysBefore, "prometheus-days-before", 14, "Number [1,15] of days before. Time 00-00-00 of that day will be used as time to query Prometheus. E.g., 1 means 00-00-00 of yesterday.")
 
 	fs.BoolVar(&o.createPR, "create-pr", false, "Create a pull request to the change made with this tool.")
 	fs.StringVar(&o.githubLogin, "github-login", githubLogin, "The GitHub username to use.")
@@ -676,7 +676,7 @@ func main() {
 		}
 	}
 
-	promVolumes, err := newPrometheusVolumes(o.PrometheusOptions, o.prometheusDaysBefore)
+	promVolumes, err := dispatcher.NewPrometheusVolumes(o.PrometheusOptions, o.prometheusDaysBefore)
 	if err != nil {
 		logrus.WithError(err).Fatal("failed to create prometheus volumes")
 	}
@@ -766,7 +766,7 @@ func main() {
 					}
 					return api.Cloud(info.Provider), nil
 				})
-			pjs, err := dispatchJobs(o.prowJobConfigDir, config, jobVolumes, blocked, promVolumes.calculateVolumeDistribution(configClusterMap), configClusterMap)
+			pjs, err := dispatchJobs(o.prowJobConfigDir, config, jobVolumes, blocked, promVolumes.CalculateVolumeDistribution(configClusterMap), configClusterMap)
 			if err != nil {
 				logrus.WithError(err).Error("failed to dispatch")
 				return

cmd/rebalancer/main.go

@@ -0,0 +1,160 @@
+package main
+
+import (
+	"flag"
+	"math"
+	"os"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+
+	"sigs.k8s.io/prow/pkg/config/secret"
+
+	"github.com/openshift/ci-tools/pkg/api"
+	"github.com/openshift/ci-tools/pkg/config"
+	"github.com/openshift/ci-tools/pkg/dispatcher"
+)
+
+type ProfilesFlag [][]string
+
+type options struct {
+	profiles             ProfilesFlag
+	prometheusDaysBefore int
+	dispatcher.PrometheusOptions
+}
+
+func (p *ProfilesFlag) Set(val string) error {
+	parts := strings.Split(val, ",")
+	*p = append(*p, parts)
+	return nil
+}
+
+func (p *ProfilesFlag) String() string {
+	var groups []string
+	for _, grp := range *p {
+		groups = append(groups, strings.Join(grp, ","))
+	}
+	return strings.Join(groups, ";")
+}
+
+func main() {
+	o := options{}
+	fs := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
+	fs.Var(&o.profiles, "profiles", "Comma-separated list of profiles; may be repeated")
+	fs.IntVar(&o.prometheusDaysBefore, "prometheus-days-before", 14,
+		"Number [1,15] of days before. Time 00-00-00 of that day will be used as time to query Prometheus. E.g., 1 means 00-00-00 of yesterday.")
+	o.PrometheusOptions.AddFlags(fs)
+	if err := fs.Parse(os.Args[1:]); err != nil {
+		logrus.WithError(err).Fatalf("cannot parse args: '%s'", os.Args[1:])
+	}
+
+	if o.PrometheusOptions.PrometheusPasswordPath != "" {
+		if err := secret.Add(o.PrometheusOptions.PrometheusPasswordPath); err != nil {
+			logrus.WithError(err).Fatal("Failed to start secrets agent")
+		}
+	}
+
+	if o.PrometheusOptions.PrometheusBearerTokenPath != "" {
+		if err := secret.Add(o.PrometheusOptions.PrometheusBearerTokenPath); err != nil {
+			logrus.WithError(err).Fatal("Failed to start secrets agent")
+		}
+	}
+	promVolumes, err := dispatcher.NewPrometheusVolumes(o.PrometheusOptions, o.prometheusDaysBefore)
+	if err != nil {
+		logrus.WithError(err).Fatal("failed to create prometheus volumes")
+	}
+
+	vol, err := promVolumes.GetJobVolumes()
+	if err != nil {
+		logrus.WithError(err).Fatal("failed to fetch prometheus volumes")
+	}
+
+	buckets := make(map[api.ClusterProfile]float64)
+	membership := make(map[api.ClusterProfile][]api.ClusterProfile)
+	for _, group := range o.profiles {
+		var profList []api.ClusterProfile
+		for _, p := range group {
+			prof := api.ClusterProfile(p)
+			if _, exists := buckets[prof]; !exists {
+				buckets[prof] = 0
+			}
+			profList = append(profList, prof)
+		}
+		for _, prof := range profList {
+			membership[prof] = profList
+		}
+	}
+
+	configsPath := config.CiopConfigInRepoPath
+	configs := make([]config.DataWithInfo, 0)
+
+	err = config.OperateOnCIOperatorConfigDir(configsPath, func(
+		c *api.ReleaseBuildConfiguration,
+		info *config.Info,
+	) error {
+		for i := range c.Tests {
+			t := &c.Tests[i]
+			ms := t.MultiStageTestConfiguration
+			if ms == nil {
+				continue
+			}
+
+			current := ms.ClusterProfile
+			group, ok := membership[current]
+			if !ok || len(group) == 0 {
+				continue
+			}
+
+			var bestProf api.ClusterProfile
+			minVal := math.MaxFloat64
+			for _, prof := range group {
+				if val := buckets[prof]; val < minVal {
+					minVal = val
+					bestProf = prof
+				}
+			}
+			if ms.ClusterProfile != bestProf {
+				logrus.Infof("reassigning test %q: %s -> %s", t.As, ms.ClusterProfile, bestProf)
+				ms.ClusterProfile = bestProf
+			}
+
+			weight := vol[getTestName(t, info)]
+			if weight == 0 {
+				continue
+			}
+			buckets[bestProf] += weight
+			configs = append(configs, config.DataWithInfo{Configuration: *c, Info: *info})
+		}
+		return nil
+	})
+	if err != nil {
+		logrus.WithError(err).Fatal("error distributing tests across profiles")
+	}
+
+	for i := range configs {
+		c := &configs[i]
+		if err := c.CommitTo(configsPath); err != nil {
+			logrus.WithError(err).Fatal("commit config")
+		}
+	}
+
+	for prof, val := range buckets {
+		logrus.WithField("weight", val).WithField("profile", prof).Info("Calculated weight")
+	}
+}
+
+func getTestName(t *api.TestStepConfiguration, info *config.Info) string {
+	test := ""
+	if t.IsPeriodic() {
+		test += "periodic-"
+	} else if t.Postsubmit {
+		test += "branch-ci-"
+	} else {
+		test += "pull-ci-"
+	}
+	test += info.Org + "-" + info.Repo + "-" + info.Branch + "-" + t.As
+	if info.Variant != "" {
+		test += "-" + info.Variant
+	}
+	return test
+}

images/autoconfigbrancher/Dockerfile

@@ -12,6 +12,7 @@ ADD registry-replacer /usr/bin/registry-replacer
 ADD ci-operator-yaml-creator /usr/bin/ci-operator-yaml-creator
 ADD clusterimageset-updater /usr/bin/clusterimageset-updater
 ADD promoted-image-governor /usr/bin/promoted-image-governor
+#ADD rebalancer /usr/bin/rebalancer TODO: enable when tool merged
 
 RUN microdnf install -y git && \
     microdnf clean all && \

cmd/prow-job-dispatcher/prometheus_volumes.go renamed to pkg/dispatcher/prometheus_volumes.go

@@ -1,4 +1,4 @@
-package main
+package dispatcher
 
 import (
 	"context"
@@ -7,11 +7,8 @@ import (
 
 	promapi "github.com/prometheus/client_golang/api"
 	prometheusapi "github.com/prometheus/client_golang/api/prometheus/v1"
-	"github.com/sirupsen/logrus"
 
 	"sigs.k8s.io/prow/pkg/config/secret"
-
-	"github.com/openshift/ci-tools/pkg/dispatcher"
 )
 
 type prometheusVolumes struct {
@@ -22,7 +19,7 @@ type prometheusVolumes struct {
 	m                    sync.Mutex
 }
 
-func newPrometheusVolumes(promOptions dispatcher.PrometheusOptions, prometheusDaysBefore int) (prometheusVolumes, error) {
+func NewPrometheusVolumes(promOptions PrometheusOptions, prometheusDaysBefore int) (prometheusVolumes, error) {
 	promClient, err := promOptions.NewPrometheusClient(secret.GetSecret)
 	if err != nil {
 		return prometheusVolumes{}, err
@@ -39,21 +36,19 @@ func (pv *prometheusVolumes) GetJobVolumes() (map[string]float64, error) {
 	pv.m.Lock()
 	defer pv.m.Unlock()
 	if len(pv.jobVolumes) != 0 && time.Since(pv.timestamp) < 24*time.Hour {
-		logrus.Info("Using cached job volumes")
 		return pv.jobVolumes, nil
 	}
 	v1api := prometheusapi.NewAPI(pv.promClient)
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	y, m, d := time.Now().Add(-time.Duration(24*pv.prometheusDaysBefore) * time.Hour).Date()
 	ts := time.Date(y, m, d, 0, 0, 0, 0, time.UTC)
-	jv, err := dispatcher.GetJobVolumesFromPrometheus(ctx, v1api, ts)
+	jv, err := GetJobVolumesFromPrometheus(ctx, v1api, ts)
 	if err != nil {
 		return nil, err
 	}
 	pv.jobVolumes = jv
 	pv.timestamp = time.Now()
-	logrus.Info("Fetched new job volumes")
 	return pv.jobVolumes, nil
 }
 
@@ -66,7 +61,7 @@ func (pv *prometheusVolumes) getTotalVolume() float64 {
 	return totalVolume
 }
 
-func (pv *prometheusVolumes) calculateVolumeDistribution(clusterMap dispatcher.ClusterMap) map[string]float64 {
+func (pv *prometheusVolumes) CalculateVolumeDistribution(clusterMap ClusterMap) map[string]float64 {
 	totalCapacity := 0
 	for _, cluster := range clusterMap {
 		totalCapacity += cluster.Capacity