added reaper to kill zombie builds

service/canceler/reaper/reaper.go

@@ -0,0 +1,111 @@
+// 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"
+	"time"
+
+	"github.com/drone/drone/core"
+)
+
+// 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
+}
+
+// TODO use multierror to aggregate errors encountered
+// TODO use trace logging
+
+func (r *Reaper) reap(ctx context.Context) error {
+	ttl := time.Hour*24
+
+	pending, err := r.Builds.Pending(ctx)
+	if err != nil {
+		return err
+	}
+	for _, build := range pending {
+		// if a build is pending for longer than the maximum
+		// pending time limit, the build is maybe cancelled.
+		if isExceeded(build.Created, ttl, buffer) {
+			err = r.reapMaybe(ctx, build)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	running, err := r.Builds.Running(ctx)
+	if err != nil {
+		return err
+	}
+	for _, build := range running {
+		// if a build is running for longer than the maximum
+		// running time limit, the build is maybe cancelled.
+		if isExceeded(build.Started, ttl, buffer) {
+			err = r.reapMaybe(ctx, build)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+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 {
+		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 {
+		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) {
+		return r.Canceler.Cancel(ctx, repo, build)
+	}
+	return nil
+}

service/canceler/reaper/reaper_test.go

@@ -0,0 +1,264 @@
+// 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()
+
+// 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) {
+
+}

service/canceler/reaper/util.go

@@ -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),
+	)
+}

service/canceler/reaper/util_test.go

@@ -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 (
+	"time"
+	"testing"
+)
+
+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
+}