darkkirb/harness-drone
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge pull request #2975 from drone/reaper

Browse source 
Reaper
This commit is contained in:
Brad Rydzewski 2020-06-08 11:26:19 -04:00 committed by GitHub
commit fcaea8f4ea
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 746 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cmd/drone-server/config/config.go
View file

@ -51,6 +51,7 @@ type (
Agent Agent
AzureBlob AzureBlob
Convert Convert
Cleanup Cleanup
Cron Cron
Cloning Cloning
Database Database
@ -96,6 +97,13 @@ type (
Pull string `envconfig:"DRONE_GIT_IMAGE_PULL" default:"IfNotExists"`
}
Cleanup struct {
Disabled bool `envconfig:"DRONE_CLEANUP_DISABLED"`
Interval time.Duration `envconfig:"DRONE_CLEANUP_INTERVAL" default:"24h"`
Running time.Duration `envconfig:"DRONE_CLEANUP_DEADLINE_RUNNING" default:"24h"`
Pending time.Duration `envconfig:"DRONE_CLEANUP_DEADLINE_PENDING" default:"24h"`
}
// Cron provides the cron configuration.
Cron struct {
Disabled bool `envconfig:"DRONE_CRON_DISABLED"`

21
cmd/drone-server/inject_service.go
View file

@ -23,6 +23,7 @@ import (
"github.com/drone/drone/metric/sink"
"github.com/drone/drone/pubsub"
"github.com/drone/drone/service/canceler"
"github.com/drone/drone/service/canceler/reaper"
"github.com/drone/drone/service/commit"
contents "github.com/drone/drone/service/content"
"github.com/drone/drone/service/content/cache"
@ -64,6 +65,7 @@ var serviceSet = wire.NewSet(
provideHookService,
provideNetrcService,
provideOrgService,
provideReaper,
provideSession,
provideStatusService,
provideSyncer,
@ -170,6 +172,25 @@ func provideSystem(config config.Config) *core.System {
}
}
// provideReaper is a Wire provider function that returns the
// zombie build reaper.
func provideReaper(
repos core.RepositoryStore,
builds core.BuildStore,
stages core.StageStore,
canceler core.Canceler,
config config.Config,
) *reaper.Reaper {
return reaper.New(
repos,
builds,
stages,
canceler,
config.Cleanup.Running,
config.Cleanup.Pending,
)
}
// provideDatadog is a Wire provider function that returns the
// datadog sink.
func provideDatadog(

16
cmd/drone-server/main.go
View file

@ -24,6 +24,7 @@ import (
"github.com/drone/drone/core"
"github.com/drone/drone/metric/sink"
"github.com/drone/drone/operator/runner"
"github.com/drone/drone/service/canceler/reaper"
"github.com/drone/drone/server"
"github.com/drone/drone/trigger/cron"
"github.com/drone/signal"
@ -114,6 +115,18 @@ func main() {
return app.cron.Start(ctx, config.Cron.Interval)
})
// launches the reaper in a goroutine. If the reaper
// is disabled, the goroutine exits immediately
// without error.
g.Go(func() (err error) {
if config.Cleanup.Disabled {
return nil
}
logrus.WithField("interval", config.Cleanup.Interval.String()).
Infoln("starting the zombie build reaper")
return app.reaper.Start(ctx, config.Cleanup.Interval)
})
// launches the build runner in a goroutine. If the local
// runner is disabled (because nomad or kubernetes is enabled)
// then the goroutine exits immediately without error.
@ -154,6 +167,7 @@ func initLogging(c config.Config) {
// application is the main struct for the Drone server.
type application struct {
cron *cron.Scheduler
reaper *reaper.Reaper
sink *sink.Datadog
runner *runner.Runner
server *server.Server
@ -163,6 +177,7 @@ type application struct {
// newApplication creates a new application struct.
func newApplication(
cron *cron.Scheduler,
reaper *reaper.Reaper,
sink *sink.Datadog,
runner *runner.Runner,
server *server.Server,
@ -173,5 +188,6 @@ func newApplication(
sink: sink,
server: server,
runner: runner,
reaper: reaper,
}
}

3
cmd/drone-server/wire_gen.go
View file

@ -63,6 +63,7 @@ func InitializeApplication(config2 config.Config) (application, error) {
validateService := provideValidatePlugin(config2)
triggerer := trigger.New(coreCanceler, configService, convertService, commitService, statusService, buildStore, scheduler, repositoryStore, userStore, validateService, webhookSender)
cronScheduler := cron2.New(commitService, cronStore, repositoryStore, userStore, triggerer)
reaper := provideReaper(repositoryStore, buildStore, stageStore, coreCanceler, config2)
coreLicense := provideLicense(client, config2)
datadog := provideDatadog(userStore, repositoryStore, buildStore, system, coreLicense, config2)
logStore := provideLogStore(db, config2)
@ -104,6 +105,6 @@ func InitializeApplication(config2 config.Config) (application, error) {
mainPprofHandler := providePprof(config2)
mux := provideRouter(server, webServer, mainRpcHandlerV1, mainRpcHandlerV2, mainHealthzHandler, metricServer, mainPprofHandler)
serverServer := provideServer(mux, config2)
mainApplication := newApplication(cronScheduler, datadog, runner, serverServer, userStore)
mainApplication := newApplication(cronScheduler, reaper, datadog, runner, serverServer, userStore)
return mainApplication, nil
}

198
service/canceler/reaper/reaper.go Normal file
View file

@ -0,0 +1,198 @@
// Copyright 2019 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 reaper
import (
"context"
"runtime/debug"
"time"
"github.com/drone/drone/core"
"github.com/hashicorp/go-multierror"
"github.com/sirupsen/logrus"
)
// Reaper finds and kills zombie jobs that are permanently
// stuck in a pending or running state.
type Reaper struct {
Repos core.RepositoryStore
Builds core.BuildStore
Stages core.StageStore
Canceler core.Canceler
Pending time.Duration // Pending is the pending pipeline deadline
Running time.Duration // Running is the running pipeline deadline
}
// New returns a new Reaper.
func New(
repos core.RepositoryStore,
builds core.BuildStore,
stages core.StageStore,
canceler core.Canceler,
running time.Duration,
pending time.Duration,
) *Reaper {
if running == 0 {
running = time.Hour * 24
}
if pending == 0 {
pending = time.Hour * 24
}
return &Reaper{
Repos: repos,
Builds: builds,
Stages: stages,
Canceler: canceler,
Pending: pending,
Running: running,
}
}
// Start starts the reaper.
func (r *Reaper) Start(ctx context.Context, dur time.Duration) error {
ticker := time.NewTicker(dur)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
r.reap(ctx)
}
}
}
func (r *Reaper) reap(ctx context.Context) error {
defer func() {
// taking the paranoid approach to recover from
// a panic that should absolutely never happen.
if r := recover(); r != nil {
logrus.Errorf("reaper: unexpected panic: %s", r)
debug.PrintStack()
}
}()
logrus.Traceln("reaper: finding zombie builds")
var result error
pending, err := r.Builds.Pending(ctx)
if err != nil {
logrus.WithError(err).
Errorf("reaper: cannot get pending builds")
result = multierror.Append(result, err)
}
for _, build := range pending {
logger := logrus.
WithField("build.id", build.ID).
WithField("build.number", build.Number).
WithField("build.repo_id", build.RepoID).
WithField("build.status", build.Status).
WithField("build.created", build.Created)
// if a build is pending for longer than the maximum
// pending time limit, the build is maybe cancelled.
if isExceeded(build.Created, r.Pending, buffer) {
logger.Traceln("reaper: cancel build: time limit exceeded")
err = r.reapMaybe(ctx, build)
if err != nil {
logger.WithError(err).
Errorln("reaper: cannot cancel build")
result = multierror.Append(result, err)
}
} else {
logger.Traceln("reaper: ignore build: time limit not exceeded")
}
}
running, err := r.Builds.Running(ctx)
if err != nil {
logrus.WithError(err).
Errorf("reaper: cannot get running builds")
result = multierror.Append(result, err)
}
for _, build := range running {
logger := logrus.
WithField("build.id", build.ID).
WithField("build.number", build.Number).
WithField("build.repo_id", build.RepoID).
WithField("build.status", build.Status).
WithField("build.created", build.Created)
// if a build is running for longer than the maximum
// running time limit, the build is maybe cancelled.
if isExceeded(build.Started, r.Running, buffer) {
logger.Traceln("reaper: cancel build: time limit exceeded")
err = r.reapMaybe(ctx, build)
if err != nil {
logger.WithError(err).
Errorln("reaper: cannot cancel build")
result = multierror.Append(result, err)
}
} else {
logger.Traceln("reaper: ignore build: time limit not exceeded")
}
}
return result
}
func (r *Reaper) reapMaybe(ctx context.Context, build *core.Build) error {
repo, err := r.Repos.Find(ctx, build.RepoID)
if err != nil {
return err
}
// if the build status is pending we can immediately
// cancel the build and all build stages.
if build.Status == core.StatusPending {
// TODO trace log entry
return r.Canceler.Cancel(ctx, repo, build)
}
stages, err := r.Stages.List(ctx, build.ID)
if err != nil {
return err
}
var started int64
for _, stage := range stages {
if stage.IsDone() {
continue
}
if stage.Started > started {
started = stage.Started
}
}
// if the build stages are all pending we can immediately
// cancel the build.
if started == 0 {
// TODO trace log entry
return r.Canceler.Cancel(ctx, repo, build)
}
// if the build stage has exceeded the timeout by a reasonable
// margin cancel the build and all build stages, else ignore.
if isExceeded(started, time.Duration(repo.Timeout)*time.Minute, buffer) {
// TODO trace log entry
return r.Canceler.Cancel(ctx, repo, build)
}
// TODO trace log entry
return nil
}

394
service/canceler/reaper/reaper_test.go Normal file
View file

@ -0,0 +1,394 @@
// Copyright 2019 Drone.IO Inc. All rights reserved.
// Use of this source code is governed by the Drone Non-Commercial License
// that can be found in the LICENSE file.
package reaper
import (
"context"
"testing"
"time"
"github.com/drone/drone/core"
"github.com/drone/drone/mock"
"github.com/golang/mock/gomock"
)
var nocontext = context.Background()
//
// reap tests
//
// this test confirms that pending builds that
// exceed the deadline are canceled, and pending
// builds that do not exceed the deadline are
// ignored.
func TestReapPending(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockRepo := &core.Repository{
ID: 2,
}
mockBuild := &core.Build{
ID: 1,
RepoID: mockRepo.ID,
Status: core.StatusPending,
Created: mustParse("2006-01-01T00:00:00").Unix(), // expire > 24 hours, must cancel
}
mockPending := []*core.Build{
mockBuild,
{
ID: 2,
RepoID: mockRepo.ID,
Status: core.StatusPending,
Created: mustParse("2006-01-02T14:30:00").Unix(), // expire < 1 hours, must ignore
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil).Times(1)
builds := mock.NewMockBuildStore(controller)
builds.EXPECT().Pending(gomock.Any()).Return(mockPending, nil)
builds.EXPECT().Running(gomock.Any()).Return(nil, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := New(
repos,
builds,
nil,
canceler,
time.Hour*24,
time.Hour*24,
)
r.reap(nocontext)
}
// this test confirms that running builds that
// exceed the deadline are canceled, and running
// builds that do not exceed the deadline are
// ignored.
func TestReapRunning(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockRepo := &core.Repository{
ID: 2,
Timeout: 60,
}
mockBuild := &core.Build{
ID: 1,
RepoID: mockRepo.ID,
Status: core.StatusRunning,
Started: mustParse("2006-01-01T00:00:00").Unix(), // expire > 24 hours, must cancel
}
mockRunning := []*core.Build{
mockBuild,
{
ID: 2,
RepoID: mockRepo.ID,
Status: core.StatusRunning,
Started: mustParse("2006-01-02T14:30:00").Unix(), // expire < 1 hours, must ignore
},
}
mockStages := []*core.Stage{
{
BuildID: mockBuild.ID,
Status: core.StatusPending,
Started: 0,
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil).Times(1)
builds := mock.NewMockBuildStore(controller)
builds.EXPECT().Pending(gomock.Any()).Return(nil, nil)
builds.EXPECT().Running(gomock.Any()).Return(mockRunning, nil)
stages := mock.NewMockStageStore(controller)
stages.EXPECT().List(gomock.Any(), mockBuild.ID).Return(mockStages, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := New(
repos,
builds,
stages,
canceler,
time.Hour*24,
time.Hour*24,
)
r.reap(nocontext)
}
//
// reap maybe tests
//
// this test confirms that the build is cancelled
// if the build status is pending.
func TestReapPendingMaybe(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
mockBuild := &core.Build{
ID: 1,
RepoID: 2,
Status: core.StatusPending,
}
mockRepo := &core.Repository{
ID: 2,
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := &Reaper{
Repos: repos,
Stages: nil,
Canceler: canceler,
}
r.reapMaybe(nocontext, mockBuild)
}
// this test confirms that the build is cancelled
// if the build status is running, and the stage
// started date is greater than the expiry date.
func TestReapRunningMaybe(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockBuild := &core.Build{
ID: 1,
RepoID: 2,
Status: core.StatusRunning,
}
mockRepo := &core.Repository{
ID: 2,
Timeout: 60,
}
mockStages := []*core.Stage{
{
Status: core.StatusRunning,
Started: mustParse("2006-01-02T13:00:00").Unix(), // running 2 hours, 1 hour longer than timeout
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil)
stages := mock.NewMockStageStore(controller)
stages.EXPECT().List(gomock.Any(), mockBuild.ID).Return(mockStages, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := &Reaper{
Repos: repos,
Stages: stages,
Canceler: canceler,
}
r.reapMaybe(nocontext, mockBuild)
}
// this test confirms that if the build status is
// running, but all stages have a pending status,
// the build is cancelled (this likely points to some
// sort of race condition, and should not happen).
func TestReapRunningMaybe_AllStagesPending(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockBuild := &core.Build{
ID: 1,
RepoID: 2,
Status: core.StatusRunning,
}
mockRepo := &core.Repository{
ID: 2,
Timeout: 60,
}
mockStages := []*core.Stage{
{
Status: core.StatusPending,
Started: 0,
},
{
Status: core.StatusPending,
Started: 0,
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil)
stages := mock.NewMockStageStore(controller)
stages.EXPECT().List(gomock.Any(), mockBuild.ID).Return(mockStages, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := &Reaper{
Repos: repos,
Stages: stages,
Canceler: canceler,
}
r.reapMaybe(nocontext, mockBuild)
}
// this test confirms that if the build status is
// running, but all stages have a finished status,
// the build is cancelled (this likely points to some
// sort of race condition, and should not happen).
func TestReapRunningMaybe_AllStagesFinished(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockBuild := &core.Build{
ID: 1,
RepoID: 2,
Status: core.StatusRunning,
}
mockRepo := &core.Repository{
ID: 2,
Timeout: 60,
}
mockStages := []*core.Stage{
{
Status: core.StatusPassing,
Started: mustParse("2006-01-02T14:40:00").Unix(),
},
{
Status: core.StatusPassing,
Started: mustParse("2006-01-02T14:50:00").Unix(),
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil)
stages := mock.NewMockStageStore(controller)
stages.EXPECT().List(gomock.Any(), mockBuild.ID).Return(mockStages, nil)
canceler := mock.NewMockCanceler(controller)
canceler.EXPECT().Cancel(gomock.Any(), mockRepo, mockBuild)
r := &Reaper{
Repos: repos,
Stages: stages,
Canceler: canceler,
}
r.reapMaybe(nocontext, mockBuild)
}
// this test confirms that if the build status is
// running, but the stage start time has not exceeded
// the timeout period, the build is NOT cancelled.
func TestReapRunningMaybe_NotExpired(t *testing.T) {
controller := gomock.NewController(t)
defer controller.Finish()
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
mockBuild := &core.Build{
ID: 1,
RepoID: 2,
Status: core.StatusRunning,
}
mockRepo := &core.Repository{
ID: 2,
Timeout: 60,
}
mockStages := []*core.Stage{
{
Status: core.StatusPassing,
Started: mustParse("2006-01-02T14:50:00").Unix(),
},
{
Status: core.StatusRunning,
Started: mustParse("2006-01-02T14:55:00").Unix(),
},
}
repos := mock.NewMockRepositoryStore(controller)
repos.EXPECT().Find(gomock.Any(), mockBuild.RepoID).Return(mockRepo, nil)
stages := mock.NewMockStageStore(controller)
stages.EXPECT().List(gomock.Any(), mockBuild.ID).Return(mockStages, nil)
r := &Reaper{
Repos: repos,
Stages: stages,
Canceler: nil,
}
r.reapMaybe(nocontext, mockBuild)
}
//
// Failure Scenarios
//
func TestReapRunningMaybe_ErrorGetRepo(t *testing.T) {
}
func TestReapRunningMaybe_ErrorListStages(t *testing.T) {
}

32
service/canceler/reaper/util.go Normal file
View file

@ -0,0 +1,32 @@
// Copyright 2019 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 reaper
import "time"
// buffer is applied when calculating whether or not the timeout
// period is exceeded. The added buffer helps prevent false positives.
var buffer = time.Minute * 30
// helper function returns the current time.
var now = time.Now
// helper function returns true if the time exceeded the
// timeout duration.
func isExceeded(unix int64, timeout, buffer time.Duration) bool {
return now().After(
time.Unix(unix, 0).Add(timeout).Add(buffer),
)
}

75
service/canceler/reaper/util_test.go Normal file
View file

@ -0,0 +1,75 @@
// Copyright 2019 Drone.IO Inc. All rights reserved.
// Use of this source code is governed by the Drone Non-Commercial License
// that can be found in the LICENSE file.
package reaper
import (
"testing"
"time"
)
func TestIsExceeded(t *testing.T) {
defer func() {
now = time.Now
}()
now = func() time.Time {
return mustParse("2006-01-02T15:00:00")
}
var tests = []struct {
unix int64
timeout time.Duration
buffer time.Duration
exceeded bool
}{
// timestamp equal to current time, not expired
{
unix: mustParse("2006-01-02T15:00:00").Unix(),
timeout: time.Minute * 60,
buffer: time.Minute * 5,
exceeded: false,
},
// timestamp is not gt current time - timeout, not expired
{
unix: mustParse("2006-01-02T14:00:00").Unix(),
timeout: time.Minute * 60,
buffer: 0,
exceeded: false,
},
// timestamp is gt current time - timeout, expired
{
unix: mustParse("2006-01-02T13:59:00").Unix(),
timeout: time.Minute * 60,
buffer: 0,
exceeded: true,
},
// timestamp is not gt current time - timeout - buffer, not expired
{
unix: mustParse("2006-01-02T13:59:00").Unix(),
timeout: time.Minute * 60,
buffer: time.Minute * 5,
exceeded: false,
},
// timestamp is gt current time - timeout - buffer, expired
{
unix: mustParse("2006-01-02T13:04:05").Unix(),
timeout: time.Minute * 60,
buffer: time.Minute * 5,
exceeded: true,
},
}
for i, test := range tests {
got, want := isExceeded(test.unix, test.timeout, test.buffer), test.exceeded
if got != want {
t.Errorf("Want exceeded %v, got %v at index %v", want, got, i)
}
}
}
func mustParse(s string) time.Time {
t, err := time.Parse("2006-01-02T15:04:05", s)
if err != nil {
panic(err)
}
return t
}