Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Return ResourceWrapper without pointer #4115

Merged
merged 5 commits into from
Oct 5, 2023
Merged

Return ResourceWrapper without pointer #4115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
847f0d9
Return ResourceWrapper without pointer
pingsutw Oct 2, 2023
fecb4cd
nit
pingsutw Oct 2, 2023
1e482f2
Add item ID to the workqueue instead
pingsutw Oct 2, 2023
1dfd610
Revert "Add item ID to the workqueue instead"
pingsutw Oct 2, 2023
ce94855
nit
pingsutw Oct 2, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,16 +112,16 @@ func (p Plugin) Create(ctx context.Context, taskCtx webapi.TaskExecutionContextR
taskTemplate.GetContainer().Args = argTemplate
}

return &ResourceMetaWrapper{
return ResourceMetaWrapper{
OutputPrefix: outputPrefix,
AgentResourceMeta: res.GetResourceMeta(),
Token: "",
TaskType: taskTemplate.Type,
}, &ResourceWrapper{State: admin.State_RUNNING}, nil
}, ResourceWrapper{State: admin.State_RUNNING}, nil
}

func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest webapi.Resource, err error) {
metadata := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
metadata := taskCtx.ResourceMeta().(ResourceMetaWrapper)

agent, err := getFinalAgent(metadata.TaskType, p.cfg)
if err != nil {
Expand All @@ -140,7 +140,7 @@ func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest weba
return nil, err
}

return &ResourceWrapper{
return ResourceWrapper{
State: res.Resource.State,
Outputs: res.Resource.Outputs,
}, nil
Expand Down Expand Up @@ -169,7 +169,7 @@ func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
}

func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase core.PhaseInfo, err error) {
resource := taskCtx.Resource().(*ResourceWrapper)
resource := taskCtx.Resource().(ResourceWrapper)
taskInfo := &core.TaskInfo{}

switch resource.State {
Expand All @@ -190,7 +190,7 @@ func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase
return core.PhaseInfoUndefined, pluginErrors.Errorf(core.SystemErrorCode, "unknown execution phase [%v].", resource.State)
}

func writeOutput(ctx context.Context, taskCtx webapi.StatusContext, resource *ResourceWrapper) error {
func writeOutput(ctx context.Context, taskCtx webapi.StatusContext, resource ResourceWrapper) error {
taskTemplate, err := taskCtx.TaskReader().Read(ctx)
if err != nil {
return err
Expand Down
12 changes: 6 additions & 6 deletions flyteplugins/go/tasks/plugins/webapi/databricks/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,12 +146,12 @@ func (p Plugin) Create(ctx context.Context, taskCtx webapi.TaskExecutionContextR
}
runID := fmt.Sprintf("%v", data["run_id"])

return &ResourceMetaWrapper{runID, p.cfg.DatabricksInstance, token},
&ResourceWrapper{StatusCode: resp.StatusCode}, nil
return ResourceMetaWrapper{runID, p.cfg.DatabricksInstance, token},
ResourceWrapper{StatusCode: resp.StatusCode}, nil
}

func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest webapi.Resource, err error) {
exec := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
exec := taskCtx.ResourceMeta().(ResourceMetaWrapper)
req, err := buildRequest(get, nil, p.cfg.databricksEndpoint,
p.cfg.DatabricksInstance, exec.Token, exec.RunID, false)
if err != nil {
Expand All @@ -176,7 +176,7 @@ func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest weba
jobID := fmt.Sprintf("%.0f", data["job_id"])
lifeCycleState := fmt.Sprintf("%s", jobState["life_cycle_state"])
resultState := fmt.Sprintf("%s", jobState["result_state"])
return &ResourceWrapper{
return ResourceWrapper{
StatusCode: resp.StatusCode,
JobID: jobID,
LifeCycleState: lifeCycleState,
Expand Down Expand Up @@ -206,8 +206,8 @@ func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
}

func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase core.PhaseInfo, err error) {
exec := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
resource := taskCtx.Resource().(*ResourceWrapper)
exec := taskCtx.ResourceMeta().(ResourceMetaWrapper)
resource := taskCtx.Resource().(ResourceWrapper)
message := resource.Message
statusCode := resource.StatusCode
jobID := resource.JobID
Expand Down
16 changes: 8 additions & 8 deletions flyteplugins/go/tasks/plugins/webapi/snowflake/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,12 +139,12 @@
queryID := fmt.Sprintf("%v", data["statementHandle"])
message := fmt.Sprintf("%v", data["message"])

return &ResourceMetaWrapper{queryID, queryInfo.Account, token},
&ResourceWrapper{StatusCode: resp.StatusCode, Message: message}, nil
return ResourceMetaWrapper{queryID, queryInfo.Account, token},
ResourceWrapper{StatusCode: resp.StatusCode, Message: message}, nil
}

func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest webapi.Resource, err error) {
exec := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
exec := taskCtx.ResourceMeta().(ResourceMetaWrapper)
req, err := buildRequest(get, QueryInfo{}, p.cfg.snowflakeEndpoint,
exec.Account, exec.Token, exec.QueryID, false)
if err != nil {
Expand All @@ -160,7 +160,7 @@
return nil, err
}
message := fmt.Sprintf("%v", data["message"])
return &ResourceWrapper{
return ResourceWrapper{
StatusCode: resp.StatusCode,
Message: message,
}, nil
Expand All @@ -170,7 +170,7 @@
if taskCtx.ResourceMeta() == nil {
return nil
}
exec := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
exec := taskCtx.ResourceMeta().(ResourceMetaWrapper)

Check warning on line 173 in flyteplugins/go/tasks/plugins/webapi/snowflake/plugin.go

View check run for this annotation

Codecov / codecov/patch

flyteplugins/go/tasks/plugins/webapi/snowflake/plugin.go#L173 

Added line #L173 was not covered by tests
req, err := buildRequest(post, QueryInfo{}, p.cfg.snowflakeEndpoint,
exec.Account, exec.Token, exec.QueryID, true)
if err != nil {
Expand All @@ -187,8 +187,8 @@
}

func (p Plugin) Status(_ context.Context, taskCtx webapi.StatusContext) (phase core.PhaseInfo, err error) {
exec := taskCtx.ResourceMeta().(*ResourceMetaWrapper)
statusCode := taskCtx.Resource().(*ResourceWrapper).StatusCode
exec := taskCtx.ResourceMeta().(ResourceMetaWrapper)
statusCode := taskCtx.Resource().(ResourceWrapper).StatusCode
if statusCode == 0 {
return core.PhaseInfoUndefined, errors.Errorf(ErrSystem, "No Status field set.")
}
Expand Down Expand Up @@ -276,7 +276,7 @@
ID: "snowflake",
SupportedTaskTypes: []core.TaskType{"snowflake"},
PluginLoader: func(ctx context.Context, iCtx webapi.PluginSetupContext) (webapi.AsyncPlugin, error) {
return &Plugin{
return Plugin{
metricScope: iCtx.MetricsScope(),
cfg: GetConfig(),
client: &http.Client{},
Expand Down
2 changes: 1 addition & 1 deletion rsts/deployment/index.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ plugins, authentication, performance tuning, and maintaining Flyte as a producti
:text: 🤖 Agent Setup
:classes: btn-block stretched-link
^^^^^^^^^^^^
Enable Flyte agents to extend Flyte's capabilities, including features like File sesnor, Databricks job, and Snowflake query services.
Enable Flyte agents to extend Flyte's capabilities, including features like File sensor, Databricks job, and Snowflake query services.

---

Expand Down
Loading