ashb commented on code in PR #56079:
URL: https://github.com/apache/airflow/pull/56079#discussion_r2378813522
##########
go-sdk/celery/app.go:
##########
@@ -80,27 +72,58 @@ func Run(ctx context.Context, config Config) error {
celery.WithBroker(broker),
)
+ tasks := &celeryTasksRunner{d}
+ fmt.Printf("%#v\n", viper.AllKeys())
+
for _, queue := range config.Queues {
app.Register(
"execute_workload",
queue,
func(ctx context.Context, p *celery.TaskParam) error {
- p.NameArgs("payload")
- payload := p.MustString("payload")
-
- var workload api.ExecuteTaskWorkload
- if err := json.Unmarshal([]byte(payload),
&workload); err != nil {
- return err
+ err := tasks.ExecuteWorkloadTask(ctx, p)
+ if err != nil {
+ slog.ErrorContext(ctx, "Celery Task
failed", "err", err)
}
- return worker.ExecuteTaskWorkload(ctx, workload)
+ return err
},
)
}
slog.Info("waiting for tasks", "queues", config.Queues)
- err = app.Run(ctx)
+ return app.Run(ctx)
+}
+
+func (state *celeryTasksRunner) ExecuteWorkloadTask(
+ ctx context.Context,
+ p *celery.TaskParam,
+) error {
+ p.NameArgs("payload")
+ payload := p.MustString("payload")
+
+ var workload bundlev1.ExecuteTaskWorkload
+ if err := json.Unmarshal([]byte(payload), &workload); err != nil {
+ return err
+ }
+
+ client, err := state.ClientForBundle(workload.BundleInfo.Name,
workload.BundleInfo.Version)
+ if err != nil {
+ // TODO: This Should write something to the log file
+ return err
+ }
+ // TODO: Don't kill the backend process here, but instead kill it after
a bit of idleness. See if we can
+ // reuse the process for multiple tasks too
+ defer client.Kill()
Review Comment:
This comment is more about reusing the plugin process (i.e. you can
absolutely use one plugin process to run multiple tasks concurrently.
So it would be a whole separate "spawn the process, run things, then once
that plugin bundle has been idle for a configured time, say 1 or 5 minutes,
_then_ kill it". So I don't think a timeout would help, as that would possibly
kill it early if it's busy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
