This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b71a2f5e8d5 Updates example go dag (#64493)
b71a2f5e8d5 is described below
commit b71a2f5e8d5615097765eeb4e33db71870116256
Author: Aritra Basu <[email protected]>
AuthorDate: Thu Apr 23 19:26:12 2026 +0530
Updates example go dag (#64493)
---
go-sdk/README.md | 2 +-
go-sdk/edge/worker.go | 5 ++++-
go-sdk/example/bundle/main.go | 8 ++++----
go-sdk/pkg/config/config.go | 8 ++++++--
scripts/in_container/bin/generate_mprocs_config.py | 2 ++
5 files changed, 17 insertions(+), 8 deletions(-)
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 7bf8c361304..c84d5529fbf 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -110,7 +110,7 @@ def simple_dag():
extract() >> transform()
-multi_language()
+simple_dag()
```
Here we see the `@task.stub` which tells the Dag parser about the "shape" of
the go tasks, and lets us define
diff --git a/go-sdk/edge/worker.go b/go-sdk/edge/worker.go
index 48e7eec18fe..3930806e99b 100644
--- a/go-sdk/edge/worker.go
+++ b/go-sdk/edge/worker.go
@@ -124,7 +124,7 @@ func NewWorker(conf config.WorkerConfig) (*worker, error) {
configOrDefault("edge.airflow_version", "3.2.0"),
))
edgeVer.FromWorkerStateBodySysinfo0(edgeapi.WorkerStateBodySysinfo0(
- configOrDefault("edge.provider_version", "3.2.0"),
+ configOrDefault("edge.provider_version", "3.3.0"),
))
concurrency.FromWorkerStateBodySysinfo1(edgeapi.WorkerStateBodySysinfo1(maxConcurrency))
freeConcurrency.FromWorkerStateBodySysinfo1(edgeapi.WorkerStateBodySysinfo1(maxConcurrency))
@@ -452,5 +452,8 @@ func (w *worker) runWorkload(
bundleClient := raw.(bundlev1client.BundleClient)
err = bundleClient.ExecuteTaskWorkload(ctx, workload)
+ if err != nil {
+ w.logger.Error("Workload failed", logging.AttrErr(err))
+ }
return err
}
diff --git a/go-sdk/example/bundle/main.go b/go-sdk/example/bundle/main.go
index 9c1e8a09d39..5e970da5495 100644
--- a/go-sdk/example/bundle/main.go
+++ b/go-sdk/example/bundle/main.go
@@ -45,10 +45,10 @@ func (m *myBundle) GetBundleVersion() v1.BundleInfo {
}
func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
- tutorial_dag := dagbag.AddDag("tutorial_dag")
- tutorial_dag.AddTask(extract)
- tutorial_dag.AddTask(transform)
- tutorial_dag.AddTask(load)
+ simpleDag := dagbag.AddDag("simple_dag")
+ simpleDag.AddTask(extract)
+ simpleDag.AddTask(transform)
+ simpleDag.AddTask(load)
return nil
}
diff --git a/go-sdk/pkg/config/config.go b/go-sdk/pkg/config/config.go
index f0a29895834..54622743735 100644
--- a/go-sdk/pkg/config/config.go
+++ b/go-sdk/pkg/config/config.go
@@ -210,7 +210,11 @@ func BindFlagsToViper(cmd *cobra.Command, viper
*viper.Viper) {
func setDefaults(viper *viper.Viper) {
viper.SetDefault("edge.api_retries", 10)
- viper.SetDefault("edge.api_retry_wait_min", "1m")
- viper.SetDefault("edge.api_retry_wait_max", "90m")
+ viper.SetDefault("edge.api_retry_wait_min", "5s")
+ viper.SetDefault("edge.api_retry_wait_max", "90s")
viper.SetDefault("api_auth.issuer", "airflow")
+ hostname, err := os.Hostname()
+ if err == nil {
+ viper.SetDefault("edge.hostname", hostname)
+ }
}
diff --git a/scripts/in_container/bin/generate_mprocs_config.py
b/scripts/in_container/bin/generate_mprocs_config.py
index 1747582671a..ee8bed9570e 100755
--- a/scripts/in_container/bin/generate_mprocs_config.py
+++ b/scripts/in_container/bin/generate_mprocs_config.py
@@ -118,6 +118,8 @@ def generate_mprocs_config() -> str:
env = {}
env["AIRFLOW__EDGE__API_URL"] = "http://localhost:8080";
env["AIRFLOW__BUNDLES__FOLDER"] = "./bin"
+ env["AIRFLOW__LOGGING__BASE_LOG_FOLDER"] = "./logs"
+ env["AIRFLOW__EXECUTION__API_URL"] = "http://localhost:8080/execution/";
# Build command with environment cleanup
go_worker_cmd_parts = [
