Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -6,3 +6,8 @@ func BuildRevisionURLMarkdown(repoFullName string, revision string) string {
// uses Markdown formatting to generate the link on GH
return fmt.Sprintf("[%s](https://github.com/%s/commit/%s)", revision, repoFullName, revision)
}

func BuildRunURLMarkdown(repoFullName string, revision string, runId int64) string {
// uses Markdown formatting to generate the link on GH
return fmt.Sprintf("[%s](https://github.com/%s/runs/%d)", revision, repoFullName, runId)
}
14 changes: 14 additions & 0 deletions server/neptune/workflows/internal/deploy/revision/queue/queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@ package queue
import (
"container/list"
"fmt"
"strings"

"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
activity "github.com/runatlantis/atlantis/server/neptune/workflows/activities/terraform"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/terraform"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/metrics"
Expand Down Expand Up @@ -85,6 +87,18 @@ func (q *Deploy) Push(msg terraform.DeploymentInfo) {
q.queue.Push(msg, Low)
}

func (q *Deploy) GetQueuedRevisionsSummary() string {
var revisions []string
if q.IsEmpty() {
return "No other runs ahead in queue."
}
for _, deploy := range q.Scan() {
runLink := github.BuildRunURLMarkdown(deploy.Repo.GetFullName(), deploy.Commit.Revision, deploy.CheckRunID)
revisions = append(revisions, runLink)
}
return fmt.Sprintf("Runs in queue: %s", strings.Join(revisions, ", "))
}

// priority is a simple 2 priority queue implementation
// priority is determined before an item enters a queue and does not change
type priority struct {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package queue

import (
"fmt"

key "github.com/runatlantis/atlantis/server/neptune/context"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"

Expand All @@ -24,12 +25,13 @@ func (u *LockStateUpdater) UpdateQueuedRevisions(ctx workflow.Context, queue *De

var actions []github.CheckRunAction
var summary string
var revisionsSummary string = queue.GetQueuedRevisionsSummary()
state := github.CheckRunQueued
if lock.Status == LockedStatus {
actions = append(actions, github.CreateUnlockAction())
state = github.CheckRunActionRequired
revisionLink := github.BuildRevisionURLMarkdown(repoFullName, lock.Revision)
summary = fmt.Sprintf("This deploy is locked from a manual deployment for revision %s. Unlock to proceed.", revisionLink)
summary = fmt.Sprintf("This deploy is locked from a manual deployment for revision %s. Unlock to proceed.\n%s", revisionLink, revisionsSummary)
}

for _, i := range infos {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
package queue_test

import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"
"testing"
"time"

"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"

"github.com/google/uuid"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
Expand All @@ -25,9 +26,13 @@ type testCheckRunClient struct {
}

func (t *testCheckRunClient) CreateOrUpdate(ctx workflow.Context, deploymentID string, request notifier.GithubCheckRunRequest) (int64, error) {
assert.Equal(t.expectedT, t.expectedRequest, request)
assert.Equal(t.expectedT, t.expectedDeploymentID, deploymentID)
switch {
case assert.Equal(t.expectedT, t.expectedRequest, request):

case assert.Equal(t.expectedT, t.expectedDeploymentID, deploymentID):
default:
t.expectedT.FailNow()
}
return 1, nil
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ type queue interface {
CanPop() bool
Pop() (terraform.DeploymentInfo, error)
SetLockForMergedItems(ctx workflow.Context, state LockState)
GetOrderedMergedItems() []terraform.DeploymentInfo
GetQueuedRevisionsSummary() string
}

type deployer interface {
Expand Down Expand Up @@ -96,7 +98,7 @@ func NewWorker(
},
}

tfWorkflowRunner := terraform.NewWorkflowRunner(tfWorkflow, notifiers, additionalNotifiers...)
tfWorkflowRunner := terraform.NewWorkflowRunner(q, tfWorkflow, githubCheckRunCache, notifiers, additionalNotifiers...)
deployer := &Deployer{
Activities: a,
TerraformWorkflowRunner: tfWorkflowRunner,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -47,10 +47,22 @@ func (q *testQueue) Push(msg internalTerraform.DeploymentInfo) {
q.Queue.PushBack(msg)
}

func (q *testQueue) GetOrderedMergedItems() []internalTerraform.DeploymentInfo {
var result []internalTerraform.DeploymentInfo
for e := q.Queue.Front(); e != nil; e = e.Next() {
result = append(result, e.Value.(internalTerraform.DeploymentInfo))
}
return result
}

func (q *testQueue) SetLockForMergedItems(ctx workflow.Context, state queue.LockState) {
q.Lock = state
}

func (q *testQueue) GetQueuedRevisionsSummary() string {
return "Revisions in queue"
}

type workerRequest struct {
Queue []internalTerraform.DeploymentInfo
ExpectedValidationErrors []*queue.ValidationError
Expand Down
6 changes: 4 additions & 2 deletions server/neptune/workflows/internal/deploy/revision/revision.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ type Queue interface {
GetLockState() queue.LockState
SetLockForMergedItems(ctx workflow.Context, state queue.LockState)
Scan() []terraform.DeploymentInfo
GetQueuedRevisionsSummary() string
}

type DeploymentStore interface {
Expand Down Expand Up @@ -136,14 +137,15 @@ func (n *Receiver) Receive(c workflow.ReceiveChannel, more bool) {
func (n *Receiver) createCheckRun(ctx workflow.Context, id, revision string, root activity.Root, repo github.Repo) int64 {
lock := n.queue.GetLockState()
var actions []github.CheckRunAction
summary := "This deploy is queued and will be processed as soon as possible."
var revisionsSummary string = n.queue.GetQueuedRevisionsSummary()
summary := "This deploy is queued and will be processed as soon as possible.\n" + revisionsSummary
state := github.CheckRunQueued

if lock.Status == queue.LockedStatus && (root.TriggerInfo.Type == activity.MergeTrigger) {
actions = append(actions, github.CreateUnlockAction())
state = github.CheckRunActionRequired
revisionLink := github.BuildRevisionURLMarkdown(repo.GetFullName(), lock.Revision)
summary = fmt.Sprintf("This deploy is locked from a manual deployment for revision %s. Unlock to proceed.", revisionLink)
summary = fmt.Sprintf("This deploy is locked from a manual deployment for revision %s. Unlock to proceed.\n%s", revisionLink, revisionsSummary)
}

cid, err := n.checkRunClient.CreateOrUpdate(ctx, id, notifier.GithubCheckRunRequest{
Expand Down
72 changes: 55 additions & 17 deletions server/neptune/workflows/internal/deploy/revision/revision_test.go
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
package revision_test

import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"
"fmt"
"strings"
"testing"
"time"

"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"

"github.com/google/uuid"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities/terraform"
Expand All @@ -23,8 +26,10 @@ type testCheckRunClient struct {
}

func (t *testCheckRunClient) CreateOrUpdate(ctx workflow.Context, deploymentID string, request notifier.GithubCheckRunRequest) (int64, error) {
assert.Equal(t.expectedT, t.expectedRequest, request)

ok := assert.Equal(t.expectedT, t.expectedRequest, request)
if !ok {
t.expectedT.FailNow()
}
return 1, nil
}

Expand All @@ -49,6 +54,21 @@ func (q *testQueue) SetLockForMergedItems(ctx workflow.Context, state queue.Lock
q.Lock = state
}

func (q *testQueue) IsEmpty() bool {
return len(q.Queue) == 0
}

func (q *testQueue) GetQueuedRevisionsSummary() string {
var revisions []string
if q.IsEmpty() {
return "No other revisions ahead In queue."
}
for _, deploy := range q.Scan() {
revisions = append(revisions, deploy.Commit.Revision)
}
return fmt.Sprintf("Revisions in queue: %s", strings.Join(revisions, ", "))
}

type testWorker struct {
Current queue.CurrentDeployment
}
Expand Down Expand Up @@ -137,10 +157,11 @@ func TestEnqueue(t *testing.T) {
env.ExecuteWorkflow(testWorkflow, req{
ID: id,
ExpectedRequest: notifier.GithubCheckRunRequest{
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Summary: "This deploy is queued and will be processed as soon as possible.\nNo other revisions ahead In queue.",
},
ExpectedT: t,
})
Expand Down Expand Up @@ -197,10 +218,11 @@ func TestEnqueue_ManualTrigger(t *testing.T) {
env.ExecuteWorkflow(testWorkflow, req{
ID: id,
ExpectedRequest: notifier.GithubCheckRunRequest{
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Summary: "This deploy is queued and will be processed as soon as possible.\nNo other revisions ahead In queue.",
},
ExpectedT: t,
})
Expand Down Expand Up @@ -263,10 +285,11 @@ func TestEnqueue_ManualTrigger_QueueAlreadyLocked(t *testing.T) {
Revision: "123334444555",
},
ExpectedRequest: notifier.GithubCheckRunRequest{
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
State: github.CheckRunQueued,
Summary: "This deploy is queued and will be processed as soon as possible.\nNo other revisions ahead In queue.",
},
ExpectedT: t,
})
Expand Down Expand Up @@ -321,8 +344,22 @@ func TestEnqueue_MergeTrigger_QueueAlreadyLocked(t *testing.T) {

id := uuid.Must(uuid.NewUUID())

deploymentInfo := terraformWorkflow.DeploymentInfo{
Commit: github.Commit{
Revision: "123334444555",
Branch: "locking-branch",
},
CheckRunID: 0,
Root: terraform.Root{Name: "root", TriggerInfo: terraform.TriggerInfo{
Type: terraform.MergeTrigger,
}, Trigger: terraform.MergeTrigger},
ID: id,
Repo: github.Repo{Name: "nish"},
}

env.ExecuteWorkflow(testWorkflow, req{
ID: id,
ID: id,
InitialElements: []terraformWorkflow.DeploymentInfo{deploymentInfo},
Lock: queue.LockState{
// ensure that the lock gets updated
Status: queue.LockedStatus,
Expand All @@ -332,7 +369,7 @@ func TestEnqueue_MergeTrigger_QueueAlreadyLocked(t *testing.T) {
Title: "atlantis/deploy: root",
Sha: rev,
Repo: github.Repo{Name: "nish"},
Summary: "This deploy is locked from a manual deployment for revision [123334444555](https://github.com//nish/commit/123334444555). Unlock to proceed.",
Summary: "This deploy is locked from a manual deployment for revision [123334444555](https://github.com//nish/commit/123334444555). Unlock to proceed.\nRevisions in queue: 123334444555",
Actions: []github.CheckRunAction{github.CreateUnlockAction()},
State: github.CheckRunActionRequired,
},
Expand All @@ -346,6 +383,7 @@ func TestEnqueue_MergeTrigger_QueueAlreadyLocked(t *testing.T) {
assert.NoError(t, err)

assert.Equal(t, []terraformWorkflow.DeploymentInfo{
deploymentInfo,
{
Commit: github.Commit{
Revision: rev,
Expand Down
9 changes: 8 additions & 1 deletion server/neptune/workflows/internal/deploy/terraform/runner.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,10 +33,17 @@ type stateReceiver interface {
Receive(ctx workflow.Context, c workflow.ReceiveChannel, deploymentInfo DeploymentInfo)
}

func NewWorkflowRunner(w Workflow, internalNotifiers []WorkflowNotifier, additionalNotifiers ...plugins.TerraformWorkflowNotifier) *WorkflowRunner {
type deployQueue interface {
GetOrderedMergedItems() []DeploymentInfo
GetQueuedRevisionsSummary() string
}

func NewWorkflowRunner(queue deployQueue, w Workflow, githubCheckRunCache CheckRunClient, internalNotifiers []WorkflowNotifier, additionalNotifiers ...plugins.TerraformWorkflowNotifier) *WorkflowRunner {
return &WorkflowRunner{
Workflow: w,
StateReceiver: &StateReceiver{
Queue: queue,
CheckRunCache: githubCheckRunCache,
InternalNotifiers: internalNotifiers,
AdditionalNotifiers: additionalNotifiers,
},
Expand Down
40 changes: 39 additions & 1 deletion server/neptune/workflows/internal/deploy/terraform/state.go
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
package terraform

import (
"fmt"
"reflect"

"github.com/pkg/errors"
"github.com/runatlantis/atlantis/server/metrics"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/notifier"

"github.com/runatlantis/atlantis/server/neptune/workflows/internal/terraform/state"
"github.com/runatlantis/atlantis/server/neptune/workflows/plugins"
"go.temporal.io/sdk/workflow"
Expand All @@ -14,10 +17,16 @@ type WorkflowNotifier interface {
Notify(workflow.Context, notifier.Info, *state.Workflow) error
}

type CheckRunClient interface {
CreateOrUpdate(ctx workflow.Context, deploymentID string, request notifier.GithubCheckRunRequest) (int64, error)
}

type StateReceiver struct {

// We have separate classes of notifiers since we can be more flexible with our internal ones in terms of the data model
// What we support externally should be well thought out so for now this is kept to a minimum.
Queue deployQueue
CheckRunCache CheckRunClient
InternalNotifiers []WorkflowNotifier
AdditionalNotifiers []plugins.TerraformWorkflowNotifier
}
Expand All @@ -39,6 +48,35 @@ func (n *StateReceiver) Receive(ctx workflow.Context, c workflow.ReceiveChannel,
}
}

if workflowState.Apply != nil &&
workflowState.Apply.Status == state.WaitingJobStatus &&
!reflect.ValueOf(workflowState.Apply.OnWaitingActions).IsZero() {
// update queue with information about current deployment pending confirm/reject action
infos := n.Queue.GetOrderedMergedItems()

revisionsSummary := n.Queue.GetQueuedRevisionsSummary()
state := github.CheckRunQueued
runLink := github.BuildRunURLMarkdown(deploymentInfo.Repo.GetFullName(), deploymentInfo.Commit.Revision, deploymentInfo.CheckRunID)
summary := fmt.Sprintf("This deploy is queued pending action on run for revision %s.\n%s", runLink, revisionsSummary)

for _, i := range infos {
request := notifier.GithubCheckRunRequest{
Title: notifier.BuildDeployCheckRunTitle(i.Root.Name),
Sha: i.Commit.Revision,
State: state,
Repo: i.Repo,
Summary: summary,
}

workflow.GetLogger(ctx).Debug(fmt.Sprintf("Updating action pending summary for deployment id: %s", i.ID.String()))
_, err := n.CheckRunCache.CreateOrUpdate(ctx, i.ID.String(), request)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this might pose some non-determinism errors, see https://community.temporal.io/t/workflow-determinism/4027, but I could be wrong. If it does, let's add some workflow versioning here


if err != nil {
workflow.GetLogger(ctx).Debug(fmt.Sprintf("updating check run for revision %s", i.Commit.Revision), err)
}
}
}

for _, notifier := range n.InternalNotifiers {
if err := notifier.Notify(ctx, deploymentInfo.ToInternalInfo(), workflowState); err != nil {
workflow.GetMetricsHandler(ctx).Counter("notifier_failure").Inc(1)
Expand Down
Loading
Loading