diff --git a/bots-common/config.go b/bots-common/config.go index 38101db..c6bf475 100644 --- a/bots-common/config.go +++ b/bots-common/config.go @@ -58,16 +58,22 @@ func ReadConfigFile(filename string) (*ConfigFile, error) { func ReadWorkflowConfig(gitea Gitea, git_project string) (*AutogitConfig, error) { hash := strings.Split(git_project, "#") - if len(hash) != 2 { - return nil, fmt.Errorf("Missing branch information in projectgit: %s", git_project) + branch := "" + if len(hash) > 1 { + branch = hash[1] } a := strings.Split(hash[0], "/") - if len(a) != 2 { + prjGitRepo := DefaultGitPrj + switch len(a) { + case 1: + case 2: + prjGitRepo = a[1] + default: return nil, fmt.Errorf("Missing org/repo in projectgit: %s", git_project) } - data, _, err := gitea.GetRepositoryFileContent(a[0], a[1], hash[1], "workflow.config") + data, _, err := gitea.GetRepositoryFileContent(a[0], prjGitRepo, branch, "workflow.config") if err != nil { return nil, fmt.Errorf("Error fetching 'workflow.config': %w", err) } @@ -77,7 +83,14 @@ func ReadWorkflowConfig(gitea Gitea, git_project string) (*AutogitConfig, error) return nil, fmt.Errorf("Error parsing config file: %w", err) } - config.GitProjectName = git_project + config.GitProjectName = a[0] + "/" + prjGitRepo + if len(branch) > 0 { + config.GitProjectName = config.GitProjectName + "#" + branch + } + if len(config.Organization) < 1 { + config.Organization = a[0] + } + log.Println(config) return &config, nil } @@ -104,6 +117,18 @@ type AutogitConfig struct { Reviewers []string // only used by `pr` workflow } +type AutogitConfigs []*AutogitConfig + +func (configs AutogitConfigs) GetPrjGitConfig(org, repo, branch string) *AutogitConfig { + for _, c := range configs { + if c.Organization == org && c.Branch == branch { + return c + } + } + + return nil +} + func ReadWorkflowConfigs(reader io.Reader) ([]*AutogitConfig, error) { data, err := io.ReadAll(reader) if err != nil { diff --git a/bots-common/gitea_utils.go b/bots-common/gitea_utils.go index 4c403bc..ebdca9b 100644 --- a/bots-common/gitea_utils.go +++ b/bots-common/gitea_utils.go @@ -479,8 +479,11 @@ func (gitea *GiteaTransport) AddReviewComment(pr *models.PullRequest, state mode } func (gitea *GiteaTransport) GetRepositoryFileContent(org, repo, hash, path string) ([]byte, string, error) { - content, err := gitea.client.Repository.RepoGetContents( - repository.NewRepoGetContentsParams().WithOwner(org).WithRepo(repo).WithFilepath(path).WithRef(&hash), + params := repository.NewRepoGetContentsParams().WithOwner(org).WithRepo(repo).WithFilepath(path) + if len(hash) > 0 { + params = params.WithRef(&hash) + } + content, err := gitea.client.Repository.RepoGetContents(params, gitea.transport.DefaultAuthentication, ) diff --git a/bots-common/listen.go b/bots-common/listen.go index a9f9cae..e680307 100644 --- a/bots-common/listen.go +++ b/bots-common/listen.go @@ -59,27 +59,31 @@ type RequestProcessor interface { } type ListenDefinitions struct { - RabbitURL string // amqps://user:password@host/queue + RabbitURL *url.URL // amqps://user:password@host/queue GitAuthor string Handlers map[string]RequestProcessor + Orgs []string + + topics []string + currentTopics []string } type RabbitMessage rabbitmq.Delivery -func processRabbitMQ(msgCh chan<- RabbitMessage, server url.URL, topics []string) error { - queueName := server.Path - server.Path = "" +func (l *ListenDefinitions) processRabbitMQ(msgCh chan<- RabbitMessage) error { + queueName := l.RabbitURL.Path + l.RabbitURL.Path = "" if len(queueName) > 0 && queueName[0] == '/' { queueName = queueName[1:] } - connection, err := rabbitmq.DialTLS(server.String(), &tls.Config{ - ServerName: server.Hostname(), + connection, err := rabbitmq.DialTLS(l.RabbitURL.String(), &tls.Config{ + ServerName: l.RabbitURL.Hostname(), }) if err != nil { - return fmt.Errorf("Cannot connect to %s . Err: %w", server.Hostname(), err) + return fmt.Errorf("Cannot connect to %s . Err: %w", l.RabbitURL.Hostname(), err) } defer connection.Close() @@ -124,7 +128,7 @@ func processRabbitMQ(msgCh chan<- RabbitMessage, server url.URL, topics []string // log.Printf("queue: %s:%d", q.Name, q.Consumers) log.Println(" -- listening to topics:") - for _, topic := range topics { + for _, topic := range l.topics { err = ch.QueueBind(q.Name, topic, "pubsub", false, nil) log.Println(" +", topic) if err != nil { @@ -148,18 +152,18 @@ func processRabbitMQ(msgCh chan<- RabbitMessage, server url.URL, topics []string } } -func connectAndProcessRabbitMQ(log *log.Logger, ch chan<- RabbitMessage, server url.URL, topics []string) { +func (l *ListenDefinitions) connectAndProcessRabbitMQ(log *log.Logger, ch chan<- RabbitMessage) { defer func() { if r := recover(); r != nil { log.Println(r) log.Println("'crash' RabbitMQ worker. Recovering... reconnecting...") time.Sleep(5 * time.Second) - go connectAndProcessRabbitMQ(log, ch, server, topics) + go l.connectAndProcessRabbitMQ(log, ch) } }() for { - err := processRabbitMQ(ch, server, topics) + err := l.processRabbitMQ(ch) if err != nil { log.Printf("Error in RabbitMQ connection. %#v", err) log.Println("Reconnecting in 2 seconds...") @@ -168,9 +172,9 @@ func connectAndProcessRabbitMQ(log *log.Logger, ch chan<- RabbitMessage, server } } -func connectToRabbitMQ(log *log.Logger, server url.URL, topics []string) chan RabbitMessage { +func (l *ListenDefinitions) connectToRabbitMQ(log *log.Logger) chan RabbitMessage { ch := make(chan RabbitMessage, 100) - go connectAndProcessRabbitMQ(log, ch, server, topics) + go l.connectAndProcessRabbitMQ(log, ch) return ch } @@ -192,33 +196,28 @@ func ProcessEvent(f RequestProcessor, request *Request) { } -func ProcessRabbitMQEvents(listenDefs ListenDefinitions, orgs []string) error { - server, err := url.Parse(listenDefs.RabbitURL) - if err != nil { - log.Panicf("cannot parse server URL. Err: %#v\n", err) - } +func (l *ListenDefinitions) ProcessRabbitMQEvents() error { + log.Println("RabbitMQ connection:", l.RabbitURL.String()) + l.topics = make([]string, 0, len(l.Handlers)*len(l.Orgs)) + log.Println(len(l.Handlers), len(l.Orgs)) - log.Println("RabbitMQ connection:", *server) - topics := make([]string, 0, len(listenDefs.Handlers)*len(orgs)) - log.Println(len(listenDefs.Handlers), len(orgs)) - - server.User = url.UserPassword(rabbitUser, rabbitPassword) + l.RabbitURL.User = url.UserPassword(rabbitUser, rabbitPassword) scope := "suse" - if server.Hostname() == "rabbit.opensuse.org" { + if l.RabbitURL.Hostname() == "rabbit.opensuse.org" { scope = "opensuse" } - for _, org := range orgs { - for requestType, _ := range listenDefs.Handlers { - topics = append(topics, fmt.Sprintf("%s.src.%s.%s.#", scope, org, requestType)) + for _, org := range l.Orgs { + for requestType, _ := range l.Handlers { + l.topics = append(l.topics, fmt.Sprintf("%s.src.%s.%s.#", scope, org, requestType)) } } - slices.Sort(topics) - topics = slices.Compact(topics) + slices.Sort(l.topics) + l.topics = slices.Compact(l.topics) - ch := connectToRabbitMQ(log.Default(), *server, topics) + ch := l.connectToRabbitMQ(log.Default()) for { msg, ok := <-ch @@ -233,18 +232,18 @@ func ProcessRabbitMQEvents(listenDefs ListenDefinitions, orgs []string) error { reqType := route[3] org := route[2] - if !slices.Contains(orgs, org) { - log.Println("Got even for unhandeled org:", org) + if !slices.Contains(l.Orgs, org) { + log.Println("Got event for unhandeled org:", org) continue } log.Println("org:", org, "type:", reqType) - if handler, found := listenDefs.Handlers[reqType]; found { -/* h, err := CreateRequestHandler() - if err != nil { - log.Println("Cannot create request handler", err) - continue - } + if handler, found := l.Handlers[reqType]; found { + /* h, err := CreateRequestHandler() + if err != nil { + log.Println("Cannot create request handler", err) + continue + } */ req, err := ParseRequestJSON(reqType, msg.Body) if err != nil { @@ -252,7 +251,7 @@ func ProcessRabbitMQEvents(listenDefs ListenDefinitions, orgs []string) error { continue } else { log.Println("processing req", req.Type) -// h.Request = req + // h.Request = req ProcessEvent(handler, req) } diff --git a/group-review/main.go b/group-review/main.go index 5eba55c..322e80f 100644 --- a/group-review/main.go +++ b/group-review/main.go @@ -1,10 +1,13 @@ package main import ( + "encoding/json" "flag" "log" "regexp" + "slices" "strconv" + "strings" "time" "src.opensuse.org/autogits/common" @@ -13,7 +16,21 @@ import ( var reviewer *models.User var groupName string -var configs []*common.AutogitConfig +var configs common.AutogitConfigs + +type ReviewGroupMember struct { + Name string +} + +func fetchReviewGroupConfig(gitea common.Gitea, org, repo, branch, groupName string) (reviewers []ReviewGroupMember, err error) { + data, _, err := gitea.GetRepositoryFileContent(org, repo, branch, groupName+".review.group") + if err != nil { + return nil, err + } + + err = json.Unmarshal(data, &reviewers) + return +} func processNotifications(notification *models.NotificationThread, gitea common.Gitea) { rx := regexp.MustCompile(`^https://src\.(?:open)?suse\.(?:org|de)/api/v\d+/repos/(?[a-zA-Z0-9]+)/(?[_a-zA-Z0-9]+)/issues/(?[0-9]+)$`) @@ -38,6 +55,8 @@ func processNotifications(notification *models.NotificationThread, gitea common. return } + config := configs.GetPrjGitConfig(org, repo, pr.Base.Name) + log.Println("PR state:", pr.State) if pr.State == "closed" { // dismiss the review @@ -52,7 +71,7 @@ func processNotifications(notification *models.NotificationThread, gitea common. return } - prs, err := common.FetchPRSet(gitea, org, repo, id) + prs, err := common.FetchPRSet(gitea, org, repo, id, config) if err != nil { log.Printf("Cannot fetch PRSet for %s/%s/%d. Error: %v\n", org, repo, id, err) return @@ -64,9 +83,29 @@ func processNotifications(notification *models.NotificationThread, gitea common. return } - fetchReviewGroupConfig(prjGitPR.Base.Repo, prjGitPR.Base.Sha, groupName) + groupMembers, err := fetchReviewGroupConfig(gitea, prjGitPR.Base.Repo.Owner.UserName, prjGitPR.Base.Repo.Name, prjGitPR.Base.Sha, groupName) + if err != nil { + log.Println("Cannot fetch ReviewGroup definition:", groupName, err) + } for _, review := range reviews { + user := "" + if !review.Stale && + review.State == common.ReviewStateApproved && + slices.ContainsFunc(groupMembers, func(g ReviewGroupMember) bool { + if g.Name == review.User.UserName { + user = g.Name + return true + } + return false + }) && + strings.Contains(review.Body, "/"+groupName+" LGTM\n") { + + gitea.AddReviewComment(pr, common.ReviewStateApproved, "Signed off by: "+user) + if err := gitea.SetNotificationRead(notification.ID); err != nil { + log.Println("Cannot set notification as read", err) + } + } } } @@ -96,12 +135,14 @@ func main() { log.Println(" group-review [OPTIONS] ") log.Println() flag.Usage() + return } groupName = args[0] - config, err := common.ReadConfigFile(*configFile) + configData, err := common.ReadConfigFile(*configFile) if err != nil { - log.Panicln("Failed to read config file") + log.Println("Failed to read config file", err) + return } if err := common.RequireGiteaSecretToken(); err != nil { @@ -113,9 +154,11 @@ func main() { } gitea := common.AllocateGiteaTransport(*giteaHost) - configs = common.ResolveWorkflowConfigs(gitea, config) + configs, err = common.ResolveWorkflowConfigs(gitea, configData) + if err != nil { + log.Panicln(err) + } - var err error reviewer, err = gitea.GetCurrentUser() if err != nil { log.Panicln("Cannot fetch review user: %w", err) @@ -125,10 +168,16 @@ func main() { *interval = 1 } - log.Println(" ** processing group reviews for group:", reviewer.UserName) + log.Println(" ** processing group reviews for group:", groupName) + log.Println(" ** username in Gitea:", reviewer.UserName) log.Println(" ** polling internval:", *interval, "min") log.Println(" ** connecting to RabbitMQ:", *rabbitMqHost) + if groupName != reviewer.UserName { + log.Println(" ***** Reviewer does not match group name. Aborting. *****") + return + } + for { periodReviewCheck(gitea) time.Sleep(time.Duration(*interval * int64(time.Minute))) diff --git a/workflow-direct/example.json b/workflow-direct/example.json index ae9a732..2d31c57 100644 --- a/workflow-direct/example.json +++ b/workflow-direct/example.json @@ -1,14 +1,6 @@ [ - { - "Workflows": ["direct"], - "Organization": "autogits", - "GitProjectName": "MyPrj" - }, - { - "Workflows": ["direct"], - "Organization": "autogits", - "GitProjectName": "HiddenPrj", - "Branch": "hidden" - } + "autogits/MyPrj", + "autogits/HiddenPrj", + "testing" ] diff --git a/workflow-direct/go.mod b/workflow-direct/go.mod index 5591747..e2437ef 100644 --- a/workflow-direct/go.mod +++ b/workflow-direct/go.mod @@ -1,6 +1,8 @@ module src.opensuse.org/autogits/workflow-direct -go 1.22.3 +go 1.23.1 + +toolchain go1.24.0 replace src.opensuse.org/autogits/common => ../bots-common diff --git a/workflow-direct/main.go b/workflow-direct/main.go index be15d8e..ba8ae7f 100644 --- a/workflow-direct/main.go +++ b/workflow-direct/main.go @@ -25,6 +25,7 @@ import ( "io/fs" "log" "math/rand" + "net/url" "os" "path" "path/filepath" @@ -248,7 +249,7 @@ func verifyProjectState(git common.Git, org string, config *common.AutogitConfig if data, err := os.ReadFile(path.Join(git.GetPath(), config.GitProjectName, common.PrjLinksFile)); err == nil { pkgLinks, err = parseProjectLinks(data) if err != nil { - log.Println("Cannot parse project links file: %s", err.Error()) + log.Println("Cannot parse project links file:", err.Error()) pkgLinks = nil } else { ResolveLinks(org, pkgLinks, gitea) @@ -459,8 +460,35 @@ func consistencyCheckProcess() { var DebugMode bool +func updateConfiguration(configFilename string, orgs *[]string) { + configFile, err := common.ReadConfigFile(configFilename) + if err != nil { + log.Fatal(err) + } + + configs, _ := common.ResolveWorkflowConfigs(gitea, configFile) + configuredRepos = make(map[string][]*common.AutogitConfig) + *orgs = make([]string, 0, 1) + for _, c := range configs { + if slices.Contains(c.Workflows, "direct") { + if DebugMode { + log.Printf(" + adding org: '%s', branch: '%s', prjgit: '%s'\n", c.Organization, c.Branch, c.GitProjectName) + } + configs := configuredRepos[c.Organization] + if configs == nil { + configs = make([]*common.AutogitConfig, 0, 1) + } + configs = append(configs, c) + configuredRepos[c.Organization] = configs + + *orgs = append(*orgs, c.Organization) + } + } + +} + func main() { - workflowConfig := flag.String("config", "", "Repository and workflow definition file") + configFilename := flag.String("config", "", "List of PrjGit") giteaHost := flag.String("gitea", "src.opensuse.org", "Gitea instance") rabbitUrl := flag.String("url", "amqps://rabbit.opensuse.org", "URL for RabbitMQ instance") flag.BoolVar(&DebugMode, "debug", false, "Extra debugging information") @@ -477,49 +505,28 @@ func main() { checkInterval = time.Duration(*checkIntervalHours) * time.Hour - if len(*workflowConfig) == 0 { - log.Fatalln("No configuratio file specified. Aborting") - } - - configs, err := common.ReadWorkflowConfigsFile(*workflowConfig) - if err != nil { - log.Fatal(err) - } - - configuredRepos = make(map[string][]*common.AutogitConfig) - orgs := make([]string, 0, 1) - for _, c := range configs { - if slices.Contains(c.Workflows, "direct") { - if DebugMode { - log.Printf(" + adding org: '%s', branch: '%s', prjgit: '%s'\n", c.Organization, c.Branch, c.GitProjectName) - } - configs := configuredRepos[c.Organization] - if configs == nil { - configs = make([]*common.AutogitConfig, 0, 1) - } - configs = append(configs, c) - configuredRepos[c.Organization] = configs - - orgs = append(orgs, c.Organization) - } - } - gitea = common.AllocateGiteaTransport(*giteaHost) CurrentUser, err := gitea.GetCurrentUser() if err != nil { log.Fatalln("Cannot fetch current user:", err) } log.Println("Current User:", CurrentUser.UserName) - go consistencyCheckProcess() var defs common.ListenDefinitions + updateConfiguration(*configFilename, &defs.Orgs) defs.GitAuthor = GitAuthor - defs.RabbitURL = *rabbitUrl + defs.RabbitURL, err = url.Parse(*rabbitUrl) + if err != nil { + log.Panicf("cannot parse server URL. Err: %#v\n", err) + } + + go consistencyCheckProcess() + log.Println("defs:", defs) defs.Handlers = make(map[string]common.RequestProcessor) defs.Handlers[common.RequestType_Push] = &PushActionProcessor{} defs.Handlers[common.RequestType_Repository] = &RepositoryActionProcessor{} - log.Fatal(common.ProcessRabbitMQEvents(defs, orgs)) + log.Fatal(defs.ProcessRabbitMQEvents()) }