diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml index fe2201e..06115e0 100644 --- a/.github/workflows/cd.yml +++ b/.github/workflows/cd.yml @@ -10,7 +10,7 @@ jobs: name: Release strategy: matrix: - go-version: [1.24.11] + go-version: [1.25.11] os: [ubuntu-latest] runs-on: ${{ matrix.os }} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c037a29..cd697f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,7 @@ jobs: name: Compile & Test strategy: matrix: - go-version: [1.24.11] + go-version: [1.25.11] os: [ubuntu-latest] runs-on: ${{ matrix.os }} diff --git a/CHANGELOG b/CHANGELOG index 586b62f..b86ee29 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,8 @@ +Version 0.7.8 (2026-06-08) +-------------------------- +Bump Go toolchain to 1.25.11 (#101) +Log EMR cluster StateChangeReason on transient mode failure (#100) + Version 0.7.7 (2026-05-22) -------------------------- Log EMR cluster StateChangeReason on failure (#98) diff --git a/README.md b/README.md index 27c3fa3..ef65870 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ 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. -[release-image]: https://img.shields.io/badge/release-0.7.7-6ad7e5.svg?style=flat +[release-image]: https://img.shields.io/badge/release-0.7.8-6ad7e5.svg?style=flat [releases]: https://github.com/snowplow/dataflow-runner/releases [license-image]: https://img.shields.io/badge/license-Apache--2-blue.svg?style=flat diff --git a/VERSION b/VERSION index 879be8a..e7c7d3c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.7.7 +0.7.8 diff --git a/go.mod b/go.mod index e3996b7..d039089 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/snowplow/dataflow-runner -go 1.24.11 +go 1.25.11 require ( github.com/aws/aws-sdk-go-v2 v1.41.5 @@ -60,7 +60,7 @@ require ( github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/snowplow-devops/go-retry v0.0.0-20210106090855-8989bbdbae1c - golang.org/x/sys v0.40.0 // indirect + golang.org/x/sys v0.44.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 341a52f..d59bd1a 100644 --- a/go.sum +++ b/go.sum @@ -268,8 +268,8 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= -golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= diff --git a/src/emr_cluster.go b/src/emr_cluster.go index d660c4a..8db3544 100644 --- a/src/emr_cluster.go +++ b/src/emr_cluster.go @@ -182,6 +182,20 @@ func clusterStateChangeReason(status *types.ClusterStatus) (string, string) { return code, msg } +// fetchStateChangeReason describes the cluster and returns its StateChangeReason code and message. +// It returns a non-nil error if the cluster cannot be described, or empty strings if no reason is set. +func (ec EmrCluster) fetchStateChangeReason(ctx context.Context, jobflowID string) (string, string, error) { + resp, err := ec.Svc.DescribeCluster(ctx, &emr.DescribeClusterInput{ClusterId: aws.String(jobflowID)}) + if err != nil { + return "", "", err + } + if resp.Cluster == nil || resp.Cluster.Status == nil { + return "", "", nil + } + code, message := clusterStateChangeReason(resp.Cluster.Status) + return code, message, nil +} + // waitForClusterReady waits for the cluster to reach RUNNING or WAITING state using SDK v2 waiter. // Returns the cluster status even on failure (needed for bootstrap failure detection). func (ec EmrCluster) waitForClusterReady(ctx context.Context, jobflowID string) (*types.ClusterStatus, error) { diff --git a/src/emr_cluster_test.go b/src/emr_cluster_test.go index 20f213d..39c77b9 100644 --- a/src/emr_cluster_test.go +++ b/src/emr_cluster_test.go @@ -228,6 +228,29 @@ func TestRunJobFlow_Success(t *testing.T) { assert.Equal(t, "j-WAITING", id) } +func TestFetchStateChangeReason(t *testing.T) { + assert := assert.New(t) + ctx := context.Background() + record, _ := CR.ParseClusterRecord([]byte(ClusterRecord1), nil, "") + ec := mockEmrCluster(*record) + + // A cluster with a StateChangeReason returns its code and message + code, message, err := ec.fetchStateChangeReason(ctx, "j-TERMINATED_WITH_ERRORS") + assert.Nil(err) + assert.Equal("VALIDATION_ERROR", code) + assert.Equal("On the master instance, application provisioning failed", message) + + // A cluster with no StateChangeReason returns empty strings and no error + code, message, err = ec.fetchStateChangeReason(ctx, "j-STARTING") + assert.Nil(err) + assert.Equal("", code) + assert.Equal("", message) + + // A DescribeCluster failure is returned as an error + _, _, err = ec.fetchStateChangeReason(ctx, "bad") + assert.NotNil(err) +} + func TestGetJobFlowInput_Success(t *testing.T) { assert := assert.New(t) diff --git a/src/main.go b/src/main.go index c8c837b..4b08c9e 100644 --- a/src/main.go +++ b/src/main.go @@ -34,7 +34,7 @@ const ( appName = "dataflow-runner" appUsage = "Run templatable playbooks of Hadoop/Spark/et al jobs on Amazon EMR" appCopyright = "(c) 2016-2026 Snowplow Analytics Ltd" - cliVersion = "0.7.7" + cliVersion = "0.7.8" varDelim = "," fEmrConfig = "emr-config" fEmrPlaybook = "emr-playbook" @@ -327,6 +327,15 @@ func main() { maxClusterWaitDuration, ) if err != nil { + // The terminated waiter only returns a generic "transitioned to Failure" error, so + // re-describe the cluster to recover and surface the underlying StateChangeReason. + if code, message, derr := emrCluster.fetchStateChangeReason(context.Background(), jobFlowSteps.JobflowID); derr != nil { + log.Warnf("could not re-describe cluster to get StateChangeReason: %v", derr) + } else if code != "" || message != "" { + log.Errorf("EMR cluster state change reason: code='%s' message=%q", code, message) + err = fmt.Errorf("EMR cluster %s terminated with errors: code=%s message=%q", + jobFlowSteps.JobflowID, code, message) + } if lock != nil && softLock != "" { lock.Unlock() }