From c05803ff14f7d7c911bdc91383703b5cd9342396 Mon Sep 17 00:00:00 2001 From: Luke Granger-Brown Date: Sat, 13 Jun 2020 23:43:09 +0100 Subject: feat(clbot): Create Gerrit watcher and basic clbot binary. gerrit.Watcher is a class which watches the Gerrit stream-events SSH connection and produces events. There's a basic CLBot binary as well, to demonstrate driving it to produce messages on the logging output. It doesn't really do anything else. Change-Id: I274fe0a77c8329f79456425405e2fbdc3ca2edf0 Reviewed-on: https://cl.tvl.fyi/c/depot/+/245 Reviewed-by: tazjin --- fun/clbot/clbot.go | 78 ++++++++ fun/clbot/default.nix | 18 ++ fun/clbot/gerrit/default.nix | 18 ++ fun/clbot/gerrit/gerritevents/default.nix | 10 + fun/clbot/gerrit/gerritevents/events.go | 321 ++++++++++++++++++++++++++++++ fun/clbot/gerrit/gerritevents/time.go | 38 ++++ fun/clbot/gerrit/gerritevents/types.go | 221 ++++++++++++++++++++ fun/clbot/gerrit/watcher.go | 277 ++++++++++++++++++++++++++ fun/clbot/gerrit/watcher_test.go | 190 ++++++++++++++++++ fun/clbot/go.mod | 11 + fun/clbot/go.sum | 19 ++ 11 files changed, 1201 insertions(+) create mode 100644 fun/clbot/clbot.go create mode 100644 fun/clbot/default.nix create mode 100644 fun/clbot/gerrit/default.nix create mode 100644 fun/clbot/gerrit/gerritevents/default.nix create mode 100644 fun/clbot/gerrit/gerritevents/events.go create mode 100644 fun/clbot/gerrit/gerritevents/time.go create mode 100644 fun/clbot/gerrit/gerritevents/types.go create mode 100644 fun/clbot/gerrit/watcher.go create mode 100644 fun/clbot/gerrit/watcher_test.go create mode 100644 fun/clbot/go.mod create mode 100644 fun/clbot/go.sum (limited to 'fun') diff --git a/fun/clbot/clbot.go b/fun/clbot/clbot.go new file mode 100644 index 0000000000..943df3bd38 --- /dev/null +++ b/fun/clbot/clbot.go @@ -0,0 +1,78 @@ +package main + +import ( + "context" + "flag" + "io/ioutil" + "os" + "time" + + "code.tvl.fyi/fun/clbot/gerrit" + "github.com/davecgh/go-spew/spew" + log "github.com/golang/glog" + "golang.org/x/crypto/ssh" +) + +var ( + gerritAddr = flag.String("gerrit_host", "cl.tvl.fyi:29418", "Gerrit SSH host:port") + gerritSSHHostKey = flag.String("gerrit_ssh_pubkey", "ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBIUNYBYPCCBNDFSd0BuCR+8kgeuJ7IA5S2nTNQmkQUYNyXK+ot5os7rHtCk96+grd5+J8jFCuFBWisUe8h8NC0Q=", "Gerrit SSH public key") + gerritSSHTimeout = flag.Duration("gerrit_tcp_timeout", 5*time.Second, "Gerrit SSH TCP connect timeout") + + required = flag.NewFlagSet("required flags", flag.ExitOnError) + gerritAuthUsername = required.String("gerrit_ssh_auth_username", "", "Gerrit SSH username") + gerritAuthKeyPath = required.String("gerrit_ssh_auth_key", "", "Gerrit SSH private key path") +) + +func mustFixedHostKey(f string) ssh.HostKeyCallback { + pk, _, _, _, err := ssh.ParseAuthorizedKey([]byte(f)) + if err != nil { + log.Exitf("ParseAuthorizedKey(%q): %v", f, err) + } + return ssh.FixedHostKey(pk) +} + +func mustPrivateKey(p string) ssh.AuthMethod { + pkBytes, err := ioutil.ReadFile(p) + if err != nil { + log.Exitf("reading SSH private key from %q: %v", p, err) + } + pk, err := ssh.ParsePrivateKey(pkBytes) + if err != nil { + log.Exitf("parsing private key from %q: %v", p, err) + } + return ssh.PublicKeys(pk) +} + +func checkRequired(fs *flag.FlagSet) { + missing := map[string]bool{} + fs.VisitAll(func(f *flag.Flag) { missing[f.Name] = true }) + fs.Visit(func(f *flag.Flag) { delete(missing, f.Name) }) + for f := range missing { + log.Errorf("flag %q was unset but is required", f) + } + if len(missing) > 0 { + os.Exit(1) + } +} + +func main() { + flag.Parse() + checkRequired(required) + + ctx := context.Background() + cfg := &ssh.ClientConfig{ + User: *gerritAuthUsername, + Auth: []ssh.AuthMethod{mustPrivateKey(*gerritAuthKeyPath)}, + HostKeyCallback: mustFixedHostKey(*gerritSSHHostKey), + Timeout: *gerritSSHTimeout, + } + cfg.SetDefaults() + gw, err := gerrit.New(ctx, "tcp", *gerritAddr, cfg) + if err != nil { + log.Errorf("gerrit.New(%q): %v", *gerritAddr, err) + } + + for e := range gw.Events() { + log.Infof("hello: %v", spew.Sdump(e)) + } +} diff --git a/fun/clbot/default.nix b/fun/clbot/default.nix new file mode 100644 index 0000000000..7a255f027a --- /dev/null +++ b/fun/clbot/default.nix @@ -0,0 +1,18 @@ +{ depot, ... }@args: + +let + clbot = depot.fun.clbot; + gopkgs = depot.third_party.gopkgs; +in +depot.nix.buildGo.program { + name = "clbot"; + srcs = [ + ./clbot.go + ]; + deps = [ + clbot.gerrit + gopkgs."github.com".davecgh.go-spew.spew.gopkg + gopkgs."github.com".golang.glog.gopkg + gopkgs."golang.org".x.crypto.ssh.gopkg + ]; +} diff --git a/fun/clbot/gerrit/default.nix b/fun/clbot/gerrit/default.nix new file mode 100644 index 0000000000..725b400e6c --- /dev/null +++ b/fun/clbot/gerrit/default.nix @@ -0,0 +1,18 @@ +{ depot, ... }: + +let + inherit (depot.fun) clbot; + inherit (depot.third_party) gopkgs; +in +depot.nix.buildGo.package { + name = "code.tvl.fyi/fun/clbot/gerrit"; + srcs = [ + ./watcher.go + ]; + deps = [ + clbot.gerrit.gerritevents + gopkgs."github.com".cenkalti.backoff.gopkg + gopkgs."github.com".golang.glog.gopkg + gopkgs."golang.org".x.crypto.ssh.gopkg + ]; +} diff --git a/fun/clbot/gerrit/gerritevents/default.nix b/fun/clbot/gerrit/gerritevents/default.nix new file mode 100644 index 0000000000..024451858b --- /dev/null +++ b/fun/clbot/gerrit/gerritevents/default.nix @@ -0,0 +1,10 @@ +{ depot, ... }: + +depot.nix.buildGo.package { + name = "code.tvl.fyi/fun/clbot/gerrit/gerritevents"; + srcs = [ + ./time.go + ./types.go + ./events.go + ]; +} diff --git a/fun/clbot/gerrit/gerritevents/events.go b/fun/clbot/gerrit/gerritevents/events.go new file mode 100644 index 0000000000..c02b30f76e --- /dev/null +++ b/fun/clbot/gerrit/gerritevents/events.go @@ -0,0 +1,321 @@ +package gerritevents + +import ( + "encoding/json" + "fmt" +) + +var events = map[string]func() Event{} + +func registerEvent(e func() Event) { + t := e().EventType() + if _, ok := events[t]; ok { + panic(fmt.Sprintf("%s already registered", t)) + } + events[t] = e +} + +// These events are taken from https://cl.tvl.fyi/Documentation/cmd-stream-events.html. + +// Event is implemented by Gerrit event structs. +type Event interface { + EventType() string +} + +type simpleEvent struct { + Type string `json:"type"` +} + +// Parse parses a Gerrit event from JSON. +func Parse(bs []byte) (Event, error) { + var s simpleEvent + if err := json.Unmarshal(bs, &s); err != nil { + return nil, fmt.Errorf("unmarshalling %q as Gerrit Event: %v", string(bs), err) + } + ef, ok := events[s.Type] + if !ok { + return nil, fmt.Errorf("unknown event type %q", s.Type) + } + e := ef() + if err := json.Unmarshal(bs, e); err != nil { + return nil, fmt.Errorf("unmarshalling %q as Gerrit Event %q: %v", string(bs), e.EventType(), err) + } + return e, nil +} + +// AssigneeChanged indicates that a change's assignee has been changed. +type AssigneeChanged struct { + Type string `json:"type"` + Change Change `json:"change"` + Changer Account `json:"changer"` + OldAssignee Account `json:"oldAssignee"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (AssigneeChanged) EventType() string { return "assignee-changed" } + +func init() { + registerEvent(func() Event { return &AssigneeChanged{} }) +} + +// ChangeAbandoned indicates that a change has been abandoned. +type ChangeAbandoned struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Abandoner Account `json:"abandoner"` + Reason string `json:"reason"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ChangeAbandoned) EventType() string { return "change-abandoned" } + +func init() { + registerEvent(func() Event { return &ChangeAbandoned{} }) +} + +// ChangeDeleted indicates that a change has been deleted. +type ChangeDeleted struct { + Type string `json:"type"` + Change Change `json:"change"` + Deleter Account `json:"deleter"` +} + +// EventType implements Event. +func (ChangeDeleted) EventType() string { return "change-deleted" } + +func init() { + registerEvent(func() Event { return &ChangeDeleted{} }) +} + +// ChangeMerged indicates that a change has been merged into the target branch. +type ChangeMerged struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Submitter Account `json:"submitter"` + NewRev string `json:"newRev"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ChangeMerged) EventType() string { return "change-merged" } + +func init() { + registerEvent(func() Event { return &ChangeMerged{} }) +} + +// ChangeRestored indicates a change has been restored (i.e. un-abandoned). +type ChangeRestored struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Restorer Account `json:"restorer"` + Reason string `json:"reason"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ChangeRestored) EventType() string { return "change-restored" } + +func init() { + registerEvent(func() Event { return &ChangeRestored{} }) +} + +// CommentAdded indicates someone has commented on a patchset. +type CommentAdded struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Author Account `json:"author"` + Approvals []Approval `json:"approvals"` + Comment string `json:"comment"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (CommentAdded) EventType() string { return "comment-added" } + +func init() { + registerEvent(func() Event { return &CommentAdded{} }) +} + +// DroppedOutput indicates that some events may be missing from the stream. +type DroppedOutput struct { + Type string `json:"type"` +} + +// EventType implements Event. +func (DroppedOutput) EventType() string { return "dropped-output" } + +func init() { + registerEvent(func() Event { return &DroppedOutput{} }) +} + +// HashtagsChanged indicates that someone has added or removed hashtags from a change. +type HashtagsChanged struct { + Type string `json:"type"` + Change Change `json:"change"` + Editor Account `json:"editor"` + Added []string `json:"added"` + Removed []string `json:"removed"` + Hashtags []string `json:"hashtags"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (HashtagsChanged) EventType() string { return "hashtags-changed" } + +func init() { + registerEvent(func() Event { return &HashtagsChanged{} }) +} + +// ProjectCreated indicates that a new project has been created. +type ProjectCreated struct { + Type string `json:"type"` + ProjectName string `json:"projectName"` + ProjectHead string `json:"projectHead"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ProjectCreated) EventType() string { return "project-created" } + +func init() { + registerEvent(func() Event { return &ProjectCreated{} }) +} + +// PatchSetCreated indicates that a new patchset has been added to a change. +type PatchSetCreated struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Uploader Account `json:"uploader"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (PatchSetCreated) EventType() string { return "patchset-created" } + +func init() { + registerEvent(func() Event { return &PatchSetCreated{} }) +} + +// RefUpdated indicates that a ref has been updated. +type RefUpdated struct { + Type string `json:"type"` + Submitter Account `json:"submitter"` + RefUpdate RefUpdate `json:"refUpdate"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (RefUpdated) EventType() string { return "ref-updated" } + +func init() { + registerEvent(func() Event { return &RefUpdated{} }) +} + +// ReviewerAdded indicates that a reviewer has been added to a change. +type ReviewerAdded struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Reviewer Account `json:"reviewer"` + Adder Account `json:"adder"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ReviewerAdded) EventType() string { return "reviewer-added" } + +func init() { + registerEvent(func() Event { return &ReviewerAdded{} }) +} + +// ReviewerDeleted indicates that a reviewer has been removed from a change, possibly removing one or more approvals. +type ReviewerDeleted struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Reviewer Account `json:"reviewer"` + Remover Account `json:"remover"` + Approvals []Approval `json:"approvals"` + Comment string `json:"comment"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (ReviewerDeleted) EventType() string { return "reviewer-deleted" } + +func init() { + registerEvent(func() Event { return &ReviewerDeleted{} }) +} + +// TopicChanged indicates that the topic attached to a change has been changed. +type TopicChanged struct { + Type string `json:"type"` + Change Change `json:"change"` + Changer Account `json:"changer"` + OldTopic string `json:"oldTopic"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (TopicChanged) EventType() string { return "topic-changed" } + +func init() { + registerEvent(func() Event { return &TopicChanged{} }) +} + +// WIPStateChanged indicates that the work-in-progress state of a change has changed. +type WIPStateChanged struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Changer Account `json:"changer"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (WIPStateChanged) EventType() string { return "wip-state-changed" } + +func init() { + registerEvent(func() Event { return &WIPStateChanged{} }) +} + +// PrivateStateChanged indicates that the private state of a change has changed. +type PrivateStateChanged struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Changer Account `json:"changer"` + EventCreatedOn Time `json:"eventCreatedOn"` +} + +// EventType implements Event. +func (PrivateStateChanged) EventType() string { return "private-state-changed" } + +func init() { + registerEvent(func() Event { return &PrivateStateChanged{} }) +} + +// VoteDeleted indicates that an approval vote has been deleted from a change. +type VoteDeleted struct { + Type string `json:"type"` + Change Change `json:"change"` + PatchSet PatchSet `json:"patchSet"` + Reviewer Account `json:"reviewer"` + Remover Account `json:"remover"` + Approvals []Approval `json:"approvals"` + Comment string `json:"comment"` +} + +// EventType implements Event. +func (VoteDeleted) EventType() string { return "vote-deleted" } + +func init() { + registerEvent(func() Event { return &VoteDeleted{} }) +} diff --git a/fun/clbot/gerrit/gerritevents/time.go b/fun/clbot/gerrit/gerritevents/time.go new file mode 100644 index 0000000000..7fbfaa3f5c --- /dev/null +++ b/fun/clbot/gerrit/gerritevents/time.go @@ -0,0 +1,38 @@ +package gerritevents + +import ( + "fmt" + "strconv" + "time" +) + +// Time is a time.Time that is formatted as a Unix timestamp in JSON. +type Time struct { + time.Time +} + +// UnmarshalJSON unmarshals a Unix timestamp into a Time. +func (t *Time) UnmarshalJSON(bs []byte) error { + if string(bs) == "null" { + return nil + } + u, err := strconv.ParseInt(string(bs), 10, 64) + if err != nil { + return err + } + t.Time = time.Unix(u, 0) + return nil +} + +// MarshalJSON marshals a Time into a Unix timestamp. +func (t *Time) MarshalJSON() ([]byte, error) { + if t.IsZero() { + return []byte("null"), nil + } + return []byte(fmt.Sprintf("%d", t.Unix())), nil +} + +// IsSet returns true if the time.Time is non-zero. +func (t *Time) IsSet() bool { + return !t.IsZero() +} diff --git a/fun/clbot/gerrit/gerritevents/types.go b/fun/clbot/gerrit/gerritevents/types.go new file mode 100644 index 0000000000..75987a2da6 --- /dev/null +++ b/fun/clbot/gerrit/gerritevents/types.go @@ -0,0 +1,221 @@ +package gerritevents + +// These types are taken from https://cl.tvl.fyi/Documentation/json.html. + +// Account is a Gerrit account (or just a Git name+email pair). +type Account struct { + Name string `json:"name"` + Email string `json:"email"` + Username string `json:"username"` +} + +// ChangeStatus represents the states a change can be in. +type ChangeStatus string + +const ( + // ChangeStatusNew is the state a change is in during review. + ChangeStatusNew ChangeStatus = "NEW" + + // ChangeStatusMerged indicates a change was merged to the target branch. + ChangeStatusMerged ChangeStatus = "MERGED" + + // ChangeStatusAbandoned indicates a change was marked as abandoned. + ChangeStatusAbandoned ChangeStatus = "ABANDONED" +) + +// Message is a message left by a reviewer. +type Message struct { + Timestamp Time `json:"timestamp"` + Reviewer Account `json:"reviewer"` + Message string `json:"message"` +} + +// TrackingID allows storing identifiers from external systems, i.e. bug trackers. +type TrackingID struct { + System string `json:"system"` + ID string `json:"id"` +} + +// ChangeKind indicates the different changes that can be made to a change. +type ChangeKind string + +const ( + // ChangeKindRework indicates a non-trivial content change. + ChangeKindRework ChangeKind = "REWORK" + + // ChangeKindTrivialRebase indicates a conflict-free merge between the new parent and the prior patch set. + ChangeKindTrivialRebase ChangeKind = "TRIVIAL_REBASE" + + // ChangeKindMergeFirstParentUpdate indicates a conflict-free change of the first parent of a merge commit. + ChangeKindMergeFirstParentUpdate ChangeKind = "MERGE_FIRST_PARENT_UPDATE" + + // ChangeKindNoCodeChange indicates no code change (the tree and parent trees are unchanged) - commit message probably changed. + ChangeKindNoCodeChange ChangeKind = "NO_CODE_CHANGE" + + // ChangeKindNoChange indicates nothing changes: the commit message, tree, and parent tree are unchanged. + ChangeKindNoChange ChangeKind = "NO_CHANGE" +) + +// Approval represents the current and past state of an approval label. +type Approval struct { + Type string `json:"type"` + Description string `json:"description"` + Value string `json:"value"` + OldValue *string `json:"oldValue"` + GrantedOn *Time `json:"grantedOn"` + By *Account `json:"by"` +} + +// PatchSetComment is a single comment left on a patchset. +type PatchSetComment struct { + File string `json:"file"` + Line int `json:"line"` + Reviewer Account `json:"reviewer"` + Message string `json:"message"` +} + +// FilePatchType represents the different modifications that can be made to a file by a patchset. +type FilePatchType string + +const ( + // FilePatchTypeAdded indicates the file did not exist, and this patchset adds it to the tree. + FilePatchTypeAdded FilePatchType = "ADDED" + + // FilePatchTypeModified indicates the file exists before and after this patchset. + FilePatchTypeModified FilePatchType = "MODIFIED" + + // FilePatchTypeDeleted indicates the file is removed by this patchset. + FilePatchTypeDeleted FilePatchType = "DELETED" + + // FilePatchTypeRenamed indicates the file has a different name before this patchset than after. + FilePatchTypeRenamed FilePatchType = "RENAMED" + + // FilePatchTypeCopied indicates the file was copied from a different file. + FilePatchTypeCopied FilePatchType = "COPIED" + + // FilePatchTypeRewrite indicates the file had a significant quantity of content changed. + FilePatchTypeRewrite FilePatchType = "REWRITE" +) + +// File represents a file in a patchset as well as how it is being modified. +type File struct { + File string `json:"file"` + FileOld string `json:"fileOld"` + Type FilePatchType `json:"type"` +} + +// PatchSet represents a single patchset within a change. +type PatchSet struct { + Number int `json:"number"` + Revision string `json:"revision"` + Parents []string `json:"parents"` + Ref string `json:"ref"` + Uploader Account `json:"uploader"` + Author Account `json:"author"` + CreatedOn Time `json:"createdOn"` + Kind ChangeKind `json:"kind"` + Approvals []Approval `json:"approvals"` + Comments []PatchSetComment `json:"comments"` + Files []File `json:"file"` + SizeInsertions int `json:"sizeInsertions"` + SizeDeletions int `json:"sizeDeletions"` +} + +// Dependency represents a change on which this change is dependent. +type Dependency struct { + ID string `json:"id"` + Number int `json:"number"` + Revision string `json:"revision"` + Ref string `json:"ref"` + IsCurrentPatchSet bool `json:"isCurrentPatchSet"` +} + +// SubmitStatus indicates whether this change has met the submit conditions and is ready to submit. +type SubmitStatus string + +const ( + // SubmitStatusOK indicates this change is ready to submit - all submit requirements are met. + SubmitStatusOK SubmitStatus = "OK" + + // SubmitStatusNotReady indicates this change cannot yet be submitted. + SubmitStatusNotReady SubmitStatus = "NOT_READY" + + // SubmitStatusRuleError indicates the submit rules could not be evaluted. Administrator intervention is required. + SubmitStatusRuleError SubmitStatus = "RULE_ERROR" +) + +// LabelStatus indicates whether this label permits submission and if the label can be granted by anyone. +type LabelStatus string + +const ( + // LabelStatusOK indicates that this label provides what is necessary for submission (e.g. CR+2). + LabelStatusOK LabelStatus = "OK" + + // LabelStatusReject indicates this label prevents submission (e.g. CR-2). + LabelStatusReject LabelStatus = "REJECT" + + // LabelStatusNeed indicates this label is required for submission, but has not been satisfied (e.g. CR0). + LabelStatusNeed LabelStatus = "NEED" + + // LabelStatusMay indicates this label is not required for submission. It may or may not be set. + LabelStatusMay LabelStatus = "MAY" + + // LabelStatusImpossible indicates this label is required for submission, but cannot be satisfied. The ACLs on this label may be set incorrectly. + LabelStatusImpossible LabelStatus = "IMPOSSIBLE" +) + +// Label represents the status of a particular label. +type Label struct { + Label string `json:"label"` + Status LabelStatus `json:"status"` + By Account `json:"by"` +} + +// Requirement represents a submit requirement. +type Requirement struct { + FallbackText string `json:"fallbackText"` + Type string `json:"type"` + // TODO(lukegb): data +} + +// SubmitRecord represents the current submission state of a change. +type SubmitRecord struct { + Status SubmitStatus `json:"status"` + Labels []Label `json:"labels"` + Requirements []Requirement `json:"requirements"` +} + +// Change represents a Gerrit CL. +type Change struct { + Project string `json:"project"` + Branch string `json:"branch"` + Topic string `json:"topic"` + ID string `json:"id"` + Number int `json:"number"` + Subject string `json:"subject"` + Owner Account `json:"owner"` + URL string `json:"url"` + CommitMessage string `json:"commitMessage"` + CreatedOn Time `json:"createdOn"` + LastUpdated *Time `json:"lastUpdated"` + Open bool `json:"open"` + Status ChangeStatus `json:"status"` + Private bool `json:"private"` + WIP bool `json:"wip"` + Comments []Message `json:"comments"` + TrackingIDs []TrackingID `json:"trackingIds"` + CurrentPatchSet *PatchSet `json:"currentPatchSet"` + PatchSets []PatchSet `json:"patchSets"` + DependsOn []Dependency `json:"dependsOn"` + NeededBy []Dependency `json:"neededBy"` + SubmitRecords []SubmitRecord `json:"submitRecord"` + AllReviewers []Account `json:"allReviewers"` +} + +// RefUpdate represents a change in a ref. +type RefUpdate struct { + OldRev string `json:"oldRev"` + NewRev string `json:"newRev"` + RefName string `json:"refName"` + Project string `json:"project"` +} diff --git a/fun/clbot/gerrit/watcher.go b/fun/clbot/gerrit/watcher.go new file mode 100644 index 0000000000..80a431f922 --- /dev/null +++ b/fun/clbot/gerrit/watcher.go @@ -0,0 +1,277 @@ +// Package gerrit implements a watcher for Gerrit events. +package gerrit + +import ( + "context" + "errors" + "fmt" + "net" + "strings" + "time" + + "code.tvl.fyi/fun/clbot/gerrit/gerritevents" + "github.com/cenkalti/backoff/v4" + log "github.com/golang/glog" + "golang.org/x/crypto/ssh" +) + +// zeroStartingBackOff is a backoff.BackOff that returns "0" as the first Duration after a reset. +// This is useful for constructing loops and just enforcing a backoff duration on every loop, rather than incorporating this logic into the loop directly. +type zeroStartingBackOff struct { + bo backoff.BackOff + initial bool +} + +// NextBackOff returns the next back off duration to use. +// For the first call after a call to Reset(), this is 0. For each subsequent duration, the underlying BackOff is consulted. +func (bo *zeroStartingBackOff) NextBackOff() time.Duration { + if bo.initial == true { + bo.initial = false + return 0 + } + return bo.bo.NextBackOff() +} + +// Reset resets to the initial state, and also passes a Reset through to the underlying BackOff. +func (bo *zeroStartingBackOff) Reset() { + bo.initial = true + bo.bo.Reset() +} + +// closer provides an embeddable implementation of Close which awaits a main loop acknowledging it has stopped. +type closer struct { + stop chan struct{} + stopped chan struct{} +} + +// newCloser returns a closer with the channels initialised. +func newCloser() closer { + return closer{ + stop: make(chan struct{}), + stopped: make(chan struct{}), + } +} + +// Close stops the main loop, waiting for the main loop to stop until it stops or the context is cancelled, whichever happens first. +func (c *closer) Close(ctx context.Context) error { + select { + case <-c.stopped: + return nil + case <-c.stop: + return nil + case <-ctx.Done(): + return ctx.Err() + default: + } + close(c.stop) + select { + case <-c.stopped: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// lineWriter is an io.Writer which splits on \n and outputs each line (with no trailing newline) to its output channel. +type lineWriter struct { + buf string + out chan string +} + +// Write accepts a slice of bytes containing zero or more new lines. +// If the contained channel is non-buffering or is full, this will block. +func (w *lineWriter) Write(p []byte) (n int, err error) { + w.buf += string(p) + pieces := strings.Split(w.buf, "\n") + w.buf = pieces[len(pieces)-1] + for n := 0; n < len(pieces)-1; n++ { + w.out <- pieces[n] + } + return len(p), nil +} + +// restartingClient is a simple SSH client that repeatedly connects to an SSH server, runs a command, and outputs the lines output by it on stdout onto a channel. +type restartingClient struct { + closer + + network string + addr string + cfg *ssh.ClientConfig + + exec string + output chan string + shutdown func() +} + +var ( + errStopConnect = errors.New("gerrit: told to stop reconnecting by remote server") +) + +func (c *restartingClient) runOnce() error { + netConn, err := net.Dial(c.network, c.addr) + if err != nil { + return fmt.Errorf("connecting to %v/%v: %w", c.network, c.addr, err) + } + defer netConn.Close() + + sshConn, newCh, newReq, err := ssh.NewClientConn(netConn, c.addr, c.cfg) + if err != nil { + return fmt.Errorf("creating SSH connection to %v/%v: %w", c.network, c.addr, err) + } + defer sshConn.Close() + + goAway := false + passedThroughReqs := make(chan *ssh.Request) + go func() { + defer close(passedThroughReqs) + for req := range newReq { + if req.Type == "goaway" { + goAway = true + log.Warningf("remote end %v/%v told me to go away!", c.network, c.addr) + sshConn.Close() + netConn.Close() + } + passedThroughReqs <- req + } + }() + + cl := ssh.NewClient(sshConn, newCh, passedThroughReqs) + + sess, err := cl.NewSession() + if err != nil { + return fmt.Errorf("NewSession on %v/%v: %w", c.network, c.addr, err) + } + defer sess.Close() + + sess.Stdout = &lineWriter{out: c.output} + + if err := sess.Start(c.exec); err != nil { + return fmt.Errorf("Start(%q) on %v/%v: %w", c.exec, c.network, c.addr, err) + } + + log.Infof("connected to %v/%v", c.network, c.addr) + + done := make(chan struct{}) + go func() { + sess.Wait() + close(done) + }() + go func() { + select { + case <-c.stop: + sess.Close() + case <-done: + } + return + }() + <-done + + if goAway { + return errStopConnect + } + return nil +} + +func (c *restartingClient) run() { + defer close(c.stopped) + ebo := backoff.NewExponentialBackOff() + ebo.MaxElapsedTime = 0 + bo := &zeroStartingBackOff{bo: ebo, initial: true} + for { + timer := time.NewTimer(bo.NextBackOff()) + select { + case <-c.stop: + timer.Stop() + return + case <-timer.C: + break + } + if err := c.runOnce(); err == errStopConnect { + if c.shutdown != nil { + c.shutdown() + return + } + } else if err != nil { + log.Errorf("SSH: %v", err) + } else { + bo.Reset() + } + } +} + +// Output returns the channel on which each newline-delimited string output by the executed command's stdout can be received. +func (c *restartingClient) Output() <-chan string { + return c.output +} + +// dialRestartingClient creates a new restartingClient. +func dialRestartingClient(network, addr string, config *ssh.ClientConfig, exec string, shutdown func()) (*restartingClient, error) { + c := &restartingClient{ + closer: newCloser(), + network: network, + addr: addr, + cfg: config, + exec: exec, + output: make(chan string), + shutdown: shutdown, + } + go c.run() + return c, nil +} + +// Watcher watches +type Watcher struct { + closer + c *restartingClient + + output chan gerritevents.Event +} + +// Close shuts down the SSH client connection, if any, and closes the output channel. +// It blocks until shutdown is complete or until the context is cancelled, whichever comes first. +func (w *Watcher) Close(ctx context.Context) { + w.c.Close(ctx) + w.closer.Close(ctx) +} + +func (w *Watcher) run() { + defer close(w.stopped) + defer close(w.output) + for { + select { + case <-w.stop: + return + case o := <-w.c.Output(): + ev, err := gerritevents.Parse([]byte(o)) + if err != nil { + log.Errorf("failed to parse event %v: %v", o, err) + continue + } + w.output <- ev + } + } +} + +// Events returns the channel upon which parsed Gerrit events can be received. +func (w *Watcher) Events() <-chan gerritevents.Event { + return w.output +} + +// New returns a running Watcher from which events can be read. +// It will begin connecting to the provided address immediately. +func New(ctx context.Context, network, addr string, cfg *ssh.ClientConfig) (*Watcher, error) { + wc := newCloser() + rc, err := dialRestartingClient(network, addr, cfg, "gerrit stream-events", func() { + wc.Close(context.Background()) + }) + if err != nil { + return nil, fmt.Errorf("dialRestartingClient: %w", err) + } + w := &Watcher{ + closer: wc, + c: rc, + output: make(chan gerritevents.Event), + } + go w.run() + return w, nil +} diff --git a/fun/clbot/gerrit/watcher_test.go b/fun/clbot/gerrit/watcher_test.go new file mode 100644 index 0000000000..ae69b2fc4c --- /dev/null +++ b/fun/clbot/gerrit/watcher_test.go @@ -0,0 +1,190 @@ +package gerrit + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/subtle" + "fmt" + "net" + "testing" + "time" + + "code.tvl.fyi/fun/clbot/gerrit/gerritevents" + log "github.com/golang/glog" + "github.com/google/go-cmp/cmp" + "golang.org/x/crypto/ssh" +) + +var ( + sshServerSigner, sshServerPublicKey = mustNewKey() + sshClientSigner, sshClientPublicKey = mustNewKey() +) + +func mustNewKey() (ssh.Signer, ssh.PublicKey) { + key, err := ecdsa.GenerateKey(elliptic.P384(), rand.Reader) + if err != nil { + panic(err) + } + signer, err := ssh.NewSignerFromKey(key) + if err != nil { + panic(err) + } + publicKey, err := ssh.NewPublicKey(key.Public()) + if err != nil { + panic(err) + } + return signer, publicKey +} + +func newSSHServer(lines string) (addr string, cleanup func(), err error) { + config := &ssh.ServerConfig{ + PublicKeyCallback: func(c ssh.ConnMetadata, pubKey ssh.PublicKey) (*ssh.Permissions, error) { + pkBytes := pubKey.Marshal() + wantPKBytes := sshClientPublicKey.Marshal() + if subtle.ConstantTimeCompare(pkBytes, wantPKBytes) == 0 { + return nil, fmt.Errorf("unauthorized") + } + return &ssh.Permissions{}, nil + }, + } + config.AddHostKey(sshServerSigner) + + ln, err := net.Listen("tcp", ":0") + if err != nil { + log.Fatalf("Listen on tcp/:0: %v", err) + } + handle := func(conn net.Conn) { + defer conn.Close() + + sc, newchch, newreqch, err := ssh.NewServerConn(conn, config) + if err != nil { + log.Fatalf("NewServerConn: %v", err) + } + go ssh.DiscardRequests(newreqch) + for newCh := range newchch { + if newCh.ChannelType() != "session" { + newCh.Reject(ssh.UnknownChannelType, "unknown channel type") + continue + } + + channel, reqs, err := newCh.Accept() + if err != nil { + log.Fatalf("Could not accept channel: %v", err) + } + go func(in <-chan *ssh.Request) { + for req := range in { + req.Reply(req.Type == "exec", nil) + } + }(reqs) + channel.Write([]byte(lines)) + sc.SendRequest("goaway", false, nil) + } + } + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go handle(conn) + } + }() + + cleanup = func() { + ln.Close() + } + return ln.Addr().String(), cleanup, err +} + +func ts(s string) gerritevents.Time { + t, err := time.Parse("2006-01-02 15:04:05 -0700 MST", s) + if err != nil { + panic(err) + } + return gerritevents.Time{t} +} + +func optStr(s string) *string { return &s } + +func TestWatcher(t *testing.T) { + tcs := []struct { + name string + lines string + want []gerritevents.Event + }{{ + name: "no events", + }, { + name: "single test event", + lines: `{"author":{"name":"tazjin","email":"mail@tazj.in","username":"tazjin"},"approvals":[{"type":"Code-Review","description":"Code-Review","value":"2","oldValue":"0"}],"comment":"Patch Set 3: Code-Review+2","patchSet":{"number":3,"revision":"6fe272d3f82c6efdfe1167fab98bf918efc03fe5","parents":["d984b6018cf68c7e8b7169b475d90134fbcee767"],"ref":"refs/changes/44/244/3","uploader":{"name":"tazjin","email":"mail@tazj.in","username":"tazjin"},"createdOn":1592081910,"author":{"name":"tazjin","email":"mail@tazj.in","username":"tazjin"},"kind":"REWORK","sizeInsertions":83,"sizeDeletions":-156},"change":{"project":"depot","branch":"master","id":"I546c701145fa204b7ba7518a8a56a783588629e0","number":244,"subject":"refactor(ops/nixos): Move my NixOS configurations to //users/tazjin","owner":{"name":"tazjin","email":"mail@tazj.in","username":"tazjin"},"url":"https://cl.tvl.fyi/c/depot/+/244","commitMessage":"refactor(ops/nixos): Move my NixOS configurations to //users/tazjin\n\nNixOS modules move one level up because it\u0027s unlikely that //ops/nixos\nwill contain actual systems at this point (they\u0027re user-specific).\n\nThis is the first users folder, so it is also added to the root\nreadTree invocation for the repository.\n\nChange-Id: I546c701145fa204b7ba7518a8a56a783588629e0\n","createdOn":1592081577,"status":"NEW"},"project":"depot","refName":"refs/heads/master","changeKey":{"id":"I546c701145fa204b7ba7518a8a56a783588629e0"},"type":"comment-added","eventCreatedOn":1592081929} +`, + want: []gerritevents.Event{ + &gerritevents.CommentAdded{ + Type: "comment-added", + Change: gerritevents.Change{ + Project: "depot", + Branch: "master", + ID: "I546c701145fa204b7ba7518a8a56a783588629e0", + Number: 244, + Subject: "refactor(ops/nixos): Move my NixOS configurations to //users/tazjin", + Owner: gerritevents.Account{Name: "tazjin", Email: "mail@tazj.in", Username: "tazjin"}, + URL: "https://cl.tvl.fyi/c/depot/+/244", + CommitMessage: "refactor(ops/nixos): Move my NixOS configurations to //users/tazjin\n\nNixOS modules move one level up because it's unlikely that //ops/nixos\nwill contain actual systems at this point (they're user-specific).\n\nThis is the first users folder, so it is also added to the root\nreadTree invocation for the repository.\n\nChange-Id: I546c701145fa204b7ba7518a8a56a783588629e0\n", + CreatedOn: ts("2020-06-13 21:52:57 +0100 BST"), + Status: "NEW", + }, + PatchSet: gerritevents.PatchSet{ + Number: 3, + Revision: "6fe272d3f82c6efdfe1167fab98bf918efc03fe5", + Parents: []string{"d984b6018cf68c7e8b7169b475d90134fbcee767"}, + Ref: "refs/changes/44/244/3", + Uploader: gerritevents.Account{Name: "tazjin", Email: "mail@tazj.in", Username: "tazjin"}, + Author: gerritevents.Account{Name: "tazjin", Email: "mail@tazj.in", Username: "tazjin"}, + CreatedOn: ts("2020-06-13 21:58:30 +0100 BST"), + Kind: "REWORK", + SizeInsertions: 83, + SizeDeletions: -156, + }, + Author: gerritevents.Account{Name: "tazjin", Email: "mail@tazj.in", Username: "tazjin"}, + Approvals: []gerritevents.Approval{{Type: "Code-Review", Description: "Code-Review", Value: "2", OldValue: optStr("0")}}, + Comment: "Patch Set 3: Code-Review+2", + EventCreatedOn: ts("2020-06-13 21:58:49 +0100 BST"), + }, + }, + }} + for _, tc := range tcs { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + serverAddr, cleanup, err := newSSHServer(tc.lines) + if err != nil { + t.Fatalf("newSSHServer: %v", err) + } + t.Cleanup(cleanup) + + config := &ssh.ClientConfig{ + User: "bert", + Auth: []ssh.AuthMethod{ssh.PublicKeys(sshClientSigner)}, + HostKeyCallback: ssh.FixedHostKey(sshServerPublicKey), + Timeout: 10 * time.Millisecond, + } + w, err := New(ctx, "tcp", serverAddr, config) + if err != nil { + t.Fatalf("New: %v", err) + } + + var gotEvents []gerritevents.Event + for ev := range w.Events() { + gotEvents = append(gotEvents, ev) + } + if diff := cmp.Diff(gotEvents, tc.want); diff != "" { + t.Errorf("got events != want events: diff:\n%v", diff) + } + }) + } +} diff --git a/fun/clbot/go.mod b/fun/clbot/go.mod new file mode 100644 index 0000000000..f954c752cb --- /dev/null +++ b/fun/clbot/go.mod @@ -0,0 +1,11 @@ +module code.tvl.fyi/fun/clbot + +go 1.14 + +require ( + github.com/cenkalti/backoff/v4 v4.0.2 + github.com/davecgh/go-spew v1.1.1 + github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b + github.com/google/go-cmp v0.4.1 + golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9 +) diff --git a/fun/clbot/go.sum b/fun/clbot/go.sum new file mode 100644 index 0000000000..0c7fb33cb9 --- /dev/null +++ b/fun/clbot/go.sum @@ -0,0 +1,19 @@ +github.com/cenkalti/backoff v1.1.0 h1:QnvVp8ikKCDWOsFheytRCoYWYPO/ObCTBGxT19Hc+yE= +github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4= +github.com/cenkalti/backoff/v4 v4.0.2 h1:JIufpQLbh4DkbQoii76ItQIUFzevQSqOLZca4eamEDs= +github.com/cenkalti/backoff/v4 v4.0.2/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= +github.com/google/go-cmp v0.4.1 h1:/exdXoGamhu5ONeUJH0deniYLWYvQwW66yvlfiiKTu0= +github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9 h1:vEg9joUBmeBcK9iSJftGNf3coIG4HqZElCPehJsfAYM= +golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -- cgit 1.4.1