Skip to content

Commit

Permalink
move lockstate struct to separate package so terraform package can us…
Browse files Browse the repository at this point in the history
…e it without cyclic dependency
  • Loading branch information
tlin4194 committed Nov 9, 2024
1 parent d5ef226 commit bee9ab2
Show file tree
Hide file tree
Showing 14 changed files with 113 additions and 88 deletions.
15 changes: 15 additions & 0 deletions server/neptune/workflows/internal/deploy/lock/lock.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
package lock

type LockStatus int

type LockState struct {
Revision string
Status LockStatus
}

const (
UnlockedStatus LockStatus = iota
LockedStatus

QueueDepthStat = "queue.depth"
)
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/activities"
"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/revision/queue"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
"github.com/slack-go/slack"
"go.temporal.io/sdk/workflow"
)
Expand All @@ -24,7 +25,7 @@ type Slack struct {
func (s *Slack) Notify(ctx workflow.Context) error {
state := s.DeployQueue.GetLockState()

if state.Status == queue.UnlockedStatus {
if state.Status == lock.UnlockedStatus {
return nil
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/revision/notifier"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/revision/queue"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/terraform"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/metrics"
"github.com/slack-go/slack"
"github.com/stretchr/testify/assert"
Expand All @@ -20,7 +21,7 @@ import (
)

type request struct {
LockState queue.LockState
LockState lock.LockState
InitialItems []terraform.DeploymentInfo
}

Expand Down Expand Up @@ -51,7 +52,7 @@ func testWorkflow(ctx workflow.Context, request request) error {

func TestNotifier(t *testing.T) {
t.Run("empty queue", func(t *testing.T) {
state := queue.LockState{Status: queue.UnlockedStatus}
state := lock.LockState{Status: lock.UnlockedStatus}
ts := testsuite.WorkflowTestSuite{}
env := ts.NewTestWorkflowEnvironment()

Expand All @@ -68,7 +69,7 @@ func TestNotifier(t *testing.T) {
})

t.Run("locked state", func(t *testing.T) {
state := queue.LockState{Status: queue.LockedStatus}
state := lock.LockState{Status: lock.LockedStatus}
ts := testsuite.WorkflowTestSuite{}
env := ts.NewTestWorkflowEnvironment()

Expand All @@ -85,7 +86,7 @@ func TestNotifier(t *testing.T) {
})

t.Run("no slack config", func(t *testing.T) {
state := queue.LockState{Status: queue.LockedStatus}
state := lock.LockState{Status: lock.LockedStatus}
ts := testsuite.WorkflowTestSuite{}
env := ts.NewTestWorkflowEnvironment()

Expand Down Expand Up @@ -121,7 +122,7 @@ func TestNotifier(t *testing.T) {
})

t.Run("activity called", func(t *testing.T) {
state := queue.LockState{Status: queue.LockedStatus}
state := lock.LockState{Status: lock.LockedStatus}
ts := testsuite.WorkflowTestSuite{}
env := ts.NewTestWorkflowEnvironment()

Expand Down
29 changes: 8 additions & 21 deletions server/neptune/workflows/internal/deploy/revision/queue/queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,20 +10,7 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/terraform"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/metrics"
"go.temporal.io/sdk/workflow"
)

type LockStatus int

type LockState struct {
Revision string
Status LockStatus
}

const (
UnlockedStatus LockStatus = iota
LockedStatus

QueueDepthStat = "queue.depth"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

type Deploy struct {
Expand All @@ -32,7 +19,7 @@ type Deploy struct {
scope metrics.Scope

// mutable: default is unlocked
lock LockState
lock lock.LockState
}

func NewQueue(callback func(workflow.Context, *Deploy), scope metrics.Scope) *Deploy {
Expand All @@ -43,12 +30,12 @@ func NewQueue(callback func(workflow.Context, *Deploy), scope metrics.Scope) *De
}
}

func (q *Deploy) GetLockState() LockState {
func (q *Deploy) GetLockState() lock.LockState {
return q.lock
}

func (q *Deploy) SetLockForMergedItems(ctx workflow.Context, state LockState) {
if state.Status == LockedStatus {
func (q *Deploy) SetLockForMergedItems(ctx workflow.Context, state lock.LockState) {
if state.Status == lock.LockedStatus {
q.scope.Counter("locked").Inc(1)
} else {
q.scope.Counter("unlocked").Inc(1)
Expand All @@ -58,11 +45,11 @@ func (q *Deploy) SetLockForMergedItems(ctx workflow.Context, state LockState) {
}

func (q *Deploy) CanPop() bool {
return q.queue.HasItemsOfPriority(High) || (q.lock.Status == UnlockedStatus && !q.queue.IsEmpty())
return q.queue.HasItemsOfPriority(High) || (q.lock.Status == lock.UnlockedStatus && !q.queue.IsEmpty())
}

func (q *Deploy) Pop() (terraform.DeploymentInfo, error) {
defer q.scope.Gauge(QueueDepthStat).Update(float64(q.queue.Size()))
defer q.scope.Gauge(lock.QueueDepthStat).Update(float64(q.queue.Size()))
return q.queue.Pop()
}

Expand All @@ -79,7 +66,7 @@ func (q *Deploy) IsEmpty() bool {
}

func (q *Deploy) Push(msg terraform.DeploymentInfo) {
defer q.scope.Gauge(QueueDepthStat).Update(float64(q.queue.Size()))
defer q.scope.Gauge(lock.QueueDepthStat).Update(float64(q.queue.Size()))
if msg.Root.TriggerInfo.Type == activity.ManualTrigger {
q.queue.Push(msg, High)
return
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/test"
"github.com/stretchr/testify/assert"
"go.temporal.io/sdk/workflow"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

func noopCallback(ctx workflow.Context, q *queue.Deploy) {}
Expand Down Expand Up @@ -38,8 +39,8 @@ func TestQueue(t *testing.T) {
q := queue.NewQueue(func(ctx workflow.Context, d *queue.Deploy) {
called = true
}, metrics.NewNullableScope())
q.SetLockForMergedItems(test.Background(), queue.LockState{
Status: queue.LockedStatus,
q.SetLockForMergedItems(test.Background(), lock.LockState{
Status: lock.LockedStatus,
})

assert.True(t, called)
Expand All @@ -52,17 +53,17 @@ func TestQueue(t *testing.T) {

t.Run("can pop empty queue locked", func(t *testing.T) {
q := queue.NewQueue(noopCallback, metrics.NewNullableScope())
q.SetLockForMergedItems(test.Background(), queue.LockState{
Status: queue.LockedStatus,
q.SetLockForMergedItems(test.Background(), lock.LockState{
Status: lock.LockedStatus,
})
assert.Equal(t, false, q.CanPop())
})
t.Run("can pop manual trigger locked", func(t *testing.T) {
q := queue.NewQueue(noopCallback, metrics.NewNullableScope())
msg1 := wrap("1", activity.ManualTrigger)
q.Push(msg1)
q.SetLockForMergedItems(test.Background(), queue.LockState{
Status: queue.LockedStatus,
q.SetLockForMergedItems(test.Background(), lock.LockState{
Status: lock.LockedStatus,
})
assert.Equal(t, true, q.CanPop())
})
Expand All @@ -76,8 +77,8 @@ func TestQueue(t *testing.T) {
q := queue.NewQueue(noopCallback, metrics.NewNullableScope())
msg1 := wrap("1", activity.MergeTrigger)
q.Push(msg1)
q.SetLockForMergedItems(test.Background(), queue.LockState{
Status: queue.LockedStatus,
q.SetLockForMergedItems(test.Background(), lock.LockState{
Status: lock.LockedStatus,
})
assert.Equal(t, false, q.CanPop())
})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (

"github.com/runatlantis/atlantis/server/neptune/workflows/activities/github"
"go.temporal.io/sdk/workflow"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

type CheckRunClient interface {
Expand All @@ -20,17 +21,17 @@ type LockStateUpdater struct {
}

func (u *LockStateUpdater) UpdateQueuedRevisions(ctx workflow.Context, queue *Deploy, repoFullName string) {
lock := queue.GetLockState()
queueLock := queue.GetLockState()
infos := queue.GetOrderedMergedItems()

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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"github.com/stretchr/testify/mock"
"go.temporal.io/sdk/testsuite"
"go.temporal.io/sdk/workflow"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

type testCheckRunClient struct {
Expand Down Expand Up @@ -126,8 +127,8 @@ func TestLockStateUpdater_locked_new_version(t *testing.T) {

env.ExecuteWorkflow(testUpdaterWorkflow, updaterReq{
Queue: []terraform.DeploymentInfo{info},
Lock: queue.LockState{
Status: queue.LockedStatus,
Lock: lock.LockState{
Status: lock.LockedStatus,
Revision: "1234",
},
ExpectedRequest: notifier.GithubCheckRunRequest{
Expand All @@ -152,7 +153,7 @@ func TestLockStateUpdater_locked_new_version(t *testing.T) {

type updaterReq struct {
Queue []terraform.DeploymentInfo
Lock queue.LockState
Lock lock.LockState
ExpectedRequest notifier.GithubCheckRunRequest
ExpectedDeploymentID string
ExpectedT *testing.T
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,17 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/plugins"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

type queue interface {
IsEmpty() bool
CanPop() bool
Pop() (terraform.DeploymentInfo, error)
SetLockForMergedItems(ctx workflow.Context, state LockState)
SetLockForMergedItems(ctx workflow.Context, state lock.LockState)
GetOrderedMergedItems() []terraform.DeploymentInfo
GetQueuedRevisionsSummary() string
GetLockState() lock.LockState
}

type deployer interface {
Expand Down Expand Up @@ -114,8 +116,8 @@ func NewWorker(
// we don't persist lock state anywhere so in the case of workflow completion we need to rebuild
// the lock state
if latestDeployment != nil && latestDeployment.Root.Trigger == string(tfModel.ManualTrigger) {
q.SetLockForMergedItems(ctx, LockState{
Status: LockedStatus,
q.SetLockForMergedItems(ctx, lock.LockState{
Status: lock.LockedStatus,
Revision: latestDeployment.Revision,
})
}
Expand Down Expand Up @@ -181,8 +183,8 @@ func (w *Worker) Work(ctx workflow.Context) {
workflow.GetMetricsHandler(ctx).WithTags(map[string]string{metricNames.SignalNameTag: UnlockSignalName}).
Counter(metricNames.SignalReceive).
Inc(1)
w.Queue.SetLockForMergedItems(ctx, LockState{
Status: UnlockedStatus,
w.Queue.SetLockForMergedItems(ctx, lock.LockState{
Status: lock.UnlockedStatus,
})
continue
default:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/revision/queue"
internalTerraform "github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/terraform"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/metrics"
"github.com/runatlantis/atlantis/server/neptune/workflows/internal/deploy/lock"
)

type testBlockingDeployer struct{}
Expand Down Expand Up @@ -47,7 +48,7 @@ func testStateWorkflow(ctx workflow.Context, r workerRequest) (workerResponse, e
Queue: list.New(),
}

q.SetLockForMergedItems(ctx, queue.LockState{Status: r.InitialLockStatus})
q.SetLockForMergedItems(ctx, lock.LockState{Status: r.InitialLockStatus})

workflow.Go(ctx, func(ctx workflow.Context) {
ch := workflow.GetSignalChannel(ctx, "add-to-queue")
Expand Down Expand Up @@ -109,8 +110,8 @@ func TestWorker_StartsWithEmptyQueue(t *testing.T) {
assert.NoError(t, err)

assert.True(t, q.QueueIsEmpty)
assert.Equal(t, queue.LockState{
Status: queue.UnlockedStatus,
assert.Equal(t, lock.LockState{
Status: lock.UnlockedStatus,
}, q.Lock)
assert.Equal(t, queue.WaitingWorkerState, q.State)
}, 2*time.Second)
Expand Down Expand Up @@ -145,8 +146,8 @@ func TestWorker_StartsWithEmptyQueue(t *testing.T) {
assert.NoError(t, err)

assert.True(t, q.QueueIsEmpty)
assert.Equal(t, queue.LockState{
Status: queue.UnlockedStatus,
assert.Equal(t, lock.LockState{
Status: lock.UnlockedStatus,
}, q.Lock)
assert.Equal(t, queue.WorkingWorkerState, q.State)
}, 6*time.Second)
Expand Down
Loading

0 comments on commit bee9ab2

Please sign in to comment.