[APIE-1040] Add generic --wait framework; refactor Flink statement create #3361
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,7 +16,7 @@ | |
| "github.com/confluentinc/cli/v4/pkg/errors" | ||
| "github.com/confluentinc/cli/v4/pkg/flink/types" | ||
| "github.com/confluentinc/cli/v4/pkg/output" | ||
| "github.com/confluentinc/cli/v4/pkg/wait" | ||
| ) | ||
|
|
||
| func (c *command) newStatementCreateCommandOnPrem() *cobra.Command { | ||
|
|
@@ -36,7 +36,8 @@ | |
| cmd.Flags().String("catalog", "", "The name of the default catalog.") | ||
| cmd.Flags().String("database", "", "The name of the default database.") | ||
| cmd.Flags().String("flink-configuration", "", "The file path to hold the Flink configuration for the statement.") | ||
| pcmd.AddWaitFlag(cmd) | ||
| pcmd.AddWaitTimeoutFlag(cmd, time.Minute) | ||
|
Contributor
There was a problem hiding this comment. TF doesn't have the user configurable wait timeout, as it's purely handled inside an internal function with hardcoded value, since we want to have unified UX for CLI and TF users, when designing please take this into consideration, we need to either alter the existing TF behavior or let CLI follow the CLI behavior. |
||
| addCmfFlagSet(cmd) | ||
| pcmd.AddOutputFlag(cmd) | ||
|
|
||
|
|
@@ -47,7 +48,7 @@ | |
| return cmd | ||
| } | ||
|
|
||
| func (c *command) statementCreateOnPrem(cmd *cobra.Command, args []string) error { | ||
|
Check failure on line 51 in internal/flink/command_statement_create_onprem.go
|
||
| // Flink statement name can be automatically generated or provided by the user | ||
| name := types.GenerateStatementNameForOnPrem() | ||
| if len(args) == 1 { | ||
|
|
@@ -114,7 +115,7 @@ | |
| Stopped: cmfsdk.PtrBool(false), | ||
| }, | ||
| } | ||
| shouldWait, err := cmd.Flags().GetBool("wait") | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
@@ -124,22 +125,24 @@ | |
| return err | ||
| } | ||
|
|
||
| if shouldWait { | ||
| timeout, err := cmd.Flags().GetDuration("wait-timeout") | ||
| if err != nil { | ||
| return err | ||
| } | ||
| finalStatement, err = wait.Poll(cmd.Context(), wait.Options[cmfsdk.Statement]{ | ||
|
Comment on lines
+128
to
+133
|
||
| Fetch: func() (cmfsdk.Statement, error) { | ||
| return client.GetStatement(c.createContext(), environment, name) | ||
| }, | ||
| IsTerminal: func(s cmfsdk.Statement) bool { | ||
| return s.GetStatus().Phase != "PENDING" | ||
| }, | ||
| Tick: 2 * time.Second, | ||
|
Contributor
There was a problem hiding this comment.
Also, please investigate on how TF handles the polling interval and timeout, is it purely manual setting, is it semi-manual setting with default, is it something we can derive from OpenAPI spec, is it something we can write to registry.yaml as override value, once you have a proposal, please sync with Kostya Linou (@linouk23) for the unified implementation. |
||
| Timeout: timeout, | ||
| }) | ||
| if err != nil { | ||
| return errors.NewErrorWithSuggestions(err.Error(), "Increase `--wait-timeout` or omit `--wait`.") | ||
|
Comment on lines
+143
to
+144
|
||
| } | ||
| } | ||
|
|
||
| if output.GetFormat(cmd) == output.Human { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package wait | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "time" | ||
| ) | ||
|
|
||
| // Options describes a single Poll invocation. T is the polled resource type. | ||
| type Options[T any] struct { | ||
| Fetch func() (T, error) | ||
| IsTerminal func(T) bool | ||
| IsFailed func(T) bool | ||
|
Contributor
There was a problem hiding this comment.
|
||
| Tick time.Duration | ||
| Timeout time.Duration | ||
| } | ||
|
|
||
| var ( | ||
| ErrTimeout = errors.New("wait timed out") | ||
| ErrFailed = errors.New("resource entered failed state") | ||
| ) | ||
|
|
||
| // Poll calls opts.Fetch immediately, then every opts.Tick until IsFailed | ||
| // returns true (ErrFailed), IsTerminal returns true (success), opts.Timeout | ||
| // elapses (ErrTimeout), ctx is cancelled (ctx.Err()), or Fetch returns a | ||
| // non-nil error. Always returns the most recent successfully-fetched T. | ||
| func Poll[T any](ctx context.Context, opts Options[T]) (T, error) { | ||
| var last T | ||
|
|
||
| check := func() (T, bool, error) { | ||
| v, err := opts.Fetch() | ||
| if err != nil { | ||
| return last, true, err | ||
|
Comment on lines
+31
to
+33
|
||
| } | ||
| last = v | ||
| if opts.IsFailed != nil && opts.IsFailed(v) { | ||
| return last, true, ErrFailed | ||
| } | ||
| if opts.IsTerminal(v) { | ||
| return last, true, nil | ||
| } | ||
| return last, false, nil | ||
| } | ||
|
|
||
| if v, done, err := check(); done { | ||
| return v, err | ||
| } | ||
|
|
||
| ticker := time.NewTicker(opts.Tick) | ||
| defer ticker.Stop() | ||
| deadline := time.After(opts.Timeout) | ||
|
|
||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| if v, done, err := check(); done { | ||
| return v, err | ||
| } | ||
| case <-deadline: | ||
| return last, ErrTimeout | ||
| case <-ctx.Done(): | ||
| return last, ctx.Err() | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| package wait | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| type fakeResource struct { | ||
| phase string | ||
| } | ||
|
|
||
| func TestPoll_ImmediateReady(t *testing.T) { | ||
|
Check warning on line 17 in pkg/wait/wait_test.go
|
||
| calls := 0 | ||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| calls++ | ||
| return fakeResource{phase: "READY"}, nil | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase == "READY" }, | ||
| Tick: time.Millisecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "READY", v.phase) | ||
| require.Equal(t, 1, calls) | ||
| } | ||
|
|
||
| func TestPoll_EventuallyReady(t *testing.T) { | ||
|
Check warning on line 33 in pkg/wait/wait_test.go
|
||
| calls := 0 | ||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| calls++ | ||
| if calls < 3 { | ||
| return fakeResource{phase: "PENDING"}, nil | ||
| } | ||
| return fakeResource{phase: "READY"}, nil | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase != "PENDING" }, | ||
| Tick: time.Nanosecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.NoError(t, err) | ||
| require.Equal(t, "READY", v.phase) | ||
| require.Equal(t, 3, calls) | ||
| } | ||
|
|
||
| func TestPoll_Failed(t *testing.T) { | ||
|
Check warning on line 52 in pkg/wait/wait_test.go
|
||
| calls := 0 | ||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| calls++ | ||
| if calls == 1 { | ||
| return fakeResource{phase: "PENDING"}, nil | ||
| } | ||
| return fakeResource{phase: "FAILED"}, nil | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase == "READY" || r.phase == "FAILED" }, | ||
| IsFailed: func(r fakeResource) bool { return r.phase == "FAILED" }, | ||
| Tick: time.Nanosecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.ErrorIs(t, err, ErrFailed) | ||
| require.Equal(t, "FAILED", v.phase) | ||
| } | ||
|
|
||
| func TestPoll_Timeout(t *testing.T) { | ||
|
Check warning on line 71 in pkg/wait/wait_test.go
|
||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| return fakeResource{phase: "PENDING"}, nil | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase != "PENDING" }, | ||
| Tick: time.Millisecond, | ||
| Timeout: 5 * time.Millisecond, | ||
| }) | ||
| require.ErrorIs(t, err, ErrTimeout) | ||
| require.Equal(t, "PENDING", v.phase) | ||
| } | ||
|
|
||
| func TestPoll_FetchError(t *testing.T) { | ||
|
Check warning on line 84 in pkg/wait/wait_test.go
|
||
| calls := 0 | ||
| fetchErr := fmt.Errorf("transient fetch failure") | ||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| calls++ | ||
| if calls == 1 { | ||
| return fakeResource{phase: "PENDING"}, nil | ||
| } | ||
| return fakeResource{}, fetchErr | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase != "PENDING" }, | ||
| Tick: time.Nanosecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.ErrorIs(t, err, fetchErr) | ||
| require.Equal(t, "PENDING", v.phase) | ||
| } | ||
|
|
||
| func TestPoll_FetchErrorOnFirstCall(t *testing.T) { | ||
|
Check warning on line 103 in pkg/wait/wait_test.go
|
||
| fetchErr := fmt.Errorf("initial fetch failure") | ||
| v, err := Poll(context.Background(), Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| return fakeResource{}, fetchErr | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return true }, | ||
| Tick: time.Nanosecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.ErrorIs(t, err, fetchErr) | ||
| require.Equal(t, fakeResource{}, v) | ||
| } | ||
|
|
||
| func TestPoll_ContextCancelled(t *testing.T) { | ||
|
Check warning on line 117 in pkg/wait/wait_test.go
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
|
Check warning on line 118 in pkg/wait/wait_test.go
|
||
| go func() { | ||
| time.Sleep(5 * time.Millisecond) | ||
| cancel() | ||
| }() | ||
| _, err := Poll(ctx, Options[fakeResource]{ | ||
| Fetch: func() (fakeResource, error) { | ||
| return fakeResource{phase: "PENDING"}, nil | ||
| }, | ||
| IsTerminal: func(r fakeResource) bool { return r.phase != "PENDING" }, | ||
| Tick: time.Millisecond, | ||
| Timeout: time.Second, | ||
| }) | ||
| require.True(t, errors.Is(err, context.Canceled)) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
"PENDING"may not be the only intermediate status, there are"FAILING"and"STOPPING"for Flink statement:https://docs.confluent.io/cloud/current/ccloud/get-sqlv-1-statement/?_highlight=statement
If we think even one level higher, is there a way we can retrieve information from OpenAPI spec to determine which states are pending states, and which states are terminal states, this is the question we need to give answer to ourselves.
For example, a naive idea would be, if the status ends with
"ING", we need to wait, otherwise we should terminate.https://github.com/confluentinc/api/blob/master/flink-gateway/v1/openapi.yaml#L3313
Can you check if such idea can be applied to other async resources?