// Copyright 2022 Woodpecker Authors // Copyright 2021 Informatyka Boguslawski sp. z o.o. sp.k., http://www.ib.pl/ // Copyright 2018 Drone.IO Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package api import ( "errors" "fmt" "net/http" "strconv" "github.com/gin-gonic/gin" "github.com/rs/zerolog/log" "go.woodpecker-ci.org/woodpecker/v3/server" "go.woodpecker-ci.org/woodpecker/v3/server/forge" "go.woodpecker-ci.org/woodpecker/v3/server/forge/types" "go.woodpecker-ci.org/woodpecker/v3/server/model" "go.woodpecker-ci.org/woodpecker/v3/server/pipeline" "go.woodpecker-ci.org/woodpecker/v3/server/store" "go.woodpecker-ci.org/woodpecker/v3/shared/token" ) // GetQueueInfo // // @Summary Get pipeline queue information // @Description Returns pipeline queue information with agent details // @Router /queue/info [get] // @Produce json // @Success 200 {object} QueueInfo // @Tags Pipeline queues // @Param Authorization header string true "Insert your personal access token" default(Bearer ) func GetQueueInfo(c *gin.Context) { info := server.Config.Services.Queue.Info(c) _store := store.FromContext(c) // Create a map to store agent names by ID agentNameMap := make(map[int64]string) // Process tasks and add agent names pendingWithAgents, err := processQueueTasks(_store, info.Pending, agentNameMap) if err != nil { c.String(http.StatusInternalServerError, err.Error()) return } waitingWithAgents, err := processQueueTasks(_store, info.WaitingOnDeps, agentNameMap) if err != nil { c.String(http.StatusInternalServerError, err.Error()) return } runningWithAgents, err := processQueueTasks(_store, info.Running, agentNameMap) if err != nil { c.String(http.StatusInternalServerError, err.Error()) return } // Create response with agent-enhanced tasks response := model.QueueInfo{ Pending: pendingWithAgents, WaitingOnDeps: waitingWithAgents, Running: runningWithAgents, Stats: struct { WorkerCount int `json:"worker_count"` PendingCount int `json:"pending_count"` WaitingOnDepsCount int `json:"waiting_on_deps_count"` RunningCount int `json:"running_count"` }{ WorkerCount: info.Stats.Workers, PendingCount: info.Stats.Pending, WaitingOnDepsCount: info.Stats.WaitingOnDeps, RunningCount: info.Stats.Running, }, Paused: info.Paused, } c.IndentedJSON(http.StatusOK, response) } // getAgentName finds an agent's name, utilizing a map as a cache. func getAgentName(store store.Store, agentNameMap map[int64]string, agentID int64) (string, bool) { // 1. Check the cache first. name, exists := agentNameMap[agentID] if exists { return name, true } // 2. If not in cache, query the store. agent, err := store.AgentFind(agentID) if err != nil || agent == nil { // Agent not found or an error occurred. return "", false } // 3. Found the agent, update the cache and return the name. if agent.Name != "" { agentNameMap[agentID] = agent.Name return agent.Name, true } return "", false } // processQueueTasks converts tasks to QueueTask structs and adds agent names. func processQueueTasks(store store.Store, tasks []*model.Task, agentNameMap map[int64]string) ([]model.QueueTask, error) { result := make([]model.QueueTask, 0, len(tasks)) for _, task := range tasks { taskResponse := model.QueueTask{ Task: *task, } if task.AgentID != 0 { name, ok := getAgentName(store, agentNameMap, task.AgentID) if !ok { return nil, fmt.Errorf("agent not found for task %s", task.ID) } taskResponse.AgentName = name } if task.PipelineID != 0 { p, err := store.GetPipeline(task.PipelineID) if err != nil { return nil, fmt.Errorf("pipeline not found for task %s", task.ID) } taskResponse.PipelineNumber = p.Number } result = append(result, taskResponse) } return result, nil } // PauseQueue // // @Summary Pause the pipeline queue // @Router /queue/pause [post] // @Produce plain // @Success 204 // @Tags Pipeline queues // @Param Authorization header string true "Insert your personal access token" default(Bearer ) func PauseQueue(c *gin.Context) { server.Config.Services.Queue.Pause() c.Status(http.StatusNoContent) } // ResumeQueue // // @Summary Resume the pipeline queue // @Router /queue/resume [post] // @Produce plain // @Success 204 // @Tags Pipeline queues // @Param Authorization header string true "Insert your personal access token" default(Bearer ) func ResumeQueue(c *gin.Context) { server.Config.Services.Queue.Resume() c.Status(http.StatusNoContent) } // BlockTilQueueHasRunningItem // // @Summary Block til pipeline queue has a running item // @Router /queue/norunningpipelines [get] // @Produce plain // @Success 204 // @Tags Pipeline queues // @Param Authorization header string true "Insert your personal access token" default(Bearer ) func BlockTilQueueHasRunningItem(c *gin.Context) { for { info := server.Config.Services.Queue.Info(c) if info.Stats.Running == 0 { break } } c.Status(http.StatusNoContent) } // PostHook // // @Summary Incoming webhook from forge // @Router /hook [post] // @Produce plain // @Success 200 // @Tags System // @Param hook body object true "the webhook payload; forge is automatically detected" func PostHook(c *gin.Context) { _store := store.FromContext(c) // // 1. Check if the webhook is valid and authorized // var repo *model.Repo _, err := token.ParseRequest([]token.Type{token.HookToken}, c.Request, func(t *token.Token) (string, error) { var err error repo, err = getRepoFromToken(_store, t) if err != nil { return "", err } return repo.Hash, nil }) if err != nil { msg := "failure to parse token from hook" log.Error().Err(err).Msg(msg) c.String(http.StatusBadRequest, msg) return } if repo == nil { msg := "failure to get repo from token" log.Error().Msg(msg) c.String(http.StatusBadRequest, msg) return } _forge, err := server.Config.Services.Manager.ForgeFromRepo(repo) if err != nil { log.Error().Err(err).Int64("repo-id", repo.ID).Msgf("Cannot get forge with id: %d", repo.ForgeID) c.AbortWithStatus(http.StatusInternalServerError) return } // // 2. Parse the webhook data // repoFromForge, pipelineFromForge, err := _forge.Hook(c, c.Request) if err != nil { if errors.Is(err, &types.ErrIgnoreEvent{}) { msg := fmt.Sprintf("forge driver: %s", err) log.Debug().Err(err).Msg(msg) c.String(http.StatusOK, msg) return } msg := "failure to parse hook" log.Debug().Err(err).Msg(msg) c.String(http.StatusBadRequest, msg) return } if pipelineFromForge == nil { msg := "ignoring hook: hook parsing resulted in empty pipeline" log.Debug().Msg(msg) c.String(http.StatusOK, msg) return } if repoFromForge == nil { msg := "failure to ascertain repo from hook" log.Debug().Msg(msg) c.String(http.StatusBadRequest, msg) return } // // 3. Check the repo from the token is matching the repo returned by the forge // if repo.ForgeRemoteID != repoFromForge.ForgeRemoteID { log.Warn().Msgf("ignoring hook: repo %s does not match the repo from the token", repo.FullName) c.String(http.StatusBadRequest, "failure to parse token from hook") return } // // 4. Check if the repo is active and has an owner // if !repo.IsActive { log.Debug().Msgf("ignoring hook: repo %s is inactive", repoFromForge.FullName) c.Status(http.StatusNoContent) return } if repo.UserID == 0 { log.Warn().Msgf("ignoring hook. repo %s has no owner.", repo.FullName) c.Status(http.StatusNoContent) return } user, err := _store.GetUser(repo.UserID) if err != nil { handleDBError(c, err) return } forge.Refresh(c, _forge, _store, user) // // 4. Update the repo // if repo.FullName != repoFromForge.FullName { // create a redirection err = _store.CreateRedirection(&model.Redirection{RepoID: repo.ID, FullName: repo.FullName}) if err != nil { _ = c.AbortWithError(http.StatusInternalServerError, err) return } } repo.Update(repoFromForge) err = _store.UpdateRepo(repo) if err != nil { c.String(http.StatusInternalServerError, err.Error()) return } // // 5. Check if pull requests are allowed for this repo // if pipelineFromForge.IsPullRequest() && !repo.AllowPull { log.Debug().Str("repo", repo.FullName).Msg("ignoring hook: pull requests are disabled for this repo in woodpecker") c.Status(http.StatusNoContent) return } // // 6. Finally create a pipeline // pl, err := pipeline.Create(c, _store, repo, pipelineFromForge) if err != nil { handlePipelineErr(c, err) } else { c.JSON(http.StatusOK, pl) } } func getRepoFromToken(store store.Store, t *token.Token) (*model.Repo, error) { if t.Get("repo-forge-remote-id") != "" { forgeID, err := strconv.ParseInt(t.Get("forge-id"), 10, 64) if err != nil { return nil, err } return store.GetRepoForgeID(forgeID, model.ForgeRemoteID(t.Get("repo-forge-remote-id"))) } // get the repo by the repo-id // TODO: remove in next major repoID, err := strconv.ParseInt(t.Get("repo-id"), 10, 64) if err != nil { return nil, err } return store.GetRepo(repoID) }