Extract reading datafeed state

Author: tobio

internal/elasticsearch/ml/datafeed/delete.go

@@ -2,7 +2,6 @@ package datafeed
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients/elasticsearch"
 	"github.com/hashicorp/terraform-plugin-framework/diag"
@@ -44,22 +43,18 @@ func (r *datafeedResource) delete(ctx context.Context, req resource.DeleteReques
 }
 
 func (r *datafeedResource) maybeStopDatafeed(ctx context.Context, datafeedId string) (bool, diag.Diagnostics) {
-	var diags diag.Diagnostics
-
 	// Check current state
-	currentState, err := r.getDatafeedState(ctx, datafeedId)
-	if err != nil {
-		// If we can't get the state, try to extract the error details
-		if err.Error() == fmt.Sprintf("datafeed %s not found", datafeedId) {
-			// Datafeed does not exist, nothing to stop
-			return false, diags
-		}
-		diags.AddError("Failed to get datafeed state", err.Error())
+	currentState, diags := GetDatafeedState(ctx, r.client, datafeedId)
+	if diags.HasError() {
 		return false, diags
 	}
 
+	if currentState == nil {
+		return false, nil
+	}
+
 	// If the datafeed is not running, nothing to stop
-	if currentState != "started" && currentState != "starting" {
+	if *currentState != "started" && *currentState != "starting" {
 		return false, diags
 	}
 
@@ -71,9 +66,9 @@ func (r *datafeedResource) maybeStopDatafeed(ctx context.Context, datafeedId str
 	}
 
 	// Wait for the datafeed to reach stopped state
-	err = r.waitForDatafeedState(ctx, datafeedId, "stopped")
-	if err != nil {
-		diags.AddError("Failed to wait for datafeed to stop", fmt.Sprintf("Datafeed %s did not stop within timeout: %s", datafeedId, err.Error()))
+	_, waitDiags := WaitForDatafeedState(ctx, r.client, datafeedId, "stopped")
+	diags.Append(waitDiags...)
+	if diags.HasError() {
 		return true, diags
 	}
 

internal/elasticsearch/ml/datafeed/state_utils.go

@@ -2,35 +2,65 @@ package datafeed
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/elastic/terraform-provider-elasticstack/internal/asyncutils"
+	"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/hashicorp/terraform-plugin-framework/diag"
 )
 
-// getDatafeedState returns the current state of a datafeed
-func (r *datafeedResource) getDatafeedState(ctx context.Context, datafeedId string) (string, error) {
-	statsResponse, diags := elasticsearch.GetDatafeedStats(ctx, r.client, datafeedId)
+// GetDatafeedState returns the current state of a datafeed
+func GetDatafeedState(ctx context.Context, client *clients.ApiClient, datafeedId string) (*string, diag.Diagnostics) {
+	statsResponse, diags := elasticsearch.GetDatafeedStats(ctx, client, datafeedId)
 	if diags.HasError() {
-		return "", fmt.Errorf("failed to get datafeed stats: %v", diags)
+		return nil, diags
 	}
 
 	if statsResponse == nil {
-		return "", fmt.Errorf("datafeed %s not found", datafeedId)
+		return nil, nil
 	}
 
-	return statsResponse.State, nil
+	return &statsResponse.State, nil
 }
 
-// waitForDatafeedState waits for a datafeed to reach the desired state
-func (r *datafeedResource) waitForDatafeedState(ctx context.Context, datafeedId, desiredState string) error {
+var terminalDatafeedStates = map[string]struct{}{
+	"stopped": {},
+	"started": {},
+}
+
+var errDatafeedInUndesiredState = errors.New("datafeed stuck in undesired state")
+
+// WaitForDatafeedState waits for a datafeed to reach the desired state
+func WaitForDatafeedState(ctx context.Context, client *clients.ApiClient, datafeedId, desiredState string) (bool, diag.Diagnostics) {
 	stateChecker := func(ctx context.Context) (bool, error) {
-		currentState, err := r.getDatafeedState(ctx, datafeedId)
-		if err != nil {
-			return false, err
+		currentState, diags := GetDatafeedState(ctx, client, datafeedId)
+		if diags.HasError() {
+			return false, diagutil.FwDiagsAsError(diags)
+		}
+
+		if currentState == nil {
+			return false, fmt.Errorf("datafeed %s not found", datafeedId)
+		}
+
+		if *currentState == desiredState {
+			return true, nil
+		}
+
+		_, isInTerminalState := terminalDatafeedStates[*currentState]
+		if isInTerminalState {
+			return false, fmt.Errorf("%w: datafeed is in state [%s] but desired state is [%s]", errDatafeedInUndesiredState, *currentState, desiredState)
 		}
-		return currentState == desiredState, nil
+
+		return false, nil
+	}
+
+	err := asyncutils.WaitForStateTransition(ctx, "datafeed", datafeedId, stateChecker)
+	if errors.Is(err, errDatafeedInUndesiredState) {
+		return false, nil
 	}
 
-	return asyncutils.WaitForStateTransition(ctx, "datafeed", datafeedId, stateChecker)
+	return err == nil, diagutil.FrameworkDiagFromError(err)
 }

internal/elasticsearch/ml/datafeed/update.go

@@ -2,7 +2,6 @@ package datafeed
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients/elasticsearch"
 	"github.com/elastic/terraform-provider-elasticstack/internal/diagutil"
@@ -62,9 +61,9 @@ func (r *datafeedResource) update(ctx context.Context, req resource.UpdateReques
 		}
 
 		// Wait for the datafeed to reach started state
-		err := r.waitForDatafeedState(ctx, datafeedId, "started")
-		if err != nil {
-			resp.Diagnostics.AddError("Failed to wait for datafeed to start", fmt.Sprintf("Datafeed %s did not start within timeout: %s", datafeedId, err.Error()))
+		_, waitDiags := WaitForDatafeedState(ctx, r.client, datafeedId, "started")
+		resp.Diagnostics.Append(waitDiags...)
+		if resp.Diagnostics.HasError() {
 			return
 		}
 	}