initial import

submitqueue/serie.go

@@ -0,0 +1,105 @@
+package submitqueue
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/tweag/gerrit-queue/gerrit"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// Serie represents a list of successive changesets with an unbroken parent -> child relation,
+// starting from the parent.
+type Serie struct {
+	ChangeSets []*gerrit.Changeset
+}
+
+// GetParentCommitIDs returns the parent commit IDs
+func (s *Serie) GetParentCommitIDs() ([]string, error) {
+	if len(s.ChangeSets) == 0 {
+		return nil, fmt.Errorf("Can't return parent on a serie with zero ChangeSets")
+	}
+	return s.ChangeSets[0].ParentCommitIDs, nil
+}
+
+// GetLeafCommitID returns the commit id of the last commit in ChangeSets
+func (s *Serie) GetLeafCommitID() (string, error) {
+	if len(s.ChangeSets) == 0 {
+		return "", fmt.Errorf("Can't return leaf on a serie with zero ChangeSets")
+	}
+	return s.ChangeSets[len(s.ChangeSets)-1].CommitID, nil
+}
+
+// CheckIntegrity checks that the series contains a properly ordered and connected chain of commits
+func (s *Serie) CheckIntegrity() error {
+	logger := log.WithFields(log.Fields{
+		"serie": s,
+	})
+	// an empty serie is invalid
+	if len(s.ChangeSets) == 0 {
+		return fmt.Errorf("An empty serie is invalid")
+	}
+
+	previousCommitID := ""
+	for i, changeset := range s.ChangeSets {
+		// we can't really check the parent of the first commit
+		// so skip verifying that one
+		logger.WithFields(log.Fields{
+			"changeset":        changeset.String(),
+			"previousCommitID": fmt.Sprintf("%.7s", previousCommitID),
+		}).Debug(" - verifying changeset")
+
+		parentCommitIDs := changeset.ParentCommitIDs
+		if len(parentCommitIDs) == 0 {
+			return fmt.Errorf("Changesets without any parent are not supported")
+		}
+		// we don't check parents of the first changeset in a series
+		if i != 0 {
+			if len(parentCommitIDs) != 1 {
+				return fmt.Errorf("Merge commits in the middle of a series are not supported (only at the beginning)")
+			}
+			if parentCommitIDs[0] != previousCommitID {
+				return fmt.Errorf("changesets parent commit id doesn't match previous commit id")
+			}
+		}
+		// update previous commit id for the next loop iteration
+		previousCommitID = changeset.CommitID
+	}
+	return nil
+}
+
+func (s *Serie) String() string {
+	var sb strings.Builder
+	sb.WriteString(fmt.Sprintf("Serie[%d]", len(s.ChangeSets)))
+	if len(s.ChangeSets) == 0 {
+		sb.WriteString("()\n")
+		return sb.String()
+	}
+	parentCommitIDs, err := s.GetParentCommitIDs()
+	if err == nil {
+		if len(parentCommitIDs) == 1 {
+			sb.WriteString(fmt.Sprintf("(parent: %.7s)", parentCommitIDs[0]))
+		} else {
+			sb.WriteString("(merge: ")
+
+			for i, parentCommitID := range parentCommitIDs {
+				sb.WriteString(fmt.Sprintf("%.7s", parentCommitID))
+				if i < len(parentCommitIDs) {
+					sb.WriteString(", ")
+				}
+			}
+
+			sb.WriteString(")")
+
+		}
+	}
+	sb.WriteString(fmt.Sprintf("(%.7s..%.7s)",
+		s.ChangeSets[0].CommitID,
+		s.ChangeSets[len(s.ChangeSets)-1].CommitID))
+	return sb.String()
+}
+
+func shortCommitID(commitID string) string {
+	return commitID[:6]
+}

submitqueue/series.go

@@ -0,0 +1,125 @@
+package submitqueue
+
+import (
+	"sort"
+
+	"github.com/tweag/gerrit-queue/gerrit"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// AssembleSeries consumes a list of `Changeset`, and groups them together to series
+//
+// As we have no control over the order of the passed changesets,
+// we maintain two lookup tables,
+// mapLeafToSerie, which allows to lookup a serie by its leaf commit id,
+// to append to an existing serie
+// and mapParentToSeries, which allows to lookup all series having a certain parent commit id,
+// to prepend to any of the existing series
+// if we can't find anything, we create a new series
+func AssembleSeries(changesets []*gerrit.Changeset) ([]*Serie, error) {
+	series := make([]*Serie, 0)
+	mapLeafToSerie := make(map[string]*Serie, 0)
+
+	for _, changeset := range changesets {
+		logger := log.WithFields(log.Fields{
+			"changeset": changeset.String(),
+		})
+
+		logger.Debug("creating initial serie")
+		serie := &Serie{
+			ChangeSets: []*gerrit.Changeset{changeset},
+		}
+		series = append(series, serie)
+		mapLeafToSerie[changeset.CommitID] = serie
+	}
+
+	// Combine series using a fixpoint approach, with a max iteration count.
+	log.Debug("glueing together phase")
+	for i := 1; i < 100; i++ {
+		didUpdate := false
+		log.Debugf("at iteration %d", i)
+		for _, serie := range series {
+			logger := log.WithField("serie", serie.String())
+			parentCommitIDs, err := serie.GetParentCommitIDs()
+			if err != nil {
+				return series, err
+			}
+			if len(parentCommitIDs) != 1 {
+				// We can't append merge commits to other series
+				logger.Infof("No single parent, skipping.")
+				continue
+			}
+			parentCommitID := parentCommitIDs[0]
+			logger.Debug("Looking for a predecessor.")
+			// if there's another serie that has this parent as a leaf, glue together
+			if otherSerie, ok := mapLeafToSerie[parentCommitID]; ok {
+				if otherSerie == serie {
+					continue
+				}
+				logger := logger.WithField("otherSerie", otherSerie)
+
+				myLeafCommitID, err := serie.GetLeafCommitID()
+				if err != nil {
+					return series, err
+				}
+
+				// append our changesets to the other serie
+				logger.Debug("Splicing together.")
+				otherSerie.ChangeSets = append(otherSerie.ChangeSets, serie.ChangeSets...)
+
+				delete(mapLeafToSerie, parentCommitID)
+				mapLeafToSerie[myLeafCommitID] = otherSerie
+
+				// orphan our serie
+				serie.ChangeSets = []*gerrit.Changeset{}
+				// remove the orphaned serie from the lookup table
+				delete(mapLeafToSerie, myLeafCommitID)
+
+				didUpdate = true
+			} else {
+				logger.Debug("Not found.")
+			}
+		}
+		series = removeOrphanedSeries(series)
+		if !didUpdate {
+			log.Infof("converged after %d iterations", i)
+			break
+		}
+	}
+
+	// Check integrity, just to be on the safe side.
+	for _, serie := range series {
+		logger := log.WithFields(log.Fields{
+			"serie": serie.String(),
+		})
+		logger.Debugf("checking integrity")
+		err := serie.CheckIntegrity()
+		if err != nil {
+			logger.Errorf("checking integrity failed: %s", err)
+		}
+	}
+	return series, nil
+}
+
+// removeOrphanedSeries removes all empty series (that contain zero changesets)
+func removeOrphanedSeries(series []*Serie) []*Serie {
+	newSeries := []*Serie{}
+	for _, serie := range series {
+		if len(serie.ChangeSets) != 0 {
+			newSeries = append(newSeries, serie)
+		}
+	}
+	return newSeries
+}
+
+// SortSeries sorts a list of series by the number of changesets in each serie, descending
+func SortSeries(series []*Serie) []*Serie {
+	newSeries := make([]*Serie, len(series))
+	copy(newSeries, series)
+	sort.Slice(newSeries, func(i, j int) bool {
+		// the weight depends on the amount of changesets series changeset size
+		return len(series[i].ChangeSets) > len(series[j].ChangeSets)
+	})
+	return newSeries
+}

submitqueue/submitqueue.go

@@ -0,0 +1,220 @@
+package submitqueue
+
+import (
+	"fmt"
+
+	"github.com/tweag/gerrit-queue/gerrit"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// SubmitQueueTag is the tag used to determine whether something
+// should be considered by the submit queue or not
+const SubmitQueueTag = "submit_me"
+
+// SubmitQueue contains a list of series, a gerrit connection, and some project configuration
+type SubmitQueue struct {
+	Series         []*Serie
+	gerrit         gerrit.IClient
+	ProjectName    string
+	BranchName     string
+	HEAD           string
+	SubmitQueueTag string // the tag used to submit something to the submit queue
+}
+
+// MakeSubmitQueue builds a new submit queue
+func MakeSubmitQueue(gerritClient gerrit.IClient, projectName string, branchName string, submitQueueTag string) SubmitQueue {
+	return SubmitQueue{
+		Series:         make([]*Serie, 0),
+		gerrit:         gerritClient,
+		ProjectName:    projectName,
+		BranchName:     branchName,
+		SubmitQueueTag: submitQueueTag,
+	}
+}
+
+// LoadSeries fills Series by searching and filtering changesets, and assembling them to Series.
+func (s *SubmitQueue) LoadSeries() error {
+	// Normally, we'd like to use a queryString like
+	// "status:open project:myproject branch:mybranch hashtag:submitQueueTag label:verified=+1 label:code-review=+2"
+	// to avoid filtering client-side
+	// Due to https://github.com/andygrunwald/go-gerrit/issues/71,
+	// we need to do this on the client (filterChangesets)
+	var queryString = fmt.Sprintf("status:open project:%s branch:%s", s.ProjectName, s.BranchName)
+	log.Debugf("Running query %s", queryString)
+
+	// Download changesets from gerrit
+	changesets, err := s.gerrit.SearchChangesets(queryString)
+	if err != nil {
+		return err
+	}
+	// // Filter to contain the SubmitQueueTag
+	// changesets = gerrit.FilterChangesets(changesets, func(c *gerrit.Changeset) bool {
+	// 	return c.HasTag(SubmitQueueTag)
+	// })
+	// Filter to be code reviewed and verified
+	changesets = gerrit.FilterChangesets(changesets, func(c *gerrit.Changeset) bool {
+		return c.IsCodeReviewed && c.IsVerified
+	})
+
+	// Assemble to series
+	series, err := AssembleSeries(changesets)
+	if err != nil {
+		return err
+	}
+
+	// Sort by size
+	s.Series = SortSeries(series)
+	return nil
+}
+
+// UpdateHEAD updates the HEAD field with the commit ID of the current HEAD
+func (s *SubmitQueue) UpdateHEAD() error {
+	HEAD, err := s.gerrit.GetHEAD(s.ProjectName, s.BranchName)
+	if err != nil {
+		return err
+	}
+	s.HEAD = HEAD
+	return nil
+}
+
+// TODO: clear submit queue tag if missing +1/+2?
+
+// DoSubmit submits changes that can be submitted,
+// and updates `Series` to contain the remaining ones
+// Also updates `BranchCommitID`.
+func (s *SubmitQueue) DoSubmit() error {
+	var remainingSeries []*Serie
+
+	for _, serie := range s.Series {
+		serieParentCommitIDs, err := serie.GetParentCommitIDs()
+		if err != nil {
+			return err
+		}
+		// we can only submit series with a single parent commit (otherwise they're not rebased)
+		if len(serieParentCommitIDs) != 1 {
+			return fmt.Errorf("%s has more than one parent commit", serie.String())
+		}
+		// if serie is rebased on top of current master…
+		if serieParentCommitIDs[0] == s.HEAD {
+			// submit the last changeset of the series, which submits intermediate ones too
+			_, err := s.gerrit.SubmitChangeset(serie.ChangeSets[len(serie.ChangeSets)-1])
+			if err != nil {
+				// this might fail, for various reasons:
+				//  - developers could have updated the changeset meanwhile, clearing +1/+2 bits
+				//  - master might have advanced, so this changeset isn't rebased on top of master
+				// TODO: we currently bail out entirely, but should be fine on the
+				// next loop. We might later want to improve the logic to be a bit more
+				// smarter (like log and try with the next one)
+				return err
+			}
+			// advance head to the leaf of the current serie for the next iteration
+			newHead, err := serie.GetLeafCommitID()
+			if err != nil {
+				return err
+			}
+			s.HEAD = newHead
+		} else {
+			remainingSeries = append(remainingSeries, serie)
+		}
+	}
+
+	s.Series = remainingSeries
+	return nil
+}
+
+// DoRebase rebases all remaining series on top of each other
+// they should still be ordered by series size
+// TODO: this will produce a very large series on the next run, so we might want to preserve individual series over multiple runs
+func (s *SubmitQueue) DoRebase() error {
+	newSeries := make([]*Serie, len(s.Series))
+	futureHEAD := s.HEAD
+	for _, serie := range s.Series {
+		//TODO: don't rebase everything, just pick a "good candidate"
+
+		logger := log.WithFields(log.Fields{
+			"serie": serie,
+		})
+		logger.Infof("rebasing %s on top of %s", serie, futureHEAD)
+		newSerie, err := s.RebaseSerie(serie, futureHEAD)
+		if err != nil {
+			logger.Warnf("unable to rebase serie %s", err)
+			// TODO: we want to skip on trivial rebase errors instead of bailing out.
+			// skip means adding that serie as it is to newSeries, without advancing previousLeafCommitId
+
+			// TODO: we also should talk about when to remove the submit-queue tag
+			// just because we scheduled a conflicting submit plan, doesn't mean this is not submittable.
+			// so just removing the submit-queue tag would be unfair
+			return err
+		}
+		newSeries = append(newSeries, newSerie)
+
+		// prepare for next iteration
+		futureHEAD, err = newSerie.GetLeafCommitID()
+		if err != nil {
+			// This should never happen
+			logger.Errorf("new serie shouldn't be empty: %s", newSerie)
+			return err
+		}
+
+	}
+	s.Series = newSeries
+	return nil
+}
+
+// Run starts the submit and rebase logic.
+func (s *SubmitQueue) Run() error {
+	//TODO: log decisions made and add to some ring buffer
+	var err error
+
+	commitID, err := s.gerrit.GetHEAD(s.ProjectName, s.BranchName)
+	if err != nil {
+		log.Errorf("Unable to retrieve HEAD of branch %s at project %s: %s", s.BranchName, s.ProjectName, err)
+		return err
+	}
+	s.HEAD = commitID
+
+	err = s.LoadSeries()
+	if err != nil {
+		return err
+	}
+	if len(s.Series) == 0 {
+		// Nothing to do!
+		log.Warn("Nothing to do here")
+		return nil
+	}
+	err = s.DoSubmit()
+	if err != nil {
+		return err
+	}
+	err = s.DoRebase()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// RebaseSerie rebases a whole serie on top of a given ref
+// TODO: only rebase a single changeset. we don't really want to join disconnected series, by rebasing them on top of each other.
+func (s *SubmitQueue) RebaseSerie(serie *Serie, ref string) (*Serie, error) {
+	newSeries := &Serie{
+		ChangeSets: make([]*gerrit.Changeset, len(serie.ChangeSets)),
+	}
+
+	rebaseOnto := ref
+	for _, changeset := range serie.ChangeSets {
+		newChangeset, err := s.gerrit.RebaseChangeset(changeset, rebaseOnto)
+
+		if err != nil {
+			// uh-oh…
+			// TODO: think about error handling
+			// TODO: remove the submit queue tag if the rebase fails (but only then, not on other errors)
+			return newSeries, err
+		}
+		newSeries.ChangeSets = append(newSeries.ChangeSets, newChangeset)
+
+		// the next changeset should be rebased on top of the current commit
+		rebaseOnto = newChangeset.CommitID
+	}
+	return newSeries, nil
+}