IntenseWebs/opentofu
3
0
Fork 0
mirror of https://github.com/opentofu/opentofu.git synced 2025-02-25 18:45:20 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor task results to run as a task stage

Browse Source
This commit is contained in:
mrinalirao 2022-11-29 11:57:51 +11:00
parent 2b14670dfd
commit d7c7f3689c
8 changed files with 322 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
go.mod
View File

@ -39,7 +39,7 @@ require (
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/go-plugin v1.4.3
github.com/hashicorp/go-retryablehttp v0.7.1
github.com/hashicorp/go-tfe v1.12.0
github.com/hashicorp/go-tfe v1.14.0
github.com/hashicorp/go-uuid v1.0.3
github.com/hashicorp/go-version v1.6.0
github.com/hashicorp/hcl v0.0.0-20170504190234-a4b07c25de5f
@ -145,7 +145,7 @@ require (
github.com/hashicorp/go-msgpack v0.5.4 // indirect
github.com/hashicorp/go-rootcerts v1.0.2 // indirect
github.com/hashicorp/go-safetemp v1.0.0 // indirect
github.com/hashicorp/go-slug v0.10.0 // indirect
github.com/hashicorp/go-slug v0.10.1 // indirect
github.com/hashicorp/golang-lru v0.5.1 // indirect
github.com/hashicorp/jsonapi v0.0.0-20210826224640-ee7dae0fb22d // indirect
github.com/hashicorp/serf v0.9.5 // indirect

8
go.sum
View File

@ -371,13 +371,13 @@ github.com/hashicorp/go-rootcerts v1.0.2 h1:jzhAVGtqPKbwpyCPELlgNWhE1znq+qwJtW5O
github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
github.com/hashicorp/go-safetemp v1.0.0 h1:2HR189eFNrjHQyENnQMMpCiBAsRxzbTMIgBhEyExpmo=
github.com/hashicorp/go-safetemp v1.0.0/go.mod h1:oaerMy3BhqiTbVye6QuFhFtIceqFoDHxNAB65b+Rj1I=
github.com/hashicorp/go-slug v0.10.0 h1:mh4DDkBJTh9BuEjY/cv8PTo7k9OjT4PcW8PgZnJ4jTY=
github.com/hashicorp/go-slug v0.10.0/go.mod h1:Ib+IWBYfEfJGI1ZyXMGNbu2BU+aa3Dzu41RKLH301v4=
github.com/hashicorp/go-slug v0.10.1 h1:05SCRWCBpCxOeP7stQHvMgOz0raCBCekaytu8Rg/RZ4=
github.com/hashicorp/go-slug v0.10.1/go.mod h1:Ib+IWBYfEfJGI1ZyXMGNbu2BU+aa3Dzu41RKLH301v4=
github.com/hashicorp/go-sockaddr v1.0.0 h1:GeH6tui99pF4NJgfnhp+L6+FfobzVW3Ah46sLo0ICXs=
github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
github.com/hashicorp/go-tfe v1.12.0 h1:2l7emKW8rNTTbnxYHNVj6b46iJzOEp2G/3xIHfGSDnc=
github.com/hashicorp/go-tfe v1.12.0/go.mod h1:thYtIxtgBpDDNdf/2yYPdBJ94Fz5yT5XCNZvGtTGHAU=
github.com/hashicorp/go-tfe v1.14.0 h1:FZKKkwlyTxw8/OE3e7NiFQLcgGXTHra9ogGhMTotxh8=
github.com/hashicorp/go-tfe v1.14.0/go.mod h1:77snluBqtTTvMrY0w/mxQA5jlHQ8NT44AqQ8UdrPf0o=
github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=

2
internal/cloud/backend_common.go
View File

@ -207,7 +207,7 @@ func (b *Cloud) waitTaskStage(stopCtx, cancelCtx context.Context, op *backend.Op
Op: op,
Run: r,
}
return b.runTasks(integration, integration.BeginOutput(outputTitle), stageID)
return b.runTaskStage(integration, integration.BeginOutput(outputTitle), stageID)
}
func (b *Cloud) costEstimate(stopCtx, cancelCtx context.Context, op *backend.Operation, r *tfe.Run) error {

149
internal/cloud/backend_runTasks.go
View File

@ -1,149 +0,0 @@
package cloud
import (
"context"
"fmt"
"strings"
"github.com/hashicorp/go-tfe"
)
type taskResultSummary struct {
unreachable bool
pending int
failed int
failedMandatory int
passed int
}
type taskStageReadFunc func(b *Cloud, stopCtx context.Context) (*tfe.TaskStage, error)
func summarizeTaskResults(taskResults []*tfe.TaskResult) *taskResultSummary {
var pendingCount, errCount, errMandatoryCount, passedCount int
for _, task := range taskResults {
if task.Status == "unreachable" {
return &taskResultSummary{
unreachable: true,
}
} else if task.Status == "running" || task.Status == "pending" {
pendingCount++
} else if task.Status == "passed" {
passedCount++
} else {
// Everything else is a failure
errCount++
if task.WorkspaceTaskEnforcementLevel == "mandatory" {
errMandatoryCount++
}
}
}
return &taskResultSummary{
unreachable: false,
pending: pendingCount,
failed: errCount,
failedMandatory: errMandatoryCount,
passed: passedCount,
}
}
func (b *Cloud) runTasksWithTaskResults(context *IntegrationContext, output IntegrationOutputWriter, fetchTaskStage taskStageReadFunc) error {
return context.Poll(func(i int) (bool, error) {
stage, err := fetchTaskStage(b, context.StopContext)
if err != nil {
return false, generalError("Failed to retrieve task stage", err)
}
summary := summarizeTaskResults(stage.TaskResults)
if summary.unreachable {
output.Output("Skipping task results.")
output.End()
return false, nil
}
if summary.pending > 0 {
pendingMessage := "%d tasks still pending, %d passed, %d failed ... "
message := fmt.Sprintf(pendingMessage, summary.pending, summary.passed, summary.failed)
if i%4 == 0 {
if i > 0 {
output.OutputElapsed(message, len(pendingMessage)) // Up to 2 digits are allowed by the max message allocation
}
}
return true, nil
}
// No more tasks pending/running. Print all the results.
// Track the first task name that is a mandatory enforcement level breach.
var firstMandatoryTaskFailed *string = nil
if i == 0 {
output.Output(fmt.Sprintf("All tasks completed! %d passed, %d failed", summary.passed, summary.failed))
} else {
output.OutputElapsed(fmt.Sprintf("All tasks completed! %d passed, %d failed", summary.passed, summary.failed), 50)
}
output.Output("")
for _, t := range stage.TaskResults {
capitalizedStatus := string(t.Status)
capitalizedStatus = strings.ToUpper(capitalizedStatus[:1]) + capitalizedStatus[1:]
status := "[green]" + capitalizedStatus
if t.Status != "passed" {
level := string(t.WorkspaceTaskEnforcementLevel)
level = strings.ToUpper(level[:1]) + level[1:]
status = fmt.Sprintf("[red]%s (%s)", capitalizedStatus, level)
if t.WorkspaceTaskEnforcementLevel == "mandatory" && firstMandatoryTaskFailed == nil {
firstMandatoryTaskFailed = &t.TaskName
}
}
title := fmt.Sprintf(`%s ⸺ %s`, t.TaskName, status)
output.SubOutput(title)
if len(t.Message) > 0 {
output.SubOutput(fmt.Sprintf("[dim]%s", t.Message))
}
if len(t.URL) > 0 {
output.SubOutput(fmt.Sprintf("[dim]Details: %s", t.URL))
}
output.SubOutput("")
}
// If a mandatory enforcement level is breached, return an error.
var taskErr error = nil
var overall string = "[green]Passed"
if firstMandatoryTaskFailed != nil {
overall = "[red]Failed"
if summary.failedMandatory > 1 {
taskErr = fmt.Errorf("the run failed because %d mandatory tasks are required to succeed", summary.failedMandatory)
} else {
taskErr = fmt.Errorf("the run failed because the run task, %s, is required to succeed", *firstMandatoryTaskFailed)
}
} else if summary.failed > 0 { // we have failures but none of them mandatory
overall = "[green]Passed with advisory failures"
}
output.SubOutput("")
output.SubOutput("[bold]Overall Result: " + overall)
output.End()
return false, taskErr
})
}
func (b *Cloud) runTasks(ctx *IntegrationContext, output IntegrationOutputWriter, stageID string) error {
return b.runTasksWithTaskResults(ctx, output, func(b *Cloud, stopCtx context.Context) (*tfe.TaskStage, error) {
options := tfe.TaskStageReadOptions{
Include: []tfe.TaskStageIncludeOpt{tfe.TaskStageTaskResults},
}
return b.client.TaskStages.Read(ctx.StopContext, stageID, &options)
})
}

147
internal/cloud/backend_taskStage_taskResults.go Normal file
View File

@ -0,0 +1,147 @@
package cloud
import (
"fmt"
"strings"
"github.com/hashicorp/go-tfe"
)
type taskResultSummary struct {
unreachable bool
pending int
failed int
failedMandatory int
passed int
}
type taskResultSummarizer struct {
finished bool
cloud *Cloud
counter int
}
func newTaskResultSummarizer(b *Cloud, ts *tfe.TaskStage) taskStageSummarizer {
if len(ts.TaskResults) == 0 {
return nil
}
return &taskResultSummarizer{
finished: false,
cloud: b,
}
}
func (trs *taskResultSummarizer) Summarize(context *IntegrationContext, output IntegrationOutputWriter, ts *tfe.TaskStage) (bool, *string, error) {
if trs.finished {
return false, nil, nil
}
trs.counter++
counts := summarizeTaskResults(ts.TaskResults)
if counts.pending != 0 {
pendingMessage := "%d tasks still pending, %d passed, %d failed ... "
message := fmt.Sprintf(pendingMessage, counts.pending, counts.passed, counts.failed)
return true, &message, nil
}
if counts.unreachable {
output.Output("Skipping task results.")
output.End()
return false, nil, nil
}
// Print out the summary
trs.runTasksWithTaskResults(output, ts.TaskResults, counts)
// Mark as finished
trs.finished = true
return false, nil, nil
}
func summarizeTaskResults(taskResults []*tfe.TaskResult) *taskResultSummary {
var pendingCount, errCount, errMandatoryCount, passedCount int
for _, task := range taskResults {
if task.Status == "unreachable" {
return &taskResultSummary{
unreachable: true,
}
} else if task.Status == "running" || task.Status == "pending" {
pendingCount++
} else if task.Status == "passed" {
passedCount++
} else {
// Everything else is a failure
errCount++
if task.WorkspaceTaskEnforcementLevel == "mandatory" {
errMandatoryCount++
}
}
}
return &taskResultSummary{
unreachable: false,
pending: pendingCount,
failed: errCount,
failedMandatory: errMandatoryCount,
passed: passedCount,
}
}
func (trs *taskResultSummarizer) runTasksWithTaskResults(output IntegrationOutputWriter, taskResults []*tfe.TaskResult, count *taskResultSummary) {
// Track the first task name that is a mandatory enforcement level breach.
var firstMandatoryTaskFailed *string = nil
if trs.counter == 0 {
output.Output(fmt.Sprintf("All tasks completed! %d passed, %d failed", count.passed, count.failed))
} else {
output.OutputElapsed(fmt.Sprintf("All tasks completed! %d passed, %d failed", count.passed, count.failed), 50)
}
output.Output("")
for _, t := range taskResults {
capitalizedStatus := string(t.Status)
capitalizedStatus = strings.ToUpper(capitalizedStatus[:1]) + capitalizedStatus[1:]
status := "[green]" + capitalizedStatus
if t.Status != "passed" {
level := string(t.WorkspaceTaskEnforcementLevel)
level = strings.ToUpper(level[:1]) + level[1:]
status = fmt.Sprintf("[red]%s (%s)", capitalizedStatus, level)
if t.WorkspaceTaskEnforcementLevel == "mandatory" && firstMandatoryTaskFailed == nil {
firstMandatoryTaskFailed = &t.TaskName
}
}
title := fmt.Sprintf(`%s ⸺ %s`, t.TaskName, status)
output.SubOutput(title)
if len(t.Message) > 0 {
output.SubOutput(fmt.Sprintf("[dim]%s", t.Message))
}
if len(t.URL) > 0 {
output.SubOutput(fmt.Sprintf("[dim]Details: %s", t.URL))
}
output.SubOutput("")
}
// If a mandatory enforcement level is breached, return an error.
var overall string = "[green]Passed"
if firstMandatoryTaskFailed != nil {
overall = "[red]Failed"
if count.failedMandatory > 1 {
output.Output(fmt.Sprintf("[reset][bold][red]Error:[reset][bold]the run failed because %d mandatory tasks are required to succeed", count.failedMandatory))
} else {
output.Output(fmt.Sprintf("[reset][bold][red]Error: [reset][bold]the run failed because the run task, %s, is required to succeed", *firstMandatoryTaskFailed))
}
} else if count.failed > 0 { // we have failures but none of them mandatory
overall = "[green]Passed with advisory failures"
}
output.SubOutput("")
output.SubOutput("[bold]Overall Result: " + overall)
output.End()
}

77
internal/cloud/backend_runTasks_test.go → internal/cloud/backend_taskStage_taskResults_test.go
View File

@ -82,16 +82,20 @@ func TestCloud_runTasksWithTaskResults(t *testing.T) {
integrationContext, writer := newMockIntegrationContext(b, t)
cases := map[string]struct {
taskResults []*tfe.TaskResult
taskStage func() *tfe.TaskStage
context *IntegrationContext
writer *testIntegrationOutput
expectedOutputs []string
isError bool
}{
"all-succeeded": {
taskResults: []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "advisory"},
taskStage: func() *tfe.TaskStage {
ts := &tfe.TaskStage{}
ts.TaskResults = []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "advisory"},
}
return ts
},
writer: writer,
context: integrationContext,
@ -99,9 +103,13 @@ func TestCloud_runTasksWithTaskResults(t *testing.T) {
isError: false,
},
"mandatory-failed": {
taskResults: []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "500 Error", Status: "failed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "advisory"},
taskStage: func() *tfe.TaskStage {
ts := &tfe.TaskStage{}
ts.TaskResults = []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "500 Error", Status: "failed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "advisory"},
}
return ts
},
writer: writer,
context: integrationContext,
@ -109,9 +117,13 @@ func TestCloud_runTasksWithTaskResults(t *testing.T) {
isError: true,
},
"advisory-failed": {
taskResults: []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "500 Error", Status: "failed", WorkspaceTaskEnforcementLevel: "advisory"},
taskStage: func() *tfe.TaskStage {
ts := &tfe.TaskStage{}
ts.TaskResults = []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "A-OK", Status: "passed", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "500 Error", Status: "failed", WorkspaceTaskEnforcementLevel: "advisory"},
}
return ts
},
writer: writer,
context: integrationContext,
@ -119,9 +131,13 @@ func TestCloud_runTasksWithTaskResults(t *testing.T) {
isError: false,
},
"unreachable": {
taskResults: []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "", Status: "unreachable", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "", Status: "unreachable", WorkspaceTaskEnforcementLevel: "advisory"},
taskStage: func() *tfe.TaskStage {
ts := &tfe.TaskStage{}
ts.TaskResults = []*tfe.TaskResult{
{ID: "1", TaskName: "Mandatory", Message: "", Status: "unreachable", WorkspaceTaskEnforcementLevel: "mandatory"},
{ID: "2", TaskName: "Advisory", Message: "", Status: "unreachable", WorkspaceTaskEnforcementLevel: "advisory"},
}
return ts
},
writer: writer,
context: integrationContext,
@ -130,27 +146,24 @@ func TestCloud_runTasksWithTaskResults(t *testing.T) {
},
}
for caseName, c := range cases {
for _, c := range cases {
c.writer.output.Reset()
err := b.runTasksWithTaskResults(c.context, writer, func(b *Cloud, stopCtx context.Context) (*tfe.TaskStage, error) {
return &tfe.TaskStage{
TaskResults: c.taskResults,
}, nil
})
if c.isError && err == nil {
t.Fatalf("Expected %s to be error", caseName)
trs := taskResultSummarizer{
cloud: b,
}
if !c.isError && err != nil {
t.Errorf("Expected %s to not be error but received %s", caseName, err)
}
output := c.writer.output.String()
for _, expected := range c.expectedOutputs {
if !strings.Contains(output, expected) {
t.Fatalf("Expected output to contain '%s' but it was:\n\n%s", expected, output)
c.context.Poll(func(i int) (bool, error) {
cont, _, _ := trs.Summarize(c.context, c.writer, c.taskStage())
if cont {
return true, nil
}
}
output := c.writer.output.String()
for _, expected := range c.expectedOutputs {
if !strings.Contains(output, expected) {
t.Fatalf("Expected output to contain '%s' but it was:\n\n%s", expected, output)
}
}
return false, nil
})
}
}

80
internal/cloud/backend_taskStages.go
View File

@ -2,6 +2,7 @@ package cloud
import (
"context"
"fmt"
"strings"
tfe "github.com/hashicorp/go-tfe"
@ -9,6 +10,16 @@ import (
type taskStages map[tfe.Stage]*tfe.TaskStage
type taskStageSummarizer interface {
// Summarize takes an IntegrationContext, IntegrationOutputWriter for
// writing output and a pointer to a tfe.TaskStage object as arguments.
// This function summarizes and outputs the results of the task stage.
// It returns a boolean which signifies whether we should continue polling
// for results, an optional message string to print while it is polling
// and an error if any.
Summarize(*IntegrationContext, IntegrationOutputWriter, *tfe.TaskStage) (bool, *string, error)
}
func (b *Cloud) runTaskStages(ctx context.Context, client *tfe.Client, runId string) (taskStages, error) {
taskStages := make(taskStages, 0)
result, err := client.Runs.ReadWithOptions(ctx, runId, &tfe.RunReadOptions{
@ -30,3 +41,72 @@ func (b *Cloud) runTaskStages(ctx context.Context, client *tfe.Client, runId str
return taskStages, nil
}
func (b *Cloud) getTaskStageWithAllOptions(ctx *IntegrationContext, stageID string) (*tfe.TaskStage, error) {
options := tfe.TaskStageReadOptions{
Include: []tfe.TaskStageIncludeOpt{tfe.TaskStageTaskResults},
}
stage, err := b.client.TaskStages.Read(ctx.StopContext, stageID, &options)
if err != nil {
return nil, generalError("Failed to retrieve task stage", err)
} else {
return stage, nil
}
}
func (b *Cloud) runTaskStage(ctx *IntegrationContext, output IntegrationOutputWriter, stageID string) error {
var errs multiErrors
// Create our summarizers
summarizers := make([]taskStageSummarizer, 0)
ts, err := b.getTaskStageWithAllOptions(ctx, stageID)
if err != nil {
return err
}
if s := newTaskResultSummarizer(b, ts); s != nil {
summarizers = append(summarizers, s)
}
return ctx.Poll(func(i int) (bool, error) {
options := tfe.TaskStageReadOptions{
Include: []tfe.TaskStageIncludeOpt{tfe.TaskStageTaskResults},
}
stage, err := b.client.TaskStages.Read(ctx.StopContext, stageID, &options)
if err != nil {
return false, generalError("Failed to retrieve task stage", err)
}
switch stage.Status {
case tfe.TaskStagePending:
// Waiting for it to start
return true, nil
// Note: Terminal statuses need to print out one last time just in case
case tfe.TaskStageRunning, tfe.TaskStagePassed, tfe.TaskStageCanceled, tfe.TaskStageErrored, tfe.TaskStageFailed:
for _, s := range summarizers {
cont, msg, err := s.Summarize(ctx, output, stage)
if cont {
if msg != nil {
if i%4 == 0 {
if i > 0 {
output.OutputElapsed(*msg, len(*msg)) // Up to 2 digits are allowed by the max message allocation
}
}
}
return true, nil
}
if err != nil {
errs.Append(err)
}
}
case "unreachable":
return false, nil
default:
return false, fmt.Errorf("Invalid Task stage status: %s ", stage.Status)
}
if len(errs) > 0 {
return false, errs.Err()
}
return false, nil
})
}

43
internal/cloud/errors.go
View File

@ -58,3 +58,46 @@ func incompatibleWorkspaceTerraformVersion(message string, ignoreVersionConflict
description := strings.TrimSpace(fmt.Sprintf("%s\n\n%s", message, suggestion))
return tfdiags.Sourceless(severity, "Incompatible Terraform version", description)
}
type multiErrors []error
// Append errs to e only if the individual error in errs is not nil
//
// If any of the errs is itself multiErrors, each individual error in errs is appended.
func (e *multiErrors) Append(errs ...error) {
for _, err := range errs {
if err == nil {
continue
}
if errs, ok := err.(multiErrors); ok {
*e = append(*e, errs...)
}
*e = append(*e, err)
}
}
// multiErrors returns an error string by joining
// all of its nonnil errors with colon separator.
func (e multiErrors) Error() string {
if len(e) == 0 {
return ""
}
es := make([]string, 0, len(e))
for _, err := range e {
if err != nil {
es = append(es, err.Error())
}
}
return strings.Join(es, ":")
}
// Err returns e as an error or returns nil if no errors were collected.
func (e multiErrors) Err() error {
// Only return self if we have at least one nonnil error.
for _, err := range e {
if err != nil {
return e
}
}
return nil
}