PR feedback

Author: tobio

internal/diagutil/context.go

@@ -0,0 +1,19 @@
+package diagutil
+
+import (
+	"context"
+
+	fwdiag "github.com/hashicorp/terraform-plugin-framework/diag"
+	"github.com/hashicorp/terraform-plugin-log/tflog"
+)
+
+var contextDeadlineExceededDiags = FrameworkDiagFromError(context.DeadlineExceeded)
+
+func ContainsContextDeadlineExceeded(ctx context.Context, diags fwdiag.Diagnostics) bool {
+	if len(contextDeadlineExceededDiags) == 0 {
+		tflog.Error(ctx, "Expected context deadline exceeded diagnostics to contain at least one error")
+		return false
+	}
+
+	return diags.Contains(contextDeadlineExceededDiags[0])
+}

internal/elasticsearch/ml/datafeed_state/create.go

@@ -25,7 +25,7 @@ func (r *mlDatafeedStateResource) Create(ctx context.Context, req resource.Creat
 	}
 
 	diags = r.update(ctx, req.Plan, &resp.State, createTimeout)
-	if diags.Contains(diagutil.FrameworkDiagFromError(context.DeadlineExceeded)[0]) {
+	if diagutil.ContainsContextDeadlineExceeded(ctx, diags) {
 		diags.AddError("Operation timed out", fmt.Sprintf("The operation to create the ML datafeed state timed out after %s. You may need to allocate more free memory within ML nodes by either closing other jobs, or increasing the overall ML memory. You may retry the operation.", createTimeout))
 	}
 

internal/elasticsearch/ml/datafeed_state/models.go

@@ -2,7 +2,9 @@ package datafeed_state
 
 import (
 	"strconv"
+	"time"
 
+	"github.com/elastic/terraform-provider-elasticstack/internal/models"
 	"github.com/elastic/terraform-provider-elasticstack/internal/utils"
 	"github.com/elastic/terraform-provider-elasticstack/internal/utils/customtypes"
 	"github.com/hashicorp/terraform-plugin-framework-timeouts/resource/timeouts"
@@ -23,15 +25,15 @@ type MLDatafeedStateData struct {
 	Timeouts                timeouts.Value       `tfsdk:"timeouts"`
 }
 
-func (d MLDatafeedStateData) GetStartAsString() (string, diag.Diagnostics) {
+func (d *MLDatafeedStateData) GetStartAsString() (string, diag.Diagnostics) {
 	return d.getTimeAttributeAsString(d.Start)
 }
 
-func (d MLDatafeedStateData) GetEndAsString() (string, diag.Diagnostics) {
+func (d *MLDatafeedStateData) GetEndAsString() (string, diag.Diagnostics) {
 	return d.getTimeAttributeAsString(d.End)
 }
 
-func (d MLDatafeedStateData) getTimeAttributeAsString(val timetypes.RFC3339) (string, diag.Diagnostics) {
+func (d *MLDatafeedStateData) getTimeAttributeAsString(val timetypes.RFC3339) (string, diag.Diagnostics) {
 	if !utils.IsKnown(val) {
 		return "", nil
 	}
@@ -42,3 +44,20 @@ func (d MLDatafeedStateData) getTimeAttributeAsString(val timetypes.RFC3339) (st
 	}
 	return strconv.FormatInt(valTime.Unix(), 10), nil
 }
+
+func (d *MLDatafeedStateData) SetStartAndEndFromAPI(datafeedStats *models.DatafeedStats) diag.Diagnostics {
+	var diags diag.Diagnostics
+	if datafeedStats.RunningState == nil {
+		diags.AddWarning("Running state was empty for a started datafeed", "The Elasticsearch API returned an empty running state for a Datafeed which was successfully started. Ignoring start and end response values.")
+		return diags
+	}
+
+	d.Start = timetypes.NewRFC3339TimeValue(time.UnixMilli(datafeedStats.RunningState.SearchInterval.StartMS))
+	if datafeedStats.RunningState.RealTimeConfigured {
+		d.End = timetypes.NewRFC3339Null()
+	} else {
+		d.End = timetypes.NewRFC3339TimeValue(time.UnixMilli(datafeedStats.RunningState.SearchInterval.EndMS))
+	}
+
+	return diags
+}

internal/elasticsearch/ml/datafeed_state/read.go

@@ -2,12 +2,11 @@ package datafeed_state
 
 import (
 	"context"
-	"time"
 
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients"
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients/elasticsearch"
+	"github.com/elastic/terraform-provider-elasticstack/internal/diagutil"
 	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/ml/datafeed"
-	"github.com/hashicorp/terraform-plugin-framework-timetypes/timetypes"
 	"github.com/hashicorp/terraform-plugin-framework/diag"
 	"github.com/hashicorp/terraform-plugin-framework/resource"
 	"github.com/hashicorp/terraform-plugin-framework/types"
@@ -56,17 +55,17 @@ func (r *mlDatafeedStateResource) read(ctx context.Context, data MLDatafeedState
 	// Update the data with current information
 	data.State = types.StringValue(datafeedStats.State)
 
-	if datafeed.State(datafeedStats.State) == datafeed.StateStarted {
-		if datafeedStats.RunningState == nil {
-			diags.AddWarning("Running state was empty for a started datafeed", "The Elasticsearch API returned an empty running state for a Datafeed which was successfully started. Ignoring start and end response values.")
-		}
+	// Regenerate composite ID to ensure it's current
+	compId, sdkDiags := client.ID(ctx, datafeedId)
+	diags.Append(diagutil.FrameworkDiagsFromSDK(sdkDiags)...)
+	if diags.HasError() {
+		return nil, diags
+	}
+
+	data.Id = types.StringValue(compId.String())
 
-		data.Start = timetypes.NewRFC3339TimeValue(time.UnixMilli(datafeedStats.RunningState.SearchInterval.StartMS))
-		if datafeedStats.RunningState.RealTimeConfigured {
-			data.End = timetypes.NewRFC3339Null()
-		} else {
-			data.End = timetypes.NewRFC3339TimeValue(time.UnixMilli(datafeedStats.RunningState.SearchInterval.EndMS))
-		}
+	if datafeed.State(datafeedStats.State) == datafeed.StateStarted {
+		diags.Append(data.SetStartAndEndFromAPI(datafeedStats)...)
 	}
 
 	return &data, diags

internal/elasticsearch/ml/datafeed_state/update.go

@@ -9,6 +9,7 @@ import (
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients/elasticsearch"
 	"github.com/elastic/terraform-provider-elasticstack/internal/diagutil"
 	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/ml/datafeed"
+	"github.com/elastic/terraform-provider-elasticstack/internal/models"
 	"github.com/hashicorp/terraform-plugin-framework-timetypes/timetypes"
 	"github.com/hashicorp/terraform-plugin-framework/diag"
 	"github.com/hashicorp/terraform-plugin-framework/resource"
@@ -33,7 +34,7 @@ func (r *mlDatafeedStateResource) Update(ctx context.Context, req resource.Updat
 	}
 
 	diags = r.update(ctx, req.Plan, &resp.State, updateTimeout)
-	if diags.Contains(diagutil.FrameworkDiagFromError(context.DeadlineExceeded)[0]) {
+	if diagutil.ContainsContextDeadlineExceeded(ctx, diags) {
 		diags.AddError("Operation timed out", fmt.Sprintf("The operation to update the ML datafeed state timed out after %s. You may need to allocate more free memory within ML nodes by either closing other jobs, or increasing the overall ML memory. You may retry the operation.", updateTimeout))
 	}
 
@@ -106,33 +107,12 @@ func (r *mlDatafeedStateResource) update(ctx context.Context, plan tfsdk.Plan, s
 			return diags
 		}
 	} else {
-		statsAfterUpdate, fwDiags := elasticsearch.GetDatafeedStats(ctx, client, datafeedId)
-		diags.Append(fwDiags...)
+		var updateDiags diag.Diagnostics
+		finalData, updateDiags = r.updateAfterMissedTransition(ctx, client, data, datafeedStats)
+		diags.Append(updateDiags...)
 		if diags.HasError() {
 			return diags
 		}
-
-		if datafeedStats == nil {
-			diags.AddError(
-				"ML Datafeed not found",
-				fmt.Sprintf("ML datafeed %s does not exist after successful update", datafeedId),
-			)
-			return diags
-		}
-
-		if statsAfterUpdate.TimingStats.SearchCount < datafeedStats.TimingStats.SearchCount {
-			diags.AddError(
-				"Datafeed did not successfully transition to the desired state",
-				fmt.Sprintf("[%s] datafeed did not settle into the [%s] state. The current state is [%s]", datafeedId, desiredState, statsAfterUpdate.State),
-			)
-			return diags
-		}
-
-		if data.Start.IsUnknown() {
-			data.Start = timetypes.NewRFC3339Null()
-		}
-
-		finalData = &data
 	}
 
 	if finalData == nil {
@@ -144,6 +124,42 @@ func (r *mlDatafeedStateResource) update(ctx context.Context, plan tfsdk.Plan, s
 	return diags
 }
 
+func (r *mlDatafeedStateResource) updateAfterMissedTransition(ctx context.Context, client *clients.ApiClient, data MLDatafeedStateData, datafeedStats *models.DatafeedStats) (*MLDatafeedStateData, diag.Diagnostics) {
+	datafeedId := data.DatafeedId.ValueString()
+	statsAfterUpdate, diags := elasticsearch.GetDatafeedStats(ctx, client, datafeedId)
+	if diags.HasError() {
+		return nil, diags
+	}
+
+	if statsAfterUpdate == nil {
+		diags.AddError(
+			"ML Datafeed not found",
+			fmt.Sprintf("ML datafeed %s does not exist after successful update", datafeedId),
+		)
+		return nil, diags
+	}
+
+	// It's possible that the datafeed starts, and then immediately stops if there is no (or very little) data to process.
+	// In this case, the state transition may occur too quickly to be detected by the wait function.
+	// To handle this, we check if the search count has increased to determine if the datafeed actually started since the update.
+	if statsAfterUpdate.TimingStats == nil || datafeedStats.TimingStats == nil {
+		diags.AddWarning("Expected Datafeed to contain timing stats",
+			fmt.Sprintf("Stats for datafeed %s did not contain timing stats either before or after the update. Before %v - After %v", datafeedId, datafeedStats, statsAfterUpdate))
+	} else if statsAfterUpdate.TimingStats.SearchCount <= datafeedStats.TimingStats.SearchCount {
+		diags.AddError(
+			"Datafeed did not successfully transition to the desired state",
+			fmt.Sprintf("[%s] datafeed did not settle into the [%s] state. The current state is [%s]", datafeedId, data.State.ValueString(), statsAfterUpdate.State),
+		)
+		return nil, diags
+	}
+
+	if data.Start.IsUnknown() {
+		data.Start = timetypes.NewRFC3339Null()
+	}
+
+	return &data, nil
+}
+
 // performStateTransition handles the ML datafeed state transition process
 func (r *mlDatafeedStateResource) performStateTransition(ctx context.Context, client *clients.ApiClient, data MLDatafeedStateData, currentState datafeed.State) (bool, diag.Diagnostics) {
 	datafeedId := data.DatafeedId.ValueString()

internal/elasticsearch/ml/job_state/create.go

@@ -25,7 +25,7 @@ func (r *mlJobStateResource) Create(ctx context.Context, req resource.CreateRequ
 	}
 
 	diags = r.update(ctx, req.Plan, &resp.State, createTimeout)
-	if diags.Contains(diagutil.FrameworkDiagFromError(context.DeadlineExceeded)[0]) {
+	if diagutil.ContainsContextDeadlineExceeded(ctx, diags) {
 		diags.AddError("Operation timed out", fmt.Sprintf("The operation to create the ML job state timed out after %s. You may need to allocate more free memory within ML nodes by either closing other jobs, or increasing the overall ML memory. You may retry the operation.", createTimeout))
 	}
 

internal/elasticsearch/ml/job_state/update.go

@@ -31,7 +31,7 @@ func (r *mlJobStateResource) Update(ctx context.Context, req resource.UpdateRequ
 	}
 
 	diags = r.update(ctx, req.Plan, &resp.State, updateTimeout)
-	if diags.Contains(diagutil.FrameworkDiagFromError(context.DeadlineExceeded)[0]) {
+	if diagutil.ContainsContextDeadlineExceeded(ctx, diags) {
 		diags.AddError("Operation timed out", fmt.Sprintf("The operation to update the ML job state timed out after %s. You may need to allocate more free memory within ML nodes by either closing other jobs, or increasing the overall ML memory. You may retry the operation.", updateTimeout))
 	}